diff --git a/Detectors/TPC/workflow/CMakeLists.txt b/Detectors/TPC/workflow/CMakeLists.txt
index f64a223f683d8..3f8c25b3d8ea1 100644
--- a/Detectors/TPC/workflow/CMakeLists.txt
+++ b/Detectors/TPC/workflow/CMakeLists.txt
@@ -198,6 +198,11 @@ o2_add_executable(idc-test-ft
                 SOURCES test/test_ft_EPN_Aggregator.cxx
                 PUBLIC_LINK_LIBRARIES O2::TPCWorkflow)
 
+o2_add_executable(cmv-test-generator
+                COMPONENT_NAME tpc
+                SOURCES test/test_cmv_generator.cxx
+                PUBLIC_LINK_LIBRARIES O2::TPCWorkflow)
+
 o2_add_executable(miptrack-filter
                   COMPONENT_NAME tpc
                   SOURCES src/tpc-miptrack-filter.cxx
diff --git a/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeCMVSpec.h b/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeCMVSpec.h
index af576b2f30a5b..8496265b68601 100644
--- a/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeCMVSpec.h
+++ b/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeCMVSpec.h
@@ -149,7 +149,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task
     // check which buffer to use for current incoming data
     const bool currentBuffer = (tf > mTFEnd[mBuffer]) ? !mBuffer : mBuffer;
     if (mTFStart[currentBuffer] > tf) {
-      LOGP(detail, "All CRUs for current TF {} already received. Skipping this TF", tf);
+      LOGP(warning, "Current TF {} is older than start of currentBuffer {}. Skipping this TF", tf, mTFStart[currentBuffer]);
       return;
     }
 
@@ -166,6 +166,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task
     }
 
     if (mProcessedCRU[currentBuffer][relTF] == mCRUs.size()) {
+      LOGP(warning, "All CRUs for current TF {} (relTF {}) already received. Skipping this TF", tf, relTF);
       return;
     }
 
@@ -182,17 +183,19 @@ class TPCDistributeCMVSpec : public o2::framework::Task
 
     forwardOrbitInfo(pc, currentBuffer, relTF, currentOutLane);
 
