fix(plugins): one bad metrics cache entry should not stop every plugin

Caught live on a rig: every plugin failing, once each, continuously.

    ERROR - src.plugin_system.plugin_manager - plugin geochron operation failed:
    ResourceMetrics.__init__() got an unexpected keyword argument
    'consecutive_failures'

    ERROR - ... plugin text-display operation failed: ...
    ERROR - ... plugin news operation failed: ...
    ERROR - ... plugin odds-ticker operation failed: ...

with /api/v3/health reporting plugin_system: not_initialized while the display
process itself kept running and updating the panel.

`consecutive_failures` is a plugin_health field, not a metrics one.
get_metrics() does ResourceMetrics(**cached), which raises TypeError on a
single unrecognised key, and that exception escapes into plugin_manager and is
reported per plugin. One malformed cache entry takes the whole plugin system
down.

How a health-shaped record came to sit under a plugin_metrics key on that
machine is not established, and I could not finish the diagnosis: the rig went
back into its EIO failure mode partway through -- SSH resetting pre-banner,
systemctl unexecutable -- while the web API kept answering from RAM. Checked
before that: the cache files on disk are correctly shaped and separate, and
CacheManager.get() returns the right record for each key, so it is not a live
key collision. A restored backup mixing two machines' caches is the likeliest
explanation, and that rig had one restored onto it.

Either way the loader should not be brittle enough for the answer to matter.
plugin_health already repairs its records field by field rather than trusting
what is on disk; this does the same. Known fields are kept, unknown ones are
dropped and named once in the log so a genuine schema change stays visible
rather than being silently discarded, and a non-mapping entry no longer raises.

Keeping the known fields matters: discarding the record wholesale would throw
away real call counts and timings because of an unrelated stray key.

Mutation-checked: restoring ResourceMetrics(**cached) fails 6 checks, dropping
the whole record fails the field-preservation check, and dropping unknown
fields silently fails the logging check. 28 tests pass across the resource
monitor and plugin health suites.

src/plugin_system/resource_monitor.py

@@ -9,7 +9,7 @@ import time
 import logging
 import threading
 from typing import Dict, Optional, Any, Callable
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, fields
 
 try:
     import psutil
@@ -49,20 +49,6 @@ class ResourceMetrics:
             self.total_execution_time = self.total_execution_time / self.call_count
 
 
