feat: type-checked merge inputs

CHANGELOG.md

@@ -27,6 +27,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   thread-safe blackboard for side-channel data between stages
   (`set` / `get` / `getOr` / `contains` / `erase` / `update` / `clear`). It is a
   polymorphic base; derived contexts are recovered with `as<Derived>()`.
+- **`TypedMergeFilter<OutputType, InputTypes...>`** /
+  **`UniformMergeFilter<InputType, OutputType>`** /
+  **`registerTypedMergeFilter`** (`MergeFilter.hpp`) — merge stages that declare
+  their slot types. The DSL checks the edges of their fan-in group when the
+  graph is built (`slot 2 of 'Summarize' expects ... but edge 'msg' carries ...`,
+  `stage 'Summarize' takes 2 inputs but the group has 3`), so a mis-wired merge
+  no longer fails with a `std::bad_any_cast` on the first message. The stages
+  receive their slots as typed `std::optional`s (`std::nullopt` for a hole),
+  with no `any_cast` in user code. `MergeFilter` / `registerMergeFilter` declare
+  no slot types and keep their current, unchecked behaviour.
+- **`MergeStage` / `MergeSlotTypes`** (`MergeStage.hpp`, new header) — how a
+  merge stage declares its slot types; `AnyMessageFilter::mergeInputTypes()`
+  exposes them to the DSL. `MergeInputs` moved here from `MergeFilter.hpp`
+  (which still provides it).
 - **`MessageFilter::setContext` / `context()` / `sharedContext()`** — stages
   receive the graph's `GraphContext`. `FilterGraph`, `DslFilterGraph`,
   `JsonFilterGraph`, `AnyFilterChain`, `FanoutFilter` and `JoinFilter` create an

EXAMPLE.md

@@ -217,6 +217,56 @@ The `MinLength=100` path drops the 19-character input, so its slot is a
 the missing path. A merge is skipped only when *every* one of its edges is
 empty — then there is nothing to combine, and the drop propagates.
 
+### Typed slots
+
+`Concat` takes its slots untyped: nothing checks that the group really carries
+three strings, and swapping two edges of the same type would pass validation
+and quietly produce a different result. A merge that declares its slot types
+avoids both. `TypedMergeFilter<Out, Ins...>` fixes the number and type of the
+slots and hands them over as `std::optional`s:
+
+```cpp
+class ReportFilter : public TypedMergeFilter<std::string, std::string, std::size_t>
+{
+public:
+    std::optional<std::string> merge(std::optional<std::string>&& upper,
+                                     std::optional<std::size_t>&& length) override
+    {
+        return std::format("{} ({} chars)", upper.value_or("-"), length.value_or(0));
+    }
+};
+static FilterRegistrar<ReportFilter> registerReport("Report");
+```
+
+```cpp
+DslFilterGraph<std::string, int> pipeline(R"dsl(
+    in -> Uppercase -> upper
+    in -> Length -> length
+    (upper, length) -> Report -> reported -> Print(prefix="[typed] ") -> out
+)dsl");
+
+pipeline.filter(std::string{"Hello, filterGraph!"});
+```
+
+```text
+[typed] HELLO, FILTERGRAPH! (19 chars)
+```
+
+Writing the group the other way round is now rejected when the graph is built,
+instead of throwing `std::bad_any_cast` on the first message:
+
+```text
+3:20: slot 1 of 'Report' expects 'class std::basic_string<char,...>' but edge 'length' carries 'unsigned __int64'
+3:20: slot 2 of 'Report' expects 'unsigned __int64' but edge 'upper' carries 'class std::basic_string<char,...>'
+```
+
+(The type names come from `typeid(...).name()`, so their spelling depends on the
+compiler; they are shortened here.)
+
+`registerTypedMergeFilter<Out, Ins...>(name, lambda)` does the same from a
+lambda, and `UniformMergeFilter<In, Out>` covers the other shape: any number of
+slots, all of the same type (N variants of one computation, combined).
+
 ## 5. Several named outputs
 
 A graph can have more than one output. `out.<key>` names each one, and the
@@ -312,7 +362,9 @@ What gets checked:
   both types (type names come from `typeid`, so their spelling depends on the
   compiler).
 - **Fan-in and `Void`** — groups must feed merge stages, merge stages need a
-  group, and a stage producing `Void` must route to `end`.
+  group, and a stage producing `Void` must route to `end`. A merge that
+  declares its slot types (`TypedMergeFilter`, `UniformMergeFilter`) also has
+  the number and type of its group's edges checked.
 - **Outputs** — the number and type of outputs must fit the graph's
   `OutputType`.
 

README.md

@@ -89,6 +89,10 @@ drops the message, and everything downstream of that edge is skipped.
 - **`MergeFilter<OutputType>`** / **`registerMergeFilter`** — fan-in stages:
   a C++ combiner turns the values of several edges into one, seeing an empty
   slot ("hole") for every edge whose path dropped the message.