-    for (auto& ref : o2::framework::InputRecordWalker(pc.inputs(), mFilter)) {
-      auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
+    auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mFilter);
+    for (auto it = inputs.begin(); it != inputs.end(); ++it) {
+      auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(*it);
       const unsigned int cru = tpcCRUHeader->subSpecification >> 7;
 
       // check if cru is specified in input cru list
-      if (!(std::binary_search(mCRUs.begin(), mCRUs.end(), cru))) {
+      if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {
         LOGP(debug, "Received data from CRU: {} which was not specified as input. Skipping", cru);
         continue;
       }
 
       if (mProcessedCRUs[currentBuffer][relTF][cru]) {
+        LOGP(warning, "CRU {} for current TF {} (relTF {}) already processed. Skipping ...", cru, tf, relTF);
         continue;
       }
       // count total number of processed CRUs for given TF
@@ -200,7 +203,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task
       // to keep track of processed CRUs
       mProcessedCRUs[currentBuffer][relTF][cru] = true;
 
-      sendOutput(pc, currentOutLane, cru, pc.inputs().get<o2::pmr::vector<uint16_t>>(ref));
+      forwardData(pc, o2::framework::Output{o2::header::gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, cru, it, [&] { sendEmptyCMVOutput(pc, currentOutLane, cru); });
     }
 
     LOGP(detail, "Number of received CRUs for current TF: {} Needed a total number of processed CRUs of: {} Current TF: {}", mProcessedCRU[currentBuffer][relTF], mCRUs.size(), tf);
@@ -247,7 +250,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task
   const unsigned int mTimeFrames{};                                                    ///< number of TFs per aggregation interval
   const int mNTFsBuffer{1};                                                            ///< number of TFs for which the CMVs will be buffered (must match TPCFLPCMVSpec)
   const unsigned int mOutLanes{};                                                      ///< number of parallel aggregate pipelines this distributor feeds
-  std::array<unsigned int, 2> mProcessedTFs{{0, 0}};                                   ///< number of processed timeframes per buffer; triggers sendOutput when it reaches mTimeFrames
+  std::array<unsigned int, 2> mProcessedTFs{{0, 0}};                                   ///< number of processed timeframes per buffer; triggers finishInterval when it reaches mTimeFrames
   std::array<std::vector<unsigned int>, 2> mProcessedCRU{};                            ///< counter of received CRUs per (buffer, relTF); used to detect when a relTF is complete
   std::array<std::vector<std::unordered_map<unsigned int, bool>>, 2> mProcessedCRUs{}; ///< per-CRU received flag ([buffer][relTF][CRU]); prevents double-counting when a CRU re-sends
   std::array<long, 2> mTFStart{};                                                      ///< absolute TF counter of the first TF in each buffer interval
@@ -274,14 +277,26 @@ class TPCDistributeCMVSpec : public o2::framework::Task
   /// Returns the total number of real TFs per buffer interval (= mNTFsBuffer * mTimeFrames)
   unsigned int getNRealTFs() const { return mNTFsBuffer * mTimeFrames; }
 
-  void sendOutput(o2::framework::ProcessingContext& pc, const unsigned int currentOutLane, const unsigned int cru, o2::pmr::vector<uint16_t> cmvs)
+  void sendEmptyCMVOutput(o2::framework::ProcessingContext& pc, const unsigned int currentOutLane, const unsigned int cru)
+  {
+    pc.outputs().adoptContainer(o2::framework::Output{o2::header::gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, o2::pmr::vector<uint16_t>());
+  }
+
+  void sendEmptyOrbitInfo(o2::framework::ProcessingContext& pc, const unsigned int outLane)
   {
-    pc.outputs().adoptContainer(o2::framework::Output{o2::header::gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, std::move(cmvs));
+    pc.outputs().snapshot(o2::framework::Output{o2::header::gDataOriginTPC, mOrbitDescrOut[outLane], header::DataHeader::SubSpecificationType{outLane}}, static_cast<uint64_t>(0));
   }
 
-  void sendOrbitInfo(o2::framework::ProcessingContext& pc, const unsigned int outLane, const uint64_t orbitInfo)
+  template <typename EmptyFn>
+  void forwardData(o2::framework::ProcessingContext& pc, o2::framework::Output outSpec, const unsigned int cru, auto& it, EmptyFn&& sendEmptyData)
   {
-    pc.outputs().snapshot(o2::framework::Output{o2::header::gDataOriginTPC, mOrbitDescrOut[outLane], header::DataHeader::SubSpecificationType{outLane}}, orbitInfo);
+    if (auto* payloadMsg = it.getPayload()) {
+      pc.outputs().forwardPayload(outSpec, *payloadMsg);
+    } else [[unlikely]] {
+      // this should never happen
+      LOGP(warning, "No payload for CRU {}; sending empty {} payload", cru, outSpec.description.as<std::string>());
+      sendEmptyData();
+    }
   }
 
   void forwardOrbitInfo(o2::framework::ProcessingContext& pc, const bool currentBuffer, const unsigned int relTF, const unsigned int currentOutLane)
@@ -290,14 +305,15 @@ class TPCDistributeCMVSpec : public o2::framework::Task
       return;
     }
 
-    for (auto& ref : o2::framework::InputRecordWalker(pc.inputs(), mOrbitFilter)) {
-      auto const* hdr = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
-      const unsigned int cru = hdr->subSpecification >> 7;
+    auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mOrbitFilter);
+    for (auto it = inputs.begin(); it != inputs.end(); ++it) {
+      auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(*it);
+      const unsigned int cru = tpcCRUHeader->subSpecification >> 7;
       if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {
         continue;
       }
 
-      sendOrbitInfo(pc, currentOutLane, pc.inputs().get<uint64_t>(ref));
+      forwardData(pc, o2::framework::Output{o2::header::gDataOriginTPC, mOrbitDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{currentOutLane}}, cru, it, [&] { sendEmptyOrbitInfo(pc, currentOutLane); });
       mOrbitInfoForwarded[currentBuffer][relTF] = true;
       break;
     }
@@ -319,24 +335,30 @@ class TPCDistributeCMVSpec : public o2::framework::Task
     }
 
     if (!mOrbitInfoForwarded[mBuffer].empty()) {
-      for (auto& ref : o2::framework::InputRecordWalker(pc.inputs(), mOrbitFilter)) {
-        auto const* hdr = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
-        const unsigned int cru = hdr->subSpecification >> 7;
+      auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mOrbitFilter);
+      for (auto it = inputs.begin(); it != inputs.end(); ++it) {
+        auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(*it);
+        const unsigned int cru = tpcCRUHeader->subSpecification >> 7;
         if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {
           continue;
         }
-        sendOrbitInfo(pc, currentOutLane, pc.inputs().get<uint64_t>(ref));
+
+        forwardData(pc, o2::framework::Output{o2::header::gDataOriginTPC, mOrbitDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{currentOutLane}}, cru, it, [&] { sendEmptyOrbitInfo(pc, currentOutLane); });
         break;
       }
     }
 
-    for (auto& ref : o2::framework::InputRecordWalker(pc.inputs(), mFilter)) {
-      auto const* hdr = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
-      const unsigned int cru = hdr->subSpecification >> 7;
+    auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mFilter);
+    for (auto it = inputs.begin(); it != inputs.end(); ++it) {
+      auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(*it);
+      const unsigned int cru = tpcCRUHeader->subSpecification >> 7;
+
+      // check if cru is specified in input cru list
       if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {
         continue;
       }
-      sendOutput(pc, currentOutLane, cru, pc.inputs().get<o2::pmr::vector<uint16_t>>(ref));
+
+      forwardData(pc, o2::framework::Output{o2::header::gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, cru, it, [&] { sendEmptyCMVOutput(pc, currentOutLane, cru); });
     }
   }
 
@@ -368,15 +390,15 @@ class TPCDistributeCMVSpec : public o2::framework::Task
 
       // if the last buffer has a smaller time range than expected, flush its remaining uncompleted TFs
       if ((mTFStart[currentBuffer] > mTFStart[!currentBuffer]) && (relTF > mNTFsDataDrop)) {
-        LOGP(warning, "Checking last buffer from {} to {}", mStartNTFsDataDrop[!currentBuffer], mProcessedCRU[!currentBuffer].size());
+        LOGP(warning, "Checking last buffer from relTF {} to {}", mStartNTFsDataDrop[!currentBuffer], mProcessedCRU[!currentBuffer].size());
         const unsigned int lastLane = (currentOutLane == 0) ? (mOutLanes - 1) : (currentOutLane - 1);
         checkMissingData(pc, !currentBuffer, mStartNTFsDataDrop[!currentBuffer], mProcessedCRU[!currentBuffer].size(), lastLane);
-        LOGP(detail, "All empty TFs for TF {} for current buffer filled with dummy and sent. Clearing buffer", tf);
+        LOGP(warning, "All empty TFs of last buffer [{}, {}] filled with dummy and sent, triggered by data from TF {} (relTF {}). Clearing buffer", mTFStart[!currentBuffer], mTFEnd[!currentBuffer], tf, relTF);
         finishInterval(pc, lastLane, !currentBuffer, tf);
       }
 
       const int tfEndCheck = std::clamp(static_cast<int>(relTF) - mNTFsDataDrop, 0, static_cast<int>(mProcessedCRU[currentBuffer].size()));
-      LOGP(detail, "Checking current buffer from {} to {}", mStartNTFsDataDrop[currentBuffer], tfEndCheck);
+      LOGP(detail, "Checking current buffer from relTF {} to {}", mStartNTFsDataDrop[currentBuffer], tfEndCheck);
       checkMissingData(pc, currentBuffer, mStartNTFsDataDrop[currentBuffer], tfEndCheck, currentOutLane);
       mStartNTFsDataDrop[currentBuffer] = tfEndCheck;
     }
@@ -386,7 +408,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task
   {
     for (int iTF = startTF; iTF < endTF; ++iTF) {
       if (mProcessedCRU[currentBuffer][iTF] != mCRUs.size()) {
-        LOGP(warning, "CRUs for lane {} rel. TF: {} curr TF {} are missing! Processed {} CRUs out of {}", outLane, iTF, mTFStart[currentBuffer] + static_cast<long>(iTF) * mNTFsBuffer, mProcessedCRU[currentBuffer][iTF], mCRUs.size());
+        LOGP(warning, "CRUs for lane {} rel. TF: {} curr TF {} are missing! Processed {} CRUs out of {}", outLane, iTF, mTFStart[currentBuffer] + static_cast<long>(iTF) * mNTFsBuffer + mNTFsBuffer - 1, mProcessedCRU[currentBuffer][iTF], mCRUs.size());
         ++mProcessedTFs[currentBuffer];
         mProcessedCRU[currentBuffer][iTF] = mCRUs.size();
 
@@ -394,13 +416,13 @@ class TPCDistributeCMVSpec : public o2::framework::Task
         for (auto& it : mProcessedCRUs[currentBuffer][iTF]) {
           if (!it.second) {
             it.second = true;
-            sendOutput(pc, outLane, it.first, o2::pmr::vector<uint16_t>());
+            sendEmptyCMVOutput(pc, outLane, it.first);
           }
         }
 
         // send zero orbit placeholder for missing TF so the aggregate lane can still reconstruct timing
         if (!mOrbitInfoForwarded[currentBuffer][iTF]) {
-          sendOrbitInfo(pc, outLane, 0);
+          sendEmptyOrbitInfo(pc, outLane);
           mOrbitInfoForwarded[currentBuffer][iTF] = true;
         }
       }
@@ -420,7 +442,7 @@ class TPCDistributeCMVSpec : public o2::framework::Task
       }
     }
 
-    LOGP(detail, "All TFs {} for current buffer received. Clearing buffer", tf);
+    LOGP(info, "All TFs for buffer [{}, {}] received at data from TF {}. Clearing buffer", mTFStart[buffer], mTFEnd[buffer], tf);
     clearBuffer(buffer);
     mStartNTFsDataDrop[buffer] = 0;
     mSendOutputStartInfo[buffer] = true;
diff --git a/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeIDCSpec.h b/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeIDCSpec.h
index 6e589cd6c4e8b..9fe49fc07fb10 100644
--- a/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeIDCSpec.h
+++ b/Detectors/TPC/workflow/include/TPCWorkflow/TPCDistributeIDCSpec.h
@@ -136,7 +136,7 @@ class TPCDistributeIDCSpec : public o2::framework::Task
     // check which buffer to use for current incoming data
     const bool currentBuffer = (tf > mTFEnd[mBuffer]) ? !mBuffer : mBuffer;
     if (mTFStart[currentBuffer] > tf) {
-      LOGP(info, "all CRUs for current TF {} already received. Skipping this TF", tf);
+      LOGP(warning, "Current TF {} is older than start of currentBuffer {}. Skipping this TF", tf, mTFStart[currentBuffer]);
       return;
     }
 
@@ -154,6 +154,7 @@ class TPCDistributeIDCSpec : public o2::framework::Task
     }
 
     if (mProcessedCRU[currentBuffer][relTF] == mCRUs.size()) {
+      LOGP(warning, "All CRUs for current TF {} (relTF {}) already received. Skipping this TF", tf, relTF);
       return;
     }
 
@@ -169,17 +170,19 @@ class TPCDistributeIDCSpec : public o2::framework::Task
       pc.outputs().snapshot(Output{gDataOriginTPC, getDataDescriptionIDCOrbitReset(), header::DataHeader::SubSpecificationType{currentOutLane}}, dataformats::Pair<long, int>{o2::base::GRPGeomHelper::instance().getOrbitResetTimeMS(), o2::base::GRPGeomHelper::instance().getNHBFPerTF()});
     }
 
