From c75d48d79418959831b6a0ad4349dd3a071d21ce Mon Sep 17 00:00:00 2001
From: Maarten Sebregts <msebregts@ignitioncomputing.com>
Date: Wed, 2 Sep 2026 14:22:02 +0200
Subject: [PATCH 1/8] WIP: add reducer filters

---
 integration_test/test_filters.py              |  28 +++++
 src/python/libmuscle/communicator.py          |  91 ++++++++++++--
 src/python/libmuscle/mpp_message.py           |   2 +-
 .../test/test_communicator_conduit_filters.py | 115 +++++++++++++++++-
 src/python/libmuscle/timeline_manager.py      |   5 +
 5 files changed, 230 insertions(+), 11 deletions(-)

diff --git a/integration_test/test_filters.py b/integration_test/test_filters.py
index 6ee76b7e..520a02e9 100644
--- a/integration_test/test_filters.py
+++ b/integration_test/test_filters.py
@@ -115,6 +115,24 @@ def repeat_s(filters: str) -> None:
     assert micro_received == len(EXPECTED_COUNTS)
 
 
+def postprocess() -> None:
+    instance = Instance({Operator.F_INIT: ["macro", "meso", "micro"]})
+
+    while instance.reuse_instance():
+        macro = instance.receive("macro")
+        meso = instance.receive("meso")
+        micro = instance.receive("micro")
+
+        print("Received:", macro.data, meso.data, micro.data)
+        if macro.data[1] == 0:
+            # meso and micro didn't run in this iteration, so we get an empty message
+            assert meso.data is None
+            assert micro.data is None
+        else:
+            assert meso.data[1] == macro.data[1] - 1
+            assert micro.data[1] == 1
+
+
 config = """
 ymmsl_version: v0.2
 models:
@@ -149,19 +167,27 @@ def repeat_s(filters: str) -> None:
         ports:
           f_init: macro meso micro
         implementation: pico
+      postprocess:
+        description: Postprocessing of final actor outputs
+        ports:
+          f_init: macro meso micro
+        implementation: postprocess
     conduits:
       macro.out:
       - meso.in
       - {filters} pico.macro
       - {filters} repeat_s.macro
+      - postprocess.macro
       meso.out:
       - micro.in
       - repeat pico.meso
       - repeat_s.meso
       - repeat repeat_s.repeated_meso
+      - last postprocess.meso
       micro.out:
       - pico.micro
       - repeat_s.micro
+      - last last postprocess.micro
 """
 
 
@@ -173,6 +199,7 @@ def test_repeater_filters(tmp_path, filters):
         "micro": ("python", micro),
         "repeat_s": ("python", repeat_s, filters),
         "pico": ("python", pico, filters),
+        "postprocess": ("python", postprocess),
     }
     run_manager_with_actors(config.format(filters=filters), tmp_path, actors)
 
@@ -186,6 +213,7 @@ def test_repeater_filters_cpp(tmp_path, filters):
         "micro": ("python", micro),
         "repeat_s": ("cpp", "conduit_filters_test", "repeat_s", filters),
         "pico": ("cpp", "conduit_filters_test", "pico", filters),
+        "postprocess": ("python", postprocess),
     }
     run_manager_with_actors(config.format(filters=filters), tmp_path, actors)
 
diff --git a/src/python/libmuscle/communicator.py b/src/python/libmuscle/communicator.py
index 0225c937..3fc98621 100644
--- a/src/python/libmuscle/communicator.py
+++ b/src/python/libmuscle/communicator.py
@@ -155,6 +155,18 @@ def __init__(
         self._message_cache: MPPCacheType = {}
         """Message cache for pre-received messages."""
 
+        self._reduce_count: dict[Reference, int] = {}
+        """Reduce count for each peer port (component + port_name).
+        
+        The reduce count is relative to the component timeline:
+        - 0: send once per reuse loop
+        - 1: send once per parent reuse loop
+        - 2: send once per grandparent reuse loop
+        - etc.
+        """
+        self._reducer_cache: dict[Reference, MPPMessage] = {}
+        """Message cache for reducer filters."""
+
     def get_locations(self) -> list[str]:
         """Returns a list of locations that we can be reached at.
 
@@ -278,7 +290,11 @@ def send_message(
             )
             encoded_message = mpp_message.encoded()
             profile_event.message_size = len(memoryview(encoded_message))
-            self._server.deposit(recv_endpoint.ref(), encoded_message)
+            peer_port = recv_endpoint.kernel + recv_endpoint.port
+            if peer_port in self._reduce_count:
+                self._handle_reducer_filter_on_send(recv_endpoint, mpp_message)
+            else:
+                self._server.deposit(recv_endpoint.ref(), encoded_message)
 
         profile_event.stop()
         if port.is_vector():
@@ -289,6 +305,54 @@ def send_message(
         elif message.data.is_final_milestone():
             port.set_closed(slot)
 
+    def _handle_reducer_filter_on_send(
+        self, recv_endpoint: Endpoint, message: MPPMessage
+    ) -> None:
+        """Send a message on Conduits with a reducer filter applied."""
+        message.message_number = -1  # Disabled checkpointing for reducer filter GH#411
+
+        peer_port = recv_endpoint.kernel + recv_endpoint.port
+        reduce_count = self._reduce_count[peer_port]
+        component_count = len(self._timeline_manager.get_iteration())
+
+        if not isinstance(message.data, Milestone):  # User-provided message: cache it
+            # mpp_message.iteration = Milestone when this message should be sent:
+            message.iteration = message.iteration[: component_count - reduce_count]
+            self._reducer_cache[recv_endpoint.ref()] = message
+
+        else:
+            # Decide whether to send the milestone, ignore it, or send a cached message.
+            # For example, if our current iteration is [1, 2, 3], we can have:
+            # - Milestone([1, 2, 3]):
+            #   - reduce_count=0 (last of O_I): we need to send the last message from
+            #     the cache (message [1, 2, 3]). The peer doesn't need the milestone, so
+            #     we do not send that.
+            #   - reduce_count>0 (last last of O_I / last of O_F): don't send anything,
+            #     our peer expects [1, 2] as next message.
+            # - Milestone([1, 2]):
+            #   - reduce_count=0 (last of O_I): we need to send the milestone
+            #   - reduce_count=1 (last last of O_I / last of O_F): send the cached
+            #     message
+            #   - reduce_count>1 don't send anything
+            # etc.
+            n_milestone = len(message.data.iteration)
+            if n_milestone + reduce_count < component_count:  # Send the milestone
+                self._server.deposit(recv_endpoint.ref(), message.encoded())
+            elif n_milestone + reduce_count > component_count:  # Ignore the milestone
+                pass
+            else:  # Send cached message
+                cached_msg = self._reducer_cache.pop(recv_endpoint.ref(), None)
+                if cached_msg is None:
+                    _logger.info(
+                        "No cached message available to send because this instance did "
+                        "not run. Sending an empty message instead to %s.",
+                        recv_endpoint.ref(),
+                    )
+                    cached_msg = message
+                    cached_msg.data = None
+                assert cached_msg.iteration == message.iteration
+                self._server.deposit(recv_endpoint.ref(), cached_msg.encoded())
+
     def pre_receive(self) -> FInitCacheType:
         """Pre-receive on all connected F_INIT ports and S ports with repeat filters.
 
@@ -315,8 +379,6 @@ def pre_receive(self) -> FInitCacheType:
                     if message.iteration != milestone_iteration
                 }
 
-                # TODO: send buffered message for reducer filters
-
                 if milestone.is_final_milestone():
                     raise PortClosed()
 
@@ -529,7 +591,10 @@ def _receive_message(self, port_name: str, slot: Optional[int]) -> MPPMessage:
             self._profiler.record_event(receive_event)
 
         expected_message_number = port.get_num_messages(slot)
