feat: add finish(), the end-of-stream lifecycle hook

CHANGELOG.md

@@ -37,6 +37,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   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.
+- **`MessageFilter::finish()`** — end-of-stream hook, called once after the last
+  message, so that a stage holding state can flush it (write a report, close a
+  file, publish a result to the `GraphContext`). It defaults to a no-op and
+  produces no message. `DslFilterGraph::finish()` finishes every stage in run
+  order; `FilterGraph`, `JsonFilterGraph`, `AnyFilterChain`, `FanoutFilter`,
+  `JoinFilter` and nested graphs forward it. Every stage is finished even if one
+  throws; the first exception is rethrown afterwards. There is no `tick()`: use
+  an in-band tick message (see README).
 - **`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`
@@ -51,9 +59,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   the `std::shared_ptr` for composites that forward it.
 
 ### Changed
-- **Breaking:** `AnyMessageFilter` has a new pure virtual
-  `setContext(std::shared_ptr<GraphContext>)`; custom implementations must
-  forward the context to the stages they wrap.
+- **Breaking:** `AnyMessageFilter` has new pure virtuals
+  `setContext(std::shared_ptr<GraphContext>)` and `finish()`; custom
+  implementations must forward both to the stages they wrap. They are pure
+  rather than no-ops on purpose: a composite that forgot to forward them would
+  otherwise fail silently.
 - README, EXAMPLE.md and `apps/textPipeline` now describe runtime graphs in the
   DSL first; the JSON format is documented as a supported alternative.
 - `dsl::toMermaid` now uses generated node ids (instead of edge names such as

README.md

@@ -72,6 +72,9 @@ drops the message, and everything downstream of that edge is skipped.
 - **`GraphContext`** — a graph-scoped, type-keyed, thread-safe blackboard, so
   stages can share values without knowing who produced them. See
   [Graph context](#graph-context).
+- **`MessageFilter::finish()`** — the end-of-stream hook: a stage that
+  accumulates state flushes it when the graph is finished. See
+  [Ending a run](#ending-a-run).
 
 ### Runtime graphs in the text DSL
 
@@ -439,6 +442,57 @@ Custom composite stages override it to forward the context to their inner
 stages; `MessageFilter::sharedContext()` returns the `std::shared_ptr` to pass
 on.
 
+## Ending a run
+
+A stage acts when a message reaches it, so a stage that accumulates something
+(statistics, a batch, an open file) has no natural point at which to flush it.
+`MessageFilter::finish()` is that point:
+
+```cpp
+class WriteReport : public MessageFilter<Record>
+{
+public:
+    std::optional<Record> filter(Record&& record) override
+    {
+        mSeen.push_back(record);
+        return std::move(record);
+    }
+
+    void finish() override // called once, after the last message
+    {
+        writeJson(mPath, mSeen);
+    }
+
+private:
+    std::vector<Record> mSeen;
+    std::filesystem::path mPath;
+};
+```
+
+```cpp
+DslFilterGraph<Record, Void> graph(text);
+for (auto&& record : records) { graph.filter(std::move(record)); }
+graph.finish(); // finishes every stage, in run order
+```
+
+- `finish()` defaults to a no-op, so existing stages are unaffected.
+- Every graph type and composite stage (`FilterGraph`, `DslFilterGraph`,
+  `JsonFilterGraph`, `AnyFilterChain`, `FanoutFilter`, `JoinFilter`, nested
+  graphs) forwards it to its stages. A `DslFilterGraph` finishes its stages in
+  run order, so a stage is finished after the stages it reads from.
+- **Every stage is finished even if one of them throws**; the first exception is
+  rethrown once the others have run.
+- `finish()` produces no message, so nothing is routed downstream. A stage that
+  wants to hand a final result to the application publishes it through the
+  [graph context](#graph-context) (or its own side channel).
+- Calling `finish()` is the owner's decision: a graph that is never finished
+  never flushes, and finishing twice finishes every stage twice.
+
+There is no `tick()`. Time is domain-specific (event time, wall clock, a sensor
+clock), so a stage that must act while no messages arrive is better served by an
+**in-band tick message**: make the graph's input type a variant with a `Tick`
+alternative and feed ticks through the graph like any other message.
+
 ## JSON format
 
 The JSON format predates the DSL and remains fully supported; both use the

libs/filterGraph/core/filterGraph/AnyFilterChain.hpp

@@ -103,6 +103,16 @@ class AnyFilterChain : public AnyMessageFilter
         }
     }
 
+    void finish() override
+    {
+        detail::FinishScope finished;
+        for (auto& stage : mStages)
+        {
+            finished.run([&stage] { stage->finish(); });
+        }
+        finished.rethrow();
+    }
+
 private:
     std::vector<std::shared_ptr<AnyMessageFilter>> mStages;
     // Shared by the stages of this chain until a graph hands down its own, so

libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp

@@ -1,3 +1,3 @@
 #pragma once
 
 #include <filterGraph/core/filterGraph/MergeStage.hpp>
@@ -32,6 +32,10 @@
     // MessageFilter::setContext.
     virtual void setContext(std::shared_ptr<GraphContext> context) = 0;
 
+    // Finishes the wrapped stage(s); see MessageFilter::finish. Pure, like
+    // setContext, so that a composite cannot silently fail to forward it.
+    virtual void finish() = 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
@@ -77,6 +81,11 @@
         mFilter->setContext(std::move(context));
     }
 
+    void finish() override
+    {
+        mFilter->finish();
+    }
+
     MergeSlotTypes mergeInputTypes() const override
     {
         if (const auto* merge = dynamic_cast<const MergeStage*>(mFilter.get()))

libs/filterGraph/core/filterGraph/DslFilterGraph.hpp

@@ -1,3 +1,3 @@
 #pragma once
 
 #include <filterGraph/core/filterGraph/AnyMessageFilter.hpp>
@@ -484,6 +484,18 @@
         }
     }
 
