fix(aggregators): flush trailing log-run repeat counts

src/rai_core/rai/aggregators/ros2/aggregators.py

@@ -33,21 +33,33 @@ class ROS2LogsAggregator(BaseAggregator[Log]):
 
     def get(self) -> HumanMessage:
         msgs = self.get_buffer()
-        buffer = []
-        prev_parsed = None
+        buffer: List[str] = []
+        prev_parsed: str | None = None
         counter = 0
+
+        def flush_run() -> None:
+            nonlocal counter, prev_parsed
+            if prev_parsed is None:
+                return
+            buffer.append(prev_parsed)
+            if counter > 0:
+                buffer.append(f"Log above repeated {counter} times")
+            counter = 0
+
         for log in msgs:
             level = self.levels[log.level]
             parsed = f"[{log.name}] [{level}] [{log.function}] {log.msg}"
+            if prev_parsed is None:
+                prev_parsed = parsed
+                counter = 0
+                continue
             if parsed == prev_parsed:
                 counter += 1
                 continue
-            else:
-                if counter != 0:
-                    parsed = f"Log above repeated {counter} times"
-            buffer.append(parsed)
-            counter = 0
+            flush_run()
             prev_parsed = parsed
+        flush_run()
+
         result = f"Logs summary: {list(dict.fromkeys(buffer))}"
         self.clear_buffer()
         return HumanMessage(content=result)

tests/aggregators/test_ros2_logs_aggregator.py

@@ -58,11 +58,61 @@ def test_ros2_logs_aggregator_deduplicates_and_clears_buffer():
     assert isinstance(summary, HumanMessage)
     assert (
         summary.content
-        == "Logs summary: ['[demo_node] [WARNING] [do_work] System warming up', 'Log above repeated 1 times']"
+        == "Logs summary: ['[demo_node] [WARNING] [do_work] System warming up', "
+        "'Log above repeated 1 times', "
+        "'[demo_node] [ERROR] [do_work] System failure detected']"
     )
     assert aggregator.get_buffer() == []
 
 
+def test_ros2_logs_aggregator_trailing_run_flushes_repeat_count():
+    aggregator = ROS2LogsAggregator()
+    for _ in range(3):
+        aggregator(
+            DummyLog(
+                level=30,
+                name="demo_node",
+                function="do_work",
+                msg="System warming up",
+            )
+        )
+
+    summary = aggregator.get()
+    assert summary.content == (
+        "Logs summary: ['[demo_node] [WARNING] [do_work] System warming up', "
+        "'Log above repeated 2 times']"
+    )
+    assert aggregator.get_buffer() == []
+
+
+def test_ros2_logs_aggregator_trailing_second_run():
+    aggregator = ROS2LogsAggregator()
+    aggregator(
+        DummyLog(level=30, name="demo_node", function="do_work", msg="warm")
+    )
+    for _ in range(3):
+        aggregator(
+            DummyLog(level=40, name="demo_node", function="do_work", msg="fail")
+        )
+
+    summary = aggregator.get()
+    assert summary.content == (
+        "Logs summary: ['[demo_node] [WARNING] [do_work] warm', "
+        "'[demo_node] [ERROR] [do_work] fail', "
+        "'Log above repeated 2 times']"
+    )
+
+
+def test_ros2_logs_aggregator_single_log_has_no_repeat_line():
+    aggregator = ROS2LogsAggregator()
+    aggregator(
+        DummyLog(level=20, name="n", function="f", msg="once")
+    )
+    summary = aggregator.get()
+    assert summary.content == "Logs summary: ['[n] [INFO] [f] once']"
+    assert "repeated" not in summary.content
+
+
 def test_ros2_logs_aggregator_str():
     aggregator = ROS2LogsAggregator()
     assert str(aggregator) == "ROS2LogsAggregator(len=0)"