+- **`TypedMergeFilter<Out, Ins...>`** / **`UniformMergeFilter<In, Out>`** /
+  **`registerTypedMergeFilter`** — fan-in stages that declare their slot types,
+  so the edges of a group are checked when the graph is built and the stage
+  receives typed `std::optional`s instead of `std::any`.
 - **`validateDslGraph<In, Out>(text)`** — the same checks as construction,
   returned as a list of `line:column` diagnostics instead of a thrown
   **`GraphError`**.
@@ -241,9 +245,10 @@ edge -> Stage -> edge -> Stage(key=value) -> edge
 - **Fan-out:** read the same edge in several statements. Every reader gets its
   own copy of the value.
 - **Fan-in:** `(a, b, c) -> Merge -> merged` gathers several edges into a merge
-  stage registered with `registerMergeFilter<OutputType>(name, combiner)`. The
-  combiner receives `MergeInputs` (`std::vector<std::any>`), one slot per edge,
-  in the order listed.
+  stage, one slot per edge, in the order listed. A merge either declares its
+  slot types (see [Typed merges](#typed-merges)) or takes them untyped as
+  `MergeInputs` (`std::vector<std::any>`), as
+  `registerMergeFilter<OutputType>(name, combiner)` does.
 - `#` starts a comment that runs to the end of the line.
 
 ### How a graph runs
@@ -260,6 +265,48 @@ For every message:
    skipped only when *all* of its edges are empty.
 5. Values routed to `end` are discarded.
 
+### Typed merges
+
+A merge stage may declare the type of each of its slots. The DSL then checks
+the edges of its group when the graph is built, like every other edge, instead
+of failing with a `std::bad_any_cast` on the first message — and the stage
+receives typed `std::optional`s, so it needs no `any_cast` of its own:
+
+```cpp
+// Fixed arity, one type per slot.
+class Summarize : public TypedMergeFilter<Stats, Message, Valid>
+{
+public:
+    std::optional<Stats> merge(std::optional<Message>&& message, std::optional<Valid>&& valid) override
+    {
+        return Stats{...}; // an empty optional is a hole: that path dropped the message
+    }
+};
+static FilterRegistrar<Summarize> registerSummarize("Summarize");
+
+// The same from a lambda, without a subclass.
+registerTypedMergeFilter<Stats, Message, Valid>(
+    "Summarize",
+    [](std::optional<Message>&& message, std::optional<Valid>&& valid) -> std::optional<Stats> { ... });
+
+// Any number of slots, all of one type: N variants of the same computation.
+class PickBest : public UniformMergeFilter<Candidate, Result>
+{
+public:
+    std::optional<Result> merge(std::vector<std::optional<Candidate>>&& candidates) override { ... }
+};
+```
+
+Mis-wiring a group is then a build-time diagnostic, not a wrong result:
+
+```text
+4:34: slot 2 of 'Summarize' expects 'struct Valid' but edge 'msg' carries 'struct Message'
+4:34: stage 'Summarize' takes 2 inputs but the group has 3
+```
+
+`MergeFilter` / `registerMergeFilter` declare no slot types; their edges stay
+unchecked, and the combiner reads the slots with `std::any_cast`.
+
 ### Choosing the output type
 
 `DslFilterGraph<InputType, OutputType>`'s `OutputType` states what the graph's

apps/textPipeline/main.cpp

@@ -1,3 +1,3 @@
 // Generic, domain-unrelated example: a small text-processing pipeline that
 // demonstrates the core building blocks of filterGraph:
 //  - MessageFilter: the base stage interface
@@ -31,6 +31,7 @@
 using filterGraph::JsonFilterGraph;
 using filterGraph::MergeInputs;
 using filterGraph::MessageFilter;
+using filterGraph::TypedMergeFilter;
 using filterGraph::registerFanoutFilter;
 using filterGraph::registerMergeFilter;
 using filterGraph::validateDslGraph;
@@ -157,6 +158,21 @@
     return true;
 }();
 
+// A typed merge stage for `(upper, length) -> Report`: it declares the type of
+// each slot, so the DSL checks the group's edges when the graph is built, and
+// the slots arrive as std::optionals (empty = a hole) instead of std::any.
+class ReportFilter : public TypedMergeFilter<std::string, std::string, std::size_t>
+{
+public:
+    std::optional<std::string> merge(std::optional<std::string>&& upper,
+                                     std::optional<std::size_t>&& length) override
+    {
+        return std::format("{} ({} chars)", upper.value_or("-"), length.value_or(0));
+    }
+};
+
+static FilterRegistrar<ReportFilter> registerReport("Report");
+
 // Only needed by the JSON example: in the DSL, fan-out is built in.
 static const bool sRegisterFanout = [] {
     registerFanoutFilter<std::string>("Fanout");
@@ -203,6 +219,27 @@
         pipeline.filter(std::string{"Hello, filterGraph!"});
     }
 
+    // 3b) The same fan-in with a typed merge: Report declares its slot types,
+    //     so a group in the wrong order is rejected when the graph is built
+    //     instead of throwing std::bad_any_cast on the first message.
+    {
+        DslFilterGraph<std::string, int> pipeline(R"dsl(
+            in -> Uppercase -> upper
+            in -> Length -> length
+            (upper, length) -> Report -> reported -> Print(prefix="[typed] ") -> out
+        )dsl");
+
+        pipeline.filter(std::string{"Hello, filterGraph!"});
+
+        for (const auto& diagnostic : validateDslGraph<std::string, int>(
+                 "in -> Uppercase -> upper\n"
+                 "in -> Length -> length\n"
+                 "(length, upper) -> Report -> reported -> Print -> out\n"))
+        {
+            std::cout << "[typed check] " << dsl::formatDiagnostic(diagnostic) << '\n';
+        }
+    }
+
     // 4) Several named outputs of different types, returned as GraphOutputs.
     {
         DslFilterGraph<std::string> analysis(R"dsl(

libs/filterGraph/CMakeLists.txt

@@ -14,6 +14,7 @@ target_sources(${PROJECT_NAME}
     core/filterGraph/JsonFilterGraph.hpp
     core/filterGraph/FanoutFilter.hpp
     core/filterGraph/JoinFilter.hpp
+    core/filterGraph/MergeStage.hpp
     core/filterGraph/MergeFilter.hpp
     core/filterGraph/GraphValidator.hpp
     core/filterGraph/GraphLang.hpp

libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <filterGraph/core/filterGraph/MergeStage.hpp>
 #include <filterGraph/core/filterGraph/MessageFilter.hpp>
 
 #include <any>
@@ -30,6 +31,13 @@
     // Hands the graph's context to the wrapped stage(s); see
     // MessageFilter::setContext.
     virtual void setContext(std::shared_ptr<GraphContext> context) = 0;
+
+    // The slot types of a merge stage (see MergeStage); empty for an untyped
+    // merge and for every stage that is not a merge.
+    virtual MergeSlotTypes mergeInputTypes() const
+    {
+        return {};
+    }
 };
 
 // Adapts a concrete MessageFilter<Filter::InType, Filter::OutType> to the
@@ -69,6 +77,15 @@
         mFilter->setContext(std::move(context));
     }
 
+    MergeSlotTypes mergeInputTypes() const override
+    {
+        if (const auto* merge = dynamic_cast<const MergeStage*>(mFilter.get()))
+        {
+            return merge->mergeInputTypes();
+        }
+        return {};
+    }
+
 private:
     std::shared_ptr<Filter> mFilter;
 };

libs/filterGraph/core/filterGraph/DslFilterGraph.hpp

@@ -1,3 +1,3 @@
 #pragma once
 
 #include <filterGraph/core/filterGraph/AnyMessageFilter.hpp>
@@ -229,6 +229,47 @@
     return std::format("edge '{}'", edge);
 }
 
