fix: bound ErrorBudget._events with deque(maxlen=N) (#172)

Author: ArokyaMatthew

packages/agent-sre/src/agent_sre/slo/objectives.py

@@ -46,6 +46,10 @@ class ErrorBudget:
 
     Error Budget = 1 - SLO target
     Burn Rate = actual error rate / allowed error rate
+
+    Events are stored in a bounded ``deque(maxlen=max_events)`` to
+    prevent unbounded memory growth in long-running SLOs.  When the
+    buffer is full, the oldest events are silently evicted on append.
     """
 
     total: float = 0.0  # Set from SLO target
@@ -58,6 +62,11 @@ class ErrorBudget:
     _events: collections.deque = field(default_factory=lambda: collections.deque(maxlen=100_000))
     _monotonic_offset: float = field(default_factory=lambda: time.time() - time.monotonic())
 
+    def __post_init__(self) -> None:
+        """Ensure _events is a bounded deque with the correct maxlen."""
+        if not isinstance(self._events, collections.deque) or self._events.maxlen != self.max_events:
+            self._events = collections.deque(self._events, maxlen=self.max_events)
+
     @property
     def remaining(self) -> float:
         """Remaining error budget as a fraction (0.0 to 1.0)."""
@@ -76,11 +85,24 @@ def is_exhausted(self) -> bool:
         return self.consumed >= self.total
 
     def record_event(self, good: bool) -> None:
-        """Record a good or bad event against the budget."""
+        """Record a good or bad event against the budget.
+
+        Events are stored in a bounded deque.  When ``max_events`` is
+        reached, the oldest event is silently evicted.
+        """
         if not good:
             self.consumed += 1.0
         self._events.append({"good": good, "timestamp": time.monotonic()})
 
+    def clear_events(self) -> None:
+        """Clear all recorded events (does **not** reset ``consumed``)."""
+        self._events.clear()
+
+    @property
+    def event_count(self) -> int:
+        """Number of events currently in the buffer."""
+        return len(self._events)
+
     def burn_rate(self, window_seconds: int | None = None) -> float:
         """Calculate current burn rate within a time window.
 

packages/agent-sre/tests/unit/test_objectives.py

@@ -57,6 +57,33 @@ def test_alerts(self) -> None:
         assert "burn_rate_warning" in names
         assert "burn_rate_critical" in names
 
+    def test_events_bounded_by_max_events(self) -> None:
+        """Events beyond max_events are silently evicted (oldest first)."""
+        budget = ErrorBudget(total=10.0, max_events=100)
+        for _ in range(200):
+            budget.record_event(good=True)
+        assert budget.event_count == 100  # capped at maxlen
+
+    def test_custom_max_events(self) -> None:
+        """Custom max_events parameter is respected."""
+        budget = ErrorBudget(total=10.0, max_events=5)
+        for i in range(10):
+            budget.record_event(good=(i % 2 == 0))
+        assert budget.event_count == 5
+        # The deque should contain only the last 5 events
+        assert budget._events.maxlen == 5
+
+    def test_clear_events(self) -> None:
+        """clear_events() empties the deque but does not reset consumed."""
+        budget = ErrorBudget(total=10.0)
+        for _ in range(5):
+            budget.record_event(good=False)
+        assert budget.consumed == 5.0
+        assert budget.event_count == 5
+        budget.clear_events()
+        assert budget.event_count == 0
+        assert budget.consumed == 5.0  # consumed is NOT reset
+
 
 class TestSLO:
     def test_creation(self) -> None: