From 44ec5a3fff0c65547faaf683120af2be10598b35 Mon Sep 17 00:00:00 2001
From: Kyle Knoepfel <knoepfel@fnal.gov>
Date: Tue, 25 Aug 2026 14:43:14 -0500
Subject: [PATCH 1/8] Promote resource-limiting TBB preview check to top-level
 directory

---
 CMakeLists.txt                  |  3 ++
 Modules/private/PhlexTBB.cmake  | 39 +++++++++++++++++++++
 phlex/core/CMakeLists.txt       |  5 +++
 test/tbb-preview/CMakeLists.txt | 61 ++++++---------------------------
 4 files changed, 57 insertions(+), 51 deletions(-)
 create mode 100644 Modules/private/PhlexTBB.cmake

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 332669d09..72cda4921 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -157,10 +157,13 @@ set(CMAKE_INSTALL_RPATH_USE_LINK_PATH ON)
 # add_compile_options(-fprofile-instr-generate -fcoverage-mapping)
 
 list(PREPEND CMAKE_MODULE_PATH ${PROJECT_SOURCE_DIR}/Modules)
+include(${PROJECT_SOURCE_DIR}/Modules/private/PhlexTBB.cmake)
 
 # Dependencies required by public API
 find_package(Boost REQUIRED COMPONENTS json EXPORT)
 find_package(TBB REQUIRED EXPORT)
+phlex_check_tbb_resource_limiting()
+
 find_package(fmt REQUIRED EXPORT)
 find_package(spdlog REQUIRED EXPORT)
 find_package(Microsoft.GSL REQUIRED EXPORT)
diff --git a/Modules/private/PhlexTBB.cmake b/Modules/private/PhlexTBB.cmake
new file mode 100644
index 000000000..8ec81b9ef
--- /dev/null
+++ b/Modules/private/PhlexTBB.cmake
@@ -0,0 +1,39 @@
+# Provides phlex_check_tbb_resource_limiting(), which verifies that the
+# TBB preview resource-limiting API required by Phlex is available.
+
+include_guard()
+
+include(CheckCXXSourceCompiles)
+
+function(phlex_check_tbb_resource_limiting)
+  set(_phlex_required_libraries_save ${CMAKE_REQUIRED_LIBRARIES})
+  set(_phlex_required_includes_save ${CMAKE_REQUIRED_INCLUDES})
+  set(_phlex_required_flags_save "${CMAKE_REQUIRED_FLAGS}")
+
+  get_target_property(_phlex_tbb_includes TBB::tbb INTERFACE_INCLUDE_DIRECTORIES)
+  if(_phlex_tbb_includes)
+    set(CMAKE_REQUIRED_INCLUDES ${_phlex_tbb_includes})
+  endif()
+  set(CMAKE_REQUIRED_LIBRARIES TBB::tbb)
+  set(CMAKE_REQUIRED_FLAGS "-std=c++${CMAKE_CXX_STANDARD}")
+
+  check_cxx_source_compiles(
+    "
+    #define TBB_PREVIEW_FLOW_GRAPH_RESOURCE_LIMITING 1
+    #include <oneapi/tbb/flow_graph.h>
+
+    using type = oneapi::tbb::flow::resource_limiter<int>;
+
+    int main() {}
+    "
+    HAVE_TBB_RESOURCE_LIMITING
+  )
+
+  set(CMAKE_REQUIRED_LIBRARIES ${_phlex_required_libraries_save})
+  set(CMAKE_REQUIRED_INCLUDES ${_phlex_required_includes_save})
+  set(CMAKE_REQUIRED_FLAGS "${_phlex_required_flags_save}")
+
+  if(NOT HAVE_TBB_RESOURCE_LIMITING)
+    message(FATAL_ERROR "Phlex requires TBB with flow::resource_limiter support")
+  endif()
+endfunction()
diff --git a/phlex/core/CMakeLists.txt b/phlex/core/CMakeLists.txt
index b35c04c0f..1b8ae6a98 100644
--- a/phlex/core/CMakeLists.txt
+++ b/phlex/core/CMakeLists.txt
@@ -79,6 +79,7 @@ install(
   DESTINATION include/phlex/core/detail
 )
 target_include_directories(phlex_core PRIVATE ${PROJECT_SOURCE_DIR})
+target_compile_definitions(phlex_core PUBLIC TBB_PREVIEW_FLOW_GRAPH_RESOURCE_LIMITING=1)
 phlex_apply_symbol_visibility(phlex_core)
 phlex_apply_optimizations(phlex_core)
 
@@ -96,6 +97,10 @@ phlex_make_internal_library(
   PUBLIC TBB::tbb phlex::metaprogramming phlex_model_internal phlex_utilities_internal
   PRIVATE Boost::json spdlog::spdlog
 )
+target_compile_definitions(
+  phlex_core_internal
+  INTERFACE TBB_PREVIEW_FLOW_GRAPH_RESOURCE_LIMITING=1
+)
 add_library(phlex::core_internal ALIAS phlex_core_internal)
 
 # Interface library
diff --git a/test/tbb-preview/CMakeLists.txt b/test/tbb-preview/CMakeLists.txt
index 4fd8abb10..852cd58a8 100644
--- a/test/tbb-preview/CMakeLists.txt
+++ b/test/tbb-preview/CMakeLists.txt
@@ -1,52 +1,11 @@
-include(CheckCXXSourceCompiles)
-
-# Check if flow::resource_limiter is available in TBB (preview feature)
-set(CMAKE_REQUIRED_LIBRARIES_SAVE ${CMAKE_REQUIRED_LIBRARIES})
-set(CMAKE_REQUIRED_INCLUDES_SAVE ${CMAKE_REQUIRED_INCLUDES})
-set(CMAKE_REQUIRED_FLAGS_SAVE "${CMAKE_REQUIRED_FLAGS}")
-set(CMAKE_TRY_COMPILE_VERBOSE_SAVE ${CMAKE_TRY_COMPILE_VERBOSE})
-
-get_target_property(_TBB_INCLUDES TBB::tbb INTERFACE_INCLUDE_DIRECTORIES)
-get_target_property(_TBB_LOCATION TBB::tbb LOCATION)
-if(_TBB_INCLUDES)
-  set(CMAKE_REQUIRED_INCLUDES ${_TBB_INCLUDES})
-endif()
-if(_TBB_LOCATION)
-  set(CMAKE_REQUIRED_LIBRARIES ${_TBB_LOCATION})
-endif()
-set(CMAKE_REQUIRED_FLAGS "-std=c++${CMAKE_CXX_STANDARD}")
-
-set(CMAKE_TRY_COMPILE_VERBOSE ON)
-
-check_cxx_source_compiles(
-  "
-  #define TBB_PREVIEW_FLOW_GRAPH_RESOURCE_LIMITING 1
-  #include <oneapi/tbb/flow_graph.h>
-
-  using type = oneapi::tbb::flow::resource_limiter<int>;
-
-  int main() {}
-"
-  HAVE_TBB_RESOURCE_LIMITING
+cet_test(
+  resource_limiting
+  USE_CATCH2_MAIN
+  SOURCE
+  resource_limiting_test.cpp
+  LIBRARIES
+  phlex::core
+  TBB::tbb
+  spdlog::spdlog
 )
-
-set(CMAKE_REQUIRED_LIBRARIES ${CMAKE_REQUIRED_LIBRARIES_SAVE})
-set(CMAKE_REQUIRED_INCLUDES ${CMAKE_REQUIRED_INCLUDES_SAVE})
-set(CMAKE_REQUIRED_FLAGS "${CMAKE_REQUIRED_FLAGS_SAVE}")
-set(CMAKE_TRY_COMPILE_VERBOSE ${CMAKE_TRY_COMPILE_VERBOSE_SAVE})
-
-if(HAVE_TBB_RESOURCE_LIMITING)
-  cet_test(
-    resource_limiting
-    USE_CATCH2_MAIN
-    SOURCE
-    resource_limiting_test.cpp
-    LIBRARIES
-    phlex::core
-    TBB::tbb
-    spdlog::spdlog
-  )
-  target_compile_definitions(resource_limiting PRIVATE TBB_PREVIEW_FLOW_GRAPH_RESOURCE_LIMITING=1)
-else()
-  message(STATUS "Skipping resource_limiting test: flow::resource_limiter not available in TBB")
-endif()
+target_compile_definitions(resource_limiting PRIVATE TBB_PREVIEW_FLOW_GRAPH_RESOURCE_LIMITING=1)

From d3ca3a362671783bfb1f93e6b3f3bf27978334cd Mon Sep 17 00:00:00 2001
From: Kyle Knoepfel <knoepfel@fnal.gov>
Date: Wed, 26 Aug 2026 16:09:44 -0500
Subject: [PATCH 2/8] feat(resources): add mutable resource registration

---
 phlex/core/CMakeLists.txt       |   1 +
 phlex/core/framework_graph.hpp  |  18 +++-
 phlex/core/glue.hpp             |  52 +++++++++--
 phlex/core/graph_proxy.hpp      |  36 +++++---
 phlex/core/registration_api.hpp |  31 +++++--
 phlex/core/resource_api.hpp     | 153 ++++++++++++++++++++++++++++++++
 phlex/source.hpp                |   5 +-
 test/CMakeLists.txt             |   3 +
 test/resources_test.cpp         |  38 ++++++++
 9 files changed, 305 insertions(+), 32 deletions(-)
 create mode 100644 phlex/core/resource_api.hpp
 create mode 100644 test/resources_test.cpp

diff --git a/phlex/core/CMakeLists.txt b/phlex/core/CMakeLists.txt
index 1b8ae6a98..68363ba5d 100644
--- a/phlex/core/CMakeLists.txt
+++ b/phlex/core/CMakeLists.txt
@@ -65,6 +65,7 @@ install(
     provider_node.hpp
     registrar.hpp
     registration_api.hpp
+    resource_api.hpp
     source.hpp
     upstream_predicates.hpp
   DESTINATION include/phlex/core
diff --git a/phlex/core/framework_graph.hpp b/phlex/core/framework_graph.hpp
index d01670d19..126469b27 100644
--- a/phlex/core/framework_graph.hpp
+++ b/phlex/core/framework_graph.hpp
@@ -8,6 +8,7 @@
 #include "phlex/core/index_router.hpp"
 #include "phlex/core/message.hpp"
 #include "phlex/core/node_catalog.hpp"
+#include "phlex/core/resource_api.hpp"
 #include "phlex/driver.hpp"
 #include "phlex/model/data_cell_tracker.hpp"
 #include "phlex/model/data_layer_hierarchy.hpp"
@@ -31,6 +32,7 @@
 #include <string>
 #include <string_view>
 #include <tuple>
+#include <typeindex>
 #include <utility>
 #include <vector>
 
@@ -71,7 +73,7 @@ namespace phlex::detail {
 
     module_graph_proxy<void_tag> module_proxy(configuration const& config)
     {
-      return {config, graph_, nodes_, registration_errors_};
+      return {config, graph_, nodes_, registration_errors_, resources_};
     }
 
     source_bundle source_proxy(configuration const& config)
@@ -79,6 +81,7 @@ namespace phlex::detail {
       return {.config = config,
               .graph = graph_,
               .nodes = nodes_,
+              .resources = resources_,
               .registration_errors = registration_errors_};
     }
 
@@ -146,6 +149,13 @@ namespace phlex::detail {
       return make_glue().template add_source<Source>(name, std::forward<Args>(args)...);
     }
 
+    template <typename Resource, typename... Args>
+      requires(!std::is_const_v<Resource> && std::constructible_from<Resource, Args...>)
+    void add_resource(Args&&... args)
+    {
+      resources_.template add<Resource>(std::forward<Args>(args)...);
+    }
+
     template <typename T, typename... Args>
     glue<T> make(Args&&... args)
     {
@@ -185,7 +195,7 @@ namespace phlex::detail {
       if constexpr (is_bound_object<T> && Construct) {
         bound_object = std::make_shared<T>(std::forward<Args>(args)...);
       }
-      return {graph_, nodes_, std::move(bound_object), registration_errors_};
+      return {graph_, nodes_, std::move(bound_object), registration_errors_, resources_};
     }
 
     void run();
@@ -200,10 +210,12 @@ namespace phlex::detail {
     resource_usage graph_resource_usage_;
     max_allowed_parallelism parallelism_limit_;
     fixed_hierarchy fixed_hierarchy_;
+    // The graph_ object uses the filters_, nodes_, resources_, and hierarchy_ objects implicitly.
+    // These must be declared before graph_ so that they outlive it during destruction.
     data_layer_hierarchy hierarchy_{};
+    resource_catalog resources_;
     node_catalog nodes_;
     std::map<std::string, filter> filters_;
-    // The graph_ object uses the filters_, nodes_, and hierarchy_ objects implicitly.
     tbb::flow::graph graph_{};
     std::optional<framework_driver> driver_;
     std::vector<std::string> registration_errors_;
diff --git a/phlex/core/glue.hpp b/phlex/core/glue.hpp
index 88ea9d653..a99418662 100644
--- a/phlex/core/glue.hpp
+++ b/phlex/core/glue.hpp
@@ -7,6 +7,7 @@
 #include "phlex/core/concepts.hpp"
 #include "phlex/core/registrar.hpp"
 #include "phlex/core/registration_api.hpp"
+#include "phlex/core/resource_api.hpp"
 #include "phlex/core/source.hpp"
 #include "phlex/metaprogramming/delegate.hpp"
 
@@ -48,8 +49,14 @@ namespace phlex::detail {
          node_catalog& nodes,
          std::shared_ptr<T> bound_obj,
          std::vector<std::string>& errors,
+         resource_catalog& resources,
          configuration const* config = nullptr) :
-      graph_{g}, nodes_{nodes}, bound_obj_{std::move(bound_obj)}, errors_{errors}, config_{config}
+      graph_{g},
+      nodes_{nodes},
+      bound_obj_{std::move(bound_obj)},
+      errors_{errors},
+      resources_{resources},
+      config_{config}
     {
     }
 
@@ -70,6 +77,7 @@ namespace phlex::detail {
                       graph_,
                       nodes_,
                       errors_,
+                      resources_,
                       std::move(partition),
                       std::forward<InitArgs>(init_args)...};
     }
@@ -82,8 +90,14 @@ namespace phlex::detail {
                  concurrency c)
     {
       internal::verify_name(name, config_);
-      return make_registration<observer_node>(
-        config_, name, algorithm_bits{bound_obj_, std::move(f)}, c, graph_, nodes_, errors_);
+      return make_registration<observer_node>(config_,
+                                              name,
+                                              algorithm_bits{bound_obj_, std::move(f)},
+                                              c,
+                                              graph_,
+                                              nodes_,
+                                              errors_,
+                                              resources_);
     }
 
     // 'f' is a by-value sink: it is moved into algorithm_bits.  The clang-tidy
@@ -94,8 +108,14 @@ namespace phlex::detail {
                  concurrency c)
     {
       internal::verify_name(name, config_);
-      return provider_api{
-        config_, name, algorithm_bits{bound_obj_, std::move(f)}, c, graph_, nodes_, errors_};
+      return provider_api{config_,
+                          name,
+                          algorithm_bits{bound_obj_, std::move(f)},
+                          c,
+                          graph_,
+                          nodes_,
+                          errors_,
+                          resources_};
     }
 
     // 'f' is a by-value sink: it is moved into algorithm_bits.  The clang-tidy
@@ -106,8 +126,14 @@ namespace phlex::detail {
                    concurrency c)
     {
       internal::verify_name(name, config_);
-      return make_registration<transform_node>(
-        config_, name, algorithm_bits{bound_obj_, std::move(f)}, c, graph_, nodes_, errors_);
+      return make_registration<transform_node>(config_,
+                                               name,
+                                               algorithm_bits{bound_obj_, std::move(f)},
+                                               c,
+                                               graph_,
+                                               nodes_,
+                                               errors_,
+                                               resources_);
     }
 
     // 'f' is a by-value sink: it is moved into algorithm_bits.  The clang-tidy
@@ -118,8 +144,14 @@ namespace phlex::detail {
                    concurrency c)
     {
       internal::verify_name(name, config_);
-      return make_registration<predicate_node>(
-        config_, name, algorithm_bits{bound_obj_, std::move(f)}, c, graph_, nodes_, errors_);
+      return make_registration<predicate_node>(config_,
+                                               name,
+                                               algorithm_bits{bound_obj_, std::move(f)},
+                                               c,
+                                               graph_,
+                                               nodes_,
+                                               errors_,
+                                               resources_);
     }
 
     auto unfold(std::string_view name,
@@ -139,6 +171,7 @@ namespace phlex::detail {
         graph_,
         nodes_,
         errors_,
+        resources_,
         std::move(destination_data_layer)};
     }
 
@@ -168,6 +201,7 @@ namespace phlex::detail {
     node_catalog& nodes_;     // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members)
     std::shared_ptr<T> bound_obj_;
     std::vector<std::string>& errors_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members)
+    resource_catalog& resources_;      // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members)
     configuration const* config_;
   };
 }
diff --git a/phlex/core/graph_proxy.hpp b/phlex/core/graph_proxy.hpp
index ed429def2..c2d557771 100644
--- a/phlex/core/graph_proxy.hpp
+++ b/phlex/core/graph_proxy.hpp
@@ -39,7 +39,8 @@ namespace phlex::detail {
     graph_proxy(configuration const& config,
                 tbb::flow::graph& g,
                 node_catalog& nodes,
-                std::vector<std::string>& errors)
+                std::vector<std::string>& errors,
+                resource_catalog& resources)
       requires(not is_bound_object<T>);
 
     /// @brief Binds a user algorithm object of type @p U to this proxy.
@@ -110,7 +111,8 @@ namespace phlex::detail {
                 tbb::flow::graph& g,
                 node_catalog& nodes,
                 std::shared_ptr<T> bound_obj,
-                std::vector<std::string>& errors)
+                std::vector<std::string>& errors,
+                resource_catalog& resources)
       requires(is_bound_object<T>);
 
   private:
@@ -123,15 +125,17 @@ namespace phlex::detail {
     node_catalog& nodes_;     // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members)
     std::shared_ptr<T> bound_obj_;
     std::vector<std::string>& errors_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members)
+    resource_catalog& resources_;      // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members)
   };
 
   template <typename T>
   graph_proxy<T>::graph_proxy(configuration const& config,
                               tbb::flow::graph& g,
                               node_catalog& nodes,
-                              std::vector<std::string>& errors)
+                              std::vector<std::string>& errors,
+                              resource_catalog& resources)
     requires(not is_bound_object<T>)
