fix(plugins): retain state history by age, with the count as a ceiling

Follow-up to the cap in this PR. A flat entry count answers the wrong
question: what a reader wants from this history is "the last couple of
hours", and how many transitions that is depends entirely on the plugin's
update interval. On a real board those span 2s to 3600s, so 200 entries is

    interval   200 entries covers
        2s          3.3 minutes     (flights, live)
       10s         16.7 minutes     (jellyfin)
       60s          1.7 hours       (default)
      300s          8.3 hours       (news)
     3600s          4.2 days

-- the plugin churning hardest, the one worth looking at, keeps the least.

So transitions are now trimmed by AGE first
(STATE_HISTORY_MAX_AGE_SECONDS, two hours), which makes the retained window
comparable whatever the cadence, and the count cap becomes purely a memory
ceiling for pollers fast enough to exceed it inside that window. The
ceiling rises 200 -> 2000: at ~230 bytes an entry that is ~0.5MB per plugin
worst case, and only plugins updating faster than roughly every 4s can
reach it. Steady-state memory is unchanged for everything slower, since the
age trim binds first.

Two details worth stating:

  - The trim reads time.monotonic(), stored alongside each transition,
    rather than the datetime already inside it. A DST shift or an NTP step
    would otherwise make every entry look ancient and flush the history in
    one go. The human-readable timestamp is untouched and still what
    get_state_history() returns.

  - Trimming happens on append, so a plugin that goes quiet keeps its last
    window until it writes again. That is deliberate: it is bounded either
    way, and a lazy trim costs nothing on the hot scheduling path. The
    guarantee is therefore about the SPAN of retained history, not its age
    against the current clock, and the test asserts it that way.

The public shape is unchanged: get_state_history() still returns the same
list of transition dicts, and state_history_count is still the lifetime
total.

test_plugin_state_history_retention.py adds 7 tests. Verified against this
branch with only the age trim removed: 4 fail, 3 pass -- the three that
survive are testing the count ceiling and the monotonic clock, which this
commit does not change. Full suite 3753 passed, 60 skipped.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01STMbQE4YctTacQXfbYqKuW

src/plugin_system/plugin_state.py

@@ -6,22 +6,38 @@ with state transitions and queries.
 """
 
 import threading
+import time
 from collections import deque
 from enum import Enum
-from typing import Optional, Dict, Any, Deque, List
+from typing import Optional, Dict, Any, Deque, List, Tuple
 from datetime import datetime
 import logging
 
 from src.logging_config import get_logger
 
 
-# Transitions retained per plugin. The history is diagnostic only -- nothing
-# reads the entries themselves, just their count -- but it is appended to on the
-# hot scheduling path: every update cycle records RUNNING on reserve and ENABLED
-# on finish. Unbounded, that is 2,880 entries per plugin per day at the default
-# 60s interval, which on a 1 GB Pi exhausts memory in weeks. Keep the recent
-# tail for debugging and let the rest age out.
-MAX_STATE_HISTORY_PER_PLUGIN = 200
+# The history is diagnostic only -- nothing reads the entries themselves, just
+# their count -- but it is appended to on the hot scheduling path: every update
+# cycle records RUNNING on reserve and ENABLED on finish. Unbounded, that is
+# 2,880 entries per plugin per day at the default 60s interval, which on a 1 GB
+# Pi exhausts memory in weeks.
+#
+# Two limits, because a single entry count answers the wrong question. What a
+# reader wants is "the last couple of hours", and how many transitions that is
+# depends entirely on the plugin's update interval -- which on a real board
+# spans 2s to 3600s. A flat 200 entries is 4.2 days for the slowest plugin and
+# 3.3 minutes for the fastest, so the plugin churning hardest, the one worth
+# looking at, keeps the least history.
+#
+# So: trim by AGE first, which makes the retained window comparable across
+# plugins whatever their cadence...
+STATE_HISTORY_MAX_AGE_SECONDS = 2 * 60 * 60
+
+# ...and cap by COUNT second, purely as a memory ceiling for the fast pollers
+# whose age window would otherwise run to thousands of entries. At ~230 bytes
+# an entry this is ~0.5 MB per plugin worst case, and only plugins updating
+# faster than roughly every 4s can reach it.
+MAX_STATE_HISTORY_PER_PLUGIN = 2000
 
 
 class PluginState(Enum):
@@ -47,7 +63,10 @@ class PluginStateManager:
         self.logger = logger or get_logger(__name__)
         self._lock = threading.RLock()
         self._states: Dict[str, PluginState] = {}
-        self._state_history: Dict[str, Deque[Dict[str, Any]]] = {}
+        # (monotonic timestamp, transition). The clock is monotonic so a DST
+        # shift or an NTP step cannot make entries look old and flush the
+        # history; the human-readable timestamp lives inside the transition.
+        self._state_history: Dict[str, Deque[Tuple[float, Dict[str, Any]]]] = {}
         # Lifetime transition totals, kept separately so the count reported by
         # get_state_info() stays truthful once the history above starts rolling.
         self._state_transition_counts: Dict[str, int] = {}
@@ -70,7 +89,13 @@ class PluginStateManager:
         if history is None:
             history = deque(maxlen=MAX_STATE_HISTORY_PER_PLUGIN)
             self._state_history[plugin_id] = history
-        history.append(transition)
+        now = time.monotonic()
+        history.append((now, transition))
+        # Age out first; the deque's maxlen is the backstop for plugins that
+        # produce more than the ceiling within the window.
+        cutoff = now - STATE_HISTORY_MAX_AGE_SECONDS
+        while history and history[0][0] < cutoff:
+            history.popleft()
         self._state_transition_counts[plugin_id] = (
             self._state_transition_counts.get(plugin_id, 0) + 1
         )
@@ -160,8 +185,10 @@ class PluginStateManager:
         """
         Get state transition history for a plugin.
 
