Implement event handler system for message passing and callbacks (#587)

* feat: add event handler system for message passing procedure

Introduce a new event handler mechanism that allows users to hook into
the message passing procedure via `Event`, `handle_event`, and
`broadcast_event` functions. Events are broadcast before and after
message rule calls, enabling debugging and monitoring capabilities.

Key changes:
- Add `event_handler.jl` with event types and handler interfaces
- Extend `MessageMapping` to include an event handler parameter
- Support both custom handler types and NamedTuple-based handlers
- Add documentation for the events system
- Include comprehensive tests for event handler functionality

* type in docstring

* inject the event handler in the factor node activation options

* make format

* document the existing events better

* rename event handler to callbacks to better match RxInfer

* update docs

* use make format

* fix typo

* allow to merge callbacks

* allow reduce the result of the callbacks

* add per-event callback reducer

* make format

* remove test method

* start reimplement of the product of two messages

this is required to inject the callbacks properly, plus we do the breaking
release already. Could break (and improve) more things as well then

* fix initial integration with RxInfer.jl

* fix Aqua tests

* caught a small bug in RxInfer tests

* caught another bug from RxInfer.jl tests

* merge stricter formatting

* Refactor the variables, add docstrings, add labels

- `compute_product_of_messages` now accepts the `AbstractVariable`,
  makes it easier to identify the variable inside the callback

* fix documentation build

* 2prev

* add before/after product of two messages callbacks

* fix failing tests

* add before/after product of messages callbacks. add before/after form constraints

* add before/after marginal compute callbacks

* add error hint in case of wrong passed callbacks

* fix test

* temporary fix for the logscale switch rule

* support Dict as callback handler

* fix warnings in the tests

* make format

* refactor to use an abstract Event{E} structure

* add new method for event_name

* use mutable fields instead

* update CHANGELOG

Author: bvdmitri

CHANGELOG.md

@@ -7,9 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- Callback/event system for hooking into message passing steps (rule calls, message products, form constraints, marginal computation)
+- `MessageProductContext` struct to bundle product computation settings and callbacks
+- Labels for variables (`RandomVariable`, `ConstVariable`, `DataVariable`)
+- Docstrings for variable types, form constraints, and related functions
+- Documentation page for callbacks
+- `MethodError` hint for mismatched `handle_event` signatures
+
 ### Changed
 - Switched from `ReTestItems` to `TestItemRunner` for tests ([#584](https://github.com/ReactiveBayes/ReactiveMP.jl/pull/584))
 - Made formatting checks stricter
+- Renamed `variables/variable.jl` to `variables/generic.jl`
+- Replaced hardcoded `DefaultMessageProdFn`/`DefaultMarginalProdFn` with `MessageProductContext`
 
 ## [5.6.6] - 2026-03-13
 

docs/make.jl

@@ -14,9 +14,11 @@ makedocs(
         "Introduction"    => "index.md",
         "Library" => [
             "Factor nodes"         => "lib/nodes.md",
+            "Variables"             => "lib/variables.md",
             "Messages"             => "lib/message.md",
             "Marginals"            => "lib/marginal.md",
             "Message update rules" => "lib/rules.md",
+            "Callbacks"            => "lib/callbacks.md",
             "Helper utils"         => "lib/helpers.md",
             "Algebra utils"        => "lib/algebra.md",
             "Specific factor nodes" => [

docs/src/lib/callbacks.md

@@ -0,0 +1,56 @@
+# Callbacks in the Message Passing Procedure
+
+ReactiveMP provides a way to "hook" into the message passing procedure and listen to various events
+via "callbacks". This can be useful, for example, to debug messages or monitor the order of computations.
+
+```@docs
+ReactiveMP.Event
+ReactiveMP.event_name
+ReactiveMP.handle_event
+ReactiveMP.invoke_callback
+ReactiveMP.merge_callbacks
+ReactiveMP.MergedCallbacks
+```
+
+## Event naming convention
+
+Every event in ReactiveMP is a concrete subtype of [`ReactiveMP.Event{E}`](@ref) where `E` is a `Symbol` identifying the event.
+The naming convention is straightforward: for an event identified by the symbol `:event_name`, the corresponding struct is called `EventNameEvent`.
+For example:
+
+| Symbol | Struct |
+|--------|--------|
+| `:before_message_rule_call` | [`ReactiveMP.BeforeMessageRuleCallEvent`](@ref) |
+| `:after_product_of_two_messages` | [`ReactiveMP.AfterProductOfTwoMessagesEvent`](@ref) |
+| `:before_form_constraint_applied` | [`ReactiveMP.BeforeFormConstraintAppliedEvent`](@ref) |
+
+Each event struct carries the relevant data as fields, so you can inspect what happened during inference.
+You can use [`ReactiveMP.event_name`](@ref) to retrieve the symbol from any event type:
+
+```@example callbacks
+using ReactiveMP #hide
+ReactiveMP.event_name(ReactiveMP.BeforeProductOfTwoMessagesEvent)
+```
+
+To see which fields an event carries, use the standard Julia introspection:
+
+```julia
+julia> ?ReactiveMP.BeforeProductOfTwoMessagesEvent
+```
+
+## All defined events
+
+Here is the list of predefined event types, to which a custom callback handler can react to.
+
+```@docs
+ReactiveMP.BeforeMessageRuleCallEvent
+ReactiveMP.AfterMessageRuleCallEvent
+ReactiveMP.BeforeProductOfTwoMessagesEvent
+ReactiveMP.AfterProductOfTwoMessagesEvent
+ReactiveMP.BeforeProductOfMessagesEvent
+ReactiveMP.AfterProductOfMessagesEvent
+ReactiveMP.BeforeFormConstraintAppliedEvent
+ReactiveMP.AfterFormConstraintAppliedEvent
+ReactiveMP.BeforeMarginalComputationEvent
+ReactiveMP.AfterMarginalComputationEvent
+```

docs/src/lib/message.md

@@ -73,13 +73,15 @@ is_clamped(message), is_initial(message)
 ### [Product of messages](@id lib-messages-product)
 
 In message passing framework, in order to compute a posterior we must compute a normalized product of two messages.
-For this purpose the `ReactiveMP.jl` uses the `multiply_messages` function, which internally uses the `prod` function
-defined in `BayesBase.jl` with various product strategies. We refer an interested reader to the documentation of the 
-`BayesBase.jl` package for more information.
+For this purpose the `ReactiveMP.jl` uses the [`ReactiveMP.MessageProductContext`](@ref) structure, together with the [`ReactiveMP.compute_product_of_messages`](@ref) and [`ReactiveMP.compute_product_of_two_messages`](@ref) functions. Both functions accept a [`ReactiveMP.AbstractVariable`](@ref) as the first argument to identify which variable the product is being computed for — this is useful for callbacks (e.g. [`ReactiveMP.BeforeProductOfTwoMessagesEvent`](@ref)). The [`ReactiveMP.compute_product_of_two_messages`](@ref) function internally uses the `prod` function
+defined in `BayesBase.jl` with various product strategies. We refer an interested reader to the documentation of the `BayesBase.jl` package for more information.
 
 ```@docs
-ReactiveMP.multiply_messages
-ReactiveMP.messages_prod_fn
+ReactiveMP.MessageProductContext
+ReactiveMP.compute_product_of_two_messages
+ReactiveMP.compute_product_of_messages
+ReactiveMP.MessagesProductFromLeftToRight
+ReactiveMP.MessagesProductFromRightToLeft
 ```
 
 
@@ -95,4 +97,4 @@ A *message mapping* defines how messages are transformed or mapped during the pr
 
 ```@docs
 ReactiveMP.MessageMapping
-```
+```

docs/src/lib/nodes.md

@@ -113,27 +113,6 @@ ReactiveMP.RequireMarginalFunctionalDependencies
 ReactiveMP.RequireEverythingFunctionalDependencies
 ```
 
-## [Customizing Dependencies with Metadata](@id lib-node-metadata-dependencies)
-
-The functional dependencies of a node can be customized at runtime using options during node activation. This allows for runtime customization of the functional dependencies, e.g. to test different message passing schemes or implement specialized behavior for specific instances of a node type:
-
-```julia
-# Define custom dependencies based on metadata
-function ReactiveMP.collect_functional_dependencies(::Type{MyNode}, options::FactorNodeActivationOptions)
-    if some_condition(options) # a user can specify dependencies based, for example, on metadata
-        return CustomDependencies()
-    end
-    # Fall back to default dependencies
-    return ReactiveMP.collect_functional_dependencies(MyNode, getdependecies(options))
-end
-
-# Use custom dependencies during activation
-node = factornode(MyNode, ...)
-activate!(node, FactorNodeActivationOptions(:custom_behavior, ...))
-```
-
-This feature is particularly useful for testing different message passing schemes or implementing specialized behavior for specific instances of a node type.
-
 ## [Node traits](@id lib-node-traits)
 
 Each factor node has to define the [`ReactiveMP.is_predefined_node`](@ref) trait function and to specify a [`ReactiveMP.PredefinedNodeFunctionalForm`](@ref) 

docs/src/lib/variables.md

@@ -0,0 +1,36 @@
+
+# [Variables](@id lib-variables)
+
+Variables are fundamental building blocks of a factor graph. Each variable represents either a latent quantity to be inferred, an observed data point, or a fixed constant. All variable types are subtypes of [`ReactiveMP.AbstractVariable`](@ref).
+
+```@docs
+ReactiveMP.AbstractVariable
+```
+
+## [Random variables](@id lib-variables-random)
+
+Random variables represent latent (unobserved) quantities in the model. During inference, messages flow through them to update the marginal belief.
+
+```@docs
+ReactiveMP.RandomVariable
+ReactiveMP.randomvar
+```
+
+## [Data variables](@id lib-variables-data)
+
+Data variables represent observed quantities. Their value is not fixed at creation time and can be updated later via [`update!`](@ref).
+
+```@docs
+ReactiveMP.DataVariable
+ReactiveMP.datavar
+ReactiveMP.update!
+```
+
+## [Constant variables](@id lib-variables-constant)
+
+Constant variables hold a fixed value, wrapped in a `PointMass` distribution. Messages from constant variables are always marked as clamped.
+
+```@docs
+ReactiveMP.ConstVariable
+ReactiveMP.constvar
+```

ext/ReactiveMPProjectionExt/layout/cvi_projection.jl

@@ -52,6 +52,7 @@ function deltafn_apply_layout(
     scheduler,
     addons,
     rulefallback,
+    callbacks,
 )
     return deltafn_apply_layout(
         DeltaFnDefaultRuleLayout(),
@@ -62,6 +63,7 @@ function deltafn_apply_layout(
         scheduler,
         addons,
         rulefallback,
+        callbacks,
     )
 end
 
@@ -75,6 +77,7 @@ function deltafn_apply_layout(
     scheduler,
     addons,
     rulefallback,
+    callbacks,
 )
     return deltafn_apply_layout(
         DeltaFnDefaultRuleLayout(),
@@ -85,6 +88,7 @@ function deltafn_apply_layout(
         scheduler,
         addons,
         rulefallback,
+        callbacks,
     )
 end
 
@@ -98,6 +102,7 @@ function deltafn_apply_layout(
     scheduler,
     addons,
     rulefallback,
+    callbacks,
 )
     let interface = factornode.out
         msgs_names      = Val{(:out,)}()
@@ -125,6 +130,7 @@ function deltafn_apply_layout(
                     addons,
                     factornode,
                     rulefallback,
+                    callbacks,
                 )
                 (dependencies) -> DeferredMessage(
                     dependencies[1], dependencies[2], messagemap
@@ -152,6 +158,7 @@ function deltafn_apply_layout(
     scheduler,
     addons,
     rulefallback,
+    callbacks,
 )
     return deltafn_apply_layout(
         DeltaFnDefaultRuleLayout(),
@@ -162,5 +169,6 @@ function deltafn_apply_layout(
         scheduler,
         addons,
         rulefallback,
+        callbacks,
     )
 end

src/ReactiveMP.jl

@@ -21,6 +21,8 @@ include("helpers/algebra/standard_basis_vector.jl")
 
 include("constraints/form.jl")
 
+include("callbacks.jl")
+include("variable.jl")
 include("message.jl")
 include("marginal.jl")
 include("addons.jl")
@@ -76,7 +78,7 @@ include("approximations/cvi_projection.jl")
 # Equality node is a special case and needs to be included before random variable implementation
 include("nodes/equality.jl")
 
-include("variables/variable.jl")
+include("variables/generic.jl")
 include("variables/random.jl")
 include("variables/constant.jl")
 include("variables/data.jl")
@@ -113,6 +115,27 @@ function __init__()
             """
             println(io, errmsg)
         end
+        if exc.f === ReactiveMP.handle_event && length(argtypes) >= 2
+            event_type = argtypes[2]
+            event_hint = if event_type <: ReactiveMP.Event
+                "Event{$(repr(ReactiveMP.event_name(event_type)))}"
+            else
+                string(event_type)
+            end
+            errmsg = """
+
+            `ReactiveMP.handle_event` was called with a callback handler of type `$(argtypes[1])` for event `$(event_type)`, but no matching method was found. This can happen if:
+
+            1. You implemented a custom callback handler but forgot to define `handle_event` for this specific event type.
+               Make sure your handler has a method like:
+                 ReactiveMP.handle_event(::$(argtypes[1]), event::$(event_hint)) = ...
+
+            2. You meant to pass a `NamedTuple` as the callbacks handler but forgot the trailing comma.
+               In Julia, `(key = value)` is parsed as a plain assignment, not a NamedTuple.
+               Use `(key = value,)` (with a trailing comma) instead.
+            """
+            println(io, errmsg)
+        end
     end
 end
 

src/approximations/unscented.jl

@@ -135,13 +135,13 @@ function statistic_estimation(
     # Compute `C_tilde` only if `C === true`
     @inbounds C_tilde = if C
         reshape(
-        sum(
-            wc * (xi - m) * (yi - m_tilde) for
-            (wc, xi, yi) in zip(weights_c, sigma_points, g_sigma)
-        ),
-        1,
-        d_out,
-    )
+            sum(
+                wc * (xi - m) * (yi - m_tilde) for
+                (wc, xi, yi) in zip(weights_c, sigma_points, g_sigma)
+            ),
+            1,
+            d_out,
+        )
     else
         nothing
     end
@@ -224,10 +224,10 @@ function unscented_statistics(
     # Compute `C_tilde` only if `C === true`
     @inbounds C_tilde = if C
         sum(
-        weights_c[k + 1] *
-        (sigma_points[k + 1] - m) *
-        (g_sigma[k + 1] - m_tilde)' for k in 0:(2d)
-    )
+            weights_c[k + 1] *
+            (sigma_points[k + 1] - m) *
+            (g_sigma[k + 1] - m_tilde)' for k in 0:(2d)
+        )
     else
         nothing
     end
@@ -258,10 +258,10 @@ function unscented_statistics(
     # Compute `C_tilde` only if `C === true`
     @inbounds C_tilde = if C
         sum(
-        weights_c[k + 1] *
-        (sigma_points[k + 1] - m) *
-        (g_sigma[k + 1] - m_tilde)' for k in 0:(2d)
-    )
+            weights_c[k + 1] *
+            (sigma_points[k + 1] - m) *
+            (g_sigma[k + 1] - m_tilde)' for k in 0:(2d)
+        )
     else
         nothing
     end