+    // Finishes every stage in run order, so that a stage sees its upstream
+    // stages finished before itself.
+    void finish()
+    {
+        filterGraph::detail::FinishScope finished;
+        for (auto& stage : mStages)
+        {
+            finished.run([&stage] { stage.filter->finish(); });
+        }
+        finished.rethrow();
+    }
+
 private:
     static constexpr std::size_t kInputSlot = 0;
 
@@ -636,6 +648,14 @@
         MessageFilter<InputType, OutputType>::setContext(std::move(context));
     }
 
+    // Finishes every stage of the graph, in run order; nested graphs are
+    // stages, so they finish their own stages in turn. See
+    // MessageFilter::finish.
+    void finish() override
+    {
+        mPlan.finish();
+    }
+
     // The parsed graph, e.g. for dsl::toMermaid.
     const dsl::GraphProgram& program() const
     {

libs/filterGraph/core/filterGraph/FanoutFilter.hpp

@@ -66,6 +66,16 @@ class FanoutFilter : public MessageFilter<InputType, InputType>
         MessageFilter<InputType, InputType>::setContext(std::move(context));
     }
 
+    void finish() override
+    {
+        detail::FinishScope finished;
+        for (auto& receiver : mReceivers)
+        {
+            finished.run([&receiver] { receiver->finish(); });
+        }
+        finished.rethrow();
+    }
+
 private:
     std::vector<std::shared_ptr<AnyMessageFilter>> mReceivers;
 };

libs/filterGraph/core/filterGraph/FilterGraph.hpp

@@ -1,3 +1,3 @@
 #pragma once
 
 #include <filterGraph/core/filterGraph/MessageFilter.hpp>
@@ -46,6 +46,11 @@
         MessageFilter<InType, OutType>::setContext(std::move(context));
     }
 
+    void finish() override
+    {
+        mFilter->finish();
+    }
+
 private:
     std::shared_ptr<Filter> mFilter;
 };
@@ -83,6 +88,14 @@
         MessageFilter<InType, OutType>::setContext(std::move(context));
     }
 
+    void finish() override
+    {
+        detail::FinishScope finished;
+        finished.run([this] { mFilter->finish(); });
+        finished.run([this] { mNext.finish(); });
+        finished.rethrow();
+    }
+
 private:
     std::shared_ptr<Filter> mFilter;
     FilterGraph<Rest...>    mNext;

libs/filterGraph/core/filterGraph/JoinFilter.hpp

@@ -73,6 +73,16 @@ class JoinFilter : public MessageFilter<InputType, OutputType>
         MessageFilter<InputType, OutputType>::setContext(std::move(context));
     }
 
+    void finish() override
+    {
+        detail::FinishScope finished;
+        for (auto& path : mPaths)
+        {
+            finished.run([&path] { path->finish(); });
+        }
+        finished.rethrow();
+    }
+
     std::optional<OutputType> filter(InputType&& data) override
     {
         std::vector<std::any> gathered;

libs/filterGraph/core/filterGraph/JsonFilterGraph.hpp

@@ -1,3 +1,3 @@
 #pragma once
 
 #include <filterGraph/core/filterGraph/AnyFilterChain.hpp>
@@ -68,6 +68,11 @@
         MessageFilter<InputType, OutputType>::setContext(std::move(context));
     }
 
+    void finish() override
+    {
+        mChain.finish();
+    }
+
 private:
     AnyFilterChain mChain;
 };

libs/filterGraph/core/filterGraph/MessageFilter.hpp

@@ -1,14 +1,53 @@
 #pragma once
 
 #include <filterGraph/core/filterGraph/GraphContext.hpp>
 
+#include <exception>
 #include <functional>
 #include <memory>
 #include <optional>
 #include <utility>
 
 namespace filterGraph {
 
+namespace detail {
+
+// Finishes a group of stages: every stage is finished even if an earlier one
+// throws, and the first exception is rethrown afterwards. Composite stages use
+// it so that one failing stage cannot keep the others from flushing.
+class FinishScope
+{
+public:
+    template <typename Fn>
+    void run(Fn&& fn)
+    {
+        try
+        {
+            std::forward<Fn>(fn)();
+        }
+        catch (...)
+        {
+            if (!mFirstError)
+            {
+                mFirstError = std::current_exception();
+            }
+        }
+    }
+
+    void rethrow() const
+    {
+        if (mFirstError)
+        {
+            std::rethrow_exception(mFirstError);
+        }
+    }
+
+private:
+    std::exception_ptr mFirstError;
+};
+
+} // namespace detail
+
 // Base interface for a single processing stage. A MessageFilter consumes an
 // InputType (by rvalue reference, so it may move from / mutate it freely) and
 // produces an std::optional<OutputType>: returning std::nullopt allows a
@@ -31,6 +70,18 @@
     virtual ~MessageFilter()                                    = default;
     virtual std::optional<OutputType> filter(InputType&& input) = 0;
 
+    // Called once after the last message, so that a stage holding state can
+    // flush it: write a report, close a file, publish a result to the
+    // GraphContext. It produces no message, so nothing is routed downstream;
+    // stages that need to emit a final value should publish it through the
+    // context or their own side channel.
+    //
+    // A graph calls finish() on its stages when the graph itself is finished.
+    // Composite stages override this to forward it to their inner stages.
+    // Calling it is the owner's decision: a graph that is never finished simply
+    // never flushes, and finishing twice finishes every stage twice.
+    virtual void finish() {}
+
     // Composite stages override this to forward the context to their inner
     // stages. Not meant to be called while messages are being processed.
     // Passing nullptr installs a fresh empty context rather than none, so the