-    for (auto& ref : InputRecordWalker(pc.inputs(), mFilter)) {
-      auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(ref);
+    auto inputs = o2::framework::InputRecordWalker(pc.inputs(), mFilter);
+    for (auto it = inputs.begin(); it != inputs.end(); ++it) {
+      auto const* tpcCRUHeader = o2::framework::DataRefUtils::getHeader<o2::header::DataHeader*>(*it);
       const unsigned int cru = tpcCRUHeader->subSpecification >> 7;
 
       // check if cru is specified in input cru list
-      if (!(std::binary_search(mCRUs.begin(), mCRUs.end(), cru))) {
+      if (!std::binary_search(mCRUs.begin(), mCRUs.end(), cru)) {
         LOGP(debug, "Received data from CRU: {} which was not specified as input. Skipping", cru);
         continue;
       }
 
       if (mProcessedCRUs[currentBuffer][relTF][cru]) {
+        LOGP(warning, "CRU {} for current TF {} (relTF {}) already processed. Skipping ...", cru, tf, relTF);
         continue;
       } else {
         // count total number of processed CRUs for given TF
@@ -189,8 +192,14 @@ class TPCDistributeIDCSpec : public o2::framework::Task
         mProcessedCRUs[currentBuffer][relTF][cru] = true;
       }
 
-      // sending IDCs
-      sendOutput(pc, currentOutLane, cru, pc.inputs().get<pmr::vector<float>>(ref));
+      // forward payload by shallow copy
+      if (auto* payloadMsg = it.getPayload()) {
+        pc.outputs().forwardPayload(Output{gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, *payloadMsg);
+      } else [[unlikely]] {
+        // this should never happen
+        LOGP(warning, "No IDCGROUP payload for CRU {} (TF {}, relTF {}); sending empty IDCAGG{} payload", cru, tf, relTF, currentOutLane);
+        sendEmptyIDCOutput(pc, currentOutLane, cru);
+      }
     }
 
     LOGP(info, "number of received CRUs for current TF: {}    Needed a total number of processed CRUs of: {}   Current TF: {}", mProcessedCRU[currentBuffer][relTF], mCRUs.size(), tf);
@@ -247,9 +256,9 @@ class TPCDistributeIDCSpec : public o2::framework::Task
   std::vector<InputSpec> mFilter{};                                                    ///< filter for looping over input data
   std::vector<header::DataDescription> mDataDescrOut{};
 
-  void sendOutput(o2::framework::ProcessingContext& pc, const unsigned int currentOutLane, const unsigned int cru, o2::pmr::vector<float> idcs)
+  void sendEmptyIDCOutput(o2::framework::ProcessingContext& pc, const unsigned int currentOutLane, const unsigned int cru)
   {
-    pc.outputs().adoptContainer(Output{gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, std::move(idcs));
+    pc.outputs().adoptContainer(Output{gDataOriginTPC, mDataDescrOut[currentOutLane], header::DataHeader::SubSpecificationType{cru}}, pmr::vector<float>());
   }
 
   /// returns the output lane to which the data will be send
@@ -284,19 +293,19 @@ class TPCDistributeIDCSpec : public o2::framework::Task
   void checkIntervalsForMissingData(o2::framework::ProcessingContext& pc, const bool currentBuffer, const long relTF, const unsigned int currentOutLane, const uint32_t tf)
   {
     if (!(mProcessedTotalData++ % mCheckEveryNData)) {
-      LOGP(info, "Checking for dropped packages...");
+      LOGP(detail, "Checking for dropped packages...");
 
       // if last buffer has smaller time range check the whole last buffer
       if ((mTFStart[currentBuffer] > mTFStart[!currentBuffer]) && (relTF > mNTFsDataDrop)) {
-        LOGP(warning, "checking last buffer from {} to {}", mStartNTFsDataDrop[!currentBuffer], mProcessedCRU[!currentBuffer].size());
+        LOGP(warning, "Checking last buffer from relTF {} to {}", mStartNTFsDataDrop[!currentBuffer], mProcessedCRU[!currentBuffer].size());
         const unsigned int lastLane = (currentOutLane == 0) ? (mOutLanes - 1) : (currentOutLane - 1);
         checkMissingData(pc, !currentBuffer, mStartNTFsDataDrop[!currentBuffer], mProcessedCRU[!currentBuffer].size(), lastLane);
-        LOGP(info, "All empty TFs for TF {} for current buffer filled with dummy and sent. Clearing buffer", tf);
+        LOGP(warning, "All empty TFs of last buffer [{}, {}] filled with dummy and sent, triggered by data from TF {} (relTF {}). Clearing buffer", mTFStart[!currentBuffer], mTFEnd[!currentBuffer], tf, relTF);
         finishInterval(pc, lastLane, !currentBuffer, tf);
       }
 
       const int tfEndCheck = std::clamp(static_cast<int>(relTF) - mNTFsDataDrop, 0, static_cast<int>(mProcessedCRU[currentBuffer].size()));
-      LOGP(info, "checking current buffer from {} to {}", mStartNTFsDataDrop[currentBuffer], tfEndCheck);
+      LOGP(detail, "Checking current buffer from relTF {} to {}", mStartNTFsDataDrop[currentBuffer], tfEndCheck);
       checkMissingData(pc, currentBuffer, mStartNTFsDataDrop[currentBuffer], tfEndCheck, currentOutLane);
       mStartNTFsDataDrop[currentBuffer] = tfEndCheck;
     }
@@ -306,7 +315,7 @@ class TPCDistributeIDCSpec : public o2::framework::Task
   {
     for (int iTF = startTF; iTF < endTF; ++iTF) {
       if (mProcessedCRU[currentBuffer][iTF] != mCRUs.size()) {
-        LOGP(warning, "CRUs for lane {}  rel. TF: {}  curr TF {} are missing! Processed {} CRUs out of {}", outLane, iTF, mTFStart[currentBuffer] + iTF, mProcessedCRU[currentBuffer][iTF], mCRUs.size());
+        LOGP(warning, "CRUs for lane {} rel. TF: {} curr TF {} are missing! Processed {} CRUs out of {}", outLane, iTF, mTFStart[currentBuffer] + static_cast<long>(iTF) * mNTFsBuffer + mNTFsBuffer - 1, mProcessedCRU[currentBuffer][iTF], mCRUs.size());
         ++mProcessedTFs[currentBuffer];
         mProcessedCRU[currentBuffer][iTF] = mCRUs.size();
 
@@ -314,7 +323,7 @@ class TPCDistributeIDCSpec : public o2::framework::Task
         for (auto& it : mProcessedCRUs[currentBuffer][iTF]) {
           if (!it.second) {
             it.second = true;
-            sendOutput(pc, outLane, it.first, pmr::vector<float>());
+            sendEmptyIDCOutput(pc, outLane, it.first);
           }
         }
       }
@@ -334,7 +343,7 @@ class TPCDistributeIDCSpec : public o2::framework::Task
       }
     }
 
-    LOGP(info, "All TFs {} for current buffer received. Clearing buffer", tf);
+    LOGP(info, "All TFs for buffer [{}, {}] received at data from TF {}. Clearing buffer", mTFStart[buffer], mTFEnd[buffer], tf);
     clearBuffer(buffer);
     mStartNTFsDataDrop[buffer] = 0;
     mSendOutputStartInfo[buffer] = true;
diff --git a/Detectors/TPC/workflow/test/test_cmv_generator.cxx b/Detectors/TPC/workflow/test/test_cmv_generator.cxx
new file mode 100644
index 0000000000000..0a670f8d4c0e1
--- /dev/null
+++ b/Detectors/TPC/workflow/test/test_cmv_generator.cxx
@@ -0,0 +1,426 @@
+// Copyright 2019-2020 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.
+
+/// \file  test_cmv_generator.cxx
+/// \brief DPL source workflow that generates dummy CMV data for testing the CMV FLP pipeline.
+///
+/// Replaces o2-tpc-cmv-to-vector in tests; directly emits CMVVECTOR and CMVORBITS
+/// messages per CRU per TF so the workflow can be piped straight into o2-tpc-cmv-flp:
+///
+///   o2-tpc-cmv-test-generator --crus 0-359 --timeframes 100 \
+///   | o2-tpc-cmv-flp --crus 0-359 --n-TFs-buffer 10        \
+///   | o2-dpl-output-proxy --dataspec "downstream:TPC/CMVGROUP;downstream:TPC/CMVORBITINFO" ...
+///
+/// \author Ernst Hellbar <ernst.hellbar@cern.ch>
+
+#include "Framework/DataProcessorSpec.h"
+#include "Framework/Task.h"
+#include "Framework/ControlService.h"
+#include "Framework/ConfigParamRegistry.h"
+#include "Framework/ConfigParamSpec.h"
+#include "Framework/Logger.h"
+#include "Headers/DataHeader.h"
+#include "Algorithm/RangeTokenizer.h"
+#include "TPCBase/CRU.h"
+#include "DataFormatsTPC/CMV.h"
+#include "TPCCalibration/CMVHelper.h"
+#include "TPCCalibration/CMVContainer.h"
+#include "TPCWorkflow/ProcessingHelpers.h"
+#include "CommonUtils/TreeStreamRedirector.h"
+#include "CommonUtils/ConfigurableParam.h"
+#include "DetectorsRaw/HBFUtilsInitializer.h"
+#include "DetectorsRaw/HBFUtils.h"
+#include <fmt/format.h>
+#include <fmt/ranges.h>
+
+#include <vector>
+#include <chrono>
+#include <thread>
+#include <cmath>
+#include <memory>
+#include <limits>
+#include <random>
+#include <stdexcept>
+#include <string>
+#include <unordered_set>
+
+using namespace o2::framework;
+using o2::header::gDataOriginTPC;
+
+// ─────────────────────────────────────────────────────────────────────────────
+// workflow options
+// ─────────────────────────────────────────────────────────────────────────────
+void customize(std::vector<ConfigParamSpec>& workflowOptions)
+{
+  const std::string cruDefault = "0-" + std::to_string(o2::tpc::CRU::MaxCRU - 1);
+  std::vector<ConfigParamSpec> options{
+    {"crus", VariantType::String, cruDefault.c_str(), {"List of CRUs, comma-separated ranges, e.g. 0-3,7,9-15"}},
+    {"timeframes", VariantType::Int, 100, {"Number of TFs to generate; use -1 to run indefinitely"}},
+    {"delay", VariantType::Bool, false, {"Add delay after sending all CRUs"}},
+    {"delayTime", VariantType::Int, 1, {"Duration of the global per-TF delay in ms (requires --delay true)"}},
+    {"delayEveryN", VariantType::Int, 1, {"Apply the global delay only on average once every N TFs, randomly chosen (1 = every TF, requires --delay true)"}},
+    {"delayCRUs", VariantType::String, "", {"CRUs for which to add an extra per-CRU delay before sending, comma-separated ranges"}},
+    {"delayTimeCRUs", VariantType::Int, 1, {"Duration of the per-CRU delay in ms (requires --delayCRUs)"}},
+    {"dropTFsRandom", VariantType::Int, 0, {"Drop a whole TF randomly: on average one every N TFs (0 = disabled)"}},
+    {"dropTFsRange", VariantType::String, "", {"Drop all TFs in this range, e.g. 10-12"}},
+    {"tfLength", VariantType::Float, 0.f, {"Minimum wall-clock time between consecutively sent TFs in ms (rate limiter); the generator sleeps if a TF is produced faster than this (0 = disabled)"}},
+    {"seed", VariantType::Int, 42, {"RNG seed for CMV value generation"}},
+    {"amplitude", VariantType::Float, 5.0f, {"Amplitude of the sinusoidal CMV signal (ADC units); ignored when --input-file is set"}},
+    {"noise", VariantType::Float, 1.0f, {"Gaussian noise std-dev added per time bin (ADC units); used as the smearing width in --input-file mode"}},
+    {"input-file", VariantType::String, "", {"ROOT file with a CMV 'ccdb_object' tree; the template TF (see --input-entry) is decoded once and re-emitted, smeared per generated TF. Empty = synthetic sinusoidal signal"}},
+    {"input-entry", VariantType::Int, 0, {"Tree entry (TF index) used as the template when --input-file is set"}},
+    {"configKeyValues", VariantType::String, "", {"Semicolon separated key=value strings"}}};
+  o2::raw::HBFUtilsInitializer::addConfigOption(options, "hbfutils");
+  std::swap(workflowOptions, options);
+}
+
+#include "Framework/runDataProcessing.h"
+
+// ─────────────────────────────────────────────────────────────────────────────
+// generator device
+// ─────────────────────────────────────────────────────────────────────────────
+class CMVGeneratorDevice : public o2::framework::Task
+{
+ public:
+  static constexpr uint32_t sOrbitsPerPacket = 8; ///< each CMV packet covers 8 heartbeat orbits
+
+  CMVGeneratorDevice(const std::vector<uint32_t>& crus,
+                     const std::unordered_set<uint32_t>& delayCRUs,
+                     unsigned int maxTFs,
+                     bool delay,
+                     int delayTime,
+                     int delayEveryN,
+                     int delayTimeCRUs,
+                     int dropTFsRandom,
+                     const std::vector<int>& rangeTFsDrop,
+                     float tfLength,
+                     float amplitude,
+                     float noise,
+                     int seed,
+                     const std::string& inputFile,
+                     long long inputEntry)
+    : mCRUs(crus), mDelayCRUs(delayCRUs), mMaxTFs(maxTFs), mDelay(delay), mDelayTime(delayTime), mDelayEveryN(delayEveryN), mDelayTimeCRUs(delayTimeCRUs), mDropTFsRandom(dropTFsRandom), mRangeTFsDrop(rangeTFsDrop), mTFLength(tfLength), mAmplitude(amplitude), mNoise(noise), mRng(static_cast<std::mt19937::result_type>(seed)), mInputFileName(inputFile), mInputEntry(inputEntry) {}
+
+  void init(o2::framework::InitContext& ic) final
+  {
+    mTimer100TFs = std::chrono::high_resolution_clock::now();
+    mLastTFTime = std::chrono::high_resolution_clock::now();
+
+    if (!mCRUs.empty()) {
+      LOGP(info, "crus: {}", fmt::join(mCRUs, ", "));
+    }
+    if (!mDelayCRUs.empty()) {
+      const std::vector<uint32_t> delayCRUsSorted(mDelayCRUs.begin(), mDelayCRUs.end());
+      LOGP(info, "delayCRUs: {}", fmt::join(delayCRUsSorted, ", "));
+    }
+
+    mWriteDebug = ic.options().get<bool>("write-debug");
+    if (mWriteDebug) {
+      mDebugStreamFileName = ic.options().get<std::string>("debug-file-name");
+      LOGP(info, "Creating debug stream {}", mDebugStreamFileName);
+      mDebugStream = std::make_unique<o2::utils::TreeStreamRedirector>(mDebugStreamFileName.data(), "recreate");
+    }
+
+    if (!mInputFileName.empty()) {
+      o2::tpc::CMVFileHandle handle;
+      if (!handle.open(mInputFileName)) {
+        throw std::runtime_error("CMV generator: failed to open input file " + mInputFileName);
+      }
+      const auto nEntries = handle.tree->GetEntries();
+      if (mInputEntry < 0 || mInputEntry >= nEntries) {
+        const auto msg = fmt::format("CMV generator: --input-entry {} out of range [0, {}) in {}", mInputEntry, nEntries, mInputFileName);
+        handle.close();
+        throw std::runtime_error(msg);
+      }
+      const o2::tpc::CMVPerTF* tmpl = handle.getEntry(mInputEntry);
+      if (!tmpl) {
+        handle.close();
+        throw std::runtime_error("CMV generator: failed to read/decode entry from " + mInputFileName);
+      }
+      // When noise is enabled we keep the per-CRU float template and re-encode it
+      // (template + noise) every TF. When noise is disabled the output is identical
+      // for every TF, so we encode it once here and just re-snapshot it in run().
+      const bool addNoise = (mNoise > 0.f);
+      if (addNoise) {
+        mBaseCMVFloat.resize(mCRUs.size());
+        for (size_t iCRU = 0; iCRU < mCRUs.size(); ++iCRU) {
+          const auto cru = mCRUs[iCRU];
+          auto& base = mBaseCMVFloat[iCRU];
+          base.resize(o2::tpc::cmv::NTimeBinsPerTF);
+          for (uint32_t tb = 0; tb < o2::tpc::cmv::NTimeBinsPerTF; ++tb) {
+            base[tb] = tmpl->getCMVFloat(static_cast<int>(cru), static_cast<int>(tb));
+          }
+        }
+      } else {
+        mBaseCMVEncoded.resize(mCRUs.size());
+        for (size_t iCRU = 0; iCRU < mCRUs.size(); ++iCRU) {
+          const auto cru = mCRUs[iCRU];
+          auto& enc = mBaseCMVEncoded[iCRU];
+          enc.resize(o2::tpc::cmv::NTimeBinsPerTF);
+          for (uint32_t tb = 0; tb < o2::tpc::cmv::NTimeBinsPerTF; ++tb) {
+            o2::tpc::cmv::Data d;
+            d.setCMVFloat(tmpl->getCMVFloat(static_cast<int>(cru), static_cast<int>(tb)));
+            enc[tb] = d.getCMV();
+          }
+        }
+      }
+      handle.close();
+      mUseInputFile = true;
+      LOGP(info, "Loaded CMV template from {} (entry {}): {} CRUs x {} bins, noise sigma {} ADC ({})",
+           mInputFileName, mInputEntry, mCRUs.size(), o2::tpc::cmv::NTimeBinsPerTF, mNoise,
+           addNoise ? "re-smeared per TF" : "encoded once, replayed verbatim");
+    }
+  }
+
+  void run(o2::framework::ProcessingContext& ctx) final
+  {
+    using timer = std::chrono::high_resolution_clock;
+    const auto tf = o2::tpc::processing_helpers::getCurrentTF(ctx);
+
+    // ── TF dropping ──────────────────────────────────────────────────────────
+    // Note: RangeTokenizer guarantees sorted output, so front()/back() are min/max.
+    if (!mRangeTFsDrop.empty() && tf >= static_cast<uint32_t>(mRangeTFsDrop.front()) && tf <= static_cast<uint32_t>(mRangeTFsDrop.back())) {
+      LOGP(info, "Dropping TF {} (range drop)", tf);
+      return;
+    }
+    if (mDropTFsRandom > 0 && std::uniform_int_distribution<int>{0, mDropTFsRandom - 1}(mRng) == 0) {
+      LOGP(info, "Dropping TF {} (random drop)", tf);
+      return;
+    }
+
+    auto start = timer::now();
+
+    // ── CMV values ───────────────────────────────────────────────────────────
+    // NTimeBinsPerTF = NPacketsPerTFPerCRU (4) * NTimeBinsPerPacket (3564) = 14256
+    //   - synthetic mode: shared cmvVec = sinusoidal signal + noise (same for all CRUs)
+    //   - input-file mode: per-CRU template + the shared noise vector
+    const bool addNoise = (mNoise > 0.f); // skip all RNG when --noise 0
+    std::normal_distribution<float> noiseDist{0.f, mNoise};
+    std::vector<uint16_t> cmvVec(o2::tpc::cmv::NTimeBinsPerTF);
+    std::vector<float> noiseVec; // only populated in input-file mode when noise is enabled
+    if (mUseInputFile) {
+      if (addNoise) {
+        noiseVec.resize(o2::tpc::cmv::NTimeBinsPerTF);
+        for (auto& n : noiseVec) {
+          n = noiseDist(mRng);
+        }
+      }
+    } else {
+      const float signal = -std::abs(mAmplitude * std::sin(tf * 0.05f));
+      for (auto& v : cmvVec) {
+        o2::tpc::cmv::Data d;
+        d.setCMVFloat(addNoise ? (signal + noiseDist(mRng)) : signal);
+        v = d.getCMV();
+      }
+    }
+
+    // ── Orbit / BC info (same for all CRUs) ──────────────────────────────────
+    // One packed (orbit<<32|bc) entry per CMV packet (4 per TF).
+    // Each packet covers 8 heartbeat orbits (NTimeBinsPerPacket = 3564 = 8 LHC orbits),
+    // so the orbit advances by 8 per packet and by NPacketsPerTFPerCRU*8 = 32 per TF.
+    std::vector<uint64_t> orbitBCVec(o2::tpc::cmv::NPacketsPerTFPerCRU);
+    for (uint32_t pkt = 0; pkt < o2::tpc::cmv::NPacketsPerTFPerCRU; ++pkt) {
+      const uint32_t orbit = static_cast<uint32_t>(tf * o2::tpc::cmv::NPacketsPerTFPerCRU * sOrbitsPerPacket + pkt * sOrbitsPerPacket);
+      orbitBCVec[pkt] = uint64_t(orbit) << 32; // bc = 0
+    }
+
+    for (size_t iCRU = 0; iCRU < mCRUs.size(); ++iCRU) {
+      const auto cru = mCRUs[iCRU];
+      const o2::header::DataHeader::SubSpecificationType subSpec{cru << 7};
+
+      // ── per-CRU delay ────────────────────────────────────────────────────
+      if (mDelayCRUs.count(cru)) {
+        LOGP(info, "Delaying CRU {} by {} ms (TF {})", cru, mDelayTimeCRUs, tf);
+        std::this_thread::sleep_for(std::chrono::milliseconds(mDelayTimeCRUs));
+      }
+
+      // Select the vector to emit: the precomputed template (no noise) is sent
+      // verbatim; otherwise this CRU's template is smeared with the shared noise.
+      std::vector<uint16_t>* out = &cmvVec;
+      if (mUseInputFile && !addNoise) {
+        out = &mBaseCMVEncoded[iCRU]; // encoded once in init(), reused every TF
+      } else if (mUseInputFile) {
+        const auto& base = mBaseCMVFloat[iCRU];
+        for (uint32_t tb = 0; tb < o2::tpc::cmv::NTimeBinsPerTF; ++tb) {
+          o2::tpc::cmv::Data d;
+          d.setCMVFloat(base[tb] + noiseVec[tb]);
+          cmvVec[tb] = d.getCMV();
+        }
+      }
+
+      ctx.outputs().snapshot(Output{gDataOriginTPC, "CMVVECTOR", subSpec}, *out);
+      ctx.outputs().snapshot(Output{gDataOriginTPC, "CMVORBITS", subSpec}, orbitBCVec);
+
+      if (mWriteDebug) {
+        auto& stream = (*mDebugStream) << "cmvs";
+        stream << "cru=" << cru
+               << "tfCounter=" << tf
+               << "nCMVs=" << out->size()
+               << "cmvs=" << *out
+               << "\n";
+      }
+    }
+
+    if (!(tf % 100)) {
+      const auto elapsed100 = std::chrono::duration_cast<std::chrono::milliseconds>(timer::now() - mTimer100TFs).count();
+      LOGP(info, "Generated CMV data for TF {} ({} ms for last 100 TFs)", tf, elapsed100);
+      mTimer100TFs = timer::now();
+    }
+
+    // ── global delay ─────────────────────────────────────────────────────────
+    if (mDelay && (mDelayEveryN <= 1 || std::uniform_int_distribution<int>{0, mDelayEveryN - 1}(mRng) == 0)) {
+      auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(timer::now() - start).count();
+      if (elapsed < mDelayTime) {
+        LOGP(info, "Delaying TF {} by {} ms", tf, mDelayTime - elapsed);
+        std::this_thread::sleep_for(std::chrono::milliseconds(mDelayTime - elapsed));
+      }
+    }
+
+    // ── rate limiting ────────────────────────────────────────────────────────
+    // Enforce a minimum wall-clock spacing between consecutively sent TFs.
+    if (mTFLength > 0.f) {
+      const auto elapsedSinceLast = std::chrono::duration_cast<std::chrono::microseconds>(timer::now() - mLastTFTime).count();
+      const auto tfLengthUs = static_cast<int64_t>(mTFLength * 1000.f);
+      if (elapsedSinceLast < tfLengthUs) {
+        const auto waitUs = tfLengthUs - elapsedSinceLast;
+        LOGP(info, "Rate limiting TF {}: waiting {} us (tfLength={} ms)", tf, waitUs, mTFLength);
+        std::this_thread::sleep_for(std::chrono::microseconds(waitUs));
+      }
+      mLastTFTime = timer::now();
+    }
+
+    // endOfStream() propagates the EoS signal to downstream devices (required for source devices).
+    if (mMaxTFs != std::numeric_limits<unsigned int>::max() && tf >= mMaxTFs - 1) {
+      ctx.services().get<ControlService>().endOfStream();
+      ctx.services().get<ControlService>().readyToQuit(QuitRequest::Me);
+    }
+  }
+
+  void endOfStream(o2::framework::EndOfStreamContext&) final { closeFiles(); }
+  void stop() final { closeFiles(); }
+
+ private:
+  void closeFiles()
+  {
+    if (mDebugStream) {
+      auto& stream = (*mDebugStream) << "cmvs";
+      auto& tree = stream.getTree();
+      tree.SetAlias("sector", "int(cru/10)");
+      mDebugStream->Close();
+      mDebugStream.reset(nullptr);
+    }
+  }
+
+  const std::vector<uint32_t> mCRUs{};
+  const std::unordered_set<uint32_t> mDelayCRUs{};
+  const unsigned int mMaxTFs{};
+  const bool mDelay{false};
+  const int mDelayTime{1};
+  const int mDelayEveryN{1};
+  const int mDelayTimeCRUs{1};
+  const int mDropTFsRandom{0};
+  const std::vector<int> mRangeTFsDrop{};
+  const float mTFLength{0.f};
+  const float mAmplitude{5.f};
+  const float mNoise{1.f};
+  std::mt19937 mRng{};
+  const std::string mInputFileName{};                 ///< CMV ROOT file to use as template ("" = synthetic mode)
+  const long long mInputEntry{0};                     ///< tree entry (TF) used as template
+  bool mUseInputFile{false};                          ///< true once a template has been loaded
+  std::vector<std::vector<float>> mBaseCMVFloat;      ///< decoded template CMV values [iCRU][timeBin] (noise>0 path), aligned to mCRUs
+  std::vector<std::vector<uint16_t>> mBaseCMVEncoded; ///< pre-encoded template output [iCRU][timeBin] (noise==0 path), aligned to mCRUs
+  std::chrono::high_resolution_clock::time_point mTimer100TFs{};
+  std::chrono::high_resolution_clock::time_point mLastTFTime{};
+  bool mWriteDebug{false};
+  std::string mDebugStreamFileName{};
+  std::unique_ptr<o2::utils::TreeStreamRedirector> mDebugStream{};
+};
+
+// ─────────────────────────────────────────────────────────────────────────────
+DataProcessorSpec generateCMVsCRU(const std::vector<uint32_t>& crus,
+                                  const std::unordered_set<uint32_t>& delayCRUs,
+                                  unsigned int maxTFs,
+                                  bool delay,
+                                  int delayTime,
+                                  int delayEveryN,
+                                  int delayTimeCRUs,
+                                  int dropTFsRandom,
+                                  const std::vector<int>& rangeTFsDrop,
+                                  float tfLength,
+                                  float amplitude,
+                                  float noise,
+                                  int seed,
+                                  const std::string& inputFile,
+                                  long long inputEntry)
+{
+  std::vector<OutputSpec> outputSpecs;
+  outputSpecs.reserve(crus.size() * 2);
+  for (const auto cru : crus) {
+    const o2::header::DataHeader::SubSpecificationType subSpec{cru << 7};
+    outputSpecs.emplace_back(gDataOriginTPC, "CMVVECTOR", subSpec, Lifetime::Timeframe);
+    outputSpecs.emplace_back(gDataOriginTPC, "CMVORBITS", subSpec, Lifetime::Timeframe);
+  }
+
+  return DataProcessorSpec{
+    "tpc-cmv-generator",
+    Inputs{},
+    outputSpecs,
+    AlgorithmSpec{adaptFromTask<CMVGeneratorDevice>(crus, delayCRUs, maxTFs, delay, delayTime, delayEveryN, delayTimeCRUs, dropTFsRandom, rangeTFsDrop, tfLength, amplitude, noise, seed, inputFile, inputEntry)},
+    Options{
+      {"write-debug", VariantType::Bool, false, {"Write a debug output tree"}},
+      {"debug-file-name", VariantType::String, "./cmv_generator_debug.root", {"Name of the debug output file"}},
+    }};
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+WorkflowSpec defineDataProcessing(ConfigContext const& config)
+{
+  const auto tpcCRUs = o2::RangeTokenizer::tokenize<int>(config.options().get<std::string>("crus"));
+  const std::vector<uint32_t> crus(tpcCRUs.begin(), tpcCRUs.end());
+
+  const auto delayCRUsStr = config.options().get<std::string>("delayCRUs");
+  std::unordered_set<uint32_t> delayCRUs;
+  if (!delayCRUsStr.empty()) {
+    for (const auto cru : o2::RangeTokenizer::tokenize<int>(delayCRUsStr)) {
+      delayCRUs.insert(static_cast<uint32_t>(cru));
+    }
+  }
+
+  const auto dropTFsRangeStr = config.options().get<std::string>("dropTFsRange");
+  const auto rangeTFsDrop = dropTFsRangeStr.empty() ? std::vector<int>{} : o2::RangeTokenizer::tokenize<int>(dropTFsRangeStr);
+  const int timeframesInt = config.options().get<int>("timeframes");
+  // -1 means run indefinitely; map to UINT_MAX so the termination check never fires.
+  const auto timeframes = (timeframesInt < 0) ? std::numeric_limits<unsigned int>::max() : static_cast<unsigned int>(timeframesInt);
+  const auto delay = config.options().get<bool>("delay");
+  const auto delayTime = config.options().get<int>("delayTime");
+  const auto delayEveryN = config.options().get<int>("delayEveryN");
+  const auto delayTimeCRUs = config.options().get<int>("delayTimeCRUs");
+  const auto dropTFsRandom = config.options().get<int>("dropTFsRandom");
+  const auto tfLength = config.options().get<float>("tfLength");
+  const auto seed = config.options().get<int>("seed");
+  const auto amplitude = config.options().get<float>("amplitude");
+  const auto noise = config.options().get<float>("noise");
+  const auto inputFile = config.options().get<std::string>("input-file");
+  const auto inputEntry = static_cast<long long>(config.options().get<int>("input-entry"));
+
+  o2::conf::ConfigurableParam::updateFromString(config.options().get<std::string>("configKeyValues"));
+
+  WorkflowSpec workflow;
+  workflow.emplace_back(generateCMVsCRU(crus, delayCRUs, timeframes, delay, delayTime, delayEveryN, delayTimeCRUs, dropTFsRandom, rangeTFsDrop, tfLength, amplitude, noise, seed, inputFile, inputEntry));
+
+  auto& hbfu = o2::raw::HBFUtils::Instance();
+  long startTime = hbfu.startTime > 0 ? hbfu.startTime : std::chrono::time_point_cast<std::chrono::milliseconds>(std::chrono::system_clock::now()).time_since_epoch().count();
+  o2::conf::ConfigurableParam::updateFromString(fmt::format("HBFUtils.startTime={}", startTime).data());
+  o2::conf::ConfigurableParam::updateFromString(fmt::format("HBFUtils.nHBFPerTF={}", hbfu.nHBFPerTF).data());
+  o2::raw::HBFUtilsInitializer hbfIni(config, workflow);
+
+  return workflow;
+}
diff --git a/Framework/Core/include/Framework/DataAllocator.h b/Framework/Core/include/Framework/DataAllocator.h
index 104e418cbcd19..89572fc86ca2c 100644
--- a/Framework/Core/include/Framework/DataAllocator.h
+++ b/Framework/Core/include/Framework/DataAllocator.h
@@ -455,6 +455,11 @@ class DataAllocator
   void snapshot(const Output& spec, const char* payload, size_t payloadSize,
                 o2::header::SerializationMethod serializationMethod = o2::header::gSerializationMethodNone);
 
+  /// create a shallow copy of the @a inputPayload and forward it to the output route of @a spec
+  /// if the transport types of the input and output routes are different, a real copy is created as fallback
+  void forwardPayload(const Output& spec, fair::mq::Message& inputPayload,
+                      o2::header::SerializationMethod serializationMethod = o2::header::gSerializationMethodNone);
+
   /// make an object of type T and route to output specified by OutputRef
   /// The object is owned by the framework, returned reference can be used to fill the object.
   ///
diff --git a/Framework/Core/include/Framework/InputRecord.h b/Framework/Core/include/Framework/InputRecord.h
index ff5c902ec9e40..dc8d8455bb418 100644
--- a/Framework/Core/include/Framework/InputRecord.h
+++ b/Framework/Core/include/Framework/InputRecord.h
@@ -213,6 +213,9 @@ class InputRecord
   /// O(1) access to the part described by @a indices in slot @a pos.
   [[nodiscard]] DataRef getAtIndices(int pos, DataRefIndices indices) const;
 
+  /// Return the payload as fair::mq::Message* for the part described by @a indices in slot @a slotIdx
+  fair::mq::Message* getPayloadAtIndices(size_t slotIdx, DataRefIndices indices) const;
+
   /// O(1) advance from @a current to the next part's indices in slot @a pos.
   [[nodiscard]] DataRefIndices nextIndices(int pos, DataRefIndices current) const
   {
@@ -745,6 +748,7 @@ class InputRecord
     [[nodiscard]] DataRefIndices initialIndices() const { return {0, 1}; }
     [[nodiscard]] DataRefIndices endIndices() const { return {size_t(-1), size_t(-1)}; }
     [[nodiscard]] DataRef getAtIndices(DataRefIndices idx) const { return record->getAtIndices((int)slot, idx); }
+    [[nodiscard]] fair::mq::Message* getPayloadAtIndices(DataRefIndices idx) const { return record->getPayloadAtIndices((int)slot, idx); }
     [[nodiscard]] DataRefIndices nextIndices(DataRefIndices idx) const { return record->nextIndices((int)slot, idx); }
     [[nodiscard]] size_t size() const { return record->getNofParts((int)slot); }
 
diff --git a/Framework/Core/include/Framework/InputRecordWalker.h b/Framework/Core/include/Framework/InputRecordWalker.h
index 528d5ad0c327c..88403cabc1190 100644
--- a/Framework/Core/include/Framework/InputRecordWalker.h
+++ b/Framework/Core/include/Framework/InputRecordWalker.h
@@ -115,6 +115,11 @@ class InputRecordWalker
       return not operator==(rh);
     }
 
+    fair::mq::Message* getPayload() const
+    {
+      return mCurrentRange.getPayloadAtIndices(mCurrent.indices());
+    }
+
    private:
     bool next(bool isInitialPart = false)
     {
diff --git a/Framework/Core/include/Framework/InputSpan.h b/Framework/Core/include/Framework/InputSpan.h
index d708d2e2f5dde..424ea5ad9e139 100644
--- a/Framework/Core/include/Framework/InputSpan.h
+++ b/Framework/Core/include/Framework/InputSpan.h
@@ -13,9 +13,11 @@
 
 #include "Framework/DataRef.h"
 #include <functional>
+#include <fairmq/FwdDecls.h>
 
 extern template class std::function<o2::framework::DataRef(size_t, o2::framework::DataRefIndices)>;
 extern template class std::function<o2::framework::DataRefIndices(size_t, o2::framework::DataRefIndices)>;
+extern template class std::function<fair::mq::Message*(size_t, o2::framework::DataRefIndices)>;
 
 namespace o2::framework
 {
@@ -38,6 +40,7 @@ class InputSpan
             std::function<int(size_t)> refCountGetter,
             std::function<DataRef(size_t, DataRefIndices)> indicesGetter,
             std::function<DataRefIndices(size_t, DataRefIndices)> nextIndicesGetter,
+            std::function<fair::mq::Message*(size_t, DataRefIndices)> payloadGetter,
             size_t size);
 
   /// @a i-th element of the InputSpan (O(partidx) sequential scan via indices protocol)
@@ -56,6 +59,12 @@ class InputSpan
     return mIndicesGetter(slotIdx, indices);
   }
 
+  /// Return the payload as fair::mq::Message* for the part described by @a indices in slot @a slotIdx
+  [[nodiscard]] fair::mq::Message* getPayloadAtIndices(size_t slotIdx, DataRefIndices indices) const
+  {
+    return mPayloadGetter(slotIdx, indices);
+  }
+
   /// Advance from @a current to the indices of the next part in slot @a slotIdx in O(1).
   [[nodiscard]] DataRefIndices nextIndices(size_t slotIdx, DataRefIndices current) const
   {
@@ -179,6 +188,12 @@ class InputSpan
       return mCurrentIndices.headerIdx;
     }
 
+    // return current indices
+    [[nodiscard]] DataRefIndices indices() const
+    {
+      return mCurrentIndices;
+    }
+
     // return an iterable range over all parts in the current slot
     // only available for slot-level iterators whose parent has parts(size_t)
     [[nodiscard]] auto parts() const
@@ -201,6 +216,7 @@ class InputSpan
     [[nodiscard]] DataRefIndices initialIndices() const { return {0, 1}; }
     [[nodiscard]] DataRefIndices endIndices() const { return {size_t(-1), size_t(-1)}; }
     [[nodiscard]] DataRef getAtIndices(DataRefIndices idx) const { return span->getAtIndices(slot, idx); }
+    [[nodiscard]] fair::mq::Message* getPayloadAtIndices(DataRefIndices idx) const { return span->getPayloadAtIndices(slot, idx); }
     [[nodiscard]] DataRefIndices nextIndices(DataRefIndices idx) const { return span->nextIndices(slot, idx); }
     [[nodiscard]] size_t size() const { return span->getNofParts(slot); }
 
@@ -230,6 +246,7 @@ class InputSpan
   std::function<int(size_t)> mRefCountGetter;
   std::function<DataRef(size_t, DataRefIndices)> mIndicesGetter;
   std::function<DataRefIndices(size_t, DataRefIndices)> mNextIndicesGetter;
+  std::function<fair::mq::Message*(size_t, DataRefIndices)> mPayloadGetter;
   size_t mSize;
 };
 
diff --git a/Framework/Core/src/DataAllocator.cxx b/Framework/Core/src/DataAllocator.cxx
index 0802bb8300ae7..92f7a7f8a6e8f 100644
--- a/Framework/Core/src/DataAllocator.cxx
+++ b/Framework/Core/src/DataAllocator.cxx
@@ -342,6 +342,26 @@ void DataAllocator::snapshot(const Output& spec, const char* payload, size_t pay
   addPartToContext(routeIndex, std::move(payloadMessage), spec, serializationMethod);
 }
 
+void DataAllocator::forwardPayload(const Output& spec, fair::mq::Message& inputPayload,
+                                   o2::header::SerializationMethod serializationMethod)
+{
+  auto& proxy = mRegistry.get<FairMQDeviceProxy>();
+  auto& timingInfo = mRegistry.get<TimingInfo>();
+
+  RouteIndex routeIndex = matchDataHeader(spec, timingInfo.timeslice);
+  auto* transport = proxy.getOutputTransport(routeIndex);
+
+  if (inputPayload.GetTransport() == transport) {
+    auto payloadMessage = transport->CreateMessage();
+    payloadMessage->Copy(inputPayload);
+    addPartToContext(routeIndex, std::move(payloadMessage), spec, serializationMethod);
+  } else {
+    auto payloadMessage = transport->CreateMessage(inputPayload.GetSize(), fair::mq::Alignment{64});
+    memcpy(payloadMessage->GetData(), inputPayload.GetData(), inputPayload.GetSize());
+    addPartToContext(routeIndex, std::move(payloadMessage), spec, serializationMethod);
+  }
+}
+
 Output DataAllocator::getOutputByBind(OutputRef&& ref)
 {
   if (ref.label.empty()) {
diff --git a/Framework/Core/src/DataProcessingDevice.cxx b/Framework/Core/src/DataProcessingDevice.cxx
index 4121d333f6b56..ab9a66f4e45d3 100644
--- a/Framework/Core/src/DataProcessingDevice.cxx
+++ b/Framework/Core/src/DataProcessingDevice.cxx
@@ -2203,7 +2203,14 @@ bool DataProcessingDevice::tryDispatchComputation(ServiceRegistryRef ref, std::v
       auto next = currentSetOfInputs[i] | get_next_pair{current};
       return next.headerIdx < currentSetOfInputs[i].size() ? next : DataRefIndices{size_t(-1), size_t(-1)};
     };
-    return InputSpan{nofPartsGetter, refCountGetter, indicesGetter, nextIndicesGetter, currentSetOfInputs.size()};
+    auto payloadGetter = [&currentSetOfInputs](size_t i, DataRefIndices current) -> fair::mq::Message* {
+      auto const& msgs = currentSetOfInputs[i];
+      if (msgs.size() <= current.payloadIdx || !msgs[current.payloadIdx]) {
+        return nullptr;
+      }
+      return msgs[current.payloadIdx].get();
+    };
+    return InputSpan{nofPartsGetter, refCountGetter, indicesGetter, nextIndicesGetter, payloadGetter, currentSetOfInputs.size()};
   };
 
   auto markInputsAsDone = [ref](TimesliceSlot slot) -> void {
diff --git a/Framework/Core/src/DataRelayer.cxx b/Framework/Core/src/DataRelayer.cxx
index 7adf5b5c97fbb..67c382d413783 100644
--- a/Framework/Core/src/DataRelayer.cxx
+++ b/Framework/Core/src/DataRelayer.cxx
@@ -236,7 +236,14 @@ DataRelayer::ActivityStats DataRelayer::processDanglingInputs(std::vector<Expira
         auto next = partial[idx] | get_next_pair{current};
         return next.headerIdx < partial[idx].size() ? next : DataRefIndices{size_t(-1), size_t(-1)};
       };
-      InputSpan span{nPartsGetter, refCountGetter, indicesGetter, nextIndicesGetter, static_cast<size_t>(partial.size())};
+      auto payloadGetter = [&partial](size_t idx, DataRefIndices current) -> fair::mq::Message* {
+        auto const& msgs = partial[idx];
+        if (msgs.size() <= current.payloadIdx || !msgs[current.payloadIdx]) {
+          return nullptr;
+        }
+        return msgs[current.payloadIdx].get();
+      };
+      InputSpan span{nPartsGetter, refCountGetter, indicesGetter, nextIndicesGetter, payloadGetter, static_cast<size_t>(partial.size())};
       // Setup the input span
 
       if (expirator.checker(services, timestamp.value, span) == false) {
@@ -818,7 +825,14 @@ void DataRelayer::getReadyToProcess(std::vector<DataRelayer::RecordAction>& comp
       auto next = partial[idx] | get_next_pair{current};
       return next.headerIdx < partial[idx].size() ? next : DataRefIndices{size_t(-1), size_t(-1)};
     };
-    InputSpan span{nPartsGetter, refCountGetter, indicesGetter, nextIndicesGetter, static_cast<size_t>(partial.size())};
+    auto payloadGetter = [&partial](size_t idx, DataRefIndices current) -> fair::mq::Message* {
+      auto const& msgs = partial[idx];
+      if (msgs.size() <= current.payloadIdx || !msgs[current.payloadIdx]) {
+        return nullptr;
+      }
+      return msgs[current.payloadIdx].get();
+    };
+    InputSpan span{nPartsGetter, refCountGetter, indicesGetter, nextIndicesGetter, payloadGetter, static_cast<size_t>(partial.size())};
     CompletionPolicy::CompletionOp action = mCompletionPolicy.callbackFull(span, mInputs, mContext);
 
     auto& variables = mTimesliceIndex.getVariablesForSlot(slot);
diff --git a/Framework/Core/src/InputRecord.cxx b/Framework/Core/src/InputRecord.cxx
index 7bc9907b13ba4..514f4a33b337a 100644
--- a/Framework/Core/src/InputRecord.cxx
+++ b/Framework/Core/src/InputRecord.cxx
@@ -149,6 +149,11 @@ DataRef InputRecord::getAtIndices(int pos, DataRefIndices indices) const
   return ref;
 }
 
+fair::mq::Message* InputRecord::getPayloadAtIndices(size_t slotIdx, DataRefIndices indices) const
+{
+  return mSpan.getPayloadAtIndices(slotIdx, indices);
+}
+
 size_t InputRecord::size() const
 {
   return mSpan.size();
diff --git a/Framework/Core/src/InputSpan.cxx b/Framework/Core/src/InputSpan.cxx
index ccea2d1dd66ed..e6a81d3088d72 100644
--- a/Framework/Core/src/InputSpan.cxx
+++ b/Framework/Core/src/InputSpan.cxx
@@ -13,6 +13,7 @@
 
 template class std::function<o2::framework::DataRef(size_t, o2::framework::DataRefIndices)>;
 template class std::function<o2::framework::DataRefIndices(size_t, o2::framework::DataRefIndices)>;
+template class std::function<fair::mq::Message*(size_t, o2::framework::DataRefIndices)>;
 
 namespace o2::framework
 {
@@ -20,8 +21,9 @@ InputSpan::InputSpan(std::function<size_t(size_t)> nofPartsGetter,
                      std::function<int(size_t)> refCountGetter,
                      std::function<DataRef(size_t, DataRefIndices)> indicesGetter,
                      std::function<DataRefIndices(size_t, DataRefIndices)> nextIndicesGetter,
+                     std::function<fair::mq::Message*(size_t, DataRefIndices)> payloadGetter,
                      size_t size)
-  : mNofPartsGetter{nofPartsGetter}, mRefCountGetter(refCountGetter), mIndicesGetter{std::move(indicesGetter)}, mNextIndicesGetter{std::move(nextIndicesGetter)}, mSize{size}
+  : mNofPartsGetter{nofPartsGetter}, mRefCountGetter(refCountGetter), mIndicesGetter{std::move(indicesGetter)}, mNextIndicesGetter{std::move(nextIndicesGetter)}, mPayloadGetter{std::move(payloadGetter)}, mSize{size}
 {
 }
 
diff --git a/Framework/Core/test/benchmark_InputRecord.cxx b/Framework/Core/test/benchmark_InputRecord.cxx
index e3ec00ac815ed..224dafd53d13b 100644
--- a/Framework/Core/test/benchmark_InputRecord.cxx
+++ b/Framework/Core/test/benchmark_InputRecord.cxx
@@ -52,6 +52,7 @@ static void BM_InputRecordGenericGetters(benchmark::State& state)
     nullptr,
     [](size_t, DataRefIndices) { return DataRef{nullptr, nullptr, nullptr}; },
     [](size_t, DataRefIndices) -> DataRefIndices { return {size_t(-1), size_t(-1)}; },
+    nullptr,
     0};
   ServiceRegistry registry;
   InputRecord emptyRecord(schema, span, registry);
@@ -92,6 +93,7 @@ static void BM_InputRecordGenericGetters(benchmark::State& state)
     nullptr,
     [&inputs](size_t i, DataRefIndices idx) { return DataRef{nullptr, static_cast<char const*>(inputs[2 * i + idx.headerIdx]), static_cast<char const*>(inputs[2 * i + idx.payloadIdx])}; },
     [](size_t, DataRefIndices) -> DataRefIndices { return {size_t(-1), size_t(-1)}; },
+    nullptr,
     inputs.size() / 2};
   InputRecord record{schema, span2, registry};
 
diff --git a/Framework/Core/test/test_CompletionPolicy.cxx b/Framework/Core/test/test_CompletionPolicy.cxx
index cc16ba95ba8f2..ee7dc7e91df24 100644
--- a/Framework/Core/test/test_CompletionPolicy.cxx
+++ b/Framework/Core/test/test_CompletionPolicy.cxx
@@ -60,6 +60,7 @@ TEST_CASE("TestCompletionPolicy_callback")
     nullptr,
     [&ref](size_t, DataRefIndices) -> DataRef { return ref; },
     [](size_t, DataRefIndices) -> DataRefIndices { return {size_t(-1), size_t(-1)}; },
+    nullptr,
     1};
   std::vector<InputSpec> specs;
   ServiceRegistryRef servicesRef{services};
diff --git a/Framework/Core/test/test_InputRecord.cxx b/Framework/Core/test/test_InputRecord.cxx
index 5dff09409325f..6633bd40a30f8 100644
--- a/Framework/Core/test/test_InputRecord.cxx
+++ b/Framework/Core/test/test_InputRecord.cxx
@@ -51,6 +51,7 @@ TEST_CASE("TestInputRecord")
     nullptr,
     [](size_t, DataRefIndices) { return DataRef{nullptr, nullptr, nullptr}; },
     [](size_t, DataRefIndices) -> DataRefIndices { return {size_t(-1), size_t(-1)}; },
+    nullptr,
     0};
   ServiceRegistry registry;
   InputRecord emptyRecord(schema, span, registry);
@@ -99,6 +100,7 @@ TEST_CASE("TestInputRecord")
     nullptr,
     [&inputs](size_t i, DataRefIndices idx) { return DataRef{nullptr, static_cast<char const*>(inputs[2 * i + idx.headerIdx]), static_cast<char const*>(inputs[2 * i + idx.payloadIdx])}; },
     [](size_t, DataRefIndices) -> DataRefIndices { return {size_t(-1), size_t(-1)}; },
+    nullptr,
     inputs.size() / 2};
   InputRecord record{schema, span2, registry};
 
diff --git a/Framework/Core/test/test_InputRecordWalker.cxx b/Framework/Core/test/test_InputRecordWalker.cxx
index 1fcfea1ba1587..bfbbd3651b5d3 100644
--- a/Framework/Core/test/test_InputRecordWalker.cxx
+++ b/Framework/Core/test/test_InputRecordWalker.cxx
@@ -40,7 +40,7 @@ struct DataSet {
                                                            auto payload = static_cast<char const*>(this->messages[i].second.at(idx.payloadIdx)->data());
                                                            return DataRef{nullptr, header, payload}; }, [this](size_t i, DataRefIndices current) -> DataRefIndices {
                                                            size_t next = current.headerIdx + 2;
-                                                           return next < this->messages[i].second.size() ? DataRefIndices{next, next + 1} : DataRefIndices{size_t(-1), size_t(-1)}; }, this->messages.size()}, record{schema, span, registry}, values{std::move(v)}
+                                                           return next < this->messages[i].second.size() ? DataRefIndices{next, next + 1} : DataRefIndices{size_t(-1), size_t(-1)}; }, nullptr, this->messages.size()}, record{schema, span, registry}, values{std::move(v)}
   {
     REQUIRE(messages.size() == schema.size());
   }
diff --git a/Framework/Core/test/test_InputSpan.cxx b/Framework/Core/test/test_InputSpan.cxx
index f8d043a2a48ba..9c3ae67e0e063 100644
--- a/Framework/Core/test/test_InputSpan.cxx
+++ b/Framework/Core/test/test_InputSpan.cxx
@@ -41,7 +41,7 @@ TEST_CASE("TestInputSpan")
     return next < inputs[i].size() ? DataRefIndices{next, next + 1} : DataRefIndices{size_t(-1), size_t(-1)};
   };
 