-    : config_{&config}, graph_{g}, nodes_{nodes}, errors_{errors}
+    : config_{&config}, graph_{g}, nodes_{nodes}, errors_{errors}, resources_{resources}
   {
   }
 
@@ -191,7 +195,7 @@ namespace phlex::detail {
                               std::string destination_data_layer,
                               concurrency c) const
   {
-    return glue<Splitter>{graph_, nodes_, nullptr, errors_, config_}.unfold(
+    return glue<Splitter>{graph_, nodes_, nullptr, errors_, resources_, config_}.unfold(
       name, std::move(pred), std::move(unf), c, std::move(destination_data_layer));
   }
 
@@ -216,8 +220,12 @@ namespace phlex::detail {
   Proxy<U> graph_proxy<T>::bind_to(Args&&... args) const
     requires(not is_bound_object<T>)
   {
-    return Proxy<U>{
-      config_, graph_, nodes_, std::make_shared<U>(std::forward<Args>(args)...), errors_};
+    return Proxy<U>{config_,
+                    graph_,
+                    nodes_,
+                    std::make_shared<U>(std::forward<Args>(args)...),
+                    errors_,
+                    resources_};
   }
 
   template <typename T>
@@ -225,16 +233,24 @@ namespace phlex::detail {
                               tbb::flow::graph& g,
                               node_catalog& nodes,
                               std::shared_ptr<T> bound_obj,
-                              std::vector<std::string>& errors)
+                              std::vector<std::string>& errors,
+                              resource_catalog& resources)
     requires(is_bound_object<T>)
-    : config_{config}, graph_{g}, nodes_{nodes}, bound_obj_{std::move(bound_obj)}, errors_{errors}
+    :
+    config_{config},
+    graph_{g},
+    nodes_{nodes},
+    bound_obj_{std::move(bound_obj)},
+    errors_{errors},
+    resources_{resources}
   {
   }
 
   template <typename T>
   glue<T> graph_proxy<T>::create_glue(bool use_bound_object) const
   {
-    return glue{graph_, nodes_, (use_bound_object ? bound_obj_ : nullptr), errors_, config_};
+    return glue{
+      graph_, nodes_, (use_bound_object ? bound_obj_ : nullptr), errors_, resources_, config_};
   }
 }
 
diff --git a/phlex/core/registration_api.hpp b/phlex/core/registration_api.hpp
index 790f60e2d..f82e57ee3 100644
--- a/phlex/core/registration_api.hpp
+++ b/phlex/core/registration_api.hpp
@@ -8,6 +8,7 @@
 #include "phlex/core/declared_fold.hpp"
 #include "phlex/core/detail/make_algorithm_name.hpp"
 #include "phlex/core/node_catalog.hpp"
+#include "phlex/core/resource_api.hpp"
 #include "phlex/core/upstream_predicates.hpp"
 #include "phlex/metaprogramming/delegate.hpp"
 #include "phlex/metaprogramming/type_deduction.hpp"
@@ -43,13 +44,15 @@ namespace phlex::detail {
                      concurrency c,
                      tbb::flow::graph& g,
                      node_catalog& nodes,
-                     std::vector<std::string>& errors) :
+                     std::vector<std::string>& errors,
+                     resource_catalog& resources) :
       config_{config},
       name_{phlex::experimental::internal::make_algorithm_name(config, name)},
       alg_{std::move(alg)},
       concurrency_{c},
       graph_{g},
-      registrar_{nodes.registrar_for<node_ptr>(errors)}
+      registrar_{nodes.registrar_for<node_ptr>(errors)},
+      resources_{resources}
     {
     }
 
@@ -98,6 +101,7 @@ namespace phlex::detail {
     // Non-owning reference to the TBB graph; this class is a short-lived registration builder.
     tbb::flow::graph& graph_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members)
     registrar<node_ptr> registrar_;
+    resource_catalog& resources_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members)
   };
 
   template <template <typename...> typename HOF, typename AlgorithmBits>
@@ -107,9 +111,11 @@ namespace phlex::detail {
                          concurrency c,
                          tbb::flow::graph& g,
                          node_catalog& nodes,
-                         std::vector<std::string>& errors)
+                         std::vector<std::string>& errors,
+                         resource_catalog& resources)
   {
-    return registration_api<HOF, AlgorithmBits>{config, name, std::move(alg), c, g, nodes, errors};
+    return registration_api<HOF, AlgorithmBits>{
+      config, name, std::move(alg), c, g, nodes, errors, resources};
   }
 
   // ====================================================================================
@@ -124,13 +130,15 @@ namespace phlex::detail {
                  concurrency c,
                  tbb::flow::graph& g,
                  node_catalog& nodes,
-                 std::vector<std::string>& errors) :
+                 std::vector<std::string>& errors,
+                 resource_catalog& resources) :
       config_{config},
       name_{phlex::experimental::internal::make_algorithm_name(config, name)},
       alg_{std::move(alg)},
       concurrency_{c},
       graph_{g},
-      registrar_{nodes.registrar_for<provider_node_ptr>(errors)}
+      registrar_{nodes.registrar_for<provider_node_ptr>(errors)},
+      resources_{resources}
     {
     }
 
@@ -171,6 +179,7 @@ namespace phlex::detail {
     // Non-owning reference to the TBB graph; this class is a short-lived registration builder.
     tbb::flow::graph& graph_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members)
     registrar<provider_node_ptr> registrar_;
+    resource_catalog& resources_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members)
   };
 
   // ====================================================================================
@@ -192,6 +201,7 @@ namespace phlex::detail {
              tbb::flow::graph& g,
              node_catalog& nodes,
              std::vector<std::string>& errors,
+             resource_catalog& resources,
              std::string partition,
              InitArgs&&... init_args) :
       config_{config},