-        Only the most recent MAX_STATE_HISTORY_PER_PLUGIN transitions are
-        retained; older ones age out.
+        Retention is by age first -- transitions older than
+        STATE_HISTORY_MAX_AGE_SECONDS are dropped -- and by count second, at
+        MAX_STATE_HISTORY_PER_PLUGIN, which only binds for plugins updating
+        fast enough to exceed it inside that window.
 
         Args:
             plugin_id: Plugin identifier
@@ -175,7 +202,7 @@ class PluginStateManager:
         with self._lock:
             return [
                 dict(transition)
-                for transition in self._state_history.get(plugin_id, ())
+                for _stamp, transition in self._state_history.get(plugin_id, ())
             ]
     
     def set_error_info(self, plugin_id: str, error_info: Dict[str, Any]) -> None:

test/test_plugin_state_history_retention.py

@@ -0,0 +1,166 @@
+"""Retention is bounded by age first and by count second.
+
+The cap added in the parent change is a flat entry count, and an entry count
+answers the wrong question. What a reader wants from this history is "the last
+couple of hours"; how many transitions that is depends entirely on the
+plugin's update interval, which on a real board spans 2s to 3600s. A flat 200
+entries is 4.2 days of history for the slowest plugin and 3.3 minutes for the
+fastest -- so the plugin churning hardest, the one actually worth looking at,
+keeps the least.
+
+Trimming by age makes the retained window comparable whatever the cadence, and
+the count then serves only as a memory ceiling for pollers fast enough to
+produce thousands of transitions inside that window.
+"""
+
+import time
+import pytest
+
+from src.plugin_system.plugin_state import (
+    PluginState,
+    PluginStateManager,
+    MAX_STATE_HISTORY_PER_PLUGIN,
+    STATE_HISTORY_MAX_AGE_SECONDS,
+)
+
+
+class FakeClock:
+    """A monotonic clock the test drives, so no test has to sleep."""
+
+    def __init__(self):
+        self.t = 1000.0
+
+    def __call__(self):
+        return self.t
+
+    def advance(self, seconds):
+        self.t += seconds
+
+
+@pytest.fixture
+def clock(monkeypatch):
+    c = FakeClock()
+    monkeypatch.setattr("src.plugin_system.plugin_state.time.monotonic", c)
+    return c
+
+
+def _cycle(manager, plugin_id, clock, interval, cycles):
+    """One update cycle: RUNNING on reserve, ENABLED on finish."""
+    for _ in range(cycles):
+        manager.set_state(plugin_id, PluginState.RUNNING)
+        manager.set_state(plugin_id, PluginState.ENABLED)
+        clock.advance(interval)
+
+
+def test_transitions_older_than_the_window_are_dropped(clock):
+    m = PluginStateManager()
+    _cycle(m, "clock", clock, interval=60, cycles=10)
+    assert len(m.get_state_history("clock")) == 20
+
+    # Nothing happens for longer than the window, then one more cycle.
+    clock.advance(STATE_HISTORY_MAX_AGE_SECONDS + 1)
+    _cycle(m, "clock", clock, interval=60, cycles=1)
+
+    assert len(m.get_state_history("clock")) == 2, (
+        "only the transitions inside the window should survive")
+
+
+def test_every_plugin_keeps_the_same_WINDOW_not_the_same_COUNT(clock):
+    """The point of the age policy, stated as the property that distinguishes it.
+
+    Run both plugins for three times the retention window. Under a flat count
+    cap the slow one would still be holding transitions from hours before the
+    window, because it never produces enough entries to evict them. Under the
+    age policy each plugin retains its own last two hours and no more --
+    different entry counts, same span of time.
+    """
+    window = STATE_HISTORY_MAX_AGE_SECONDS
+    m = PluginStateManager()
+
+    _cycle(m, "slow", clock, interval=60, cycles=(3 * window) // 60)
+    slow = len(m.get_state_history("slow"))
+
+    # Assert the property directly rather than a derived count. The guarantee
+    # is about the SPAN of retained history, not its age against the current
+    # clock: trimming happens on append, so a plugin that has gone quiet keeps
+    # its last window until it writes again. That is intentional -- it is
+    # bounded either way, and a lazy trim costs nothing on the hot path.
+    stamps = [stamp for stamp, _ in m._state_history["slow"]]
+    assert stamps[-1] - stamps[0] <= window, (
+        f"retained history spans {stamps[-1] - stamps[0]:.0f}s, "
+        f"window is {window}s")
+    assert slow < 2 * ((3 * window) // 60), (
+        f"slow plugin kept {slow} entries -- three windows' worth was retained")
+
+    clock.t = 1000.0
+    _cycle(m, "fast", clock, interval=2, cycles=(3 * window) // 2)
+    fast = len(m.get_state_history("fast"))
+
+    # Different counts, and the fast poller keeps more of them -- under a flat
+    # count cap these would be equal and the fast one would cover minutes.
+    assert fast > slow, f"fast={fast} slow={slow}"
+
+
+def test_the_count_ceiling_still_bounds_a_fast_poller(clock):
+    """Age alone would let a 2s plugin hold 7,200 entries."""
+    m = PluginStateManager()
+    _cycle(m, "flights", clock, interval=2, cycles=STATE_HISTORY_MAX_AGE_SECONDS)
+    assert len(m.get_state_history("flights")) <= MAX_STATE_HISTORY_PER_PLUGIN
+
+
+def test_a_burst_inside_the_window_is_capped_not_kept(clock):
+    """Transitions with no time between them still cannot grow without bound."""
+    m = PluginStateManager()
+    for _ in range(MAX_STATE_HISTORY_PER_PLUGIN * 3):
+        m.set_state("flapping", PluginState.RUNNING)   # clock never advances
+    assert len(m.get_state_history("flapping")) <= MAX_STATE_HISTORY_PER_PLUGIN
+
+
+def test_ageing_out_does_not_disturb_the_lifetime_count(clock):
+    m = PluginStateManager()
+    _cycle(m, "clock", clock, interval=60, cycles=10)
+    clock.advance(STATE_HISTORY_MAX_AGE_SECONDS + 1)
+    _cycle(m, "clock", clock, interval=60, cycles=1)
+
+    assert len(m.get_state_history("clock")) == 2
+    assert m.get_state_info("clock")["state_history_count"] == 22, (
+        "the lifetime total must survive trimming, it is the flap signal")
+
+
+def test_the_surviving_entries_are_the_recent_ones(clock):
+    m = PluginStateManager()
+    _cycle(m, "clock", clock, interval=60, cycles=5)
+    clock.advance(STATE_HISTORY_MAX_AGE_SECONDS + 1)
+    m.set_state("clock", PluginState.ERROR)
+
+    history = m.get_state_history("clock")
+    assert [h["to"] for h in history] == ["error"]
+
+
+def test_a_monotonic_clock_is_used_not_the_wall_clock(clock):
+    """A DST shift or NTP step must not flush the history.
+
+    The trim reads time.monotonic(); the human-readable datetime inside each
+    transition is for display only.
+    """
+    m = PluginStateManager()
+    _cycle(m, "clock", clock, interval=60, cycles=3)
+    before = len(m.get_state_history("clock"))
+
+    import datetime as real_datetime
+
+    class ShiftedDatetime(real_datetime.datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return real_datetime.datetime(1999, 1, 1)   # clock jumps backwards
+
+    import src.plugin_system.plugin_state as ps
+    original = ps.datetime
+    ps.datetime = ShiftedDatetime
+    try:
+        m.set_state("clock", PluginState.ENABLED)
+    finally:
+        ps.datetime = original
+
+    assert len(m.get_state_history("clock")) == before + 1, (
+        "a wall-clock jump must not trim anything")