+// Checks the edges of a fan-in group against the slot types the merge stage
+// declares (MergeStage); an untyped merge declares none and is not checked.
+inline void checkMergeInputs(const StageNode&                                        node,
+                             const MergeSlotTypes&                                   slots,
+                             const std::unordered_map<std::string, std::type_index>& edgeTypes,
+                             std::vector<TextDiagnostic>&                            diagnostics)
+{
+    if (slots.empty())
+    {
+        return;
+    }
+
+    if (!slots.uniform && slots.types.size() != node.inputs.size())
+    {
+        // Without a slot-to-edge correspondence, per-slot messages would only
+        // repeat this one.
+        diagnostics.push_back({node.loc,
+                               std::format("stage '{}' takes {} inputs but the group has {}",
+                                           node.type,
+                                           slots.types.size(),
+                                           node.inputs.size())});
+        return;
+    }
+
+    for (std::size_t slot = 0; slot < node.inputs.size(); ++slot)
+    {
+        const std::type_index expected = slots.uniform ? slots.types.front() : slots.types[slot];
+        const auto            type     = edgeTypes.find(node.inputs[slot]);
+        if (type != edgeTypes.end() && type->second != expected)
+        {
+            diagnostics.push_back({node.loc,
+                                   std::format("slot {} of '{}' expects '{}' but {} carries '{}'",
+                                               slot + 1,
+                                               node.type,
+                                               expected.name(),
+                                               describeEdge(node.inputs[slot]),
+                                               type->second.name())});
+        }
+    }
+}
+
 struct CompiledStage
 {
     std::shared_ptr<AnyMessageFilter> filter;
@@ -309,7 +350,11 @@
                                  node.type,
                                  node.type)});
             }
-            else if (!merge)
+            else if (merge)
+            {
+                checkMergeInputs(node, filter->mergeInputTypes(), edgeTypes, diagnostics);
+            }
+            else
             {
                 const std::string& edge = node.inputs.front();
                 auto               type = edgeTypes.find(edge);