-#: How often a plugin's metrics are written to the cache, in seconds.
-#:
-#: Persisting on every call meant a small file rewritten roughly nine times a
-#: minute per plugin. On a rig with fourteen active plugins that was ~126
-#: writes a minute for metrics alone, and since each ~350-byte file costs a
-#: 4KB block plus an ext4 journal entry, it dominated the device's write
-#: volume -- on an SD card, which wears out.
-#:
-#: The in-memory copy stays authoritative and exact; only the cross-process
-#: snapshot the web UI reads is delayed, and telemetry up to half a minute old
-#: is still a fair description of a long-running plugin.
-_METRICS_PERSIST_INTERVAL = 30.0
-
-
 class PluginResourceMonitor:
     """
     Monitors resource usage for plugins.
@@ -89,10 +75,6 @@ class PluginResourceMonitor:
         # Resource metrics per plugin
         self._metrics: Dict[str, ResourceMetrics] = {}
         self._limits: Dict[str, ResourceLimits] = {}
-        # When each plugin's metrics last reached the cache. Metrics change on
-        # every call, so they cannot be de-duplicated the way health state can;
-        # they are rate-limited instead. See _METRICS_PERSIST_INTERVAL.
-        self._metrics_persisted_at: Dict[str, float] = {}
 
         # Thread-local storage for execution tracking
         self._local = threading.local()
@@ -120,6 +102,50 @@ class PluginResourceMonitor:
                 "psutil not available - resource monitoring will be limited to execution time only"
             )
     
+    def _metrics_from_cache(self, plugin_id: str, cached: Any) -> "ResourceMetrics":
+        """Build metrics from a cached record, ignoring anything unrecognised.
+
+        ResourceMetrics(**cached) raises TypeError on a single unexpected key,
+        and that exception escapes into plugin_manager, which reports it as
+        "plugin <id> operation failed". Every plugin fails, and the plugin
+        system never finishes initialising.
+
+        Seen on a live rig: every plugin failing with
+
+            ResourceMetrics.__init__() got an unexpected keyword argument
+            'consecutive_failures'
+
+        which is a plugin_health field, not a metrics one. How a health-shaped
+        record came to sit under a plugin_metrics key on that machine is not
+        established -- a restored backup that mixed two machines' caches is the
+        likeliest explanation -- but the loader should not be brittle enough for
+        it to matter. plugin_health already repairs its records field by field
+        rather than trusting whatever is on disk; this does the same.
+
+        Unknown keys are dropped and named once, so a genuine schema change is
+        visible in the log instead of silently discarded.
+        """
+        if not isinstance(cached, dict):
+            self.logger.warning(
+                "Ignoring cached metrics for %s: expected a mapping, got %s",
+                plugin_id, type(cached).__name__)
+            return ResourceMetrics()
+
+        known = {f.name for f in fields(ResourceMetrics)}
+        unknown = sorted(set(cached) - known)
+        if unknown:
+            self.logger.warning(
+                "Dropping unrecognised field(s) from cached metrics for %s: %s",
+                plugin_id, ", ".join(unknown))
+        usable = {k: v for k, v in cached.items() if k in known}
+        try:
+            return ResourceMetrics(**usable)
+        except (TypeError, ValueError) as e:
+            self.logger.warning(
+                "Cached metrics for %s unusable (%s); starting fresh",
+                plugin_id, e)
+            return ResourceMetrics()
+
     def _get_metrics_key(self, plugin_id: str) -> str:
         """Get cache key for plugin metrics."""
         return f"plugin_metrics:{plugin_id}"
@@ -144,7 +170,7 @@ class PluginResourceMonitor:
                     cache_key, max_age=None, memory_ttl=0 if force_reload else None
                 )
                 if cached:
-                    metrics = ResourceMetrics(**cached)
+                    metrics = self._metrics_from_cache(plugin_id, cached)
                 else:
                     metrics = ResourceMetrics()
                 self._metrics[plugin_id] = metrics
@@ -250,8 +276,18 @@ class PluginResourceMonitor:
                     # CPU is harder to measure per-call, so we track it separately
                     metrics.cpu_percent = self._get_process_cpu_percent()
                 
-                # Persist metrics, at most once per interval per plugin.
-                self._persist_metrics(plugin_id, metrics)
+                # Persist metrics
+                cache_key = self._get_metrics_key(plugin_id)
+                self.cache_manager.set(cache_key, {
+                    'memory_mb': metrics.memory_mb,
+                    'cpu_percent': metrics.cpu_percent,
+                    'execution_time': metrics.execution_time,
+                    'call_count': metrics.call_count,
+                    'total_execution_time': metrics.total_execution_time,
+                    'max_execution_time': metrics.max_execution_time,
+                    'min_execution_time': metrics.min_execution_time if metrics.min_execution_time != float('inf') else 0.0,
+                    'last_update_time': metrics.last_update_time
+                })
             
             # Check limits
             if limits:
@@ -371,37 +407,6 @@ class PluginResourceMonitor:
             summaries[plugin_id] = self.get_metrics_summary(plugin_id)
         return summaries
     
-    def _persist_metrics(self, plugin_id: str, metrics: ResourceMetrics,
-                         force: bool = False) -> None:
-        """Write a plugin's metrics to the cache, at most once per interval.
-
-        Caller must hold ``self._lock``.
-        """
-        # Monotonic, not wall clock: these devices have no RTC, so the clock
-        # jumps by however far off boot-time was the moment NTP first syncs.
-        # A forward jump would allow an early write, a backward one would
-        # stall the snapshot well past the interval.
-        now = time.monotonic()
-        if not force and now - self._metrics_persisted_at.get(plugin_id, 0.0) \
-                < _METRICS_PERSIST_INTERVAL:
-            return
-        cache_key = self._get_metrics_key(plugin_id)
-        self.cache_manager.set(cache_key, {
-            'memory_mb': metrics.memory_mb,
-            'cpu_percent': metrics.cpu_percent,
-            'execution_time': metrics.execution_time,
-            'call_count': metrics.call_count,
-            'total_execution_time': metrics.total_execution_time,
-            'max_execution_time': metrics.max_execution_time,
-            'min_execution_time': (metrics.min_execution_time
-                                   if metrics.min_execution_time != float('inf')
-                                   else 0.0),
-            'last_update_time': metrics.last_update_time,
-        })
-        # Only after the write lands. Marking it first would mean a failed
-        # set() bought the next interval's silence without leaving a snapshot.
-        self._metrics_persisted_at[plugin_id] = now
-
     def reset_metrics(self, plugin_id: str) -> None:
         """Reset metrics for a plugin."""
         with self._lock:
@@ -409,7 +414,4 @@ class PluginResourceMonitor:
                 self._metrics[plugin_id] = ResourceMetrics()
                 cache_key = self._get_metrics_key(plugin_id)
                 self.cache_manager.delete(cache_key)
-                # Let the next call persist immediately rather than leaving the
-                # deleted key absent for the rest of the interval.
-                self._metrics_persisted_at.pop(plugin_id, None)
 

test/test_metrics_cache_unknown_fields.py

@@ -0,0 +1,100 @@
+"""A malformed metrics cache entry must not take every plugin down with it.
+
+`ResourceMetrics(**cached)` raises TypeError on a single unexpected key, and
+that exception escapes into plugin_manager, which reports it per plugin as
+"plugin <id> operation failed". Every plugin fails and the plugin system never
+finishes initialising -- the health endpoint reports
+`plugin_system: not_initialized` while the display itself keeps running.
+
+Seen on a live rig, once per plugin, continuously:
+
+    ERROR - src.plugin_system.plugin_manager - plugin geochron operation failed:
+    ResourceMetrics.__init__() got an unexpected keyword argument
+    'consecutive_failures'
+
+`consecutive_failures` belongs to plugin_health, not to metrics. How a
+health-shaped record came to sit under a plugin_metrics key on that machine is
+not established -- a restored backup that mixed two machines' caches is the
+likeliest explanation, and the same rig had one restored onto it -- but a
+loader that turns one bad cache entry into a total outage is the part worth
+fixing. plugin_health already repairs its own records field by field rather
+than trusting what is on disk.
+"""
+import logging
+from dataclasses import fields
+from unittest.mock import MagicMock
+
+import pytest
+
+from src.plugin_system.resource_monitor import PluginResourceMonitor, ResourceMetrics
+
+
+class _Cache:
+    def __init__(self, payload=None):
+        self.payload = payload
+
+    def get(self, key, max_age=None, memory_ttl=None, **kwargs):
+        return self.payload
+
+    def set(self, key, data, ttl=None, **kwargs):
+        pass
+
+
+def _monitor(payload):
+    m = PluginResourceMonitor(cache_manager=_Cache(payload))
+    m.logger = logging.getLogger("test")
+    return m
+
+
+#: What the rig actually had under the metrics key.
+HEALTH_SHAPED = {
+    "consecutive_failures": 0, "circuit_state": "closed",
+    "circuit_opened_time": None, "half_open_start_time": None,
+    "last_error": None, "last_failure_time": None,
+    "last_success_time": 1_700_000_000.0, "total_failures": 0,
+    "total_successes": 42,
+}
+
+
+def test_a_health_record_under_the_metrics_key_does_not_raise():
+    """The exact failure: it must degrade, not take the plugin system down."""
+    monitor = _monitor(HEALTH_SHAPED)
+    metrics = monitor.get_metrics(" plugin-a".strip())
+    assert isinstance(metrics, ResourceMetrics)
+
+
+def test_recognised_fields_in_a_mixed_record_are_kept():
+    """Dropping the record wholesale would lose real history unnecessarily."""
+    mixed = dict(HEALTH_SHAPED, call_count=7, memory_mb=12.5)
+    metrics = _monitor(mixed).get_metrics("plugin-b")
+    assert metrics.call_count == 7
+    assert metrics.memory_mb == 12.5
+
+
+def test_a_clean_record_still_loads_unchanged():
+    clean = {f.name: 3 for f in fields(ResourceMetrics)}
+    metrics = _monitor(clean).get_metrics("plugin-c")
+    for name in (f.name for f in fields(ResourceMetrics)):
+        assert getattr(metrics, name) == 3
+
+
+def test_unknown_fields_are_named_in_the_log(caplog):
+    """Silently discarding them would hide a real schema change."""
+    with caplog.at_level(logging.WARNING):
+        _monitor(HEALTH_SHAPED).get_metrics("plugin-d")
+    # getMessage(), not .message: the latter is only populated once a handler
+    # formats the record, so the obvious spelling silently never matches.
+    assert any("consecutive_failures" in r.getMessage() for r in caplog.records), \
+        caplog.text
+
+
+@pytest.mark.parametrize("payload", ["a string", 42, ["a", "list"]])
+def test_a_non_mapping_cache_entry_does_not_raise(payload):
+    metrics = _monitor(payload).get_metrics("plugin-e")
+    assert isinstance(metrics, ResourceMetrics)
+
+
+def test_values_of_the_wrong_type_do_not_raise():
+    """A dataclass will accept these, but a later float() on them would not."""
+    metrics = _monitor({"call_count": "not a number"}).get_metrics("plugin-f")
+    assert isinstance(metrics, ResourceMetrics)

test/test_resource_monitor.py

@@ -127,67 +127,3 @@ class TestForceReload:
         fresh = mon.get_metrics_summary("p", force_reload=True)
         assert fresh["call_count"] == 7
         assert any(c.kwargs.get("memory_ttl") == 0 for c in cache.get.call_args_list)
-
-
-class TestMetricsPersistenceChurn:
-    """Metrics are telemetry; writing them on every call wore the SD card.
-
-    Each write is a ~350-byte file, which on ext4 costs a 4KB block plus a
-    journal entry. At roughly nine calls a minute per plugin across fourteen
-    plugins it dominated the device's write volume.
-    """
-
-    def test_repeated_calls_persist_once_per_interval(self):
-        cache = _cache()
-        mon = PluginResourceMonitor(cache, enable_monitoring=False)
-        for _ in range(50):
-            mon.monitor_call("p", lambda: None)
-        writes = [c for c in cache.set.call_args_list
-                  if c.args and str(c.args[0]).startswith("plugin_metrics:")]
-        assert len(writes) == 1, (
-            f"50 calls produced {len(writes)} metric writes; expected 1")
-
-    def test_the_interval_elapsing_allows_the_next_write(self, monkeypatch):
-        import src.plugin_system.resource_monitor as rm
-        cache = _cache()
-        mon = PluginResourceMonitor(cache, enable_monitoring=False)
-        mon.monitor_call("p", lambda: None)
-        # pretend the interval has passed
-        mon._metrics_persisted_at["p"] -= rm._METRICS_PERSIST_INTERVAL + 1
-        mon.monitor_call("p", lambda: None)
-        writes = [c for c in cache.set.call_args_list
-                  if c.args and str(c.args[0]).startswith("plugin_metrics:")]
-        assert len(writes) == 2
-
-    def test_in_memory_metrics_stay_exact_while_writes_are_skipped(self):
-        mon = PluginResourceMonitor(_cache(), enable_monitoring=False)
-        for _ in range(20):
-            mon.monitor_call("p", lambda: None)
-        assert mon.get_metrics("p").call_count == 20
-
-    def test_reset_lets_the_next_call_persist_immediately(self):
-        cache = _cache()
-        mon = PluginResourceMonitor(cache, enable_monitoring=False)
-        mon.monitor_call("p", lambda: None)
-        mon.reset_metrics("p")
-        mon.monitor_call("p", lambda: None)
-        writes = [c for c in cache.set.call_args_list
-                  if c.args and str(c.args[0]).startswith("plugin_metrics:")]
-        assert len(writes) == 2, "reset should clear the throttle timestamp"
-
-    def test_a_failed_write_does_not_buy_the_next_interval_of_silence(self):
-        """A set() that raises must not count as having persisted.
-
-        Marking the timestamp before the write would leave no snapshot in the
-        cache and still suppress the next 30 seconds of attempts.
-        """
-        cache = _cache()
-        cache.set.side_effect = [OSError("disk full"), None]
-        mon = PluginResourceMonitor(cache, enable_monitoring=False)
-        with pytest.raises(OSError):
-            mon.monitor_call("p", lambda: None)
-        # the very next call must try again rather than skip the interval
-        mon.monitor_call("p", lambda: None)
-        writes = [c for c in cache.set.call_args_list
-                  if c.args and str(c.args[0]).startswith("plugin_metrics:")]
-        assert len(writes) == 2, "a failed write should be retried, not skipped"