From 1c48ceeafa95d470dd3041b11eabfb6bee827bae Mon Sep 17 00:00:00 2001
From: Marc Paterno <paterno@fnal.gov>
Date: Wed, 21 Oct 2026 09:34:06 -0500
Subject: [PATCH] feat(core): integrate translator nodes into the graph

Register one-input, one-output translators as consumers and independent
producers on the runtime/typed-node architecture.

Preserve translator creator, registered suffix, input data-cell index,
and execution stage, with end-to-end observer regression coverage.
---
 phlex/core/CMakeLists.txt           |   3 +
 phlex/core/concepts.hpp             |  12 +++
 phlex/core/declared_translator.cpp  |  25 +++++
 phlex/core/declared_translator.hpp  |  41 ++++++++
 phlex/core/framework_graph.cpp      |   1 +
 phlex/core/framework_graph.hpp      |   7 ++
 phlex/core/glue.hpp                 |  21 +++++
 phlex/core/graph_proxy.hpp          |  13 +++
 phlex/core/node_catalog.cpp         |   5 +
 phlex/core/node_catalog.hpp         |   8 +-
 phlex/core/registration_api.hpp     |   1 +
 phlex/core/translator_node.hpp      | 139 ++++++++++++++++++++++++++++
 test/function_registration_test.cpp |  35 +++++++
 13 files changed, 309 insertions(+), 2 deletions(-)
 create mode 100644 phlex/core/declared_translator.cpp
 create mode 100644 phlex/core/declared_translator.hpp
 create mode 100644 phlex/core/translator_node.hpp