@@ -201,7 +211,8 @@ namespace phlex::detail {
       graph_{g},
       partition_{std::move(partition)},
       init_{std::forward<InitArgs>(init_args)...},
-      registrar_{nodes.registrar_for<declared_fold_ptr>(errors)}
+      registrar_{nodes.registrar_for<declared_fold_ptr>(errors)},
+      resources_{resources}
     {
     }
 
@@ -243,6 +254,7 @@ namespace phlex::detail {
     std::string partition_;
     init_tuple init_;
     registrar<declared_fold_ptr> registrar_;
+    resource_catalog& resources_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members)
   };
 
   // ====================================================================================
@@ -270,6 +282,7 @@ namespace phlex::detail {
                tbb::flow::graph& g,
                node_catalog& nodes,
                std::vector<std::string>& errors,
+               resource_catalog& resources,
                std::string destination_data_layer) :
       config_{config},
       registrar_{nodes.registrar_for<declared_unfold_ptr>(errors)},
@@ -278,7 +291,8 @@ namespace phlex::detail {
       graph_{g},
       predicate_{std::move(predicate)},
       unfold_{std::move(unfold)},
-      destination_layer_{std::move(destination_data_layer)}
+      destination_layer_{std::move(destination_data_layer)},
+      resources_{resources}
     {
     }
 
@@ -320,6 +334,7 @@ namespace phlex::detail {
     Predicate predicate_;
     Unfold unfold_;
     std::string destination_layer_;
+    resource_catalog& resources_; // NOLINT(cppcoreguidelines-avoid-const-or-ref-data-members)
   };
 
   // ====================================================================================
diff --git a/phlex/core/resource_api.hpp b/phlex/core/resource_api.hpp
new file mode 100644
index 000000000..19847d6f9
--- /dev/null
+++ b/phlex/core/resource_api.hpp
@@ -0,0 +1,153 @@
+#ifndef PHLEX_CORE_RESOURCE_API_HPP
+#define PHLEX_CORE_RESOURCE_API_HPP
+
+#include "boost/core/demangle.hpp"
+#include "boost/mp11/algorithm.hpp"
+#include "boost/mp11/list.hpp"
+#include "boost/mp11/set.hpp"
+
+#include "fmt/format.h"
+#include "oneapi/tbb/flow_graph.h"
+
+#include <cassert>
+#include <concepts>
+#include <cstddef>
+#include <map>
+#include <memory>
+#include <stdexcept>
+#include <tuple>
+#include <type_traits>
+#include <typeindex>
+#include <typeinfo>
+#include <utility>
+
+namespace phlex::detail {
+  class resource_base {
+  public:
+    virtual ~resource_base() = default;
+  };
+
+  template <typename T>
+  class wrapped_resource : public resource_base {
+  public:
+    template <typename... Args>
+    explicit wrapped_resource(Args&&... args) :
+      resource_{std::forward<Args>(args)...}, limiter_{&resource_}
+    {
+    }
+
+    using token_type = T*;
+
+    tbb::flow::resource_limiter<token_type>& limiter() noexcept { return limiter_; }
+
+  private:
+    // The framework owns the resource, while TBB uses the limiter to limit access to it.
+    // Declare resource_ first so it outlives the limiter, which holds a pointer to it.
+    T resource_;
+    tbb::flow::resource_limiter<token_type> limiter_;
+  };
+
+  class resource_catalog {
+  public:
+    template <typename T, typename... Args>
+      requires(!std::is_const_v<T> && std::constructible_from<T, Args...>)
+    void add(Args&&... args)
+    {
+      auto const type = std::type_index(typeid(T));
+      if (resources_.contains(type)) {
+        throw std::runtime_error(fmt::format("Resource of type '{}' has already been registered",
+                                             boost::core::demangle(typeid(T).name())));
+      }
+      resources_.emplace(type, std::make_unique<wrapped_resource<T>>(std::forward<Args>(args)...));
+    }
+
+    template <typename T>
+    auto& limiter_for() const
+    {
+      static_assert(!std::is_const_v<T>, "resource<T> must name a non-const resource type.");
+
+      auto const found = resources_.find(std::type_index(typeid(T)));
+      if (found == resources_.end()) {
+        throw std::runtime_error(fmt::format("Resource of type '{}' has not been registered",
+                                             boost::core::demangle(typeid(T).name())));
+      }
+      auto* wrapped = dynamic_cast<wrapped_resource<T>*>(found->second.get());
+      assert(wrapped != nullptr && "resource catalog entry has unexpected type");
+      return wrapped->limiter();
+    }
+
+  private:
+    std::map<std::type_index, std::unique_ptr<resource_base>> resources_;
+  };
+
+  template <typename T>
+  struct resource {
+    static_assert(!std::is_const_v<T>, "resource<T> must name a non-const resource type.");
+  };
+
+  namespace internal {
+    template <typename T>
+    struct is_resource_impl : std::false_type {};
+
+    template <typename T>
+    struct is_resource_impl<resource<T>> : std::true_type {};
+
+    // mp11 predicate form of is_resource: usable directly with mp_find_if, mp_all_of, etc.
+    // The remove_cvref_t is needed because the argument types deduced from a forwarding or
+    // by-value parameter pack may still carry cv-qualification.
+    template <typename T>
+    using is_resource_p = is_resource_impl<std::remove_cvref_t<T>>;
+  }
+
+  template <typename T>
+  concept is_resource = internal::is_resource_impl<std::remove_cvref_t<T>>::value;
+
+  namespace internal {
+    template <typename T>
+    struct resource_type;
+
+    template <typename T>
+    struct resource_type<resource<T>> {
+      using type = T;
+    };
+
+    template <typename T>
+    using resource_type_t = typename resource_type<std::remove_cvref_t<T>>::type;
+  }
+
+  // ====================================================================================
+  // resource_split
+  //
+  // Given the types of the arguments supplied to a registration's input_family(...), determine
+  // the index at which the trailing resource<T> arguments begin.  Everything before that index
+  // is treated as a product selector; everything from that index onward must be a resource.
+  //
+  // mp_find_if yields mp_size<L> when no element satisfies the predicate, so a list containing
+  // no resources naturally splits at the end, giving an empty resource list.
+  template <typename... Args>
+  struct resource_split {
+    using arguments = boost::mp11::mp_list<std::remove_cvref_t<Args>...>;
+
+    static constexpr std::size_t index =
+      boost::mp11::mp_find_if<arguments, internal::is_resource_p>::value;
+
+    using selector_types = boost::mp11::mp_take_c<arguments, index>;
+    using resource_types = boost::mp11::mp_drop_c<arguments, index>;
+
+    // std::tuple<resource<T>...> for the resources that follow the split.
+    using resources_type = boost::mp11::mp_rename<resource_types, std::tuple>;
+
+    static_assert(boost::mp11::mp_is_set<resource_types>::value,
+                  "Each resource type may only be specified once.");
+
+    static constexpr std::size_t number_selectors = index;
+    static constexpr std::size_t number_resources = boost::mp11::mp_size<resource_types>::value;
+
+    // Once the first resource has been seen, every remaining argument must also be a resource;
+    // otherwise the arguments cannot be partitioned into a selector group and a resource group.
+    static constexpr bool resources_are_last =
+      boost::mp11::mp_all_of<resource_types, internal::is_resource_p>::value;
+  };
+}
+
+#endif // PHLEX_CORE_RESOURCE_API_HPP
diff --git a/phlex/source.hpp b/phlex/source.hpp
index 75ba84507..cbf20fcf9 100644
--- a/phlex/source.hpp
+++ b/phlex/source.hpp
@@ -17,6 +17,7 @@ namespace phlex::detail {
     configuration const& config;
     tbb::flow::graph& graph;
     node_catalog& nodes;
+    resource_catalog& resources;
     std::vector<std::string>& registration_errors;
     // NOLINTEND(cppcoreguidelines-avoid-const-or-ref-data-members)
   };
@@ -31,7 +32,7 @@ namespace phlex::detail {
 
   public:
     providers_graph_proxy(source_bundle bundle) :
-      base{bundle.config, bundle.graph, bundle.nodes, bundle.registration_errors}
+      base{bundle.config, bundle.graph, bundle.nodes, bundle.registration_errors, bundle.resources}
     {
     }
 
@@ -58,7 +59,7 @@ namespace phlex::detail {
 
   public:
     source_graph_proxy(source_bundle bundle) :
-      base{bundle.config, bundle.graph, bundle.nodes, bundle.registration_errors}
+      base{bundle.config, bundle.graph, bundle.nodes, bundle.registration_errors, bundle.resources}
     {
     }
 
diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt
index 1ac190098..1a5982e54 100644
--- a/test/CMakeLists.txt
+++ b/test/CMakeLists.txt
@@ -265,6 +265,9 @@ cet_test(
   phlex::core_internal
   layer_generator_internal
 )
+cet_test(resources USE_CATCH2_MAIN SOURCE resources_test.cpp LIBRARIES
+         phlex::core_internal
+)
 cet_test(
   type_distinction
   USE_CATCH2_MAIN
diff --git a/test/resources_test.cpp b/test/resources_test.cpp
new file mode 100644
index 000000000..c27828d7f
--- /dev/null
+++ b/test/resources_test.cpp
@@ -0,0 +1,38 @@
+#include "phlex/core/framework_graph.hpp"
+#include "phlex/core/resource_api.hpp"
+
+#include "catch2/catch_test_macros.hpp"
+#include "catch2/matchers/catch_matchers_string.hpp"
+
+using namespace phlex;
+using namespace phlex::detail;
+
+namespace {
+  struct Catch2 {}; // NOLINT(readability-identifier-naming)
+}
+
+TEST_CASE("resource catalog", "[resource]")
+{
+  resource_catalog catalog;
+
+  SECTION("registered resources can be looked up")
+  {
+    catalog.add<Catch2>();
+    CHECK_NOTHROW(catalog.limiter_for<Catch2>());
+  }
+
+  SECTION("duplicate registrations throw")
+  {
+    catalog.add<Catch2>();
+    CHECK_THROWS_WITH(catalog.add<Catch2>(),
+                      Catch::Matchers::ContainsSubstring("Resource of type '") &&
+                        Catch::Matchers::ContainsSubstring("' has already been registered"));
+  }
+
+  SECTION("missing resources throw")
+  {
+    CHECK_THROWS_WITH(catalog.limiter_for<int>(),
+                      Catch::Matchers::ContainsSubstring("Resource of type '") &&
+                        Catch::Matchers::ContainsSubstring("' has not been registered"));
+  }
+}

From 764d2048a2ef22dc6559fd61fd15d38b477e6b5e Mon Sep 17 00:00:00 2001
From: Kyle Knoepfel <knoepfel@fnal.gov>
Date: Wed, 26 Aug 2026 16:09:45 -0500
Subject: [PATCH 3/8] feat(registration): split product selectors from
 resources

---
 phlex/core/registration_api.hpp | 205 ++++++++++++++++++++++++--------
 1 file changed, 153 insertions(+), 52 deletions(-)

diff --git a/phlex/core/registration_api.hpp b/phlex/core/registration_api.hpp
index f82e57ee3..f060c434e 100644
--- a/phlex/core/registration_api.hpp
+++ b/phlex/core/registration_api.hpp
@@ -14,10 +14,15 @@
 #include "phlex/metaprogramming/type_deduction.hpp"
 #include "phlex/model/algorithm_name.hpp"
 
+#include <array>
 #include <concepts>
+#include <cstddef>
 #include <functional>
 #include <memory>
 #include <string_view>
+#include <tuple>
+#include <type_traits>
+#include <utility>
 
 namespace phlex {
   class configuration;
@@ -25,6 +30,24 @@ namespace phlex {
 
 namespace phlex::detail {
 
+  template <typename T>
+  concept selector_or_resource =
+    std::same_as<std::remove_cvref_t<T>, product_selector> || is_resource<T>;
+
+  template <typename Split, typename... Args>
+  auto partition_selectors_and_resources(Args&&... args)
+  {
+    auto all = std::forward_as_tuple(std::forward<Args>(args)...);
+    auto selectors = [&all]<std::size_t... Is>(std::index_sequence<Is...>) {
+      return std::array<product_selector, Split::number_selectors>{std::move(std::get<Is>(all))...};
+    }(std::make_index_sequence<Split::number_selectors>{});
+    auto resources = [&all]<std::size_t... Is>(std::index_sequence<Is...>) {
+      return
+        typename Split::resources_type{std::move(std::get<Is + Split::number_selectors>(all))...};
+    }(std::make_index_sequence<Split::number_resources>{});
+    return std::pair{std::move(selectors), std::move(resources)};
+  }
+
   // ====================================================================================
   // Registration API
 
@@ -37,6 +60,8 @@ namespace phlex::detail {
     static constexpr auto num_inputs = AlgorithmBits::number_inputs;
     static constexpr auto num_outputs = hof_type::number_output_products;
 
+    static_assert(num_inputs > 0, "input_family requires at least one product selector.");
+
   public:
     registration_api(configuration const* config,
                      std::string_view name,
@@ -56,41 +81,64 @@ namespace phlex::detail {
     {
     }
 
-    auto input_family(std::array<product_selector, num_inputs> input_args)
+    template <std::size_t NProducts, typename... Resources>
+    auto input_family(std::array<product_selector, NProducts> input_args, std::tuple<Resources...>)
     {
-      populate_types<input_parameter_types>(input_args);
+      static_assert((is_resource<Resources> && ...),
+                    "The second argument to input_family(...) must be a tuple of resources.");
+      static_assert(num_inputs == NProducts + sizeof...(Resources),
+                    "The number of function parameters is not the same as the number of specified "
+                    "input arguments.");
+
+      populate_types<boost::mp11::mp_take_c<input_parameter_types, NProducts>>(input_args);
 
       if constexpr (num_outputs == 0ull) {
         registrar_.set_creator([this, inputs = std::move(input_args)](
                                  auto predicates, auto const& /* output_product_suffixes */) {
-          return std::make_unique<hof_type>(std::move(name_),
-                                            concurrency_.value,
-                                            std::move(predicates),
-                                            graph_,
-                                            std::move(alg_),
-                                            std::vector(inputs.begin(), inputs.end()));
+          using node_type = HOF<AlgorithmBits, internal::resource_type_t<Resources>...>;
+          return std::make_unique<node_type>(std::move(name_),
+                                             concurrency_.value,
+                                             std::move(predicates),
+                                             graph_,
+                                             std::move(alg_),
+                                             std::vector(std::from_range, std::move(inputs)),
+                                             resources_);
         });
       } else {
         registrar_.set_creator(
           [this, inputs = std::move(input_args)](auto predicates, auto output_product_suffixes) {
-            return std::make_unique<hof_type>(std::move(name_),
-                                              concurrency_.value,
-                                              std::move(predicates),
-                                              graph_,
-                                              std::move(alg_),
-                                              std::vector(inputs.begin(), inputs.end()),
-                                              std::move(output_product_suffixes));
+            using node_type = HOF<AlgorithmBits, internal::resource_type_t<Resources>...>;
+            return std::make_unique<node_type>(std::move(name_),
+                                               concurrency_.value,
+                                               std::move(predicates),
+                                               graph_,
+                                               std::move(alg_),
+                                               std::vector(std::from_range, std::move(inputs)),
+                                               std::move(output_product_suffixes),
+                                               resources_);
           });
       }
       return upstream_predicates<node_ptr, num_outputs>{std::move(registrar_), config_};
     }
 
-    auto input_family(std::same_as<product_selector> auto... input_args)
+    auto input_family(std::array<product_selector, num_inputs> input_args)
+    {
+      return input_family(std::move(input_args), std::tuple<>{});
+    }
+
+    auto input_family(product_selector first_selector, selector_or_resource auto... additional_args)
     {
-      static_assert(num_inputs == sizeof...(input_args),
+      using split = resource_split<decltype(first_selector), decltype(additional_args)...>;
+      static_assert(split::resources_are_last,
+                    "All resource<T> arguments to input_family(...) must follow the "
+                    "product_selector arguments.");
+      static_assert(num_inputs == split::number_selectors + split::number_resources,
                     "The number of function parameters is not the same as the number of specified "
                     "input arguments.");
-      return input_family({std::move(input_args)...});
+
+      auto [selectors, resources] = partition_selectors_and_resources<split>(
+        std::move(first_selector), std::move(additional_args)...);
+      return input_family(std::move(selectors), std::move(resources));
     }
 
   private:
@@ -193,6 +241,8 @@ namespace phlex::detail {
     static constexpr auto num_inputs = AlgorithmBits::number_inputs;
     static constexpr auto num_outputs = 1; // For now
 
+    static_assert(num_inputs > 1, "input_family requires at least one product selector.");
+
   public:
     fold_api(configuration const* config,
              std::string_view name,
@@ -216,32 +266,54 @@ namespace phlex::detail {
     {
     }
 
-    auto input_family(std::array<product_selector, num_inputs - 1> input_args)
+    template <std::size_t NProducts, typename... Resources>
+    auto input_family(std::array<product_selector, NProducts> input_args, std::tuple<Resources...>)
     {
-      populate_types<input_parameter_types>(input_args);
+      static_assert((is_resource<Resources> && ...),
+                    "The second argument to input_family(...) must be a tuple of resources.");
+      static_assert(num_inputs - 1 == NProducts + sizeof...(Resources),
+                    "The number of fold product parameters is not the same as the number of "
+                    "specified input arguments.");
+
+      using product_parameter_types = boost::mp11::mp_take_c<input_parameter_types, NProducts>;
+      populate_types<product_parameter_types>(input_args);
 
       registrar_.set_creator(
         [this, inputs = std::move(input_args)](auto predicates, auto output_product_suffixes) {
-          return std::make_unique<fold_node<AlgorithmBits, init_tuple>>(
-            std::move(name_),
-            concurrency_.value,
-            std::move(predicates),
-            graph_,
-            std::move(alg_),
-            std::move(init_),
-            std::vector(inputs.begin(), inputs.end()),
-            std::move(output_product_suffixes),
-            std::move(partition_));
+          using node_type =
+            fold_node<AlgorithmBits, init_tuple, internal::resource_type_t<Resources>...>;
+          return std::make_unique<node_type>(std::move(name_),
+                                             concurrency_.value,
+                                             std::move(predicates),
+                                             graph_,
+                                             std::move(alg_),
+                                             std::move(init_),
+                                             std::vector(std::from_range, std::move(inputs)),
+                                             std::move(output_product_suffixes),
+                                             std::move(partition_),
+                                             resources_);
         });
       return upstream_predicates<declared_fold_ptr, num_outputs>{std::move(registrar_), config_};
     }
 
-    auto input_family(std::same_as<product_selector> auto... input_args)
+    auto input_family(std::array<product_selector, num_inputs - 1> input_args)
+    {
+      return input_family(std::move(input_args), std::tuple<>{});
+    }
+
+    auto input_family(product_selector first_selector, selector_or_resource auto... additional_args)
     {
-      static_assert(num_inputs - 1 == sizeof...(input_args),
+      using split = resource_split<decltype(first_selector), decltype(additional_args)...>;
+      static_assert(split::resources_are_last,
+                    "All resource<T> arguments to input_family(...) must follow the "
+                    "product_selector arguments.");
+      static_assert(num_inputs - 1 == split::number_selectors + split::number_resources,
                     "The number of function parameters is not the same as the number of specified "
                     "input arguments.");
-      return input_family({std::move(input_args)...});
+
+      auto [selectors, resources] = partition_selectors_and_resources<split>(
+        std::move(first_selector), std::move(additional_args)...);
+      return input_family(std::move(selectors), std::move(resources));
     }
 
   private:
@@ -262,11 +334,13 @@ namespace phlex::detail {
 
   template <typename Object, typename Predicate, typename Unfold>
   class unfold_api {
-    using input_parameter_types = constructor_parameter_types<Object>;
+    using all_input_parameter_types = constructor_parameter_types<Object>;
 
-    static constexpr auto num_inputs = std::tuple_size_v<input_parameter_types>;
+    static constexpr auto num_inputs = std::tuple_size_v<all_input_parameter_types>;
     static constexpr std::size_t num_outputs = number_output_objects<Unfold>;
 
+    static_assert(num_inputs > 0, "input_family requires at least one product selector.");
+
     // FIXME: Should maybe use some type of static assert, but not in a way that
     //        constrains the arguments of the Predicate and the Unfold to be the same.
     //
@@ -296,32 +370,59 @@ namespace phlex::detail {
     {
     }
 
-    auto input_family(std::array<product_selector, num_inputs> input_args)
+    // The selectors configure Object's constructor while trailing resources configure Unfold.
+    // FIXME: Evaluate whether input_family(...) should apply only to the constructor and a
+    // separate registration clause should declare resources for the unfold operation.
+    template <std::size_t NProducts, typename... Resources>
+    auto input_family(std::array<product_selector, NProducts> input_args, std::tuple<Resources...>)
     {
-      populate_types<input_parameter_types>(input_args);
+      static_assert((is_resource<Resources> && ...),
+                    "The second argument to input_family(...) must be a tuple of resources.");
+      static_assert(num_inputs == NProducts,
+                    "The number of generator constructor parameters is not the same as the number "
+                    "of specified product selectors.");
+
+      using product_parameter_types = boost::mp11::mp_take_c<all_input_parameter_types, NProducts>;
+      populate_types<product_parameter_types>(input_args);
 
       registrar_.set_creator([this, inputs = std::move(input_args)](auto upstream_predicates,
                                                                     auto output_product_suffixes) {
-        return std::make_unique<unfold_node<Object, Predicate, Unfold>>(
-          std::move(name_),
-          concurrency_,
-          std::move(upstream_predicates),
-          graph_,
-          std::move(predicate_),
-          std::move(unfold_),
-          std::vector(inputs.begin(), inputs.end()),
-          std::move(output_product_suffixes),
-          std::move(destination_layer_));
+        using node_type =
+          unfold_node<Object, Predicate, Unfold, internal::resource_type_t<Resources>...>;
+        return std::make_unique<node_type>(std::move(name_),
+                                           concurrency_,
+                                           std::move(upstream_predicates),
+                                           graph_,
+                                           std::move(predicate_),
+                                           std::move(unfold_),
+                                           std::vector(std::from_range, std::move(inputs)),
+                                           std::move(output_product_suffixes),
+                                           std::move(destination_layer_),
+                                           resources_);
       });
       return upstream_predicates<declared_unfold_ptr, num_outputs>{std::move(registrar_), config_};
     }
 
-    auto input_family(std::same_as<product_selector> auto... input_args)
+    auto input_family(std::array<product_selector, num_inputs> input_args)
     {
-      static_assert(num_inputs == sizeof...(input_args),
-                    "The number of function parameters is not the same as the number of specified "
-                    "input arguments.");
-      return input_family({std::move(input_args)...});
+      return input_family(std::move(input_args), std::tuple<>{});
+    }
+
+    auto input_family(product_selector first_selector, selector_or_resource auto... additional_args)
+    {
+      using split = resource_split<decltype(first_selector), decltype(additional_args)...>;
+      static_assert(
+        num_inputs == split::number_selectors,
+        "The number of generator constructor parameters is not the same as the number of "
+        "specified "
+        "input arguments.");
+      static_assert(split::resources_are_last,
+                    "All resource<T> arguments to input_family(...) must follow the "
+                    "product_selector arguments.");
+
+      auto [selectors, resources] = partition_selectors_and_resources<split>(
+        std::move(first_selector), std::move(additional_args)...);
+      return input_family(std::move(selectors), std::move(resources));
     }
 
   private:

From 55fd879c058a988bbe92583be09bb6bfa6ce45ac Mon Sep 17 00:00:00 2001
From: Kyle Knoepfel <knoepfel@fnal.gov>
Date: Wed, 26 Aug 2026 16:10:14 -0500
Subject: [PATCH 4/8] feat(nodes): add resource-aware node builder

---
 phlex/core/CMakeLists.txt          |   1 +
 phlex/core/node_builder.hpp        | 174 +++++++++++++++++++++++++++++
 phlex/metaprogramming/delegate.hpp |   7 ++
 3 files changed, 182 insertions(+)
 create mode 100644 phlex/core/node_builder.hpp

diff --git a/phlex/core/CMakeLists.txt b/phlex/core/CMakeLists.txt
index 68363ba5d..1ec53f860 100644
--- a/phlex/core/CMakeLists.txt
+++ b/phlex/core/CMakeLists.txt
@@ -59,6 +59,7 @@ install(
     message.hpp
     multilayer_join_node.hpp
     index_router.hpp
+    node_builder.hpp
     node_catalog.hpp
     product_selector.hpp
     products_consumer.hpp
diff --git a/phlex/core/node_builder.hpp b/phlex/core/node_builder.hpp
new file mode 100644
index 000000000..9b282485d
--- /dev/null
+++ b/phlex/core/node_builder.hpp
@@ -0,0 +1,174 @@
+#ifndef PHLEX_CORE_NODE_BUILDER_HPP
+#define PHLEX_CORE_NODE_BUILDER_HPP
+
+#include "phlex/core/resource_api.hpp"
+
+#include "oneapi/tbb/flow_graph.h"
+
+#include <cstddef>
+#include <functional>
+#include <tuple>
+#include <utility>
+
+namespace phlex::detail {
+
+  using no_outputs_t = std::tuple<>;
+  using no_resources_t = std::tuple<>;
+
+  template <typename... Outputs>
+  struct multifunction_outputs {};
+
+  template <typename InputMessages, typename Function, typename Outputs, typename Resources>
+  struct node_builder;
+
+  template <typename InputMessages, typename Function>
+  struct node_builder<InputMessages, Function, no_outputs_t, no_resources_t> {
+    using node_t = tbb::flow::function_node<InputMessages>;
+
+    template <typename NodeBody>
+    static node_t make(tbb::flow::graph& g,
+                       std::size_t concurrency,
+                       resource_catalog&,
+                       Function ft,
+                       NodeBody node_body)
+    {
+      return {g,
+              concurrency,
+              [ft = std::move(ft), node_body = std::move(node_body)](
+                InputMessages const& messages) mutable -> oneapi::tbb::flow::continue_msg {
+                std::invoke(node_body, ft, messages);
+                return {};
+              }};
+    }
+  };
+
+  template <typename InputMessages, typename Function, typename Resource, typename... Resources>
+  struct node_builder<InputMessages, Function, no_outputs_t, std::tuple<Resource, Resources...>> {
+    using node_t = tbb::flow::resource_limited_node<InputMessages, no_outputs_t>;
+
+    template <typename NodeBody>
+    static node_t make(tbb::flow::graph& g,
+                       std::size_t concurrency,
+                       resource_catalog& resources,
+                       Function ft,
+                       NodeBody node_body)
+    {
+      return {g,
+              concurrency,
+              std::tie(resources.template limiter_for<Resource>(),
+                       resources.template limiter_for<Resources>()...),
+              [ft = std::move(ft), node_body = std::move(node_body)](
+                InputMessages const& messages, auto&, auto*... resource_tokens) mutable {
+                std::invoke(node_body, ft, messages, resource_tokens...);
+              }};
+    }
+  };
+
+  template <typename InputMessages, typename Function, typename Output>
+  struct node_builder<InputMessages, Function, std::tuple<Output>, no_resources_t> {
+    using node_t = tbb::flow::function_node<InputMessages, Output>;
+
+    template <typename NodeBody>
+    static node_t make(tbb::flow::graph& g,
+                       std::size_t concurrency,
+                       resource_catalog&,
+                       Function ft,
+                       NodeBody node_body)
+    {
+      return {g,
+              concurrency,
+              [ft = std::move(ft),
+               node_body = std::move(node_body)](InputMessages const& messages) mutable {
+                return std::invoke(node_body, ft, messages);
+              }};
+    }
+
+    static tbb::flow::sender<Output>& output_port(node_t& node) { return node; }
+  };
+
+  template <typename InputMessages,
+            typename Function,
+            typename Output,
+            typename Resource,
+            typename... Resources>
+  struct node_builder<InputMessages,
+                      Function,
+                      std::tuple<Output>,
+                      std::tuple<Resource, Resources...>> {
+    using node_t = tbb::flow::resource_limited_node<InputMessages, std::tuple<Output>>;
+
+    template <typename NodeBody>
+    static node_t make(tbb::flow::graph& g,
+                       std::size_t concurrency,
+                       resource_catalog& resources,
+                       Function ft,
+                       NodeBody node_body)
+    {
+      return {g,
+              concurrency,
+              std::tie(resources.template limiter_for<Resource>(),
+                       resources.template limiter_for<Resources>()...),
+              [ft = std::move(ft), node_body = std::move(node_body)](
+                InputMessages const& messages, auto& ports, auto*... resource_tokens) mutable {
+                std::get<0>(ports).try_put(
+                  std::invoke(node_body, ft, messages, resource_tokens...));
+              }};
+    }
+
+    static tbb::flow::sender<Output>& output_port(node_t& node)
+    {
+      return tbb::flow::output_port<0>(node);
+    }
+  };
+
+  template <typename InputMessages, typename Function, typename... Outputs>
+  struct node_builder<InputMessages, Function, multifunction_outputs<Outputs...>, no_resources_t> {
+    using node_t = tbb::flow::multifunction_node<InputMessages, std::tuple<Outputs...>>;
+
+    template <typename NodeBody>
+    static node_t make(tbb::flow::graph& g,
+                       std::size_t concurrency,
+                       resource_catalog&,
+                       Function ft,
+                       NodeBody node_body)
+    {
+      return {g,
+              concurrency,
+              [ft = std::move(ft), node_body = std::move(node_body)](InputMessages const& messages,
+                                                                     auto& ports) mutable {
+                std::invoke(node_body, ft, messages, ports);
+              }};
+    }
+  };
+
+  template <typename InputMessages,
+            typename Function,
+            typename... Outputs,
+            typename Resource,
+            typename... Resources>
+  struct node_builder<InputMessages,
+                      Function,
+                      multifunction_outputs<Outputs...>,
+                      std::tuple<Resource, Resources...>> {
+    using node_t = tbb::flow::resource_limited_node<InputMessages, std::tuple<Outputs...>>;
+
+    template <typename NodeBody>
+    static node_t make(tbb::flow::graph& g,
+                       std::size_t concurrency,
+                       resource_catalog& resources,
+                       Function ft,
+                       NodeBody node_body)
+    {
+      return {g,
+              concurrency,
+              std::tie(resources.template limiter_for<Resource>(),
+                       resources.template limiter_for<Resources>()...),
+              [ft = std::move(ft), node_body = std::move(node_body)](
+                InputMessages const& messages, auto& ports, auto*... resource_tokens) mutable {
+                std::invoke(node_body, ft, messages, ports, resource_tokens...);
+              }};
+    }
+  };
+}
+
+#endif // PHLEX_CORE_NODE_BUILDER_HPP
diff --git a/phlex/metaprogramming/delegate.hpp b/phlex/metaprogramming/delegate.hpp
index 192568375..e7582d825 100644
--- a/phlex/metaprogramming/delegate.hpp
+++ b/phlex/metaprogramming/delegate.hpp
@@ -3,6 +3,9 @@
 
 #include "phlex/metaprogramming/type_deduction.hpp"
 
+#include "boost/mp11/algorithm.hpp"
+
+#include <cstddef>
 #include <functional>
 #include <memory>
 #include <utility>
@@ -36,6 +39,10 @@ namespace phlex::detail {
     using algorithm_type = Algorithm;
     using input_parameter_types = function_parameter_types<Algorithm>;
     static constexpr auto number_inputs = std::tuple_size_v<input_parameter_types>;
+    static constexpr auto number_outputs = number_output_objects<Algorithm>;
+
+    template <std::size_t Number>
+    using input_parameters = boost::mp11::mp_take_c<input_parameter_types, Number>;
 
     // A single templated constructor handles both cases: 'object' deduces to
     // shared_ptr<void_tag> for free functions and closures, and to shared_ptr<T>

From 24e7c2bad2ac6b59d10949174992ab2a677e811e Mon Sep 17 00:00:00 2001
From: Kyle Knoepfel <knoepfel@fnal.gov>
Date: Wed, 26 Aug 2026 16:13:38 -0500
Subject: [PATCH 5/8] feat(observer): support resource tokens

---
 phlex/core/declared_observer.hpp | 60 +++++++++++++++++++-------------
 test/resources_test.cpp          | 34 ++++++++++++++++++
 2 files changed, 70 insertions(+), 24 deletions(-)

diff --git a/phlex/core/declared_observer.hpp b/phlex/core/declared_observer.hpp
index 3f00ffd9f..1b8085a88 100644
--- a/phlex/core/declared_observer.hpp
+++ b/phlex/core/declared_observer.hpp
@@ -8,8 +8,10 @@
 #include "phlex/core/input_arguments.hpp"
 #include "phlex/core/message.hpp"
 #include "phlex/core/multilayer_join_node.hpp"
+#include "phlex/core/node_builder.hpp"
 #include "phlex/core/product_selector.hpp"
 #include "phlex/core/products_consumer.hpp"
+#include "phlex/core/resource_api.hpp"
 #include "phlex/metaprogramming/type_deduction.hpp"
 #include "phlex/model/algorithm_name.hpp"
 #include "phlex/model/data_cell_index.hpp"
@@ -44,11 +46,15 @@ namespace phlex::detail {
 
   // =====================================================================================
 
-  template <typename AlgorithmBits>
+  template <typename AlgorithmBits, typename... Resources>
   class observer_node : public declared_observer {
-    using input_args = AlgorithmBits::input_parameter_types;
     using function_t = AlgorithmBits::bound_type;
-    static constexpr auto num_inputs = AlgorithmBits::number_inputs;
+    static constexpr auto num_resources = sizeof...(Resources);
+    static constexpr auto num_products = AlgorithmBits::number_inputs - num_resources;
+    using input_product_types = AlgorithmBits::template input_parameters<num_products>;
+    using builder =
+      node_builder<messages_t<num_products>, function_t, no_outputs_t, std::tuple<Resources...>>;
+    using node_t = typename builder::node_t;
 
   public:
     static constexpr auto number_output_products = 0;
@@ -59,19 +65,23 @@ namespace phlex::detail {
                   std::vector<std::string> predicates,
                   tbb::flow::graph& g,
                   AlgorithmBits alg,
-                  product_selectors input_products) :
+                  product_selectors input_products,
+                  resource_catalog& resources) :
       declared_observer{std::move(algo_name), std::move(predicates), std::move(input_products)},
-      join_{make_join_or_none<num_inputs>(g, name().to_string(), layers())},
-      observer_{g,
-                concurrency,
-                [this, ft = alg.release_algorithm()](
-                  messages_t<num_inputs> const& messages) -> oneapi::tbb::flow::continue_msg {
-                  call(ft, messages, std::make_index_sequence<num_inputs>{});
-                  ++calls_;
-                  return {};
-                }}
+      join_{make_join_or_none<num_products>(g, name().to_string(), layers())},
+      observer_{builder::make(
+        g,
+        concurrency,
+        resources,
+        alg.release_algorithm(),
+        [this](function_t const& ft,
+               messages_t<num_products> const& messages,
+               auto*... resource_tokens) {
+          call(ft, messages, std::make_index_sequence<num_products>{}, resource_tokens...);
+          ++calls_;
+        })}
     {
-      if constexpr (num_inputs > 1ull) {
+      if constexpr (num_products > 1ull) {
         make_edge(join_, observer_);
       }
     }
@@ -79,32 +89,34 @@ namespace phlex::detail {
   private:
     tbb::flow::receiver<message>& port_for(product_selector const& input_product) override
     {
-      return receiver_for<num_inputs>(join_, input(), input_product, observer_);
+      return receiver_for<num_products>(join_, input(), input_product, observer_);
     }
 
     std::vector<tbb::flow::receiver<message>*> ports() override
     {
-      return input_ports<num_inputs>(join_, observer_);
+      return input_ports<num_products>(join_, observer_);
     }
 
     template <std::size_t... Is>
     void call(function_t const& ft,
-              messages_t<num_inputs> const& messages,
-              std::index_sequence<Is...>)
+              messages_t<num_products> const& messages,
+              std::index_sequence<Is...>,
+              auto*... resource_tokens)
     {
-      if constexpr (num_inputs == 1ull) {
-        std::invoke(ft, std::get<Is>(input_).retrieve(messages)...);
+      if constexpr (num_products == 1ull) {
+        std::invoke(ft, std::get<Is>(input_).retrieve(messages)..., resource_tokens...);
       } else {
-        std::invoke(ft, std::get<Is>(input_).retrieve(std::get<Is>(messages))...);
+        std::invoke(
+          ft, std::get<Is>(input_).retrieve(std::get<Is>(messages))..., resource_tokens...);
       }
     }
 
     named_index_ports index_ports() final { return join_.index_ports(); }
     std::size_t num_calls() const final { return calls_.load(); }
 
-    input_retriever_types<input_args> input_{input_arguments<input_args>()};
-    join_or_none_t<num_inputs> join_;
-    tbb::flow::function_node<messages_t<num_inputs>> observer_;
+    input_retriever_types<input_product_types> input_{input_arguments<input_product_types>()};
+    join_or_none_t<num_products> join_;
+    node_t observer_;
     std::atomic<std::size_t> calls_;
   };
 }
diff --git a/test/resources_test.cpp b/test/resources_test.cpp
index c27828d7f..7e529b387 100644
--- a/test/resources_test.cpp
+++ b/test/resources_test.cpp
@@ -1,16 +1,50 @@
 #include "phlex/core/framework_graph.hpp"
 #include "phlex/core/resource_api.hpp"
+#include "phlex/utilities/sleep_for.hpp"
+#include "phlex/utilities/thread_counter.hpp"
 
 #include "catch2/catch_test_macros.hpp"
 #include "catch2/matchers/catch_matchers_string.hpp"
 
+#include <chrono>
+
 using namespace phlex;
 using namespace phlex::detail;
+using namespace std::chrono_literals;
 
 namespace {
   struct Catch2 {}; // NOLINT(readability-identifier-naming)
 }
 
+TEST_CASE("registering one resource with the graph", "[graph][resource]")
+{
+  auto g = framework_graph::with_default_driver();
+  g.add_resource<Catch2>();
+
+  g.provide(
+     "number_maker", [](data_cell_index const&) { return 42; }, concurrency::unlimited)
+    .output_product("number_maker", "", "job");
+
+  auto counter = thread_counter::counter_type{};
+  auto verify_number = [&counter](int const num, Catch2 const*) {
+    // Both observers share one resource token and must not run concurrently.
+    thread_counter throw_if_more_than_one_thread{counter};
+    REQUIRE(num == 42);
+    spin_for(1ms);
+  };
+
+  product_selector const number_selector{.creator = "number_maker", .layer = "job"};
+  g.observe("verify1", verify_number, concurrency::unlimited)
+    .input_family(number_selector, resource<Catch2>{});
+  g.observe("verify2", verify_number, concurrency::unlimited)
+    .input_family(number_selector, resource<Catch2>{});
+
+  g.execute();
+
+  CHECK(g.execution_count("verify1") == 1);
+  CHECK(g.execution_count("verify2") == 1);
+}
+
 TEST_CASE("resource catalog", "[resource]")
 {
   resource_catalog catalog;

From 9d59815782d43225baeb23be25b6e35e80330b52 Mon Sep 17 00:00:00 2001
From: Kyle Knoepfel <knoepfel@fnal.gov>
Date: Wed, 26 Aug 2026 16:13:40 -0500
Subject: [PATCH 6/8] feat(transform): support resource tokens in transforms
 and predicates

---
 phlex/core/declared_predicate.hpp | 75 +++++++++++++++++++------------
 phlex/core/declared_transform.hpp | 61 ++++++++++++++++---------
 test/transform_node_test.cpp      | 46 ++++++++++++++++++-
 3 files changed, 130 insertions(+), 52 deletions(-)

diff --git a/phlex/core/declared_predicate.hpp b/phlex/core/declared_predicate.hpp
index ab63465c8..7657d6fb1 100644
--- a/phlex/core/declared_predicate.hpp
+++ b/phlex/core/declared_predicate.hpp
@@ -9,8 +9,10 @@
 #include "phlex/core/input_arguments.hpp"
 #include "phlex/core/message.hpp"
 #include "phlex/core/multilayer_join_node.hpp"
+#include "phlex/core/node_builder.hpp"
 #include "phlex/core/product_selector.hpp"
 #include "phlex/core/products_consumer.hpp"
+#include "phlex/core/resource_api.hpp"
 #include "phlex/metaprogramming/type_deduction.hpp"
 #include "phlex/model/algorithm_name.hpp"
 #include "phlex/model/data_cell_index.hpp"
@@ -49,11 +51,17 @@ namespace phlex::detail {
 
   // =====================================================================================
 
-  template <typename AlgorithmBits>
+  template <typename AlgorithmBits, typename... Resources>
   class predicate_node : public declared_predicate {
-    using input_args = AlgorithmBits::input_parameter_types;
     using function_t = AlgorithmBits::bound_type;
-    static constexpr auto num_inputs = AlgorithmBits::number_inputs;
+    static constexpr auto num_resources = sizeof...(Resources);
+    static constexpr auto num_products = AlgorithmBits::number_inputs - num_resources;
+    using input_product_types = AlgorithmBits::template input_parameters<num_products>;
+    using builder = node_builder<messages_t<num_products>,
+                                 function_t,
+                                 std::tuple<predicate_result>,
+                                 std::tuple<Resources...>>;
+    using node_t = typename builder::node_t;
 
   public:
     static constexpr auto number_output_products = 0ull;
@@ -64,22 +72,28 @@ namespace phlex::detail {
                    std::vector<std::string> predicates,
                    tbb::flow::graph& g,
                    AlgorithmBits alg,
-                   product_selectors input_products) :
+                   product_selectors input_products,
+                   resource_catalog& resources) :
       declared_predicate{std::move(algo_name), std::move(predicates), std::move(input_products)},
-      join_{make_join_or_none<num_inputs>(g, name().to_string(), layers())},
-      predicate_{g,
-                 concurrency,
-                 [this, ft = alg.release_algorithm()](
-                   messages_t<num_inputs> const& messages) -> predicate_result {
-                   auto const& msg = most_derived(messages);
-                   auto const& [store, message_id] = std::tie(msg.store, msg.id);
-
-                   bool const rc = call(ft, messages, std::make_index_sequence<num_inputs>{});
-                   ++calls_;
-                   return {message_id, rc};
-                 }}
+      join_{make_join_or_none<num_products>(g, name().to_string(), layers())},
+      predicate_{builder::make(
+        g,
+        concurrency,
+        resources,
+        alg.release_algorithm(),
+        [this](function_t const& ft,
+               messages_t<num_products> const& messages,
+               auto*... resource_tokens) -> predicate_result {
+          auto const& msg = most_derived(messages);
+          auto const& [store, message_id] = std::tie(msg.store, msg.id);
+
+          bool const rc =
+            call(ft, messages, std::make_index_sequence<num_products>{}, resource_tokens...);
+          ++calls_;
+          return {message_id, rc};
+        })}
     {
-      if constexpr (num_inputs > 1ull) {
+      if constexpr (num_products > 1ull) {
         make_edge(join_, predicate_);
       }
     }
@@ -87,33 +101,38 @@ namespace phlex::detail {
   private:
     tbb::flow::receiver<message>& port_for(product_selector const& input_product) override
     {
-      return receiver_for<num_inputs>(join_, input(), input_product, predicate_);
+      return receiver_for<num_products>(join_, input(), input_product, predicate_);
+    }
+    tbb::flow::sender<predicate_result>& sender() override
+    {
+      return builder::output_port(predicate_);
     }
-    tbb::flow::sender<predicate_result>& sender() override { return predicate_; }
 
     std::vector<tbb::flow::receiver<message>*> ports() override
     {
-      return input_ports<num_inputs>(join_, predicate_);
+      return input_ports<num_products>(join_, predicate_);
     }
 
     template <std::size_t... Is>
     bool call(function_t const& ft,
-              messages_t<num_inputs> const& messages,
-              std::index_sequence<Is...>)
+              messages_t<num_products> const& messages,
+              std::index_sequence<Is...>,
+              auto*... resource_tokens)
     {
-      if constexpr (num_inputs == 1ull) {
-        return std::invoke(ft, std::get<Is>(input_).retrieve(messages)...);
+      if constexpr (num_products == 1ull) {
+        return std::invoke(ft, std::get<Is>(input_).retrieve(messages)..., resource_tokens...);
       } else {
-        return std::invoke(ft, std::get<Is>(input_).retrieve(std::get<Is>(messages))...);
+        return std::invoke(
+          ft, std::get<Is>(input_).retrieve(std::get<Is>(messages))..., resource_tokens...);
       }
     }
 
     named_index_ports index_ports() final { return join_.index_ports(); }
     std::size_t num_calls() const final { return calls_.load(); }
 
-    input_retriever_types<input_args> input_{input_arguments<input_args>()};
-    join_or_none_t<num_inputs> join_;
-    tbb::flow::function_node<messages_t<num_inputs>, predicate_result> predicate_;
+    input_retriever_types<input_product_types> input_{input_arguments<input_product_types>()};
+    join_or_none_t<num_products> join_;
+    node_t predicate_;
     std::atomic<std::size_t> calls_;
   };
 
diff --git a/phlex/core/declared_transform.hpp b/phlex/core/declared_transform.hpp
index b5fd24c71..87cf8b997 100644
--- a/phlex/core/declared_transform.hpp
+++ b/phlex/core/declared_transform.hpp
@@ -11,8 +11,10 @@
 #include "phlex/core/input_arguments.hpp"
 #include "phlex/core/message.hpp"
 #include "phlex/core/multilayer_join_node.hpp"
+#include "phlex/core/node_builder.hpp"
 #include "phlex/core/product_selector.hpp"
 #include "phlex/core/products_consumer.hpp"
+#include "phlex/core/resource_api.hpp"
 #include "phlex/metaprogramming/type_deduction.hpp"
 #include "phlex/model/algorithm_name.hpp"
 #include "phlex/model/data_cell_index.hpp"
@@ -33,6 +35,7 @@
 #include <ranges>
 #include <stdexcept>
 #include <string>
+#include <tuple>
 #include <type_traits>
 #include <utility>
 
@@ -55,13 +58,19 @@ namespace phlex::detail {
 
   // =====================================================================================
 
-  template <typename AlgorithmBits>
+  template <typename AlgorithmBits, typename... Resources>
   class transform_node : public declared_transform {
     using function_t = AlgorithmBits::bound_type;
-    using input_parameter_types = AlgorithmBits::input_parameter_types;
 
-    static constexpr auto num_inputs = AlgorithmBits::number_inputs;
-    static constexpr auto num_outputs = number_output_objects<function_t>;
+    static constexpr auto num_resources = sizeof...(Resources);
+    static constexpr auto num_products = AlgorithmBits::number_inputs - num_resources;
+    static constexpr auto num_outputs = AlgorithmBits::number_outputs;
+    using input_product_types = AlgorithmBits::template input_parameters<num_products>;
+    using builder = node_builder<messages_t<num_products>,
+                                 function_t,
+                                 std::tuple<message>,
+                                 std::tuple<Resources...>>;
+    using node_t = typename builder::node_t;
 
   public:
     using node_ptr_type = declared_transform_ptr;
@@ -73,20 +82,26 @@ namespace phlex::detail {
                    tbb::flow::graph& g,
                    AlgorithmBits alg,
                    product_selectors input_products,
-                   std::vector<std::string> output) :
+                   std::vector<std::string> output,
+                   resource_catalog& resources) :
       declared_transform{std::move(algo_name), std::move(predicates), std::move(input_products)},
       output_{
         to_product_specifications(name(), std::move(output), make_output_type_ids<function_t>())},
-      join_{make_join_or_none<num_inputs>(g, name().to_string(), layers())},
-      transform_{
+      join_{make_join_or_none<num_products>(g, name().to_string(), layers())},
+      transform_{builder::make(
         g,
         concurrency,
-        [this, ft = alg.release_algorithm()](messages_t<num_inputs> const& messages) -> message {
+        resources,
+        alg.release_algorithm(),
+        [this](function_t const& ft,
+               messages_t<num_products> const& messages,
+               auto*... resource_tokens) -> message {
           using namespace phlex::experimental::detail;
           auto const& msg = most_derived(messages);
           auto const& [store, message_id] = std::tie(msg.store, msg.id);
 
-          auto result = call(ft, messages, std::make_index_sequence<num_inputs>{});
+          auto result =
+            call(ft, messages, std::make_index_sequence<num_products>{}, resource_tokens...);
           ++calls_;
           ++product_count_[store->index()->layer_hash()];
 
@@ -96,9 +111,9 @@ namespace phlex::detail {
             store->index(), name(), std::move(new_products));
 
           return {.store = std::move(new_store), .id = message_id};
-        }}
+        })}
     {
-      if constexpr (num_inputs > 1ull) {
+      if constexpr (num_products > 1ull) {
         make_edge(join_, transform_);
       }
     }
@@ -106,26 +121,28 @@ namespace phlex::detail {
   private:
     tbb::flow::receiver<message>& port_for(product_selector const& input_product) override
     {
-      return receiver_for<num_inputs>(join_, input(), input_product, transform_);
+      return receiver_for<num_products>(join_, input(), input_product, transform_);
     }
 
     std::vector<tbb::flow::receiver<message>*> ports() override
     {
-      return input_ports<num_inputs>(join_, transform_);
+      return input_ports<num_products>(join_, transform_);
     }
 
-    tbb::flow::sender<message>& output_port() override { return transform_; }
+    tbb::flow::sender<message>& output_port() override { return builder::output_port(transform_); }
     product_specifications const& output() const override { return output_; }
 
     template <std::size_t... Is>
     auto call(function_t const& ft,
-              messages_t<num_inputs> const& messages,
-              std::index_sequence<Is...>)
+              messages_t<num_products> const& messages,
+              std::index_sequence<Is...>,
+              auto*... resource_tokens)
     {
-      if constexpr (num_inputs == 1ull) {
-        return std::invoke(ft, std::get<Is>(input_).retrieve(messages)...);
+      if constexpr (num_products == 1ull) {
+        return std::invoke(ft, std::get<Is>(input_).retrieve(messages)..., resource_tokens...);
       } else {
-        return std::invoke(ft, std::get<Is>(input_).retrieve(std::get<Is>(messages))...);
+        return std::invoke(
+          ft, std::get<Is>(input_).retrieve(std::get<Is>(messages))..., resource_tokens...);
       }
     }
 
@@ -140,10 +157,10 @@ namespace phlex::detail {
       return result;
     }
 
-    input_retriever_types<input_parameter_types> input_{input_arguments<input_parameter_types>()};
+    input_retriever_types<input_product_types> input_{input_arguments<input_product_types>()};
     product_specifications output_;
-    join_or_none_t<num_inputs> join_;
-    tbb::flow::function_node<messages_t<num_inputs>, message> transform_;
+    join_or_none_t<num_products> join_;
+    node_t transform_;
     std::atomic<std::size_t> calls_;
     tbb::concurrent_unordered_map<std::size_t, std::atomic<std::size_t>> product_count_;
   };
diff --git a/test/transform_node_test.cpp b/test/transform_node_test.cpp
index fff626ba6..8c90171d5 100644
--- a/test/transform_node_test.cpp
+++ b/test/transform_node_test.cpp
@@ -39,8 +39,15 @@ namespace {
     auto operator<=>(output_type_2 const&) const = default;
   };
 
+  struct transform_resource {};
+
   output_type_1 double_value(input_type_1 const& input) { return {input.value * 2}; }
 
+  output_type_1 increment_with_resource(input_type_1 const& input, transform_resource const*)
+  {
+    return {input.value + 1};
+  }
+
   auto number_and_label(input_type_1 const& input)
   {
     return std::tuple{output_type_1{input.value}, output_type_2{std::to_string(input.value)}};
@@ -79,9 +86,10 @@ TEST_CASE("transform_node directly transforms one input product", "[transform_no
   auto input_selector = selector<input_type_1>("input", "");
   auto input_store = store_with_product("input", "", input_type_1{21});
   auto alg = algorithm_bits_for(double_value);
+  resource_catalog resources;
 
   transform_node<decltype(alg)> node{
-    algorithm_name{"double_value"}, 1u, {}, graph, std::move(alg), {input_selector}, {}};
+    algorithm_name{"double_value"}, 1u, {}, graph, std::move(alg), {input_selector}, {}, resources};
   declared_transform& transform = node;
 
   auto const& output_specs = transform.output();
@@ -115,6 +123,7 @@ TEST_CASE("transform_node stores multiple output products", "[transform_node]")
   auto input_selector = selector<input_type_1>("input", "");
   auto input_store = store_with_product("input", "", input_type_1{7});
   auto alg = algorithm_bits_for(number_and_label);
+  resource_catalog resources;
 
   transform_node<decltype(alg)> node{algorithm_name{"number_and_label"},
                                      1u,
@@ -122,7 +131,8 @@ TEST_CASE("transform_node stores multiple output products", "[transform_node]")
                                      graph,
                                      std::move(alg),
                                      {input_selector},
-                                     {"number", "label"}};
+                                     {"number", "label"},
+                                     resources};
   declared_transform& transform = node;
 
   oneapi::tbb::flow::queue_node<message> sink{graph};
@@ -151,3 +161,35 @@ TEST_CASE("transform_node stores multiple output products", "[transform_node]")
   CHECK(transform.num_calls() == 1u);
   CHECK(transform.product_count() == 1u);
 }
+
+TEST_CASE("transform_node receives a resource token", "[transform_node][resource]")
+{
+  oneapi::tbb::flow::graph graph;
+  auto input_selector = selector<input_type_1>("input", "");
+  auto input_store = store_with_product("input", "", input_type_1{21});
+  auto alg = algorithm_bits_for(increment_with_resource);
+  resource_catalog resources;
+  resources.add<transform_resource>();
+
+  transform_node<decltype(alg), transform_resource> node{algorithm_name{"increment_with_resource"},
+                                                         1u,
+                                                         {},
+                                                         graph,
+                                                         std::move(alg),
+                                                         {input_selector},
+                                                         {},
+                                                         resources};
+  declared_transform& transform = node;
+
+  oneapi::tbb::flow::queue_node<message> sink{graph};
+  make_edge(transform.output_port(), sink);
+
+  REQUIRE(node.port(input_selector).try_put({.store = input_store, .id = message_id}));
+  graph.wait_for_all();
+
+  message output;
+  REQUIRE(sink.try_get(output));
+  CHECK(output.store->get_product<output_type_1>(transform.output()[0]) == output_type_1{22});
+  CHECK(transform.num_calls() == 1u);
+  CHECK(transform.product_count() == 1u);
+}

From 85e1bd71285884f2be62a58208ed45b05b90eef1 Mon Sep 17 00:00:00 2001
From: Kyle Knoepfel <knoepfel@fnal.gov>
Date: Wed, 26 Aug 2026 16:13:41 -0500
Subject: [PATCH 7/8] feat(fold): support resource tokens

---
 phlex/core/declared_fold.hpp | 61 ++++++++++++++++++++++--------------
 test/fold_test.cpp           | 46 +++++++++++++++++++++++++++
 2 files changed, 84 insertions(+), 23 deletions(-)

diff --git a/phlex/core/declared_fold.hpp b/phlex/core/declared_fold.hpp
index c7cfa4593..77826321e 100644
--- a/phlex/core/declared_fold.hpp
+++ b/phlex/core/declared_fold.hpp
@@ -10,6 +10,7 @@
 #include "phlex/core/fwd.hpp"
 #include "phlex/core/input_arguments.hpp"
 #include "phlex/core/message.hpp"
+#include "phlex/core/node_builder.hpp"
 #include "phlex/core/product_selector.hpp"
 #include "phlex/core/products_consumer.hpp"
 #include "phlex/model/algorithm_name.hpp"
@@ -73,15 +74,24 @@ namespace phlex::detail {
 
   // =====================================================================================
 
-  template <typename AlgorithmBits, typename InitTuple>
+  template <typename AlgorithmBits, typename InitTuple, typename... Resources>
   class fold_node : public declared_fold {
     using all_parameter_types = AlgorithmBits::input_parameter_types;
     using result_type = std::decay_t<std::tuple_element_t<0, all_parameter_types>>;
-    using input_parameter_types = skip_first_type<all_parameter_types>; // Skip fold object
-    static constexpr auto num_inputs = std::tuple_size_v<input_parameter_types>;
+    using non_result_parameter_types = skip_first_type<all_parameter_types>;
+    static constexpr std::size_t num_resources = sizeof...(Resources);
+    static constexpr std::size_t num_input_products =
+      std::tuple_size_v<non_result_parameter_types> - num_resources;
+    using input_product_types =
+      boost::mp11::mp_take_c<non_result_parameter_types, num_input_products>;
 
     static constexpr std::size_t num_outputs = 1; // hard-coded for now
     using function_t = AlgorithmBits::bound_type;
+    using builder = node_builder<accumulator_with_messages<result_type, num_input_products>,
+                                 function_t,
+                                 no_outputs_t,
+                                 std::tuple<Resources...>>;
+    using node_t = typename builder::node_t;
 
   public:
     fold_node(phlex::experimental::algorithm_name algo_name,
@@ -92,7 +102,8 @@ namespace phlex::detail {
               InitTuple initializer,
               product_selectors input_products,
               std::vector<std::string> output,
-              std::string partition_layer) :
+              std::string partition_layer,
+              resource_catalog& resources) :
       declared_fold{std::move(algo_name),
                     std::move(predicates),
                     std::move(input_products),
@@ -105,17 +116,20 @@ namespace phlex::detail {
             this->output(),
             make_initializer<result_type>(
               std::move(initializer), std::make_index_sequence<std::tuple_size_v<InitTuple>>{})},
-      fold_{g,
-            concurrency,
-            [this, ft = alg.release_algorithm()](
-              accumulator_with_messages<result_type, num_inputs> const& accum_with_msgs) {
-              std::size_t const partition_hash = apply_fold(ft, accum_with_msgs);
-
-              ++calls_;
-
-              join_.notify_result_repeater_port().try_put(partition_hash);
-              return tbb::flow::continue_msg{};
-            }}
+      fold_{builder::make(
+        g,
+        concurrency,
+        resources,
+        alg.release_algorithm(),
+        [this](function_t const& ft,
+               accumulator_with_messages<result_type, num_input_products> const& accum_with_msgs,
+               auto*... resource_tokens) {
+          std::size_t const partition_hash = apply_fold(ft, accum_with_msgs, resource_tokens...);
+
+          ++calls_;
+
+          join_.notify_result_repeater_port().try_put(partition_hash);
+        })}
     {
       make_edge(join_, fold_);
     }
@@ -138,23 +152,24 @@ namespace phlex::detail {
 
     std::size_t apply_fold(
       function_t const& ft,
-      accumulator_with_messages<result_type, num_inputs> const& accum_with_msgs)
+      accumulator_with_messages<result_type, num_input_products> const& accum_with_msgs,
+      auto*... resource_tokens)
     {
       // We have to do awkward index management until we can use structured bindings with packs.
       auto& accumulator = std::get<0>(accum_with_msgs);
       [&]<std::size_t... Is>(std::index_sequence<Is...>) {
         accumulator.partial_result->call(
-          ft, std::get<Is>(input_).retrieve(std::get<Is + 1>(accum_with_msgs))...);
-      }(std::make_index_sequence<num_inputs>{});
+          ft,
+          std::get<Is>(input_).retrieve(std::get<Is + 1>(accum_with_msgs))...,
+          resource_tokens...);
+      }(std::make_index_sequence<num_input_products>{});
       return accumulator.index->hash();
     }
 
-    input_retriever_types<input_parameter_types> input_{input_arguments<input_parameter_types>()};
+    input_retriever_types<input_product_types> input_{input_arguments<input_product_types>()};
     product_specifications output_;
-    fold_join_node<result_type, num_inputs> join_;
-    tbb::flow::function_node<accumulator_with_messages<result_type, num_inputs>,
-                             tbb::flow::continue_msg>
-      fold_;
+    fold_join_node<result_type, num_input_products> join_;
+    node_t fold_;
     std::atomic<std::size_t> calls_;
   };
 }
diff --git a/test/fold_test.cpp b/test/fold_test.cpp
index f24620b4c..3aff6e4d1 100644
--- a/test/fold_test.cpp
+++ b/test/fold_test.cpp
@@ -24,7 +24,10 @@
 // =======================================================================================
 
 #include "phlex/core/framework_graph.hpp"
+#include "phlex/core/resource_api.hpp"
 #include "phlex/model/data_cell_index.hpp"
+#include "phlex/utilities/sleep_for.hpp"
+#include "phlex/utilities/thread_counter.hpp"
 #include "plugins/layer_generator.hpp"
 
 #include "catch2/catch_test_macros.hpp"
@@ -35,6 +38,7 @@
 #include <vector>
 
 using namespace phlex;
+using namespace std::chrono_literals;
 
 namespace {
   void add(std::atomic<unsigned int>& counter, unsigned int number) { counter += number; }
@@ -53,6 +57,9 @@ namespace {
 
   // Provider algorithm
   unsigned int provide_number(data_cell_index const& id) { return id.number(); }
+
+  struct fold_resource_1 {};
+  struct fold_resource_2 {};
 }
 
 TEST_CASE("Different data layers of fold", "[graph]")
@@ -127,3 +134,42 @@ TEST_CASE("Fold output without send consumed downstream", "[graph]")
   CHECK(g.execution_count("collect_numbers") == std::size_t{index_limit} * number_limit);
   CHECK(g.execution_count("verify_collected_numbers") == index_limit);
 }
+
+TEST_CASE("Fold receives a resource token", "[graph][fold][resource]")
+{
+  constexpr auto number_runs = 2u;
+  constexpr auto number_events = 5u;
+
+  auto gen = experimental::layer_generator::make();
+  gen->add_layer("run", {.parent_layer = "job", .count = number_runs});
+  gen->add_layer("event", {.parent_layer = "run", .count = number_events});
+
+  auto g = phlex::detail::framework_graph::without_driver();
+  g.add_driver(gen);
+  g.add_resource<fold_resource_1>();
+  g.add_resource<fold_resource_2>();
+
+  g.provide("provide_number", provide_number, concurrency::unlimited)
+    .output_product("input", "number", "event");
+
+  auto counter = detail::thread_counter::counter_type{};
+  auto add_with_resources = [&counter](std::atomic<unsigned int>& sum,
+                                       unsigned int number,
+                                       fold_resource_1 const*,
+                                       fold_resource_2 const*) {
+    detail::thread_counter throw_if_more_than_one_thread{counter};
+    detail::spin_for(5ms);
+    sum += number;
+  };
+
+  auto const input = product_selector{.creator = "input", .layer = "event", .suffix = "number"};
+  g.fold("resource_add", add_with_resources, concurrency::unlimited, "run")
+    .input_family(input, detail::resource<fold_resource_1>{}, detail::resource<fold_resource_2>{});
+  g.observe("verify_resource_sum", [](unsigned int actual) { CHECK(actual == 10u); })
+    .input_family(product_selector{.creator = "resource_add", .layer = "run"});
+
+  g.execute();
+
+  CHECK(g.execution_count("resource_add") == std::size_t{number_runs} * number_events);
+  CHECK(g.execution_count("verify_resource_sum") == number_runs);
+}

From 1cd8f5b610a955151febfce4a55f84b933326721 Mon Sep 17 00:00:00 2001
From: Kyle Knoepfel <knoepfel@fnal.gov>
Date: Wed, 26 Aug 2026 16:14:08 -0500
Subject: [PATCH 8/8] feat(unfold): support resource tokens in unfold
 operations

---
 phlex/core/declared_unfold.hpp | 70 ++++++++++++++++++++++------------
 test/unfold_test.cpp           | 57 +++++++++++++++++++++++++++
 2 files changed, 102 insertions(+), 25 deletions(-)

diff --git a/phlex/core/declared_unfold.hpp b/phlex/core/declared_unfold.hpp
index aa34a6596..6b0f94403 100644
--- a/phlex/core/declared_unfold.hpp
+++ b/phlex/core/declared_unfold.hpp
@@ -8,6 +8,7 @@
 #include "phlex/core/input_arguments.hpp"
 #include "phlex/core/message.hpp"
 #include "phlex/core/multilayer_join_node.hpp"
+#include "phlex/core/node_builder.hpp"
 #include "phlex/core/products_consumer.hpp"
 #include "phlex/model/algorithm_name.hpp"
 #include "phlex/model/data_cell_index.hpp"
@@ -80,11 +81,18 @@ namespace phlex::detail {
 
   // =====================================================================================
 
-  template <typename Object, typename Predicate, typename Unfold>
+  template <typename Object, typename Predicate, typename Unfold, typename... Resources>
   class unfold_node : public declared_unfold {
-    using input_args = constructor_parameter_types<Object>;
-    static constexpr std::size_t num_inputs = std::tuple_size_v<input_args>;
+    using all_input_args = constructor_parameter_types<Object>;
+    static constexpr std::size_t num_inputs = std::tuple_size_v<all_input_args>;
+    using input_args = all_input_args;
     static constexpr std::size_t num_outputs = number_output_objects<Unfold>;
+    using function_t = std::decay_t<Unfold>;
+    using builder = node_builder<messages_t<num_inputs>,
+                                 function_t,
+                                 multifunction_outputs<message, index_message, unfold_flush>,
+                                 std::tuple<Resources...>>;
+    using node_t = typename builder::node_t;
 
   public:
     unfold_node(phlex::experimental::algorithm_name algo_name,
@@ -95,7 +103,8 @@ namespace phlex::detail {
                 Unfold&& unfold,
                 product_selectors input_products,
                 std::vector<std::string> output_product_suffixes,
-                std::string child_layer_name) :
+                std::string child_layer_name,
+                resource_catalog& resources) :
       declared_unfold{std::move(algo_name),
                       std::move(predicates),
                       std::move(input_products),
@@ -104,20 +113,29 @@ namespace phlex::detail {
                                         std::move(output_product_suffixes),
                                         make_type_ids<skip_first_type<return_type<Unfold>>>())},
       join_{make_join_or_none<num_inputs>(g, name().to_string(), layers())},
-      unfold_{g,
-              concurrency,
-              [this, p = std::move(predicate), ufold = std::move(unfold)](
-                messages_t<num_inputs> const& messages, auto& outputs) {
-                auto const& msg = most_derived(messages);
-                auto const& store = msg.store;
-
-                generator gen{store, name(), child_layer()};
-                call(
-                  p, ufold, store->index(), gen, messages, std::make_index_sequence<num_inputs>{});
-                std::get<2>(outputs).try_put({.index = store->index(),
-                                              .layer_hash = gen.child_layer_hash(),
-                                              .count = gen.child_count()});
-              }}
+      unfold_{builder::make(g,
+                            concurrency,
+                            resources,
+                            std::move(unfold),
+                            [this, p = std::move(predicate)](function_t const& ufold,
+                                                             messages_t<num_inputs> const& messages,
+                                                             auto& outputs,
+                                                             auto*... resource_tokens) {
+                              auto const& msg = most_derived(messages);
+                              auto const& store = msg.store;
+
+                              generator gen{store, name(), child_layer()};
+                              call(p,
+                                   ufold,
+                                   store->index(),
+                                   gen,
+                                   messages,
+                                   std::make_index_sequence<num_inputs>{},
+                                   resource_tokens...);
+                              std::get<2>(outputs).try_put({.index = store->index(),
+                                                            .layer_hash = gen.child_layer_hash(),
+                                                            .count = gen.child_count()});
+                            })}
     {
       if constexpr (num_inputs > 1ull) {
         make_edge(join_, unfold_);
@@ -154,7 +172,8 @@ namespace phlex::detail {
               data_cell_index_ptr const& unfolded_id,
               generator& g,
               messages_t<num_inputs> const& messages,
-              std::index_sequence<Is...>)
+              std::index_sequence<Is...>,
+              auto*... resource_tokens)
     {
       ++calls_;
       Object obj = [this, &messages]() {
@@ -169,12 +188,15 @@ namespace phlex::detail {
       while (std::invoke(predicate, obj, running_value)) {
         products new_products{num_outputs};
         auto new_id = unfolded_id->make_child(child_layer(), counter);
-        if constexpr (requires { std::invoke(unfold, obj, running_value, *new_id); }) {
-          auto [next_value, prods] = std::invoke(unfold, obj, running_value, *new_id);
+        if constexpr (requires {
+                        std::invoke(unfold, obj, running_value, *new_id, resource_tokens...);
+                      }) {
+          auto [next_value, prods] =
+            std::invoke(unfold, obj, running_value, *new_id, resource_tokens...);
           new_products.add_all(output_, std::move(prods));
           running_value = next_value;
         } else {
-          auto [next_value, prods] = std::invoke(unfold, obj, running_value);
+          auto [next_value, prods] = std::invoke(unfold, obj, running_value, resource_tokens...);
           new_products.add_all(output_, std::move(prods));
           running_value = next_value;
         }
@@ -194,9 +216,7 @@ namespace phlex::detail {
     input_retriever_types<input_args> input_{input_arguments<input_args>()};
     product_specifications output_;
     join_or_none_t<num_inputs> join_;
-    tbb::flow::multifunction_node<messages_t<num_inputs>,
-                                  std::tuple<message, index_message, unfold_flush>>
-      unfold_;
+    node_t unfold_;
     std::atomic<std::size_t> msg_counter_; // Is this sufficient?  Probably not.
     std::atomic<std::size_t> calls_;
     std::atomic<std::size_t> product_count_;
diff --git a/test/unfold_test.cpp b/test/unfold_test.cpp
index 64f53d472..9235657f0 100644
--- a/test/unfold_test.cpp
+++ b/test/unfold_test.cpp
@@ -15,7 +15,10 @@
 // =======================================================================================
 
 #include "phlex/core/framework_graph.hpp"
+#include "phlex/core/resource_api.hpp"
 #include "phlex/model/data_cell_index.hpp"
+#include "phlex/utilities/sleep_for.hpp"
+#include "phlex/utilities/thread_counter.hpp"
 #include "plugins/layer_generator.hpp"
 #include "test/products_for_output.hpp"
 
@@ -24,8 +27,10 @@
 #include <atomic>
 #include <string>
 #include <utility>
+#include <vector>
 
 using namespace phlex;
+using namespace std::chrono_literals;
 
 namespace {
   class iota {
@@ -98,6 +103,29 @@ namespace {
     unsigned int max_;
     numbers_t numbers_;
   };
+
+  // Test load-bearing resource for unfold operations.
+  struct unfold_resource {
+    detail::thread_counter::counter_type concurrent_unfolds{};
+  };
+
+  class resource_iota {
+  public:
+    explicit resource_iota(unsigned int max_number) : max_{max_number} {}
+
+    unsigned int initial_value() const { return 0; }
+    bool predicate(unsigned int i) const { return i != max_; }
+
+    auto unfold(unsigned int i, unfold_resource* resource) const
+    {
+      detail::thread_counter guard{resource->concurrent_unfolds};
+      detail::spin_for(1ms);
+      return std::make_pair(i + 1, i);
+    }
+
+  private:
+    unsigned int max_;
+  };
 }
 
 TEST_CASE("Splitting the processing", "[graph]")
@@ -240,3 +268,32 @@ TEST_CASE("Unfold deduplicates same-layer inputs for bookkeeping", "[graph]")
   CHECK(g.execution_count("add_dual_input_unfold") == 30u);
   CHECK(g.execution_count("check_dual_input_unfold_sum") == index_limit);
 }
+
+TEST_CASE("Unfold receives a resource token", "[graph][unfold][resource]")
+{
+  constexpr auto num_events = 2u;
+
+  auto gen = experimental::layer_generator::make();
+  gen->add_layer("event", {.parent_layer = "job", .count = num_events});
+
+  auto g = phlex::detail::framework_graph::without_driver();
+  g.add_driver(gen);
+  g.add_resource<unfold_resource>();
+
+  g.provide(
+     "provide_max_number", [](data_cell_index const&) { return 10u; }, concurrency::unlimited)
+    .output_product("input", "max_number", "event");
+
+  g.unfold<resource_iota>("resource_iota",
+                          &resource_iota::predicate,
+                          &resource_iota::unfold,
+                          concurrency::unlimited,
+                          "subevent")
+    .input_family(product_selector{.creator = "input", .layer = "event", .suffix = "max_number"},
+                  detail::resource<unfold_resource>{})
+    .output_product_suffixes("new_number");
+
+  g.execute();
+
+  CHECK(g.execution_count("resource_iota") == num_events);
+}