feat: add GraphContext, a graph-scoped side channel for stages

CHANGELOG.md

@@ -23,8 +23,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **`validateDslGraph<In, Out>(text)`** — every build-time diagnostic without
   throwing; `dsl::formatDiagnostic(s)` formats them as `line:column: message`.
 - `dsl::StageNode::fanIn` records whether a stage's inputs came from a group.
+- **`GraphContext`** (`GraphContext.hpp`) — a graph-scoped, type-keyed,
+  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>()`.
+- **`MessageFilter::setContext` / `context()` / `sharedContext()`** — stages
+  receive the graph's `GraphContext`. `FilterGraph`, `DslFilterGraph`,
+  `JsonFilterGraph`, `AnyFilterChain`, `FanoutFilter` and `JoinFilter` create an
+  empty context when they are built and forward it to their stages, so stages of
+  a graph share one context without any setup; `setContext` replaces it (with
+  nullptr meaning "a fresh empty one"). `context()` returns a `GraphContext&`
+  and is never null, so stages need no null check; `sharedContext()` hands out
+  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.
 - 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

@@ -69,6 +69,9 @@ drops the message, and everything downstream of that edge is skipped.
   runtime configuration overhead.
 - **`SinkFilter<InputType>`** — a generic terminal stage that forwards data to
   a caller-supplied `std::function` callback.
+- **`GraphContext`** — a graph-scoped, type-keyed, thread-safe blackboard, so
+  stages can share values without knowing who produced them. See
+  [Graph context](#graph-context).
 
 ### Runtime graphs in the text DSL
 
@@ -317,6 +320,78 @@ listing each stage's arguments.
   C++ template parameters; the types inside `GraphOutputs` are checked when
   they are read (`get<T>` throws `std::bad_any_cast` on a mismatch).
 
+## Graph context
+
+`GraphContext` (`GraphContext.hpp`) is a side channel for data that is not part
+of the message, such as a frame number published by one stage and read by
+another. The value's type is the key, so there is at most one value per type:
+
+```cpp
+struct FrameNo { std::uint64_t value; };   // wrap primitives in a dedicated type
+
+auto ctx = std::make_shared<GraphContext>();
+ctx->set(FrameNo{42});                                     // publish (replaces)
+std::optional<FrameNo> frame = ctx->get<FrameNo>();        // nullopt if unset
+FrameNo orDefault = ctx->getOr(FrameNo{0});
+ctx->update<FrameNo>([](FrameNo& f) { ++f.value; });       // atomic; false if unset
+```
+
+- Values are copied in and out under an internal lock, so a context can be
+  shared by concurrently running paths. Store `std::shared_ptr<T>` for heavy or
+  non-copyable data. The callback given to `update` must not access the context.
+- `GraphContext` is a polymorphic base: derive an application context, pass it
+  around as `std::shared_ptr<GraphContext>`, and recover it with
+  `ctx->as<AppContext>()` (`nullptr` if it is another type). Members added by a
+  derived class are not covered by the lock, and a stage calling `as<>` depends
+  on that type, so reusable stages should stick to `set` / `get`.
+- The context is graph-scoped, not per message: if the graph ever buffers or
+  reorders messages, a value such as a frame number may belong to another
+  message than the one being processed.
+
+There is always a context, so a stage never has to check for one. A stage
+reads it through `MessageFilter::context()`, which returns a `GraphContext&`:
+
+```cpp
+class StampFrame : public MessageFilter<Image>
+{
+public:
+    std::optional<Image> filter(Image&& image) override
+    {
+        image.frame = context().getOr(FrameNo{0}).value;
+        return std::move(image);
+    }
+};
+```
+
+Every graph type (`FilterGraph`, `DslFilterGraph`, `JsonFilterGraph`) creates
+an empty context when it is built and hands it to its stages, and so does every
+composite stage (`FanoutFilter`, `JoinFilter`, nested graphs), including to
+receivers and paths added later. So stages of the same graph share one context
+out of the box:
+
+```cpp
+DslFilterGraph<Image, Image> graph(text);
+graph.context().set(FrameNo{1});     // the graph's own context
+```
+
+Call `setContext` to put a different context in its place — typically an
+application's derived one, or a single context shared by several graphs:
+
+```cpp
+graph.setContext(ctx);               // replaces the graph's own context
+```
+
+A stage that is not part of a graph keeps its own empty context, which nobody
+else sees: it works, but nothing is shared. Note that the graph a stage is
+added to overwrites the stage's context with its own, so hand a shared context
+to the graph rather than to individual stages. `setContext(nullptr)` installs a
+fresh empty context rather than none.
+
+`setContext` is not meant to be called while messages are being processed.
+Custom composite stages override it to forward the context to their inner
+stages; `MessageFilter::sharedContext()` returns the `std::shared_ptr` to pass
+on.
+
 ## JSON format
 
 The JSON format predates the DSL and remains fully supported; both use the

libs/filterGraph/CMakeLists.txt

@@ -20,6 +20,7 @@ target_sources(${PROJECT_NAME}
     core/filterGraph/GraphLangLexy.hpp
     core/filterGraph/GraphLangHandwritten.hpp
     core/filterGraph/DslFilterGraph.hpp
+    core/filterGraph/GraphContext.hpp
 )
 
 target_link_libraries(${PROJECT_NAME}

libs/filterGraph/core/filterGraph/AnyFilterChain.hpp

@@ -61,6 +61,8 @@ class AnyFilterChain : public AnyMessageFilter
 
             mStages.push_back(std::move(stage));
         }
+
+        AnyFilterChain::setContext(mContext);
     }
 
     std::optional<std::any> filter(std::any&& input) override
@@ -92,8 +94,20 @@ class AnyFilterChain : public AnyMessageFilter
         return mStages.back()->outputType();
     }
 
+    void setContext(std::shared_ptr<GraphContext> context) override
+    {
+        mContext = context ? std::move(context) : std::make_shared<GraphContext>();
+        for (auto& stage : mStages)
+        {
+            stage->setContext(mContext);
+        }
+    }
+
 private:
     std::vector<std::shared_ptr<AnyMessageFilter>> mStages;
+    // Shared by the stages of this chain until a graph hands down its own, so
+    // that a chain built on its own still has one context, not one per stage.
+    std::shared_ptr<GraphContext> mContext = std::make_shared<GraphContext>();
 };
 
 } // namespace filterGraph

libs/filterGraph/core/filterGraph/AnyMessageFilter.hpp

@@ -1,3 +1,3 @@
 #pragma once
 
 #include <filterGraph/core/filterGraph/MessageFilter.hpp>
@@ -26,6 +26,10 @@
 
     virtual std::type_index inputType() const  = 0;
     virtual std::type_index outputType() const = 0;
+
+    // Hands the graph's context to the wrapped stage(s); see
+    // MessageFilter::setContext.
+    virtual void setContext(std::shared_ptr<GraphContext> context) = 0;
 };
 
 // Adapts a concrete MessageFilter<Filter::InType, Filter::OutType> to the
@@ -60,6 +64,11 @@
         return typeid(typename Filter::OutType);
     }
 
+    void setContext(std::shared_ptr<GraphContext> context) override
+    {
+        mFilter->setContext(std::move(context));
+    }
+
 private:
     std::shared_ptr<Filter> mFilter;
 };

libs/filterGraph/core/filterGraph/DslFilterGraph.hpp

@@ -431,6 +431,14 @@ class GraphPlan
         return mOutputKeys;
     }
 
+    void setContext(const std::shared_ptr<GraphContext>& context)
+    {
+        for (auto& stage : mStages)
+        {
+            stage.filter->setContext(context);
+        }
+    }
+
 private:
     static constexpr std::size_t kInputSlot = 0;
 
@@ -544,6 +552,8 @@ class DslFilterGraph : public MessageFilter<InputType, OutputType>
             dsl::detail::sortDiagnostics(diagnostics);
             throw GraphError(std::move(diagnostics));
         }
+
+        DslFilterGraph::setContext(this->sharedContext());
     }
 
     std::optional<OutputType> filter(InputType&& data) override
@@ -575,6 +585,12 @@ class DslFilterGraph : public MessageFilter<InputType, OutputType>
         }
     }
 
+    void setContext(std::shared_ptr<GraphContext> context) override
+    {
+        mPlan.setContext(context);
+        MessageFilter<InputType, OutputType>::setContext(std::move(context));
+    }
+
     // The parsed graph, e.g. for dsl::toMermaid.
     const dsl::GraphProgram& program() const
     {

libs/filterGraph/core/filterGraph/FanoutFilter.hpp

@@ -43,6 +43,7 @@ class FanoutFilter : public MessageFilter<InputType, InputType>
                 receiver->inputType().name(),
                 typeid(InputType).name()));
         }
+        receiver->setContext(this->sharedContext());
         mReceivers.push_back(std::move(receiver));
     }
 
@@ -56,6 +57,15 @@ class FanoutFilter : public MessageFilter<InputType, InputType>
         return std::move(data);
     }
 
+    void setContext(std::shared_ptr<GraphContext> context) override
+    {
+        for (auto& receiver : mReceivers)
+        {
+            receiver->setContext(context);
+        }
+        MessageFilter<InputType, InputType>::setContext(std::move(context));
+    }
+
 private:
     std::vector<std::shared_ptr<AnyMessageFilter>> mReceivers;
 };

libs/filterGraph/core/filterGraph/FilterGraph.hpp

@@ -32,13 +32,20 @@ class FilterGraph<Filter> : public MessageFilter<typename Filter::InType, typena
     explicit FilterGraph(std::shared_ptr<Filter> filter)
         : mFilter(std::move(filter))
     {
+        FilterGraph::setContext(this->sharedContext());
     }
 
     std::optional<OutType> filter(InType&& data) override
     {
         return mFilter->filter(std::move(data));
     }
 
+    void setContext(std::shared_ptr<GraphContext> context) override
+    {
+        mFilter->setContext(context);
+        MessageFilter<InType, OutType>::setContext(std::move(context));
+    }
+
 private:
     std::shared_ptr<Filter> mFilter;
 };
@@ -56,6 +63,7 @@ class FilterGraph<Filter, Rest...> : public MessageFilter<typename Filter::InTyp
         : mFilter(std::move(filter))
         , mNext(std::move(rest)...)
     {
+        FilterGraph::setContext(this->sharedContext());
     }
 
     std::optional<OutType> filter(InType&& data) override
@@ -68,6 +76,13 @@ class FilterGraph<Filter, Rest...> : public MessageFilter<typename Filter::InTyp
         return mNext.filter(std::move(*intermediate));
     }
 
+    void setContext(std::shared_ptr<GraphContext> context) override
+    {
+        mFilter->setContext(context);
+        mNext.setContext(context);
+        MessageFilter<InType, OutType>::setContext(std::move(context));
+    }
+
 private:
     std::shared_ptr<Filter> mFilter;
     FilterGraph<Rest...>    mNext;