diff --git a/phlex/core/CMakeLists.txt b/phlex/core/CMakeLists.txt
index 0496d0e4e..e3a07dce9 100644
--- a/phlex/core/CMakeLists.txt
+++ b/phlex/core/CMakeLists.txt
@@ -10,6 +10,7 @@ cet_make_library(
   declared_output.cpp
   declared_predicate.cpp
   declared_transform.cpp
+  declared_translator.cpp
   declared_unfold.cpp
   detail/filter_impl.cpp
   detail/make_algorithm_name.cpp
@@ -49,6 +50,7 @@ install(
     declared_output.hpp
     declared_predicate.hpp
     declared_transform.hpp
+    declared_translator.hpp
     declared_unfold.hpp
     fold_join_node.hpp
     fold_node.hpp
@@ -76,6 +78,7 @@ install(
     resource_api.hpp
     source.hpp
     transform_node.hpp
+    translator_node.hpp
     unfold_node.hpp
     upstream_predicates.hpp
   DESTINATION include/phlex/core
diff --git a/phlex/core/concepts.hpp b/phlex/core/concepts.hpp
index dda035697..fc1e11232 100644
--- a/phlex/core/concepts.hpp
+++ b/phlex/core/concepts.hpp
@@ -20,6 +20,12 @@ namespace phlex::detail {
   template <typename T>
   concept sendable = requires { typename phlex::experimental::sendable_type<T>; };
 
+  template <typename T, std::size_t N>
+  concept exactly_n_input_parameters = number_parameters<T> == N;
+
+  template <typename T>
+  concept exactly_one_input_parameter = exactly_n_input_parameters<T, 1>;
+
   template <typename T, std::size_t N>
   concept at_least_n_input_parameters = number_parameters<T> >= N;
 
@@ -29,6 +35,9 @@ namespace phlex::detail {
   template <typename T>
   concept at_least_two_input_parameters = at_least_n_input_parameters<T, 2>;
 
+  template <typename T>
+  concept exactly_one_output_object = number_output_objects<T> == 1ull;
+
   template <typename T>
   concept at_least_one_output_object = number_output_objects<T> >= 1ull;
 
@@ -73,6 +82,9 @@ namespace phlex::detail {
 
   template <typename T>
   concept is_transform_like = at_least_one_input_parameter<T> && at_least_one_output_object<T>;
+
+  template <typename T>
+  concept is_translator_like = exactly_one_input_parameter<T> && exactly_one_output_object<T>;
 }
 
 #endif // PHLEX_CORE_CONCEPTS_HPP
diff --git a/phlex/core/declared_translator.cpp b/phlex/core/declared_translator.cpp
new file mode 100644
index 000000000..bf2c397d4
--- /dev/null
+++ b/phlex/core/declared_translator.cpp
@@ -0,0 +1,25 @@
+#include "phlex/core/declared_translator.hpp"
+
+#include "phlex/core/products_consumer.hpp"
+
+#include <oneapi/tbb/flow_graph.h>
+
+#include <string>
+#include <utility>
+#include <vector>
+
+namespace phlex::detail {
+  declared_translator::declared_translator(phlex::experimental::algorithm_name name,
+                                           std::vector<std::string> predicates,
+                                           product_selectors input_products,
+                                           tbb::flow::graph& graph) :
+    products_consumer{std::move(name),
+                      std::move(predicates),
+                      std::move(input_products),
+                      graph,
+                      require_layers::multi_input_only}
+  {
+  }
+
+  declared_translator::~declared_translator() = default;
+}
diff --git a/phlex/core/declared_translator.hpp b/phlex/core/declared_translator.hpp
new file mode 100644
index 000000000..65c25778c
--- /dev/null
+++ b/phlex/core/declared_translator.hpp
@@ -0,0 +1,41 @@
+#ifndef PHLEX_CORE_DECLARED_TRANSLATOR_HPP
+#define PHLEX_CORE_DECLARED_TRANSLATOR_HPP
+
+#include "phlex/core/producer.hpp"
+#include "phlex/core/product_selector.hpp"
+#include "phlex/core/products_consumer.hpp"
+#include "phlex/model/algorithm_name.hpp"
+#include "phlex/phlex_core_export.hpp"
+#include "phlex/utilities/simple_ptr_map.hpp"
+
+#include <oneapi/tbb/flow_graph.h>
+
+#include <cstddef>
+#include <memory>
+#include <string>
+#include <vector>
+
+namespace phlex::detail {
+
+  class PHLEX_CORE_EXPORT declared_translator : public products_consumer, public producer {
+  public:
+    declared_translator(phlex::experimental::algorithm_name name,
+                        std::vector<std::string> predicates,
+                        product_selectors input_products,
+                        tbb::flow::graph& graph);
+    ~declared_translator() override;
+
+    phlex::experimental::algorithm_name const& name() const noexcept override
+    {
+      return products_consumer::name();
+    }
+
+    virtual std::size_t product_count() const = 0;
+  };
+
+  using declared_translator_ptr = std::unique_ptr<declared_translator>;
+  using declared_translators = simple_ptr_map<declared_translator_ptr>;
+
+}
+
+#endif // PHLEX_CORE_DECLARED_TRANSLATOR_HPP
diff --git a/phlex/core/framework_graph.cpp b/phlex/core/framework_graph.cpp
index 69f9125c0..50e778efc 100644
--- a/phlex/core/framework_graph.cpp
+++ b/phlex/core/framework_graph.cpp
@@ -230,6 +230,7 @@ namespace phlex::detail {
     filters_.merge(internal_edges_for_predicates(graph_, nodes_.predicates, nodes_.folds));
     filters_.merge(internal_edges_for_predicates(graph_, nodes_.predicates, nodes_.unfolds));
     filters_.merge(internal_edges_for_predicates(graph_, nodes_.predicates, nodes_.transforms));
+    filters_.merge(internal_edges_for_predicates(graph_, nodes_.predicates, nodes_.translators));
   }
 
   void framework_graph::make_bookkeeping_edges()
diff --git a/phlex/core/framework_graph.hpp b/phlex/core/framework_graph.hpp
index b38eb45dc..34e491ae0 100644
--- a/phlex/core/framework_graph.hpp
+++ b/phlex/core/framework_graph.hpp
@@ -138,6 +138,13 @@ namespace phlex::detail {
       return make_glue().transform(name, std::move(f), c);
     }
 
+    auto translate(std::string_view name,
+                   is_translator_like auto f,
+                   concurrency c = concurrency::serial)
+    {
+      return make_glue().translate(name, std::move(f), c);
+    }
+
     auto provide(std::string_view name, auto f, concurrency c = concurrency::serial)
     {
       return make_glue().provide(name, std::move(f), c);
diff --git a/phlex/core/glue.hpp b/phlex/core/glue.hpp
index ae63b33b8..4926f5e52 100644
--- a/phlex/core/glue.hpp
+++ b/phlex/core/glue.hpp
@@ -10,6 +10,7 @@
 #include "phlex/core/resource_api.hpp"
 #include "phlex/core/source.hpp"
 #include "phlex/core/transform_node.hpp"
+#include "phlex/core/translator_node.hpp"
 #include "phlex/metaprogramming/delegate.hpp"
 #include "phlex/phlex_core_export.hpp"
 
@@ -146,6 +147,26 @@ namespace phlex::detail {
         resources_);
     }
 
+    // 'f' is a by-value sink: it is moved into algorithm_bits.  The clang-tidy
+    // warning to take it by const reference is a false positive.
+    template <typename FT>
+    auto translate(std::string_view name,
+                   FT f, // NOLINT(performance-unnecessary-value-param)
+                   concurrency c)
+    {
+      internal::verify_name(name, config_);
+      return make_registration<translator_node, declared_translator_ptr>(
+        config_,
+        name,
+        stage_,
+        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
     // warning to take it by const reference is a false positive.
     template <typename FT>
diff --git a/phlex/core/graph_proxy.hpp b/phlex/core/graph_proxy.hpp
index 4f8ad5917..c1210728d 100644
--- a/phlex/core/graph_proxy.hpp
+++ b/phlex/core/graph_proxy.hpp
@@ -97,6 +97,11 @@ namespace phlex::detail {
                    is_transform_like auto f, // NOLINT(performance-unnecessary-value-param)
                    concurrency c = concurrency::serial) const;
 
+    /// @brief Registers a translator node.
+    auto translate(std::string_view name,
+                   is_translator_like auto f, // NOLINT(performance-unnecessary-value-param)
+                   concurrency c = concurrency::serial) const;
+
     /// @brief Registers an unfold node.
     template <typename Unfolder>
     auto unfold(std::string_view name,
@@ -211,6 +216,14 @@ namespace phlex::detail {
     return create_glue().transform(name, std::move(f), c);
   }
 
+  template <typename T>
+  auto graph_proxy<T>::translate(std::string_view name,
+                                 is_translator_like auto f,
+                                 concurrency c) const
+  {
+    return create_glue().translate(name, std::move(f), c);
+  }
+
   template <typename T>
   template <typename Unfolder>
   auto graph_proxy<T>::unfold(std::string_view name,
diff --git a/phlex/core/node_catalog.cpp b/phlex/core/node_catalog.cpp
index c1d9c39a2..6fa15cd72 100644
--- a/phlex/core/node_catalog.cpp
+++ b/phlex/core/node_catalog.cpp
@@ -30,6 +30,7 @@ namespace phlex::detail {
     result.append_range(as_product_consumers(folds));
     result.append_range(as_product_consumers(unfolds));
     result.append_range(as_product_consumers(transforms));
+    result.append_range(as_product_consumers(translators));
     return result;
   }
 
@@ -51,6 +52,9 @@ namespace phlex::detail {
     if (auto* node = transforms.get(node_name)) {
       return node->num_calls();
     }
+    if (auto* node = translators.get(node_name)) {
+      return node->num_calls();
+    }
     if (auto* node = providers.get(node_name)) {
       return node->num_calls();
     }
@@ -69,6 +73,7 @@ namespace phlex::detail {
 
     std::vector<producer*> result;
     result.append_range(as_producers(transforms));
+    result.append_range(as_producers(translators));
     result.append_range(as_producers(folds));
     result.append_range(as_producers(unfolds));
     return producer_catalog{result};
diff --git a/phlex/core/node_catalog.hpp b/phlex/core/node_catalog.hpp
index 219f83f92..ad1242c6e 100644
--- a/phlex/core/node_catalog.hpp
+++ b/phlex/core/node_catalog.hpp
@@ -6,6 +6,7 @@
 #include "phlex/core/declared_output.hpp"
 #include "phlex/core/declared_predicate.hpp"
 #include "phlex/core/declared_transform.hpp"
+#include "phlex/core/declared_translator.hpp"
 #include "phlex/core/declared_unfold.hpp"
 #include "phlex/core/producer_catalog.hpp"
 #include "phlex/core/products_consumer.hpp"
@@ -23,7 +24,7 @@ namespace phlex::detail {
   // A node_catalog is the framework_graph's registry of all the algorithm nodes
   // that make up a Phlex data-processing application.  It holds every kind of
   // node the framework knows about (predicates, observers, outputs, folds,
-  // unfolds, transforms, and providers), grouped by type, and serves
+  // unfolds, transforms, translators, and providers), grouped by type, and serves
   // as the single source of node information used to build and run the flow
   // graph. It also holds the sources, which are factories for creating provider
   // nodes.
@@ -52,7 +53,7 @@ namespace phlex::detail {
     // There are three insertion paths, all of which ultimately insert a node
     // into one of the simple_ptr_map members:
     //
-    //   1. Algorithm registration (transform, predicate, observe, fold,
+    //   1. Algorithm registration (transform, translate, predicate, observe, fold,
     //      unfold, provide, output):  Each declaration creates a short-lived
     //      *_api builder holding a registrar bound to the appropriate map (see
     //      registrar_for()).  For all but output nodes, the registrar's
@@ -111,6 +112,7 @@ namespace phlex::detail {
     simple_ptr_map<declared_fold_ptr> folds;
     simple_ptr_map<declared_unfold_ptr> unfolds;
     simple_ptr_map<declared_transform_ptr> transforms;
+    simple_ptr_map<declared_translator_ptr> translators;
     simple_ptr_map<provider_node_ptr> providers;
     simple_ptr_map<source_ptr> sources;
 
@@ -134,6 +136,8 @@ namespace phlex::detail {
         return unfolds;
       } else if constexpr (std::is_same_v<Ptr, declared_transform_ptr>) {
         return transforms;
+      } else if constexpr (std::is_same_v<Ptr, declared_translator_ptr>) {
+        return translators;
       } else if constexpr (std::is_same_v<Ptr, provider_node_ptr>) {
         return providers;
       } else if constexpr (std::is_same_v<Ptr, source_ptr>) {
diff --git a/phlex/core/registration_api.hpp b/phlex/core/registration_api.hpp
index eace1f4ba..30c25cdde 100644
--- a/phlex/core/registration_api.hpp
+++ b/phlex/core/registration_api.hpp
@@ -10,6 +10,7 @@
 #include "phlex/core/predicate_node.hpp"
 #include "phlex/core/resource_api.hpp"
 #include "phlex/core/transform_node.hpp"
+#include "phlex/core/translator_node.hpp"
 #include "phlex/core/unfold_node.hpp"
 #include "phlex/core/upstream_predicates.hpp"
 #include "phlex/metaprogramming/delegate.hpp"
diff --git a/phlex/core/translator_node.hpp b/phlex/core/translator_node.hpp
new file mode 100644
index 000000000..33d4d6b9b
--- /dev/null
+++ b/phlex/core/translator_node.hpp
@@ -0,0 +1,139 @@
+#ifndef PHLEX_CORE_TRANSLATOR_NODE_HPP
+#define PHLEX_CORE_TRANSLATOR_NODE_HPP
+
+#include "phlex/core/concepts.hpp"
+#include "phlex/core/declared_translator.hpp"
+#include "phlex/core/fwd.hpp"
+#include "phlex/core/input_arguments.hpp"
+#include "phlex/core/message.hpp"
+#include "phlex/core/multilayer_join_node.hpp"
+#include "phlex/core/product_selector.hpp"
+#include "phlex/core/resource/catalog.hpp"
+#include "phlex/metaprogramming/type_deduction.hpp"
+#include "phlex/model/algorithm_name.hpp"
+#include "phlex/model/data_cell_index.hpp"
+#include "phlex/model/handle.hpp"
+#include "phlex/model/product_specification.hpp"
+#include "phlex/model/product_store.hpp"
+
+#include <gsl/pointers>
+#include <oneapi/tbb/concurrent_unordered_map.h>
+#include <oneapi/tbb/flow_graph.h>
+
+#include <atomic>
+#include <cstddef>
+#include <functional>
+#include <memory>
+#include <ranges>
+#include <string>
+#include <tuple>
+#include <type_traits>
+#include <utility>
+#include <vector>
+
+namespace phlex::detail {
+
+  template <typename AlgorithmBits>
+  class translator_node : public declared_translator {
+    using function_t = typename AlgorithmBits::bound_type;
+    using input_parameter_types = typename AlgorithmBits::input_parameter_types;
+
+    static constexpr auto num_inputs = AlgorithmBits::number_inputs;
+    static constexpr auto num_outputs = number_output_objects<function_t>;
+
+    static_assert(num_inputs == 1, "translator_node requires exactly one input");
+    static_assert(num_outputs == 1, "translator_node requires exactly one output");
+
+    static_assert(std::tuple_size_v<input_parameter_types> == 1,
+                  "translator_node requires exactly one input parameter");
+
+  public:
+    using node_ptr_type = declared_translator_ptr;
+    static constexpr auto number_output_products = num_outputs;
+
+    translator_node(phlex::experimental::algorithm_name algo_name,
+                    phlex::experimental::identifier stage,
+                    std::size_t concurrency,
+                    std::vector<std::string> predicates,
+                    tbb::flow::graph& g,
+                    AlgorithmBits alg,
+                    product_selectors input_products,
+                    std::vector<std::string> output,
+                    resource_catalog&) :
+      declared_translator{
+        std::move(algo_name), std::move(predicates), std::move(input_products), g},
+      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())},
+      translator_{
+        g,
+        concurrency,
+        [this, ft = alg.release_algorithm(), stage = std::move(stage)](
+          messages_t<num_inputs> const& messages, auto& output) {
+          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>{});
+          ++calls_;
+          ++product_count_[store->index()->layer_hash()];
+
+          products new_products{num_outputs};
+          new_products.add_all(output_, std::move(result));
+          auto new_store = std::make_shared<phlex::experimental::product_store>(
+            store->index(), gsl::not_null{&name()}, gsl::not_null{&stage}, std::move(new_products));
+
+          std::get<0>(output).try_put({.store = std::move(new_store), .id = message_id});
+        }}
+    {
+      if constexpr (num_inputs > 1ull) {
+        make_edge(join_, translator_);
+      }
+    }
+
+  private:
+    tbb::flow::receiver<message>& port_for(product_selector const& input_product) override
+    {
+      return receiver_for<num_inputs>(join_, input(), input_product, translator_);
+    }
+
+    tbb::flow::sender<message>& output_port() override
+    {
+      return tbb::flow::output_port<0>(translator_);
+    }
+    phlex::experimental::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...>)
+    {
+      if constexpr (num_inputs == 1ull) {
+        return std::invoke(ft, std::get<Is>(input_).retrieve(messages)...);
+      } else {
+        return std::invoke(ft, std::get<Is>(input_).retrieve(std::get<Is>(messages))...);
+      }
+    }
+
+    named_index_ports index_ports() final { return join_.index_ports(); }
+    std::size_t num_calls() const final { return calls_.load(); }
+    std::size_t product_count() const final
+    {
+      std::size_t result{};
+      for (auto const& count : product_count_ | std::views::values) {
+        result += count.load();
+      }
+      return result;
+    }
+
+    input_retriever_types<input_parameter_types> input_{input_arguments<input_parameter_types>()};
+    phlex::experimental::product_specifications output_;
+    join_or_none_t<num_inputs> join_;
+    tbb::flow::multifunction_node<messages_t<num_inputs>, message_tuple<1u>> translator_;
+    std::atomic<std::size_t> calls_;
+    tbb::concurrent_unordered_map<std::size_t, std::atomic<std::size_t>> product_count_;
+  };
+
+}
+
+#endif // PHLEX_CORE_TRANSLATOR_NODE_HPP
diff --git a/test/function_registration_test.cpp b/test/function_registration_test.cpp
index d4cbe9264..cff3dfdc5 100644
--- a/test/function_registration_test.cpp
+++ b/test/function_registration_test.cpp
@@ -109,3 +109,38 @@ TEST_CASE("Call non-framework functions", "[programming model]")
 
   g.execute();
 }
+
+TEST_CASE("Register and execute a translator", "[programming model]")
+{
+  auto g = phlex::detail::framework_graph::with_default_driver("translator-test");
+  data_cell_index const* input_index{};
+
+  g.provide("provide_number", provide_number)
+    .output_product("input", "number", "job", "input-stage");
+  g.translate("translate_number",
+              [&input_index](handle<int> number) {
+                input_index = &number.data_cell_index();
+                CHECK(number.stage() == "input-stage");
+                return *number * 0.5;
+              })
+    .input_family(product_selector{.creator = "input", .layer = "job", .suffix = "number"})
+    .output_product_suffixes("translated_number");
+  g.observe("verify_translation",
+            [&input_index](handle<double> number) {
+              CHECK(*number == 1.5);
+              CHECK(number.creator().algorithm == "translate_number");
+              CHECK(number.suffix() == "translated_number");
+              CHECK(number.layer() == "job");
+              CHECK(number.stage() == "translator-test");
+              REQUIRE(input_index != nullptr);
+              CHECK(&number.data_cell_index() == input_index);
+            })
+    .input_family(product_selector{
+      .creator = "translate_number", .layer = "job", .suffix = "translated_number"});
+
+  g.execute();
+
+  CHECK(g.execution_count("provide_number") == 1u);
+  CHECK(g.execution_count("translate_number") == 1u);
+  CHECK(g.execution_count("verify_translation") == 1u);
+}