-  InputSpan span{nPartsGetter, nullptr, indicesGetter, nextIndicesGetter, inputs.size()};
+  InputSpan span{nPartsGetter, nullptr, indicesGetter, nextIndicesGetter, nullptr, inputs.size()};
   REQUIRE(span.size() == inputs.size());
   routeNo = 0;
   for (; routeNo < span.size(); ++routeNo) {
diff --git a/Framework/Utils/test/RawPageTestData.h b/Framework/Utils/test/RawPageTestData.h
index 29ac4eeba6b5b..e219f8ba84637 100644
--- a/Framework/Utils/test/RawPageTestData.h
+++ b/Framework/Utils/test/RawPageTestData.h
@@ -53,6 +53,7 @@ struct DataSet {
              size_t next = current.headerIdx + 2;
              return next < this->messages[i].size() ? DataRefIndices{next, next + 1} : DataRefIndices{size_t(-1), size_t(-1)};
            },
+           nullptr,
            this->messages.size()},
       record{schema, span, registry},
       values{std::move(v)}
diff --git a/Framework/Utils/test/test_RootTreeWriter.cxx b/Framework/Utils/test/test_RootTreeWriter.cxx
index e372fb4e1302e..ebacd0e71e5c3 100644
--- a/Framework/Utils/test/test_RootTreeWriter.cxx
+++ b/Framework/Utils/test/test_RootTreeWriter.cxx
@@ -231,6 +231,7 @@ TEST_CASE("test_RootTreeWriter")
       return DataRef{nullptr, static_cast<char const*>(store[2 * i + idx.headerIdx]->GetData()), static_cast<char const*>(store[2 * i + idx.payloadIdx]->GetData())};
     },
     [](size_t, DataRefIndices) -> DataRefIndices { return {size_t(-1), size_t(-1)}; },
+    nullptr,
     store.size() / 2};
   ServiceRegistry registry;
   InputRecord inputs{