diff --git a/Framework/Core/CMakeLists.txt b/Framework/Core/CMakeLists.txt
index 437dc5ee6478c..2d7811b098efd 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,35 @@ 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 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)
+    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}>)
+  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
 # command line option to include the output proxy
 o2_add_test(
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/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..9ed128b42b40b
--- /dev/null
+++ b/Framework/Core/include/Framework/WorkflowDefinitionContext.h
@@ -0,0 +1,75 @@
+// 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 <map>
+#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;
+  /// The plugin which defines each data processor, if not embedded in the executable.
+  std::map<std::string, std::string> processorPlugins;
+};
+
+struct WorkflowDefinition {
+  /// 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 e756c9f4b1d26..f69459c01462d 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,71 @@ 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, 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
+#define DPL_WORKFLOW_PLUGIN_EMBEDDED
+#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{
+      .customize = [](o2::framework::WorkflowDefinitionContext& workflowContext) {
+        using namespace o2::framework;
+        UserCustomizationsHelper::userDefinedCustomization(workflowContext.workflowOptions);
+        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); }};
   }
-  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*));
 
+#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, mainNoCatch);
+  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/src/DeviceSpecHelpers.cxx b/Framework/Core/src/DeviceSpecHelpers.cxx
index 4c19e7a6ff17b..4114faaec5d1e 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;
@@ -1747,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 98da004e68a6f..c7752ea3c0267 100644
--- a/Framework/Core/src/runDataProcessing.cxx
+++ b/Framework/Core/src/runDataProcessing.cxx
@@ -65,6 +65,8 @@
 #include "Framework/DeviceContext.h"
 #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"
@@ -186,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];
@@ -196,8 +204,42 @@ 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) {
+    // --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;
+    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());
+      }
+    }
+    if (workflows.empty()) {
+      throw runtime_error_f("No workflow plugin in %s.", spec);
+    }
+    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");
   int result = 1;
   if (noCatch) {
@@ -674,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,
@@ -687,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.
@@ -700,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.
@@ -729,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;
@@ -751,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");
@@ -840,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,
@@ -1066,6 +1132,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);
@@ -1146,6 +1213,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)
@@ -1560,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) {
@@ -1636,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;
@@ -1659,10 +1735,12 @@ 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:
+        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) {
@@ -2061,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
@@ -2165,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) {
@@ -2882,10 +3013,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 +3030,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()
@@ -2907,6 +3038,62 @@ 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());
+      }
+      if (!names[di].starts_with(":")) {
+        result.processorPlugins[spec.name] = names[di];
+      }
+    }
+    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:
 //
@@ -2916,17 +3103,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")) {
@@ -2944,7 +3132,8 @@ int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow,
   WorkflowInfo currentWorkflow{
     argv[0],
     currentArgs,
-    currentWorkflowOptions};
+    currentWorkflowOptions,
+    workflowContext.processorPlugins};
 
   ProcessingPolicies processingPolicies;
   enum LogParsingHelpers::LogLevel minFailureLevel;
@@ -2990,6 +3179,7 @@ int doMain(int argc, char** argv, o2::framework::WorkflowSpec const& workflow,
     ("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.");
+}
diff --git a/Framework/Core/test/test_WorkflowPluginsDump.sh b/Framework/Core/test/test_WorkflowPluginsDump.sh
new file mode 100755
index 0000000000000..c85d8b33829c3
--- /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=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\""
+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());
+}
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})