-        if expected_message_number != mpp_message.message_number:
+        if (
+            mpp_message.message_number >= 0  # GH#411: negative for reducer filters
+            and expected_message_number != mpp_message.message_number
+        ):
             if (
                 expected_message_number - 1 == mpp_message.message_number
                 and port.is_resuming(slot)
@@ -685,12 +750,11 @@ def _prepare_conduit_filters(self) -> None:
         """Check which ports are connected with a conduit filter and initialize the
         associated logic.
         """
+        peer_info = self._peer_info
         # Repeater filters
         for operator in (Operator.F_INIT, Operator.S):
             for port in self._port_manager.get_connected_ports(operator):
-                filters = self._peer_info.get_filters_for_receiver(
-                    self._kernel + port.name
-                )
+                filters = peer_info.get_filters_for_receiver(self._kernel + port.name)
                 # Only keep the repeater filters, the sending component handles reducers
                 filters = [filter for filter in filters if filter.is_repeater()]
                 if filters:
@@ -698,7 +762,18 @@ def _prepare_conduit_filters(self) -> None:
                 if operator is Operator.F_INIT or filters:
                     self._pre_receive_ports.append(port)
 
-        # TODO: reducer filters
+        # Reducer filters
+        for operator in (Operator.O_I, Operator.O_F):
+            for port in self._port_manager.get_connected_ports(operator):
+                for peer_port in peer_info.get_peer_ports(port.name):
+                    filters = peer_info.get_filters_for_receiver(peer_port)
+                    # Count the reducer filters, receiving component handles repeaters
+                    n_filters = sum(1 for filter in filters if filter.is_reducer())
+                    if n_filters:
+                        if operator is Operator.O_I:
+                            self._reduce_count[peer_port] = n_filters - 1
+                        else:
+                            self._reduce_count[peer_port] = n_filters
 
     def _pad_message(
         self, cur_iteration: IterationCount, filters: list[ConduitFilter]
diff --git a/src/python/libmuscle/mpp_message.py b/src/python/libmuscle/mpp_message.py
index e6593859..61bf7637 100644
--- a/src/python/libmuscle/mpp_message.py
+++ b/src/python/libmuscle/mpp_message.py
@@ -206,7 +206,7 @@ def __init__(
         self.message_number = message_number
         self.iteration = iteration
         if isinstance(data, np.ndarray):
-            self.data = Grid(data)
+            self.data: Any = Grid(data)
         else:
             self.data = data
 
diff --git a/src/python/libmuscle/test/test_communicator_conduit_filters.py b/src/python/libmuscle/test/test_communicator_conduit_filters.py
index 0a1ae0be..c12bb32e 100644
--- a/src/python/libmuscle/test/test_communicator_conduit_filters.py
+++ b/src/python/libmuscle/test/test_communicator_conduit_filters.py
@@ -1,12 +1,12 @@
 from typing import Union
-from unittest.mock import MagicMock, patch
+from unittest.mock import ANY, MagicMock, patch
 
 import pytest
 from ymmsl.v0_2 import Conduit, ConduitFilter, Operator, Port, Settings
 from ymmsl.v0_2 import Identifier as Id
 from ymmsl.v0_2 import Reference as Ref
 
-from libmuscle.communicator import Communicator, PortClosed
+from libmuscle.communicator import Communicator, Message, PortClosed
 from libmuscle.mpp_message import Milestone, MPPMessage
 from libmuscle.peer_info import PeerInfo
 from libmuscle.port_manager import PortManager
@@ -19,6 +19,12 @@ def mpp_client():
         yield MPPClient.return_value
 
 
+@pytest.fixture
+def mpp_server():
+    with patch("libmuscle.communicator.MPPServer") as MPPServer:
+        yield MPPServer.return_value
+
+
 @pytest.fixture(params=["repeat", "pad"])
 def repeat_filter(request):
     return request.param
@@ -63,6 +69,45 @@ def repeater_communicator(repeat_filter, mpp_client):
     communicator.shutdown()
 
 
+@pytest.fixture
+def reducer_communicator(mpp_client, mpp_server):
+    port_manager = PortManager([], None)
+    communicator = Communicator(
+        Ref("component"), [], port_manager, MagicMock(), MagicMock()
+    )
+    peer_info = PeerInfo(
+        Ref("component"),
+        [],
+        [
+            Conduit("parent.out", "component.init"),
+            Conduit("component.final", "parent.in"),
+            Conduit("component.final", "sibling.in2"),
+            # Reducer filter on O_I port
+            Conduit("component.out", "sibling.in", "last"),
+            # Reducer filter on O_F port
+            Conduit("component.final", "aunt.init", "last"),
+            # Double reducer filter on O_I port
+            Conduit("component.out", "uncle.init", "last last"),
+        ],
+        {Ref("parent"): [], Ref("aunt"): [], Ref("uncle"): [], Ref("sibling"): []},
+        {Ref("parent"): [], Ref("aunt"): [], Ref("uncle"): [], Ref("sibling"): []},
+        [
+            Port(Id("init"), Operator.F_INIT),
+            Port(Id("out"), Operator.O_I),
+            Port(Id("final"), Operator.O_F),
+        ],
+    )
+    port_manager.connect_ports(peer_info)
+    communicator.set_peer_info(peer_info)
+    assert communicator._reduce_count == {
+        "sibling.in": 0,
+        "aunt.init": 1,
+        "uncle.init": 1,
+    }
+    yield communicator
+    communicator.shutdown()
+
+
 def mock_receive_messages(
     mpp_client, data: dict[str, list[Union[IterationCount, Milestone]]]
 ):
@@ -251,3 +296,69 @@ def test_repeater_filters_no_finit(mpp_client, repeat_filter):
 
     # Cleanup
     communicator.shutdown()
+
+
+def test_reducer_filters(reducer_communicator, mpp_client, mpp_server):
+    mock_receive_messages(mpp_client, {"component.init": [[0], [1], Milestone([])]})
+
+    cache = reducer_communicator.pre_receive()
+    assert cache[("init", None)].data == [0]
+    # Send some messages on O_I
+    for i in range(5):
+        reducer_communicator.send_message("out", Message(i, data="data"))
+        mpp_server.deposit.assert_not_called()
+    # Send on O_F
+    reducer_communicator.send_message("final", Message(5, data="data"))
+    mpp_server.deposit.assert_called_with("sibling.in2", ANY)
+    mpp_server.deposit.reset_mock()
+
+    # Pre-receive will send cached LAST message to sibling.in
+    cache = reducer_communicator.pre_receive()
+    assert cache[("init", None)].data == [1]
+    # N.B. we don't send the [1] milestone to sibling.in due to the LAST filter, only
+    # the cached message
+    mpp_server.deposit.assert_called_once_with("sibling.in", ANY)
+    sent_message = MPPMessage.from_bytes(mpp_server.deposit.call_args.args[1])
+    assert sent_message.timestamp == 4  # The last message on O_I
+    mpp_server.deposit.reset_mock()
+
+    # Skip O_I and send on O_F
+    reducer_communicator.send_message("final", Message(10, data="data"))
+    mpp_server.deposit.assert_called_with("sibling.in2", ANY)
+    mpp_server.deposit.reset_mock()
+
+    # Pre-receive will first send cached LAST message to sibling.in, then receive
+    # Milestone([]) and trigger:
+    # - Cached LAST message on "final" to aunt.init
+    # - Cached LAST LAST message on "out" to uncle.init
+    # - Milestone([]) to sibling.in, sibling.in2, parent.in
+    with pytest.raises(PortClosed):
+        reducer_communicator.pre_receive()
+    assert mpp_server.deposit.call_count == 6
+
+    messages_per_peer_port = {}
+    for call in mpp_server.deposit.call_args_list:
+        msg = MPPMessage.from_bytes(call.args[1])
+        messages_per_peer_port.setdefault(call.args[0], []).append(msg)
+
+    # O_I -> last -> sibling.in
+    assert len(messages_per_peer_port["sibling.in"]) == 2
+    # No message was sent on O_I this reuse loop, so LAST generates an empty message:
+    assert messages_per_peer_port["sibling.in"][0].timestamp == float("-inf")
+    assert messages_per_peer_port["sibling.in"][0].data is None
+    assert isinstance(messages_per_peer_port["sibling.in"][1].data, Milestone)
+    assert messages_per_peer_port["sibling.in"][1].data.is_final_milestone()
+
+    # Just milestones
+    for peer_port in ["sibling.in2", "parent.in"]:
+        assert len(messages_per_peer_port[peer_port]) == 1
+        assert isinstance(messages_per_peer_port[peer_port][0].data, Milestone)
+        assert messages_per_peer_port[peer_port][0].data.is_final_milestone()
+
+    # O_I -> last last -> uncle.init
+    assert len(messages_per_peer_port["uncle.init"]) == 1
+    assert messages_per_peer_port["uncle.init"][0].timestamp == 4
+
+    # O_F -> last -> aunt.init
+    assert len(messages_per_peer_port["aunt.init"]) == 1
+    assert messages_per_peer_port["aunt.init"][0].timestamp == 10
diff --git a/src/python/libmuscle/timeline_manager.py b/src/python/libmuscle/timeline_manager.py
index f7b7b30f..8c57777a 100644
--- a/src/python/libmuscle/timeline_manager.py
+++ b/src/python/libmuscle/timeline_manager.py
@@ -272,6 +272,11 @@ def __init__(self, port_manager: PortManager) -> None:
         self._iteration: Optional[IterationCount] = None
         """Current component iteration count. Is None before the first reuse loop."""
 
+    def get_iteration(self) -> IterationCount:
+        """Get the current component iteration count."""
+        assert self._iteration is not None
+        return self._iteration
+
     def check_send_message(
         self, port_name: str, slot: Optional[int] = None
     ) -> IterationCount:

From 6904e4a93bb01326bfd4a65b334140ce8b4bbf66 Mon Sep 17 00:00:00 2001
From: Maarten Sebregts <msebregts@ignitioncomputing.com>
Date: Thu, 3 Sep 2026 12:02:18 +0200
Subject: [PATCH 2/8] Fix reducer filters with timeline information from the
 manager

---
 integration_test/test_filters.py              |   4 +-
 src/python/libmuscle/communicator.py          | 140 ++++++++++--------
 src/python/libmuscle/manager/manager.py       |  10 ++
 .../test/test_communicator_conduit_filters.py |  16 +-
 4 files changed, 99 insertions(+), 71 deletions(-)

diff --git a/integration_test/test_filters.py b/integration_test/test_filters.py
index 520a02e9..cb75cd64 100644
--- a/integration_test/test_filters.py
+++ b/integration_test/test_filters.py
@@ -129,8 +129,8 @@ def postprocess() -> None:
             assert meso.data is None
             assert micro.data is None
         else:
-            assert meso.data[1] == macro.data[1] - 1
-            assert micro.data[1] == 1
+            assert meso.data[-1] == macro.data[-1] - 1
+            assert micro.data[-1] == 1
 
 
 config = """
diff --git a/src/python/libmuscle/communicator.py b/src/python/libmuscle/communicator.py
index acd65ee0..cfbf6a4a 100644
--- a/src/python/libmuscle/communicator.py
+++ b/src/python/libmuscle/communicator.py
@@ -155,14 +155,17 @@ def __init__(
         self._message_cache: MPPCacheType = {}
         """Message cache for pre-received messages."""
 
-        self._reduce_count: dict[Reference, int] = {}
-        """Reduce count for each peer port (component + port_name).
-        
-        The reduce count is relative to the component timeline:
-        - 0: send once per reuse loop
-        - 1: send once per parent reuse loop
-        - 2: send once per grandparent reuse loop
-        - etc.
+        self._reduced_count: dict[Reference, int] = {}
+        """Size of IterationCount, after applying the reducer filters, per peer port.
+
+        Keys are references to peer ports: ``component + port``. The reduced count is
+        the size of the IterationCount after applying the reducer filters and determines
+        in which (parent) timeline these messages are sent.
+
+        If our timeline is ":macro:micro" then:
+        - reduced_count = 0: send on the root (":") timeline
+        - reduced_count = 1: send on the ":macro" timeline
+        - reduced_count = 2: send on the ":macro:micro" timeline
         """
         self._reducer_cache: dict[Reference, MPPMessage] = {}
         """Message cache for reducer filters."""
@@ -189,6 +192,7 @@ def set_peer_info(self, peer_info: PeerInfo) -> None:
         Args:
             peer_info: Information about the peers.
         """
+        self._timeline = self._manager.get_timeline()
         self._peer_info = peer_info
         self._timeline_manager = TimelineManager(self._port_manager)
         self._prepare_conduit_filters()
@@ -289,13 +293,15 @@ def send_message(
                 message.data,
                 iteration,
             )
+            peer_port = recv_endpoint.kernel + recv_endpoint.port
+            if peer_port in self._reduced_count:
+                result = self._apply_reduce_filters(peer_port, mpp_message)
+                if result is None:
+                    continue
+                mpp_message = result
             encoded_message = mpp_message.encoded()
             profile_event.message_size = len(memoryview(encoded_message))
-            peer_port = recv_endpoint.kernel + recv_endpoint.port
-            if peer_port in self._reduce_count:
-                self._handle_reducer_filter_on_send(recv_endpoint, mpp_message)
-            else:
-                self._server.deposit(recv_endpoint.ref(), encoded_message)
+            self._server.deposit(recv_endpoint.ref(), encoded_message)
 
         profile_event.stop()
         if port.is_vector():
@@ -306,53 +312,64 @@ def send_message(
         elif message.data.is_final_milestone():
             port.set_closed(slot)
 
-    def _handle_reducer_filter_on_send(
-        self, recv_endpoint: Endpoint, message: MPPMessage
-    ) -> None:
-        """Send a message on Conduits with a reducer filter applied."""
-        message.message_number = -1  # Disabled checkpointing for reducer filter GH#411
+    def _apply_reduce_filters(
+        self, peer_port: Reference, message: MPPMessage
+    ) -> Optional[MPPMessage]:
+        """Apply reduce filters to a message sent on a conduit with reduce filters.
+
+        User-provided messages (through instance.send()) will be stored (overwriting any
+        existing message). For milestones this method decides if the milestone should be
+        sent, or a cached message, or nothing at all.
 
-        peer_port = recv_endpoint.kernel + recv_endpoint.port
-        reduce_count = self._reduce_count[peer_port]
-        component_count = len(self._timeline_manager.get_iteration())
+        Args:
+            recv_endpoint: The Endpoint to send to.
+            message: MPPMessage to be checked.
 
-        if not isinstance(message.data, Milestone):  # User-provided message: cache it
-            # mpp_message.iteration = Milestone when this message should be sent:
-            message.iteration = message.iteration[: component_count - reduce_count]
-            self._reducer_cache[recv_endpoint.ref()] = message
+        Returns:
+            The MPPMessage to send, or None if we do not need to send anything.
+        """
+        message.message_number = -1  # GH#411: Disabled checkpointing for reducer filter
+
+        reduced_count = self._reduced_count[peer_port]
+
+        if not isinstance(message.data, Milestone):
+            # Reduce the message iteration count to match with the timeline we send to
+            message.iteration = message.iteration[:reduced_count]
+            self._reducer_cache[message.receiver] = message
+            _logger.debug("Message for %s stored in cache", message.receiver)
+            return None
+
+        # Decide whether to send the milestone, ignore it, or send a cached message.
+        n_milestone = len(message.data.iteration)
+        if n_milestone < reduced_count:
+            # Milestone from ancestor timeline: send it
+            return message
+
+        elif n_milestone == reduced_count:
+            # This is the target timeline after reduce filters applied: we need to
+            # send the cached message (or make up an empty one) and discard the
+            # milestone:
+            cached_msg = self._reducer_cache.pop(message.receiver, None)
+            if cached_msg is None:
+                _logger.info(
+                    "No cached message available to send because this instance did "
+                    "not run. Sending an empty message to %s instead.",
+                    message.receiver,
+                )
+                message.data = None
+                return message
+
+            assert cached_msg.iteration == message.iteration
+            _logger.debug("Sending cached message to %s", message.receiver)
+            return cached_msg
 
         else:
-            # Decide whether to send the milestone, ignore it, or send a cached message.
-            # For example, if our current iteration is [1, 2, 3], we can have:
-            # - Milestone([1, 2, 3]):
-            #   - reduce_count=0 (last of O_I): we need to send the last message from
-            #     the cache (message [1, 2, 3]). The peer doesn't need the milestone, so
-            #     we do not send that.
-            #   - reduce_count>0 (last last of O_I / last of O_F): don't send anything,
-            #     our peer expects [1, 2] as next message.
-            # - Milestone([1, 2]):
-            #   - reduce_count=0 (last of O_I): we need to send the milestone
-            #   - reduce_count=1 (last last of O_I / last of O_F): send the cached
-            #     message
-            #   - reduce_count>1 don't send anything
-            # etc.
-            n_milestone = len(message.data.iteration)
-            if n_milestone + reduce_count < component_count:  # Send the milestone
-                self._server.deposit(recv_endpoint.ref(), message.encoded())
-            elif n_milestone + reduce_count > component_count:  # Ignore the milestone
-                pass
-            else:  # Send cached message
-                cached_msg = self._reducer_cache.pop(recv_endpoint.ref(), None)
-                if cached_msg is None:
-                    _logger.info(
-                        "No cached message available to send because this instance did "
-                        "not run. Sending an empty message instead to %s.",
-                        recv_endpoint.ref(),
-                    )
-                    cached_msg = message
-                    cached_msg.data = None
-                assert cached_msg.iteration == message.iteration
-                self._server.deposit(recv_endpoint.ref(), cached_msg.encoded())
+            _logger.debug(
+                "Ignored %s for %s because of LAST filters.",
+                message.data,
+                message.receiver,
+            )
+            return None
 
     def pre_receive(self) -> FInitCacheType:
         """Pre-receive on all connected F_INIT ports and S ports with repeat filters.
@@ -769,12 +786,11 @@ def _prepare_conduit_filters(self) -> None:
                 for peer_port in peer_info.get_peer_ports(port.name):
                     filters = peer_info.get_filters_for_receiver(peer_port)
                     # Count the reducer filters, receiving component handles repeaters
-                    n_filters = sum(1 for filter in filters if filter.is_reducer())
-                    if n_filters:
-                        if operator is Operator.O_I:
-                            self._reduce_count[peer_port] = n_filters - 1
-                        else:
-                            self._reduce_count[peer_port] = n_filters
+                    n_reducers = sum(1 for filter in filters if filter.is_reducer())
+                    if n_reducers > 0:
+                        self._reduced_count[peer_port] = (
+                            len(self._timeline) + len(port.timeline) - n_reducers
+                        )
 
     def _pad_message(
         self, cur_iteration: IterationCount, filters: list[ConduitFilter]
diff --git a/src/python/libmuscle/manager/manager.py b/src/python/libmuscle/manager/manager.py
index db81bc7d..daca4d02 100644
--- a/src/python/libmuscle/manager/manager.py
+++ b/src/python/libmuscle/manager/manager.py
@@ -45,6 +45,16 @@ def __init__(
             configuration: The simulation configuration.
             run_dir: Main working directory.
         """
+        # TEMP: check for checkpoints combined with reducer filters:
+        if configuration.checkpoints and any(
+            any(filter.is_reducer() for filter in conduit.filters)
+            for conduit in configuration.root_model().conduits
+        ):
+            raise NotImplementedError(
+                "This version of MUSCLE3 does not support checkpoints when using "
+                "reducer filters. See https://github.com/multiscale/muscle3/issues/411"
+            )
+
         self._configuration = configuration
         self._run_dir = run_dir
         log_dir = self._run_dir.path if self._run_dir else Path.cwd()
diff --git a/src/python/libmuscle/test/test_communicator_conduit_filters.py b/src/python/libmuscle/test/test_communicator_conduit_filters.py
index c12bb32e..c063b683 100644
--- a/src/python/libmuscle/test/test_communicator_conduit_filters.py
+++ b/src/python/libmuscle/test/test_communicator_conduit_filters.py
@@ -2,7 +2,7 @@
 from unittest.mock import ANY, MagicMock, patch
 
 import pytest
-from ymmsl.v0_2 import Conduit, ConduitFilter, Operator, Port, Settings
+from ymmsl.v0_2 import Conduit, ConduitFilter, Operator, Port, Settings, Timeline
 from ymmsl.v0_2 import Identifier as Id
 from ymmsl.v0_2 import Reference as Ref
 
@@ -72,8 +72,10 @@ def repeater_communicator(repeat_filter, mpp_client):
 @pytest.fixture
 def reducer_communicator(mpp_client, mpp_server):
     port_manager = PortManager([], None)
+    mock_manager = MagicMock()
+    mock_manager.get_timeline.return_value = Timeline(":parent")
     communicator = Communicator(
-        Ref("component"), [], port_manager, MagicMock(), MagicMock()
+        Ref("component"), [], port_manager, MagicMock(), mock_manager
     )
     peer_info = PeerInfo(
         Ref("component"),
@@ -93,16 +95,16 @@ def reducer_communicator(mpp_client, mpp_server):
         {Ref("parent"): [], Ref("aunt"): [], Ref("uncle"): [], Ref("sibling"): []},
         [
             Port(Id("init"), Operator.F_INIT),
-            Port(Id("out"), Operator.O_I),
+            Port(Id("out"), Operator.O_I, Timeline("component")),
             Port(Id("final"), Operator.O_F),
         ],
     )
     port_manager.connect_ports(peer_info)
     communicator.set_peer_info(peer_info)
-    assert communicator._reduce_count == {
-        "sibling.in": 0,
-        "aunt.init": 1,
-        "uncle.init": 1,
+    assert communicator._reduced_count == {
+        "sibling.in": 1,
+        "aunt.init": 0,
+        "uncle.init": 0,
     }
     yield communicator
     communicator.shutdown()

From 09d27c6aabe4bd39c476e62bb933e89fcea9c0a8 Mon Sep 17 00:00:00 2001
From: Maarten Sebregts <msebregts@ignitioncomputing.com>
Date: Thu, 3 Sep 2026 14:36:26 +0200
Subject: [PATCH 3/8] Check that F_INIT iteration count matches component
 timeline

---
 src/python/libmuscle/communicator.py          |  2 +-
 .../libmuscle/test/test_communicator.py       |  6 +--
 .../test/test_communicator_conduit_filters.py | 39 ++++++++++---------
 .../libmuscle/test/test_timeline_manager.py   | 29 +++++++++++---
 src/python/libmuscle/timeline_manager.py      | 14 ++++---
 5 files changed, 56 insertions(+), 34 deletions(-)

diff --git a/src/python/libmuscle/communicator.py b/src/python/libmuscle/communicator.py
index cfbf6a4a..465512c8 100644
--- a/src/python/libmuscle/communicator.py
+++ b/src/python/libmuscle/communicator.py
@@ -194,7 +194,7 @@ def set_peer_info(self, peer_info: PeerInfo) -> None:
         """
         self._timeline = self._manager.get_timeline()
         self._peer_info = peer_info
-        self._timeline_manager = TimelineManager(self._port_manager)
+        self._timeline_manager = TimelineManager(self._port_manager, self._timeline)
         self._prepare_conduit_filters()
 
     def set_receive_timeout(self, receive_timeout: float) -> None:
diff --git a/src/python/libmuscle/test/test_communicator.py b/src/python/libmuscle/test/test_communicator.py
index 9a5eef93..c7cd67fe 100644
--- a/src/python/libmuscle/test/test_communicator.py
+++ b/src/python/libmuscle/test/test_communicator.py
@@ -112,15 +112,13 @@ def test_create_communicator(communicator, mpp_server):
     pass
 
 
-def test_set_peer_info_creates_timeline_manager(
-    communicator, connected_port_manager, timeline_manager
-):
+def test_set_peer_info_creates_timeline_manager(communicator, timeline_manager):
     peer_info = MagicMock()
 
     communicator.set_peer_info(peer_info)
 
     assert communicator._peer_info == peer_info
-    timeline_manager.assert_called_once_with(connected_port_manager)
+    timeline_manager.assert_called_once()
     assert communicator._timeline_manager == timeline_manager.return_value
 
 
diff --git a/src/python/libmuscle/test/test_communicator_conduit_filters.py b/src/python/libmuscle/test/test_communicator_conduit_filters.py
index c063b683..99dcd6c0 100644
--- a/src/python/libmuscle/test/test_communicator_conduit_filters.py
+++ b/src/python/libmuscle/test/test_communicator_conduit_filters.py
@@ -33,8 +33,10 @@ def repeat_filter(request):
 @pytest.fixture()
 def repeater_communicator(repeat_filter, mpp_client):
     port_manager = PortManager([], None)
+    mock_manager = MagicMock()
+    mock_manager.get_timeline.return_value = Timeline(":parent3:parent2:parent1")
     communicator = Communicator(
-        Ref("component"), [], port_manager, MagicMock(), MagicMock()
+        Ref("component"), [], port_manager, MagicMock(), mock_manager
     )
     peer_info = PeerInfo(
         Ref("component"),
@@ -152,16 +154,17 @@ def side_effect(peer, _):
 
 
 def test_repeater_filters(repeater_communicator, mpp_client, repeat_filter):
-    twicerepeated_messages = [[], Milestone([])]
-    repeated_messages = [[0], [1], [2], Milestone([])]
+    twicerepeated_messages = [[0], Milestone([])]
+    repeated_messages = [[0, 0], [0, 1], [0, 2], Milestone([0]), Milestone([])]
     unfiltered_messages = [
-        [0, 0],
-        [0, 1],
-        Milestone([0]),
+        [0, 0, 0],
+        [0, 0, 1],
+        Milestone([0, 0]),
         # parent is allowed to send 0 messages on its O_I port in an iteration
-        Milestone([1]),
-        [2, 0],
-        Milestone([2]),
+        Milestone([0, 1]),
+        [0, 2, 0],
+        Milestone([0, 2]),
+        Milestone([0]),
         Milestone([]),
     ]
     mock_receive_messages(
@@ -177,19 +180,19 @@ def test_repeater_filters(repeater_communicator, mpp_client, repeat_filter):
     is_padded = repeat_filter == "pad"
 
     cache = repeater_communicator.pre_receive()
-    assert cache[("unfiltered", None)].data == [0, 0]
-    assert cache[("repeated", None)].data == [0]
-    assert cache[("twicerepeated", None)].data == []
+    assert cache[("unfiltered", None)].data == [0, 0, 0]
+    assert cache[("repeated", None)].data == [0, 0]
+    assert cache[("twicerepeated", None)].data == [0]
 
     cache = repeater_communicator.pre_receive()
-    assert cache[("unfiltered", None)].data == [0, 1]
-    assert cache[("repeated", None)].data == (None if is_padded else [0])
-    assert cache[("twicerepeated", None)].data == (None if is_padded else [])
+    assert cache[("unfiltered", None)].data == [0, 0, 1]
+    assert cache[("repeated", None)].data == (None if is_padded else [0, 0])
+    assert cache[("twicerepeated", None)].data == (None if is_padded else [0])
 
     cache = repeater_communicator.pre_receive()
-    assert cache[("unfiltered", None)].data == [2, 0]
-    assert cache[("repeated", None)].data == [2]
-    assert cache[("twicerepeated", None)].data == (None if is_padded else [])
+    assert cache[("unfiltered", None)].data == [0, 2, 0]
+    assert cache[("repeated", None)].data == [0, 2]
+    assert cache[("twicerepeated", None)].data == (None if is_padded else [0])
 
     with pytest.raises(PortClosed):
         repeater_communicator.pre_receive()
diff --git a/src/python/libmuscle/test/test_timeline_manager.py b/src/python/libmuscle/test/test_timeline_manager.py
index f830a646..46f788bd 100644
--- a/src/python/libmuscle/test/test_timeline_manager.py
+++ b/src/python/libmuscle/test/test_timeline_manager.py
@@ -31,7 +31,14 @@ def include_settings(request: pytest.FixtureRequest) -> bool:
 
 
 @pytest.fixture
-def timeline_manager(has_f_init: bool, include_settings: bool) -> TimelineManager:
+def timeline(request: pytest.FixtureRequest) -> Timeline:
+    return getattr(request, "param", Timeline(":"))
+
+
+@pytest.fixture
+def timeline_manager(
+    has_f_init: bool, include_settings: bool, timeline: Timeline
+) -> TimelineManager:
     conduits = [
         Conduit("component.out_f", "peer_f.in"),
         Conduit("component.out_a1", "peer_a1.in"),
@@ -69,20 +76,20 @@ def timeline_manager(has_f_init: bool, include_settings: bool) -> TimelineManage
     peer_info = PeerInfo(Ref("component"), [], conduits, peer_dims, {}, ymmsl_ports)
     pm.connect_ports(peer_info)
 
-    tm = TimelineManager(pm)
+    tm = TimelineManager(pm, timeline)
     assert tm.start_reuse_iteration() is None
     return tm
 
 
 @pytest.fixture
-def vector_timeline_manager() -> TimelineManager:
+def vector_timeline_manager(timeline: Timeline) -> TimelineManager:
     declared_ports = {Operator.O_F: ["out_v[]"]}
     pm = PortManager([], declared_ports)
     conduits = [Conduit("component.out_v", "peer.in")]
     peer_info = PeerInfo(Ref("component"), [], conduits, {Ref("peer"): [3]}, {}, [])
     pm.connect_ports(peer_info)
 
-    tm = TimelineManager(pm)
+    tm = TimelineManager(pm, timeline)
     assert tm.start_reuse_iteration() is None
     assert tm.check_pre_received_iteration_counts([]) == []
     return tm
@@ -274,6 +281,7 @@ def test_check_send_message_o_i_blocked_when_o_i_leads_and_incomplete(
     )
 
 
+@pytest.mark.parametrize("timeline", [Timeline(":a:b:c")], indirect=True)
 def test_check_pre_receive_increments(timeline_manager: TimelineManager) -> None:
     assert timeline_manager.check_pre_received_iteration_counts(
         [[1, 2, 3], [1, 2], [1], [], [1, 2, 3], [1, 2]]
@@ -288,6 +296,14 @@ def test_check_pre_receive_increments(timeline_manager: TimelineManager) -> None
     ) == [1, 2, 4]
 
 
+def test_check_pre_receive_counts_match_timeline(
+    timeline_manager: TimelineManager,
+) -> None:
+    with pytest.raises(RuntimeError, match="iteration count with 0 elements"):
+        timeline_manager.check_pre_received_iteration_counts([[1]])
+
+
+@pytest.mark.parametrize("timeline", [Timeline(":a")], indirect=True)
 def test_check_pre_receive_iterations_when_iteration_differs(
     timeline_manager: TimelineManager,
 ) -> None:
@@ -391,6 +407,7 @@ def test_check_receive_message_s_blocked_when_s_leads_and_incomplete(
     )
 
 
+@pytest.mark.parametrize("timeline", [Timeline(":a")], indirect=True)
 def test_finish_reuse_iteration_resets_when_complete(
     timeline_manager: TimelineManager,
 ) -> None:
@@ -415,6 +432,7 @@ def test_finish_reuse_iteration_resets_when_complete(
     )
 
 
+@pytest.mark.parametrize("timeline", [Timeline(":a")], indirect=True)
 def test_finish_reuse_iteration_raises_when_incomplete(
     timeline_manager: TimelineManager,
 ) -> None:
@@ -436,6 +454,7 @@ def test_finish_reuse_iteration_raises_when_incomplete(
     )
 
 
+@pytest.mark.parametrize("timeline", [Timeline(":a")], indirect=True)
 def test_get_state_and_restore_state_round_trip(
     timeline_manager: TimelineManager,
 ) -> None:
@@ -449,7 +468,7 @@ def test_get_state_and_restore_state_round_trip(
     # Restore into a fresh TimelineManager, as would happen after loading a
     # snapshot in a new process, from an independent but identically
     # configured PortManager.
-    restored = TimelineManager(timeline_manager._port_manager)
+    restored = TimelineManager(timeline_manager._port_manager, Timeline(":"))
     restored.restore_state(timeline_state)
 
     assert restored.get_state() == timeline_state
diff --git a/src/python/libmuscle/timeline_manager.py b/src/python/libmuscle/timeline_manager.py
index 8c57777a..3ea9dc79 100644
--- a/src/python/libmuscle/timeline_manager.py
+++ b/src/python/libmuscle/timeline_manager.py
@@ -252,7 +252,7 @@ class TimelineManager:
     ports that receive (F_INIT or S).
     """
 
-    def __init__(self, port_manager: PortManager) -> None:
+    def __init__(self, port_manager: PortManager, timeline: Timeline) -> None:
         """Create a TimelineManager.
 
         The port_manager must already have its ports connected to their peers, since
@@ -263,6 +263,7 @@ def __init__(self, port_manager: PortManager) -> None:
         Args:
             port_manager: The (already connected) port manager for this instance.
         """
+        self._timeline = timeline
         self._port_manager = port_manager
         self._send = TimelinePorts(port_manager.get_connected_ports(Operator.O_F))
         subtimelines = port_manager.list_subtimelines()
@@ -272,11 +273,6 @@ def __init__(self, port_manager: PortManager) -> None:
         self._iteration: Optional[IterationCount] = None
         """Current component iteration count. Is None before the first reuse loop."""
 
-    def get_iteration(self) -> IterationCount:
-        """Get the current component iteration count."""
-        assert self._iteration is not None
-        return self._iteration
-
     def check_send_message(
         self, port_name: str, slot: Optional[int] = None
     ) -> IterationCount:
@@ -325,6 +321,12 @@ def check_pre_received_iteration_counts(
                 f"Internal error: received F_INIT iteration count {new_iteration} "
                 f"is not newer than the previous iteration {self._iteration}."
             )
+        if len(new_iteration) != len(self._timeline):
+            raise RuntimeError(
+                f"Received unexpected F_INIT iteration count: {new_iteration}. Was "
+                f"expecting an iteration count with {len(self._timeline)} elements, "
+                f"since we are in timeline {self._timeline}"
+            )
         self._iteration = new_iteration
         return self._iteration
 

From 283008d312c4a97253bcb468d7ff16a23c33f481 Mon Sep 17 00:00:00 2001
From: Maarten Sebregts <msebregts@ignitioncomputing.com>
Date: Thu, 3 Sep 2026 17:00:48 +0200
Subject: [PATCH 4/8] Add implementation and tests for reducer filters in C++

---
 src/cpp/libmuscle/communicator.cpp            |  87 ++++++++-
 src/cpp/libmuscle/communicator.hpp            |  28 +++
 .../tests/mocks/mock_timeline_manager.hpp     |   3 +-
 .../test_communicator_conduit_filters.cpp     | 171 ++++++++++++++++--
 .../libmuscle/tests/test_timeline_manager.cpp |  11 +-
 src/cpp/libmuscle/timeline_manager.cpp        |  10 +-
 src/cpp/libmuscle/timeline_manager.hpp        |   4 +-
 src/cpp/ymmsl/ports.cpp                       |  10 +
 src/cpp/ymmsl/ports.hpp                       |   4 +
 src/cpp/ymmsl/tests/test_ports.cpp            |   9 +
 src/python/libmuscle/communicator.py          |   2 +-
 .../test/test_communicator_conduit_filters.py |  18 +-
 12 files changed, 319 insertions(+), 38 deletions(-)

diff --git a/src/cpp/libmuscle/communicator.cpp b/src/cpp/libmuscle/communicator.cpp
index d082b705..a581d23f 100644
--- a/src/cpp/libmuscle/communicator.cpp
+++ b/src/cpp/libmuscle/communicator.cpp
@@ -96,8 +96,9 @@ std::vector<std::string> Communicator::get_locations() const {
 }
 
 void Communicator::set_peer_info(PeerInfo const & peer_info) {
+    timeline_ = manager_.get_timeline();
     peer_info_ = peer_info;
-    timeline_manager_ = std::make_unique<TimelineManager>(port_manager_);
+    timeline_manager_ = std::make_unique<TimelineManager>(port_manager_, timeline_.get());
     prepare_conduit_filters_();
 }
 
@@ -161,8 +162,16 @@ void Communicator::send_message(
 
         if (message.has_next_timestamp())
             mpp_message.next_timestamp = message.next_timestamp();
-
-        auto message_bytes = mpp_message.encoded();
+        
+        std::vector<char> message_bytes;
+        auto peer_port = recv_endpoint.kernel + recv_endpoint.port;
+        if (reduced_count_.count(peer_port) > 0) {
+            message_bytes = apply_reduce_filters_(peer_port, std::move(mpp_message));
+            if (message_bytes.empty())
+                continue;
+        } else {
+            message_bytes = mpp_message.encoded();
+        }
         profile_event.message_size = message_bytes.size();
         server_.deposit(recv_endpoint.ref(), std::move(message_bytes));
     }
@@ -177,6 +186,57 @@ void Communicator::send_message(
         port.set_closed(slot);
 }
 
+std::vector<char> Communicator::apply_reduce_filters_(
+        ymmsl::Reference const & peer_port, MPPMessage && message) {
+    message.message_number = -1;  // GH#411: Disabled checkpointing for reducer filter
+
+    auto reduced_count = reduced_count_.at(peer_port);
+
+    if (!is_milestone(message.data)) {
+        // Reduce the message iteration count to match with the timeline we send to
+        message.iteration.resize(reduced_count);
+        auto it = reducer_cache_.find(message.receiver);
+        if (it != reducer_cache_.end())
+            reducer_cache_.erase(it);  // Remove existing entry
+        reducer_cache_.emplace(message.receiver, message);
+        log_debug("Message for ", message.receiver, " stored in cache");
+        return {};
+    }
+
+    // Decide whether to send the milestone, ignore it, or send a cached message.
+    std::size_t n_milestone = message.iteration.size();
+    if (n_milestone < reduced_count) {
+        // Milestone from ancestor timeline: send it
+        return message.encoded();
+    } else if (n_milestone == reduced_count) {
+        // This is the target timeline after reduce filters applied: we need to
+        // send the cached message (or make up an empty one) and discard the
+        // milestone:
+        auto it = reducer_cache_.find(message.receiver);
+        if (it == reducer_cache_.end()) {
+            log_info(
+                    "No cached message available to send because this instance did ",
+                    "not run. Sending an empty message to ", message.receiver, " instead.");
+            return MPPMessage(
+                    message.sender, message.receiver, message.port_length,
+                    message.timestamp, message.next_timestamp, message.settings_overlay,
+                    message.message_number, Data(), message.iteration
+                    ).encoded();
+        }
+
+        assert(it->second.iteration == message.iteration);        
+        log_debug("Sending cached message to ", message.receiver);
+        auto encoded = it->second.encoded();
+        reducer_cache_.erase(it);
+        return encoded;
+    } else {
+        log_debug(
+                "Ignored milestone for ", message.receiver,
+                " because of LAST filters.");
+        return {};
+    }
+}
+
 Communicator::FInitCacheType Communicator::pre_receive() {
     assert(timeline_manager_);
     auto finished_iteration = timeline_manager_->start_reuse_iteration();
@@ -376,7 +436,10 @@ MPPMessage Communicator::receive_message_(
     }
 
     int expected_message_number = port.get_num_messages(slot);
-    if (expected_message_number != mpp_message.message_number) {
+    if (
+        mpp_message.message_number >= 0  // GH#411: negative for reducer filters
+        && expected_message_number != mpp_message.message_number
+    ) {
         if (expected_message_number - 1 == mpp_message.message_number and
                 port.is_resuming(slot)) {
             log_debug("Discarding received message on ", port_and_slot,
@@ -560,7 +623,21 @@ void Communicator::prepare_conduit_filters_() {
         }
     }
 
-    // TODO: reducer filters
+    // Reducer filters
+    for (auto op : {Operator::O_I, Operator::O_F}) {
+        for (Port const & port : port_manager_.get_connected_ports(op, {})) {
+            for (auto & peer_port : peer_info_.get().get_peer_ports(port.name)) {
+                auto & filters = peer_info_.get().get_filters_for_receiver(peer_port);
+                // Count the reducer filters, receiving component handles repeaters
+                auto n_reducers = std::count_if(
+                        filters.begin(), filters.end(), ::ymmsl::is_reducer);
+                if (n_reducers > 0) {
+                    std::size_t reduced_count = timeline_.get().size() + port.timeline.size() - n_reducers;
+                    reduced_count_.emplace(peer_port, reduced_count);
+                }
+            }
+        }
+    }
 }
 
 
diff --git a/src/cpp/libmuscle/communicator.hpp b/src/cpp/libmuscle/communicator.hpp
index 3ab51e9f..8c0e57ca 100644
--- a/src/cpp/libmuscle/communicator.hpp
+++ b/src/cpp/libmuscle/communicator.hpp
@@ -170,6 +170,19 @@ class Communicator {
                 Optional<int> slot = {}
                 );
 
+        /** Apply reduce filters to a message sent on a conduit with reduce filters.
+         * 
+         * User-provided messages (through instance.send()) will be stored (overwriting any
+         * existing message). For milestones this method decides if the milestone should be
+         * sent, or a cached message, or nothing at all.
+         * 
+         * @param peer_port Peer port (component + port) to send to.
+         * @param message MPPMessage to be checked.
+         * @return The encoded MPPMessage to send, or an empty vector if we do not need to send anything.
+         */
+        std::vector<char> apply_reduce_filters_(
+                ymmsl::Reference const & peer_port, MPPMessage && message);
+
         ymmsl::Reference instance_id_() const;
         MPPClient & get_client_(ymmsl::Reference const & instance);
 
@@ -242,10 +255,25 @@ class Communicator {
         Optional<PeerInfo> peer_info_;
         double receive_timeout_;
         std::unique_ptr<TimelineManager> timeline_manager_;
+        Optional<ymmsl::Timeline> timeline_;
 
         PortManager::PortReferences pre_receive_ports_;
         std::unordered_map<std::string, std::vector<::ymmsl::ConduitFilter>> repeat_filters_;
         MPPCacheType message_cache_;
+
+        /** Size of IterationCount, after applying the reducer filters, per peer port.
+         * 
+         * Keys are references to peer ports: ``component + port``. The reduced count is
+         * the size of the IterationCount after applying the reducer filters and determines
+         * in which (parent) timeline these messages are sent.
+         * If our timeline is ":macro:micro" then:
+         * - reduced_count = 0: send on the root (":") timeline
+         * - reduced_count = 1: send on the ":macro" timeline
+         * - reduced_count = 2: send on the ":macro:micro" timeline
+         */
+        std::unordered_map<::ymmsl::Reference, std::size_t> reduced_count_;
+        /** Message cache for reducer filters */
+        MPPCacheType reducer_cache_;
 };
 
 } }
diff --git a/src/cpp/libmuscle/tests/mocks/mock_timeline_manager.hpp b/src/cpp/libmuscle/tests/mocks/mock_timeline_manager.hpp
index 640da480..e6a020c2 100644
--- a/src/cpp/libmuscle/tests/mocks/mock_timeline_manager.hpp
+++ b/src/cpp/libmuscle/tests/mocks/mock_timeline_manager.hpp
@@ -31,7 +31,8 @@ class MockTimelineManager : public MockClass<MockTimelineManager> {
             init_from_return_value();
         }
 
-        explicit MockTimelineManager(PortManager const & port_manager) {
+        explicit MockTimelineManager(
+                PortManager const & port_manager, ::ymmsl::Timeline const & timeline) {
             init_from_return_value();
             constructor(&port_manager);
         }
diff --git a/src/cpp/libmuscle/tests/test_communicator_conduit_filters.cpp b/src/cpp/libmuscle/tests/test_communicator_conduit_filters.cpp
index bf8ef1e4..0efe3f69 100644
--- a/src/cpp/libmuscle/tests/test_communicator_conduit_filters.cpp
+++ b/src/cpp/libmuscle/tests/test_communicator_conduit_filters.cpp
@@ -49,6 +49,7 @@ using libmuscle::_MUSCLE_IMPL_NS::encode_iteration;
 using libmuscle::_MUSCLE_IMPL_NS::Communicator;
 using libmuscle::_MUSCLE_IMPL_NS::DataConstRef;
 using libmuscle::_MUSCLE_IMPL_NS::IterationCount;
+using libmuscle::_MUSCLE_IMPL_NS::Message;
 using libmuscle::_MUSCLE_IMPL_NS::MPPMessage;
 using libmuscle::_MUSCLE_IMPL_NS::MockLogger;
 using libmuscle::_MUSCLE_IMPL_NS::MockProfiler;
@@ -65,6 +66,7 @@ using ymmsl::Conduit;
 using ymmsl::ConduitFilter;
 using ymmsl::Operator;
 using ymmsl::Port;
+using ymmsl::Timeline;
 
 struct libmuscle_repeater_communicator
     : ::testing::TestWithParam<std::string>
@@ -81,6 +83,7 @@ struct libmuscle_repeater_communicator
         : port_manager_({}, {})
         , communicator_("component", {}, port_manager_, profiler_, manager_)
     {
+        manager_.get_timeline.return_value = Timeline(":parent3:parent2:parent1");
         std::string repeat_filter = GetParam();
         PeerInfo peer_info(
             "component",
@@ -107,6 +110,53 @@ struct libmuscle_repeater_communicator
     }
 };
 
+struct libmuscle_reducer_communicator
+    : ::testing::Test
+{
+    RESET_MOCKS(MockLogger, MockMMPClient, MockMPPClient, MockMPPServer, MockProfiler);
+    
+    MockProfiler profiler_;
+    MockMMPClient manager_;
+
+    PortManager port_manager_;
+    Communicator communicator_;
+
+    libmuscle_reducer_communicator()
+        : port_manager_({}, {})
+        , communicator_("component", {}, port_manager_, profiler_, manager_)
+    {
+        manager_.get_timeline.return_value = Timeline(":parent");
+        PeerInfo peer_info(
+            "component",
+            {},
+            {
+                Conduit("parent.out", "component.init"),
+                Conduit("component.final", "parent.in"),
+                Conduit("component.final", "sibling.in2"),
+                // Reducer filter on O_I port
+                Conduit("component.out", "sibling.in", "last"),
+                // Reducer filter on O_F port
+                Conduit("component.final", "aunt.init", "last"),
+                // Double reducer filter on O_I port
+                Conduit("component.out", "uncle.init", "last last"),
+            },
+            {{"parent", {}}, {"aunt", {}}, {"uncle", {}}, {"sibling", {}}},
+            {{"parent", {}}, {"aunt", {}}, {"uncle", {}}, {"sibling", {}}},
+            {
+                Port("init", Operator::F_INIT),
+                Port("out", Operator::O_I, Timeline("component")),
+                Port("final", Operator::O_F)
+            }
+        );
+        port_manager_.connect_ports(peer_info);
+        communicator_.set_peer_info(peer_info);
+    }
+
+    void TearDown() override {
+        communicator_.shutdown();
+    }
+};
+
 class IterationOrMilestone {
     public:
         IterationCount iteration;
@@ -164,19 +214,20 @@ void mock_receive_messages(
 TEST_P(libmuscle_repeater_communicator, repeater_filters) {
     mock_receive_messages({
         {"component.twicerepeated", {
-            I({}), M({})
+            I({0}), M({})
         }},
         {"component.repeated", {
-            I({0}), I({1}), I({2}), M({})
+            I({0, 0}), I({0, 1}), I({0, 2}), M({0}), M({})
         }},
         {"component.unfiltered", {
-            I({0, 0}),
-            I({0, 1}),
-            M({0}),
+            I({0, 0, 0}),
+            I({0, 0, 1}),
+            M({0, 0}),
             // parent is allowed to send 0 messages on its O_I port in an iteration
-            M({1}),
-            I({2, 0}),
-            M({2}),
+            M({0, 1}),
+            I({0, 2, 0}),
+            M({0, 2}),
+            M({0}),
             M({})
         }}
     });
@@ -184,27 +235,27 @@ TEST_P(libmuscle_repeater_communicator, repeater_filters) {
     bool is_padded = GetParam() == "pad";
 
     auto cache = communicator_.pre_receive();
-    ASSERT_EQ(decode_iteration(cache.at("unfiltered").data()).get(), IterationCount({0, 0}));
-    ASSERT_EQ(decode_iteration(cache.at("repeated").data()).get(), IterationCount({0}));
-    ASSERT_EQ(decode_iteration(cache.at("twicerepeated").data()).get(), IterationCount({}));
+    ASSERT_EQ(decode_iteration(cache.at("unfiltered").data()).get(), IterationCount({0, 0, 0}));
+    ASSERT_EQ(decode_iteration(cache.at("repeated").data()).get(), IterationCount({0, 0}));
+    ASSERT_EQ(decode_iteration(cache.at("twicerepeated").data()).get(), IterationCount({0}));
 
     cache = communicator_.pre_receive();
-    ASSERT_EQ(decode_iteration(cache.at("unfiltered").data()).get(), IterationCount({0, 1}));
+    ASSERT_EQ(decode_iteration(cache.at("unfiltered").data()).get(), IterationCount({0, 0, 1}));
     if (is_padded) {
         ASSERT_TRUE(cache.at("repeated").data().is_nil());
         ASSERT_TRUE(cache.at("twicerepeated").data().is_nil());
     } else {
-        ASSERT_EQ(decode_iteration(cache.at("repeated").data()).get(), IterationCount({0}));
-        ASSERT_EQ(decode_iteration(cache.at("twicerepeated").data()).get(), IterationCount({}));
+        ASSERT_EQ(decode_iteration(cache.at("repeated").data()).get(), IterationCount({0, 0}));
+        ASSERT_EQ(decode_iteration(cache.at("twicerepeated").data()).get(), IterationCount({0}));
     }
 
     cache = communicator_.pre_receive();
-    ASSERT_EQ(decode_iteration(cache.at("unfiltered").data()).get(), IterationCount({2, 0}));
-    ASSERT_EQ(decode_iteration(cache.at("repeated").data()).get(), IterationCount({2}));
+    ASSERT_EQ(decode_iteration(cache.at("unfiltered").data()).get(), IterationCount({0, 2, 0}));
+    ASSERT_EQ(decode_iteration(cache.at("repeated").data()).get(), IterationCount({0, 2}));
     if (is_padded) {
         ASSERT_TRUE(cache.at("twicerepeated").data().is_nil());
     } else {
-        ASSERT_EQ(decode_iteration(cache.at("twicerepeated").data()).get(), IterationCount({}));
+        ASSERT_EQ(decode_iteration(cache.at("twicerepeated").data()).get(), IterationCount({0}));
     }
 
     ASSERT_THROW(communicator_.pre_receive(), PortClosed);
@@ -277,3 +328,89 @@ TEST_P(libmuscle_repeater_communicator, repeater_filters_discard_messages) {
 
 INSTANTIATE_TEST_SUITE_P(
     repeated, libmuscle_repeater_communicator, ::testing::Values("repeat", "pad"));
+
+
+TEST_F(libmuscle_reducer_communicator, reducer_filters) {
+    mock_receive_messages({{"component.init", {I({0}), I({1}), M({})}}});
+
+    auto & deposit = communicator_.server_.deposit;
+
+    auto cache = communicator_.pre_receive();
+    ASSERT_EQ(decode_iteration(cache.at("init").data()).get(), IterationCount({0}));
+    // Send some messages on O_I
+    for (std::size_t i = 0; i < 5; ++i) {
+        communicator_.send_message("out", Message(i, Data(i), Settings()));
+        ASSERT_FALSE(deposit.called());
+    }
+    // Send on O_F
+    communicator_.send_message("final", Message(5, Data("data"), Settings()));
+    ASSERT_EQ(deposit.call_args_list.size(), 2);
+    ASSERT_EQ(std::get<0>(deposit.call_args_list[0]), "parent.in");
+    ASSERT_EQ(std::get<0>(deposit.call_args_list[1]), "sibling.in2");
+    deposit.call_args_list.clear();
+
+    // Pre-receive will send cached LAST message to sibling.in
+    cache = communicator_.pre_receive();
+    ASSERT_EQ(decode_iteration(cache.at("init").data()).get(), IterationCount({1}));
+    // N.B. we don't send the [1] milestone to sibling.in due to the LAST filter, only
+    // the cached message
+    ASSERT_EQ(deposit.call_args_list.size(), 1);
+    ASSERT_EQ(std::get<0>(deposit.call_args_list[0]), "sibling.in");
+    auto sent_message = std::get<1>(deposit.call_args_list[0]);
+    ASSERT_EQ(sent_message->timestamp, 4.0);
+    deposit.call_args_list.clear();
+
+    // Skip O_I and send on O_F
+    communicator_.send_message("final", Message(10, Data("data"), Settings()));
+    ASSERT_EQ(deposit.call_args_list.size(), 2);
+    ASSERT_EQ(std::get<0>(deposit.call_args_list[0]), "parent.in");
+    ASSERT_EQ(std::get<0>(deposit.call_args_list[1]), "sibling.in2");
+    deposit.call_args_list.clear();
+
+    // Pre-receive will first send cached LAST message to sibling.in, then receive
+    // Milestone([]) and trigger:
+    // - Cached LAST message on "final" to aunt.init
+    // - Cached LAST LAST message on "out" to uncle.init
+    // - Milestone([]) to sibling.in, sibling.in2, parent.in
+    ASSERT_THROW(communicator_.pre_receive(), PortClosed);
+    ASSERT_EQ(deposit.call_args_list.size(), 6.0);
+
+    std::unordered_map<std::string, std::vector<std::shared_ptr<MPPMessage>>> messages_per_peer_port;
+    for (auto & call : deposit.call_args_list) {
+        std::string peer_port(std::get<0>(call));
+        std::shared_ptr<MPPMessage> message(std::get<1>(call));
+        auto it = messages_per_peer_port.find(peer_port);
+        if (it == messages_per_peer_port.end()) {
+            // Not found
+            messages_per_peer_port.emplace(
+                    peer_port, std::vector<std::shared_ptr<MPPMessage>>({message}));
+        } else {
+            it->second.emplace_back(message);
+        }
+    }
+
+    // O_I -> last -> sibling.in
+    auto & messages = messages_per_peer_port.at("sibling.in");
+    ASSERT_EQ(messages.size(), 2);
+    // No message was sent on O_I this reuse loop, so LAST generates an empty message
+    ASSERT_EQ(messages[0]->timestamp, -std::numeric_limits<double>::infinity());
+    ASSERT_TRUE(messages[0]->data.is_nil());
+    ASSERT_TRUE(is_milestone(messages[1]->data));
+    ASSERT_TRUE(Milestone(messages[1]->data).is_final_milestone());
+
+    // Just milestones
+    for (auto & peer_port : {"sibling.in2", "parent.in"}) {
+        messages = messages_per_peer_port.at(peer_port);
+        ASSERT_EQ(messages.size(), 1);
+        ASSERT_TRUE(is_milestone(messages[0]->data));
+        ASSERT_TRUE(Milestone(messages[0]->data).is_final_milestone());
+    }
+
+    // O_I -> last last -> uncle.init
+    ASSERT_EQ(messages_per_peer_port.at("uncle.init").size(), 1);
+    ASSERT_EQ(messages_per_peer_port.at("uncle.init")[0]->timestamp, 4.0);
+
+    // O_F -> last -> aunt.init
+    ASSERT_EQ(messages_per_peer_port.at("aunt.init").size(), 1);
+    ASSERT_EQ(messages_per_peer_port.at("aunt.init")[0]->timestamp, 10.0);
+}
diff --git a/src/cpp/libmuscle/tests/test_timeline_manager.cpp b/src/cpp/libmuscle/tests/test_timeline_manager.cpp
index 67db7d75..c0f792ee 100644
--- a/src/cpp/libmuscle/tests/test_timeline_manager.cpp
+++ b/src/cpp/libmuscle/tests/test_timeline_manager.cpp
@@ -119,9 +119,9 @@ struct libmuscle_timeline_manager : ::testing::Test {
     void create(
             PortsDescription const & declared_ports,
             std::unordered_map<std::string, Timeline> const & timelines = {},
-            bool include_settings = false) {
+            bool include_settings = false, Timeline timeline = Timeline(":")) {
         port_manager_ = make_port_manager(declared_ports, timelines, {}, include_settings);
-        tm_ = std::make_unique<TimelineManager>(*port_manager_);
+        tm_ = std::make_unique<TimelineManager>(*port_manager_, timeline);
         ASSERT_FALSE(tm_->start_reuse_iteration().is_set());
     }
 };
@@ -146,7 +146,8 @@ struct libmuscle_full_timeline_manager : libmuscle_timeline_manager {
                     {"out_a2_2", Timeline(":A2")},
                     {"in_a2", Timeline(":A2")},
                 },
-                true);
+                true,
+                Timeline(":a"));
     }
 };
 
@@ -155,7 +156,7 @@ struct libmuscle_vector_timeline_manager : libmuscle_timeline_manager {
     libmuscle_vector_timeline_manager() {
         port_manager_ = make_port_manager(
                 PortsDescription{{Operator::O_F, {"out_v[]"}}}, {}, {{"out_v", {3}}});
-        tm_ = std::make_unique<TimelineManager>(*port_manager_);
+        tm_ = std::make_unique<TimelineManager>(*port_manager_, Timeline(":"));
         tm_->start_reuse_iteration();
         EXPECT_EQ(tm_->check_pre_received_iteration_counts({}), IterationCount());
     }
@@ -340,7 +341,7 @@ TEST_F(libmuscle_full_timeline_manager, get_state_and_restore_state_round_trip)
 
     TimelineState state = tm_->get_state();
 
-    TimelineManager restored(*port_manager_);
+    TimelineManager restored(*port_manager_, Timeline(":a"));
     restored.restore_state(state);
 
     TimelineState restored_state = restored.get_state();
diff --git a/src/cpp/libmuscle/timeline_manager.cpp b/src/cpp/libmuscle/timeline_manager.cpp
index 038831bb..d7b431df 100644
--- a/src/cpp/libmuscle/timeline_manager.cpp
+++ b/src/cpp/libmuscle/timeline_manager.cpp
@@ -438,8 +438,9 @@ void SubTimelineManager::missing_actions(ExpectedActions & result) const {
 // when a test mocks TimelineManager but still includes this file for them.
 #ifndef LIBMUSCLE_MOCK_TIMELINE_MANAGER
 
-TimelineManager::TimelineManager(PortManager const & port_manager)
-    : port_manager_(port_manager)
+TimelineManager::TimelineManager(PortManager const & port_manager, Timeline const & timeline)
+    : timeline_(timeline)
+    , port_manager_(port_manager)
     , receive_(port_manager.get_connected_ports(Operator::F_INIT, Optional<Timeline>()))
     , send_(port_manager.get_connected_ports(Operator::O_F, Optional<Timeline>()))
     , submanagers_()
@@ -472,6 +473,11 @@ IterationCount TimelineManager::check_pre_received_iteration_counts(
         throw std::runtime_error(
             "Internal error: received F_INIT iteration count " + to_string(new_iteration)
             + " is not newer than the previous iteration " + to_string(iteration_.get()));
+    if (new_iteration.size() != timeline_.size())
+        throw std::runtime_error(
+            "Received unexpected F_INIT iteration count: " + to_string(new_iteration)
+            + ". Was expecting an iteration count with " + std::to_string(timeline_.size())
+            + " elements, since we are in timeline " + std::string(timeline_));
     iteration_ = new_iteration;
     return new_iteration;
 }
diff --git a/src/cpp/libmuscle/timeline_manager.hpp b/src/cpp/libmuscle/timeline_manager.hpp
index 0fddda0c..8e13ae75 100644
--- a/src/cpp/libmuscle/timeline_manager.hpp
+++ b/src/cpp/libmuscle/timeline_manager.hpp
@@ -290,7 +290,8 @@ class TimelineManager {
          * @param port_manager The (already connected) port manager for this
          *      instance.
          */
-        explicit TimelineManager(PortManager const & port_manager);
+        explicit TimelineManager(
+                PortManager const & port_manager, ::ymmsl::Timeline const & timeline);
 
         /** Check and update the timeline state before sending on the given
          * port.
@@ -371,6 +372,7 @@ class TimelineManager {
     private:
         IterationCount check_send_o_f_(Port const & port, Optional<int> slot);
 
+        ::ymmsl::Timeline timeline_;
         PortManager const & port_manager_;
         TimelinePorts receive_;
         TimelinePorts send_;
diff --git a/src/cpp/ymmsl/ports.cpp b/src/cpp/ymmsl/ports.cpp
index de21ca39..3c1f9dc4 100644
--- a/src/cpp/ymmsl/ports.cpp
+++ b/src/cpp/ymmsl/ports.cpp
@@ -1,5 +1,7 @@
 #include <ymmsl/ports.hpp>
 
+#include <algorithm>
+
 
 ::std::size_t (::std::hash<::ymmsl::impl::Timeline>::operator())(
         argument_type const & timeline) const noexcept
@@ -21,6 +23,14 @@ bool Timeline::operator==(Timeline const & rhs) const {
     return timeline_ == rhs.timeline_;
 }
 
+std::size_t Timeline::size() const {
+    if (timeline_.empty() || timeline_ == ":") return 0;
+    std::size_t num_colons = std::count(timeline_.begin(), timeline_.end(), ':');
+    if (timeline_[0] != ':')
+        return num_colons + 1;
+    return num_colons;
+}
+
 Port::Port(Identifier const & name, Operator oper, Timeline const & timeline)
     : name(name)
     , oper(oper)
diff --git a/src/cpp/ymmsl/ports.hpp b/src/cpp/ymmsl/ports.hpp
index b7ed2742..dcc7a142 100644
--- a/src/cpp/ymmsl/ports.hpp
+++ b/src/cpp/ymmsl/ports.hpp
@@ -55,6 +55,10 @@ class Timeline {
          */
         bool operator==(Timeline const & rhs) const;
 
+        /** Return the number of parts in the Timeline.
+         */
+        std::size_t size() const;
+
     private:
         std::string timeline_;
 };
diff --git a/src/cpp/ymmsl/tests/test_ports.cpp b/src/cpp/ymmsl/tests/test_ports.cpp
index be052974..29ae6784 100644
--- a/src/cpp/ymmsl/tests/test_ports.cpp
+++ b/src/cpp/ymmsl/tests/test_ports.cpp
@@ -26,6 +26,15 @@ TEST(ymmsl_timeline, hash_consistent_with_equality) {
     ASSERT_EQ(hasher(Timeline(":A1")), hasher(Timeline(":A1")));
 }
 
+TEST(ymmsl_timeline, size) {
+    ASSERT_EQ(Timeline("").size(), 0);
+    ASSERT_EQ(Timeline(":").size(), 0);
+    ASSERT_EQ(Timeline("a").size(), 1);
+    ASSERT_EQ(Timeline(":a").size(), 1);
+    ASSERT_EQ(Timeline("a:b").size(), 2);
+    ASSERT_EQ(Timeline(":a:b").size(), 2);
+}
+
 TEST(ymmsl_port, test_port) {
     auto ep1 = Port(Identifier("test_in"), Operator::F_INIT);
 
diff --git a/src/python/libmuscle/communicator.py b/src/python/libmuscle/communicator.py
index 465512c8..3da4c6a2 100644
--- a/src/python/libmuscle/communicator.py
+++ b/src/python/libmuscle/communicator.py
@@ -322,7 +322,7 @@ def _apply_reduce_filters(
         sent, or a cached message, or nothing at all.
 
         Args:
-            recv_endpoint: The Endpoint to send to.
+            peer_port: Peer port (component + port) to send to.
             message: MPPMessage to be checked.
 
         Returns:
diff --git a/src/python/libmuscle/test/test_communicator_conduit_filters.py b/src/python/libmuscle/test/test_communicator_conduit_filters.py
index 99dcd6c0..592eefaf 100644
--- a/src/python/libmuscle/test/test_communicator_conduit_filters.py
+++ b/src/python/libmuscle/test/test_communicator_conduit_filters.py
@@ -1,5 +1,5 @@
 from typing import Union
-from unittest.mock import ANY, MagicMock, patch
+from unittest.mock import ANY, MagicMock, call, patch
 
 import pytest
 from ymmsl.v0_2 import Conduit, ConduitFilter, Operator, Port, Settings, Timeline
@@ -314,7 +314,10 @@ def test_reducer_filters(reducer_communicator, mpp_client, mpp_server):
         mpp_server.deposit.assert_not_called()
     # Send on O_F
     reducer_communicator.send_message("final", Message(5, data="data"))
-    mpp_server.deposit.assert_called_with("sibling.in2", ANY)
+    assert mpp_server.deposit.call_args_list == [
+        call("parent.in", ANY),
+        call("sibling.in2", ANY),
+    ]
     mpp_server.deposit.reset_mock()
 
     # Pre-receive will send cached LAST message to sibling.in
@@ -329,7 +332,10 @@ def test_reducer_filters(reducer_communicator, mpp_client, mpp_server):
 
     # Skip O_I and send on O_F
     reducer_communicator.send_message("final", Message(10, data="data"))
-    mpp_server.deposit.assert_called_with("sibling.in2", ANY)
+    assert mpp_server.deposit.call_args_list == [
+        call("parent.in", ANY),
+        call("sibling.in2", ANY),
+    ]
     mpp_server.deposit.reset_mock()
 
     # Pre-receive will first send cached LAST message to sibling.in, then receive
@@ -342,9 +348,9 @@ def test_reducer_filters(reducer_communicator, mpp_client, mpp_server):
     assert mpp_server.deposit.call_count == 6
 
     messages_per_peer_port = {}
-    for call in mpp_server.deposit.call_args_list:
-        msg = MPPMessage.from_bytes(call.args[1])
-        messages_per_peer_port.setdefault(call.args[0], []).append(msg)
+    for item in mpp_server.deposit.call_args_list:
+        msg = MPPMessage.from_bytes(item.args[1])
+        messages_per_peer_port.setdefault(item.args[0], []).append(msg)
 
     # O_I -> last -> sibling.in
     assert len(messages_per_peer_port["sibling.in"]) == 2

From 2e2a224c17c86d5fe556085613f5b5786fffe0fc Mon Sep 17 00:00:00 2001
From: Maarten Sebregts <msebregts@ignitioncomputing.com>
Date: Thu, 3 Sep 2026 17:07:00 +0200
Subject: [PATCH 5/8] Add C++ test for checking if F_INIT iteration counts
 match timeline

---
 src/cpp/libmuscle/tests/test_timeline_manager.cpp | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/src/cpp/libmuscle/tests/test_timeline_manager.cpp b/src/cpp/libmuscle/tests/test_timeline_manager.cpp
index c0f792ee..45a83e82 100644
--- a/src/cpp/libmuscle/tests/test_timeline_manager.cpp
+++ b/src/cpp/libmuscle/tests/test_timeline_manager.cpp
@@ -196,6 +196,11 @@ TEST_F(libmuscle_full_timeline_manager, record_received_message_f_init_raises_on
     ASSERT_THROW(tm_->check_pre_received_iteration_counts({{3}, {4}}), std::logic_error);
 }
 
+TEST_F(libmuscle_full_timeline_manager, check_pre_receive_counts_match_timeline) {
+    ASSERT_THROW(tm_->check_pre_received_iteration_counts({{}}), std::runtime_error);
+    ASSERT_THROW(tm_->check_pre_received_iteration_counts({{1, 2}}), std::runtime_error);
+}
+
 TEST_F(libmuscle_timeline_manager, o_f_can_send_immediately_when_no_f_init_connections) {
     create(PortsDescription{{Operator::O_F, {"out_f"}}});
     tm_->check_pre_received_iteration_counts({});

From 8c97be6107060415539221fb3b4a085aad1cf412 Mon Sep 17 00:00:00 2001
From: Iris van der Werf <ivanderwerf@ignitioncomputing.com>
Date: Tue, 8 Sep 2026 17:03:33 +0200
Subject: [PATCH 6/8] Update the tests

---
 src/cpp/libmuscle/tests/test_timeline_manager.cpp  | 4 ++--
 src/python/libmuscle/test/test_timeline_manager.py | 2 +-
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/src/cpp/libmuscle/tests/test_timeline_manager.cpp b/src/cpp/libmuscle/tests/test_timeline_manager.cpp
index 9ef95035..a435b266 100644
--- a/src/cpp/libmuscle/tests/test_timeline_manager.cpp
+++ b/src/cpp/libmuscle/tests/test_timeline_manager.cpp
@@ -197,8 +197,8 @@ TEST_F(libmuscle_full_timeline_manager, record_received_message_f_init_raises_on
 }
 
 TEST_F(libmuscle_full_timeline_manager, check_pre_receive_counts_match_timeline) {
-    ASSERT_THROW(tm_->check_pre_received_iteration_counts({{}}), std::runtime_error);
-    ASSERT_THROW(tm_->check_pre_received_iteration_counts({{1, 2}}), std::runtime_error);
+    ASSERT_THROW(tm_->record_pre_received_iteration_counts({{}}), std::runtime_error);
+    ASSERT_THROW(tm_->record_pre_received_iteration_counts({{1, 2}}), std::runtime_error);
 }
 
 TEST_F(libmuscle_timeline_manager, o_f_can_send_immediately_when_no_f_init_connections) {
diff --git a/src/python/libmuscle/test/test_timeline_manager.py b/src/python/libmuscle/test/test_timeline_manager.py
index 8be84430..88617eb4 100644
--- a/src/python/libmuscle/test/test_timeline_manager.py
+++ b/src/python/libmuscle/test/test_timeline_manager.py
@@ -300,7 +300,7 @@ def test_check_pre_receive_counts_match_timeline(
     timeline_manager: TimelineManager,
 ) -> None:
     with pytest.raises(RuntimeError, match="iteration count with 0 elements"):
-        timeline_manager.check_pre_received_iteration_counts([[1]])
+        timeline_manager.record_pre_received_iteration_counts([[1]])
 
 
 @pytest.mark.parametrize("timeline", [Timeline(":a")], indirect=True)

From d50fe836ec613392090a43725be8c0930e538fe4 Mon Sep 17 00:00:00 2001
From: Iris van der Werf <ivanderwerf@ignitioncomputing.com>
Date: Mon, 14 Sep 2026 11:13:32 +0200
Subject: [PATCH 7/8] include comments

---
 src/cpp/libmuscle/communicator.cpp            |  6 +-
 src/cpp/libmuscle/communicator.hpp            | 17 ++---
 src/python/libmuscle/communicator.py          | 20 +++---
 .../test/test_communicator_conduit_filters.py | 69 ++++++++++++++++++-
 4 files changed, 90 insertions(+), 22 deletions(-)

diff --git a/src/cpp/libmuscle/communicator.cpp b/src/cpp/libmuscle/communicator.cpp
index 7382a9ca..68379c98 100644
--- a/src/cpp/libmuscle/communicator.cpp
+++ b/src/cpp/libmuscle/communicator.cpp
@@ -165,7 +165,7 @@ void Communicator::send_message(
         
         std::vector<char> message_bytes;
         auto peer_port = recv_endpoint.kernel + recv_endpoint.port;
-        if (reduced_count_.count(peer_port) > 0) {
+        if (outgoing_timeline_length_.count(peer_port) > 0) {
             message_bytes = apply_reduce_filters_(peer_port, std::move(mpp_message));
             if (message_bytes.empty())
                 continue;
@@ -190,7 +190,7 @@ std::vector<char> Communicator::apply_reduce_filters_(
         ymmsl::Reference const & peer_port, MPPMessage && message) {
     message.message_number = -1;  // GH#411: Disabled checkpointing for reducer filter
 
-    auto reduced_count = reduced_count_.at(peer_port);
+    auto reduced_count = outgoing_timeline_length_.at(peer_port);
 
     if (!is_milestone(message.data)) {
         // Reduce the message iteration count to match with the timeline we send to
@@ -633,7 +633,7 @@ void Communicator::prepare_conduit_filters_() {
                         filters.begin(), filters.end(), ::ymmsl::is_reducer);
                 if (n_reducers > 0) {
                     std::size_t reduced_count = timeline_.get().size() + port.timeline.size() - n_reducers;
-                    reduced_count_.emplace(peer_port, reduced_count);
+                    outgoing_timeline_length_.emplace(peer_port, reduced_count);
                 }
             }
         }
diff --git a/src/cpp/libmuscle/communicator.hpp b/src/cpp/libmuscle/communicator.hpp
index 8c0e57ca..62c72110 100644
--- a/src/cpp/libmuscle/communicator.hpp
+++ b/src/cpp/libmuscle/communicator.hpp
@@ -262,16 +262,17 @@ class Communicator {
         MPPCacheType message_cache_;
 
         /** Size of IterationCount, after applying the reducer filters, per peer port.
-         * 
-         * Keys are references to peer ports: ``component + port``. The reduced count is
-         * the size of the IterationCount after applying the reducer filters and determines
-         * in which (parent) timeline these messages are sent.
+         *
+         * Keys are references to peer ports: ``component + port``. The outgoing
+         * timeline length is the size of the IterationCount after applying the
+         * reducer filters and determines in which (parent) timeline these messages
+         * are sent.
          * If our timeline is ":macro:micro" then:
-         * - reduced_count = 0: send on the root (":") timeline
-         * - reduced_count = 1: send on the ":macro" timeline
-         * - reduced_count = 2: send on the ":macro:micro" timeline
+         * - outgoing_timeline_length = 0: send on the root (":") timeline
+         * - outgoing_timeline_length = 1: send on the ":macro" timeline
+         * - outgoing_timeline_length = 2: send on the ":macro:micro" timeline
          */
-        std::unordered_map<::ymmsl::Reference, std::size_t> reduced_count_;
+        std::unordered_map<::ymmsl::Reference, std::size_t> outgoing_timeline_length_;
         /** Message cache for reducer filters */
         MPPCacheType reducer_cache_;
 };
diff --git a/src/python/libmuscle/communicator.py b/src/python/libmuscle/communicator.py
index a0fde048..4c42aa1a 100644
--- a/src/python/libmuscle/communicator.py
+++ b/src/python/libmuscle/communicator.py
@@ -155,17 +155,17 @@ def __init__(
         self._message_cache: MPPCacheType = {}
         """Message cache for pre-received messages."""
 
-        self._reduced_count: dict[Reference, int] = {}
+        self._outgoing_timeline_length: dict[Reference, int] = {}
         """Size of IterationCount, after applying the reducer filters, per peer port.
 
-        Keys are references to peer ports: ``component + port``. The reduced count is
-        the size of the IterationCount after applying the reducer filters and determines
-        in which (parent) timeline these messages are sent.
+        Keys are references to peer ports: ``component + port``. The outgoing timeline
+        length is the size of the IterationCount after applying the reducer filters and
+        determines in which (parent) timeline these messages are sent.
 
         If our timeline is ":macro:micro" then:
-        - reduced_count = 0: send on the root (":") timeline
-        - reduced_count = 1: send on the ":macro" timeline
-        - reduced_count = 2: send on the ":macro:micro" timeline
+        - outgoing_timeline_length = 0: send on the root (":") timeline
+        - outgoing_timeline_length = 1: send on the ":macro" timeline
+        - outgoing_timeline_length = 2: send on the ":macro:micro" timeline
         """
         self._reducer_cache: dict[Reference, MPPMessage] = {}
         """Message cache for reducer filters."""
@@ -294,7 +294,7 @@ def send_message(
                 iteration,
             )
             peer_port = recv_endpoint.kernel + recv_endpoint.port
-            if peer_port in self._reduced_count:
+            if peer_port in self._outgoing_timeline_length:
                 result = self._apply_reduce_filters(peer_port, mpp_message)
                 if result is None:
                     continue
@@ -330,7 +330,7 @@ def _apply_reduce_filters(
         """
         message.message_number = -1  # GH#411: Disabled checkpointing for reducer filter
 
-        reduced_count = self._reduced_count[peer_port]
+        reduced_count = self._outgoing_timeline_length[peer_port]
 
         if not isinstance(message.data, Milestone):
             # Reduce the message iteration count to match with the timeline we send to
@@ -788,7 +788,7 @@ def _prepare_conduit_filters(self) -> None:
                     # Count the reducer filters, receiving component handles repeaters
                     n_reducers = sum(1 for filter in filters if filter.is_reducer())
                     if n_reducers > 0:
-                        self._reduced_count[peer_port] = (
+                        self._outgoing_timeline_length[peer_port] = (
                             len(self._timeline) + len(port.timeline) - n_reducers
                         )
 
diff --git a/src/python/libmuscle/test/test_communicator_conduit_filters.py b/src/python/libmuscle/test/test_communicator_conduit_filters.py
index 592eefaf..8f5b5b6c 100644
--- a/src/python/libmuscle/test/test_communicator_conduit_filters.py
+++ b/src/python/libmuscle/test/test_communicator_conduit_filters.py
@@ -103,7 +103,7 @@ def reducer_communicator(mpp_client, mpp_server):
     )
     port_manager.connect_ports(peer_info)
     communicator.set_peer_info(peer_info)
-    assert communicator._reduced_count == {
+    assert communicator._outgoing_timeline_length == {
         "sibling.in": 1,
         "aunt.init": 0,
         "uncle.init": 0,
@@ -112,6 +112,48 @@ def reducer_communicator(mpp_client, mpp_server):
     communicator.shutdown()
 
 
+@pytest.fixture
+def repeater_reducer_communicator(repeat_filter, mpp_client, mpp_server):
+    conduit = Conduit("component.out", "sibling.in", "last " + repeat_filter)
+
+    port_manager = PortManager([], None)
+    mock_manager = MagicMock()
+    mock_manager.get_timeline.return_value = Timeline(":")
+    component = Communicator(
+        Ref("component"), [], port_manager, MagicMock(), mock_manager
+    )
+    peer_info = PeerInfo(
+        Ref("component"),
+        [],
+        [conduit],
+        {Ref("sibling"): []},
+        {Ref("sibling"): []},
+        [Port(Id("out"), Operator.O_I, Timeline("component"))],
+    )
+    port_manager.connect_ports(peer_info)
+    component.set_peer_info(peer_info)
+
+    sibling_port_manager = PortManager([], None)
+    sibling = Communicator(
+        Ref("sibling"), [], sibling_port_manager, MagicMock(), MagicMock()
+    )
+    sibling_peer_info = PeerInfo(
+        Ref("sibling"),
+        [],
+        [conduit],
+        {Ref("component"): []},
+        {Ref("component"): []},
+        [Port(Id("in"), Operator.S)],
+    )
+    sibling_port_manager.connect_ports(sibling_peer_info)
+    sibling.set_peer_info(sibling_peer_info)
+    assert sibling._repeat_filters == {"in": [ConduitFilter(repeat_filter)]}
+
+    component.pre_receive()
+    yield component, sibling
+    component.shutdown()
+
+
 def mock_receive_messages(
     mpp_client, data: dict[str, list[Union[IterationCount, Milestone]]]
 ):
@@ -373,3 +415,28 @@ def test_reducer_filters(reducer_communicator, mpp_client, mpp_server):
     # O_F -> last -> aunt.init
     assert len(messages_per_peer_port["aunt.init"]) == 1
     assert messages_per_peer_port["aunt.init"][0].timestamp == 10
+
+
+def test_combined_reducer_and_repeater_filters(
+    repeater_reducer_communicator, mpp_client, mpp_server, repeat_filter
+):
+    component, sibling = repeater_reducer_communicator
+
+    for i in range(3):
+        component.send_message("out", Message(i, data=f"value_{i}"))
+    mpp_server.deposit.assert_not_called()  # messages are cached, not yet forwarded
+
+    # Closing broadcasts the closing milestone, which releases the cached last message.
+    component._close_outgoing_ports()
+    assert mpp_server.deposit.call_count == 1
+    peer, encoded = mpp_server.deposit.call_args.args
+    assert peer == "sibling.in"
+
+    # Feed the bytes "component" put on the wire into "sibling"'s receive.
+    mpp_client.receive.return_value = (encoded, MagicMock())
+    sibling.pre_receive()
+
+    is_padded = repeat_filter == "pad"
+    for i in range(3):
+        msg = sibling.receive_s_message("in")
+        assert msg.data == (None if i and is_padded else "value_2")

From 26a792f4f1272161a8f8988cc49c546212a01078 Mon Sep 17 00:00:00 2001
From: Iris van der Werf <ivanderwerf@ignitioncomputing.com>
Date: Mon, 14 Sep 2026 11:53:29 +0200
Subject: [PATCH 8/8] add a combined filter within the integration test

---
 integration_test/test_filters.py | 43 ++++++++++++++++++++++++++++++--
 1 file changed, 41 insertions(+), 2 deletions(-)

diff --git a/integration_test/test_filters.py b/integration_test/test_filters.py
index cb75cd64..a87b630a 100644
--- a/integration_test/test_filters.py
+++ b/integration_test/test_filters.py
@@ -133,6 +133,27 @@ def postprocess() -> None:
             assert micro.data[-1] == 1
 
 
+def combined(filters: str) -> None:
+    is_padded = filters.split()[-1] == "pad"
+
+    instance = Instance({Operator.F_INIT: ["trigger"], Operator.S: ["in"]})
+
+    reused = 0
+    while instance.reuse_instance():
+        instance.receive("trigger")
+
+        last_value = None if reused == 0 else ["macro", reused, "meso", reused - 1]
+        for i in range(3):
+            msg = instance.receive("in")
+            if i and is_padded:
+                assert msg.data is None
+            else:
+                assert msg.data == last_value
+        reused += 1
+
+    assert reused == 4
+
+
 config = """
 ymmsl_version: v0.2
 models:
@@ -172,18 +193,26 @@ def postprocess() -> None:
         ports:
           f_init: macro meso micro
         implementation: postprocess
+      combined:
+        description: Receives meso.out with a combined reducer and repeater filter
+        ports:
+          f_init: trigger
+          s: in
+        implementation: combined
     conduits:
       macro.out:
       - meso.in
       - {filters} pico.macro
       - {filters} repeat_s.macro
       - postprocess.macro
+      - combined.trigger
       meso.out:
       - micro.in
       - repeat pico.meso
       - repeat_s.meso
       - repeat repeat_s.repeated_meso
       - last postprocess.meso
+      - last {combined_filter} combined.in
       micro.out:
       - pico.micro
       - repeat_s.micro
@@ -200,8 +229,13 @@ def test_repeater_filters(tmp_path, filters):
         "repeat_s": ("python", repeat_s, filters),
         "pico": ("python", pico, filters),
         "postprocess": ("python", postprocess),
+        "combined": ("python", combined, filters),
     }
-    run_manager_with_actors(config.format(filters=filters), tmp_path, actors)
+    run_manager_with_actors(
+        config.format(filters=filters, combined_filter=filters.split()[-1]),
+        tmp_path,
+        actors,
+    )
 
 
 @skip_if_python_only
@@ -214,8 +248,13 @@ def test_repeater_filters_cpp(tmp_path, filters):
         "repeat_s": ("cpp", "conduit_filters_test", "repeat_s", filters),
         "pico": ("cpp", "conduit_filters_test", "pico", filters),
         "postprocess": ("python", postprocess),
+        "combined": ("python", combined, filters),
     }
-    run_manager_with_actors(config.format(filters=filters), tmp_path, actors)
+    run_manager_with_actors(
+        config.format(filters=filters, combined_filter=filters.split()[-1]),
+        tmp_path,
+        actors,
+    )
 
 
 checkpoint_config = """