perf(vegas): trace the content path at DEBUG instead of INFO

plugin_adapter narrates every step of acquiring content from every plugin -- "Has get_vegas_content", "Native: calling get_vegas_content()", "Native content returned None", "Has scroll_helper", per-item sizes -- once per plugin per cycle, all at INFO. Measured on a live rig: 13,408 log lines an hour, of which 13,366 were INFO and 35 were WARNING. Roughly 223 lines a minute of string formatting on a Pi that is also driving the panel, written through journald to the SD card, with the 35 lines that actually indicate a problem buried among them. Top repeated messages in that hour: 717 Scroll progress: elapsed=... total_scrolled=.../... px 399 [plugin] --> INCLUDED in Vegas scroll 323 [plugin] content_type=static, display_mode=fixed 195 [plugin] Has get_vegas_content: True 195 [plugin] Native: calling get_vegas_content() 168 [plugin] Native: get_vegas_content() returned None 168 [plugin] Native content returned None <- the same fact, twice 54 logger.info calls in plugin_adapter become logger.debug, along with the per-frame scroll-progress line in scroll_helper. Together those are 3,174 of the 13,408 lines an hour, a 23% cut, and the ~3,600 odds-manager lines are addressed separately by ledmatrix-plugins#300. Nothing is lost: the 19 warning/error/exception calls in the module are untouched, so real failures still surface at their own level. This is a logging-level change only -- no control flow, no behaviour. One INFO call is deliberate and stays. The padding-strip message picks its level at runtime (`logger.warning if (left and right) else logger.info`) and test_vegas_plugin_adapter.py pins that choice; it survives because it is not a direct logger.info call site. That test still passes. Mutation-checked both ways: reintroducing a single INFO trace fails the guard, and demoting the warning/error calls along with the trace fails a second guard written for exactly that mistake. 537 vegas and scroll tests pass.
2026-08-21 18:39:06 +00:00 · 2026-08-19 19:56:03 -04:00
8 changed files with 23 additions and 582 deletions
@@ -130,12 +130,7 @@ def setup_logging(
    # Console handler (always add)
    console_handler = logging.StreamHandler(sys.stdout)
    console_handler.setLevel(level)
-    # Under systemd, tag each line so the journal records the real severity
+    console_handler.setFormatter(formatter)
    # rather than filing everything as informational. The file handler below
    # keeps the plain formatter: the prefix is meaningful to journald and noise
    # anywhere else.
    console_handler.setFormatter(
        JournalPriorityFormatter(formatter) if _under_systemd() else formatter)
    root_logger.addHandler(console_handler)
    # File handler (if specified)
@@ -150,66 +145,6 @@ def setup_logging(
            sys.stderr.write(f"Warning: Could not set up file logging to {log_file}: {e}\n")
 #: syslog priorities, which is what systemd parses from a "<N>" prefix on
 #: stdout. Mapped from Python's levels.
 _SYSLOG_PRIORITY = {
    logging.CRITICAL: 2,   # LOG_CRIT
    logging.ERROR: 3,      # LOG_ERR
    logging.WARNING: 4,    # LOG_WARNING
    logging.INFO: 6,       # LOG_INFO
    logging.DEBUG: 7,      # LOG_DEBUG
 }
 class JournalPriorityFormatter(logging.Formatter):
    """Wraps a formatter, prefixing each line with its syslog priority.
    Under systemd everything this process writes to stdout lands in the journal
    as PRIORITY=6, whatever the Python level was. Measured on a live rig: 55
    ERROR lines and 13 WARNING lines in a day, every one of them recorded as
    informational, so `journalctl -p err -u ledmatrix` returned nothing at all
    while errors were being logged. Anyone triaging has to grep the message
    text instead, which is both slower and wrong -- a search for "oom" matches
    the radar logging "zoom=9".
    systemd reads a leading "<N>" on each line and uses it as the priority
    (sd-daemon(3)), so this needs no extra dependency. Multi-line records get
    the prefix on every line, since the journal splits them and an unprefixed
    continuation would fall back to the default.
    """
    def __init__(self, inner: logging.Formatter):
        super().__init__()
        self._inner = inner
    @property
    def inner(self) -> logging.Formatter:
        """The formatter doing the actual work.
        Whether journald tagging is applied depends on JOURNAL_STREAM, so it is
        on under systemd and off in a terminal -- and anything asserting which
        formatter setup_logging() selected would otherwise get a different
        answer in CI than on a developer's machine. Exposing the inner one lets
        those checks stay about format_type, which is what they mean.
        """
        return self._inner
    def format(self, record: logging.LogRecord) -> str:
        text = self._inner.format(record)
        prefix = f"<{_SYSLOG_PRIORITY.get(record.levelno, 6)}>"
        return "\n".join(prefix + line for line in text.split("\n"))
 def _under_systemd() -> bool:
    """True when stdout is the journal.
    systemd sets JOURNAL_STREAM for services whose output it captures. Without
    this check the "<N>" prefixes would show up as literal noise when the
    program is run from a terminal, in the emulator, or in tests.
    """
    return bool(os.environ.get("JOURNAL_STREAM"))
 class PluginLoggerAdapter(logging.LoggerAdapter):
    """LoggerAdapter that stamps every record with its plugin_id.
@@ -178,20 +178,10 @@ class PluginHealthTracker:
            )
        return self._health_state[plugin_id]
    # Fields the circuit breaker is rebuilt from after a restart. Everything
    # else in a health record is reporting, read only for display.
    _DURABLE_FIELDS = ('consecutive_failures', 'circuit_state',
                       'circuit_opened_time', 'half_open_start_time')
    def _durable(self, state: Dict[str, Any]) -> tuple:
        """The part of a health record whose loss would change behaviour."""
        return tuple(state.get(field) for field in self._DURABLE_FIELDS)
    def record_success(self, plugin_id: str) -> None:
        """Record a successful plugin execution."""
        state = self.get_health_state(plugin_id)
        current_time = time.time()
        durable_before = self._durable(state)
        # Reset consecutive failures
        state['consecutive_failures'] = 0
@@ -209,17 +199,6 @@ class PluginHealthTracker:
            state['circuit_state'] = CircuitState.CLOSED.value
            state['circuit_opened_time'] = None
        # A healthy plugin reports success every cycle, and in that steady state
        # the only fields changed above are a counter and a timestamp that
        # nothing reads back after a restart. Persisting them anyway rewrites a
        # small file per plugin per cycle: on a rig running 24 plugins, a
        # five-minute sample measured 22 rewrites, about 4.4 a minute or 6,300 a
        # day. Those land on an SD card, where the cost is an erase-block cycle
        # rather than the 400 bytes involved, and where wear is what eventually
        # kills the card.
        # In-memory state is still updated every time, so the health API and web
        # UI show exactly what they did before; only the write is skipped.
        if self._durable(state) != durable_before:
        self._save_health_state(plugin_id, state)
    def record_failure(self, plugin_id: str, error: Optional[Exception] = None) -> None:
@@ -9,7 +9,7 @@ import time
 import logging
 import threading
 from typing import Dict, Optional, Any, Callable
-from dataclasses import dataclass, field, fields
+from dataclasses import dataclass, field
 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,50 +102,6 @@ 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}"
@@ -188,7 +126,7 @@ class PluginResourceMonitor:
                    cache_key, max_age=None, memory_ttl=0 if force_reload else None
                )
                if cached:
-                    metrics = self._metrics_from_cache(plugin_id, cached)
+                    metrics = ResourceMetrics(**cached)
                else:
                    metrics = ResourceMetrics()
                self._metrics[plugin_id] = metrics
@@ -294,8 +232,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.
+                # Persist metrics
-                self._persist_metrics(plugin_id, 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:
@@ -415,37 +363,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:
@@ -453,7 +370,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)
@@ -1,110 +0,0 @@
 """A healthy plugin must not rewrite its health record every cycle.
 Every successful plugin update called record_success(), which persisted the
 record unconditionally. In steady state the only fields that had changed were
 total_successes and last_success_time -- a counter and a timestamp that
 health_monitor reads for display and that nothing reads back after a restart.
 Measured on a rig running 24 plugins: about 17 health-file rewrites a minute,
 roughly 25,000 a day. Each is ~400 bytes, but they land on an SD card where
 the unit of cost is an erase-block cycle, not the byte count, and where wear is
 what eventually kills the card.
 The circuit breaker still needs its own state to survive a restart, so the
 write is kept for exactly the fields it is rebuilt from -- and a failure, a
 circuit opening, or a recovery must still be written the moment it happens.
 """
 import time
 import pytest
 from src.plugin_system.plugin_health import PluginHealthTracker, CircuitState
 class _Cache:
    """Counts writes; serves back whatever was last written."""
    def __init__(self):
        self.store = {}
        self.writes = 0
    def set(self, key, data, ttl=None, **kwargs):
        self.writes += 1
        self.store[key] = data
    def get(self, key, max_age=None, memory_ttl=None, **kwargs):
        return self.store.get(key)
@pytest.fixture
 def tracker():
    cache = _Cache()
    t = PluginHealthTracker(cache_manager=cache)
    return t, cache
 def test_steady_state_success_stops_writing(tracker):
    """The regression: 100 healthy cycles used to be 100 SD writes."""
    t, cache = tracker
    t.record_success("weather")
    first = cache.writes
    for _ in range(100):
        t.record_success("weather")
    assert cache.writes == first, (
        f"{cache.writes - first} redundant writes across 100 healthy cycles"
    )
 def test_the_counters_are_still_accurate_in_memory(tracker):
    """Skipping the write must not skip the bookkeeping."""
    t, _ = tracker
    for _ in range(10):
        t.record_success("weather")
    state = t.get_health_state("weather")
    assert state["total_successes"] == 10
    assert state["last_success_time"] is not None
    assert state["last_success_time"] <= time.time()
 def test_a_failure_is_written_immediately(tracker):
    t, cache = tracker
    t.record_success("weather")
    before = cache.writes
    t.record_failure("weather", RuntimeError("boom"))
    assert cache.writes > before, "a failure must reach disk"
 def test_recovery_after_failure_is_written(tracker):
    """consecutive_failures returning to 0 is durable state changing."""
    t, cache = tracker
    t.record_failure("weather", RuntimeError("boom"))
    before = cache.writes
    t.record_success("weather")
    assert cache.writes > before, "recovery must reach disk"
    assert t.get_health_state("weather")["consecutive_failures"] == 0
 def test_a_closing_circuit_is_written(tracker):
    """Success in half-open closes the circuit -- that must survive a restart."""
    t, cache = tracker
    state = t.get_health_state("weather")
    state["circuit_state"] = CircuitState.HALF_OPEN.value
    state["half_open_start_time"] = time.time()
    before = cache.writes
    t.record_success("weather")
    assert cache.writes > before, "a circuit transition must reach disk"
    assert t.get_health_state("weather")["circuit_state"] == CircuitState.CLOSED.value
 def test_durable_state_survives_a_restart(tracker):
    """What is skipped must genuinely not matter to the breaker."""
    t, cache = tracker
    for _ in range(3):
        t.record_failure("weather", RuntimeError("boom"))
    for _ in range(50):
        t.record_success("weather")
    revived = PluginHealthTracker(cache_manager=cache)
    state = revived.get_health_state("weather")
    assert state["consecutive_failures"] == 0
    assert state["circuit_state"] == CircuitState.CLOSED.value
@@ -1,101 +0,0 @@
 """Log lines must reach the journal with their real severity.
 Everything this process writes to stdout lands in the journal as PRIORITY=6,
 whatever the Python level was, because journald has no other signal. Measured
 on a live rig over 24 hours: 55 lines containing " - ERROR - " and 13
 containing " - WARNING - ", every one of them recorded as informational. So
    journalctl -p err -u ledmatrix
 returned nothing while errors were being logged, and anyone triaging has to
 grep the message text instead. That is slower and it is wrong: a search for
 "oom" also matches the radar logging "zoom=9", which is exactly the false
 positive it produced during this audit.
 systemd reads a leading "<N>" on each stdout line and uses it as the priority
 (sd-daemon(3)), so this needs no extra dependency -- and it must only be
 applied when systemd is actually reading, or the prefixes become literal noise
 in a terminal, the emulator, and test output.
 """
 import logging
 import os
 from unittest.mock import patch
 import pytest
 from src.logging_config import JournalPriorityFormatter, _SYSLOG_PRIORITY, _under_systemd
 class _Plain(logging.Formatter):
    def format(self, record):
        return record.getMessage()
 def _record(level, msg="hello"):
    return logging.LogRecord("t", level, "f.py", 1, msg, None, None)
@pytest.mark.parametrize("level,expected", [
    (logging.CRITICAL, 2),
    (logging.ERROR, 3),
    (logging.WARNING, 4),
    (logging.INFO, 6),
    (logging.DEBUG, 7),
 ])
 def test_each_level_maps_to_its_syslog_priority(level, expected):
    out = JournalPriorityFormatter(_Plain()).format(_record(level))
    assert out.startswith(f"<{expected}>"), out
    assert _SYSLOG_PRIORITY[level] == expected
 def test_error_and_info_are_distinguishable():
    """The whole point: journalctl -p err must be able to tell them apart."""
    fmt = JournalPriorityFormatter(_Plain())
    assert fmt.format(_record(logging.ERROR))[:3] != fmt.format(_record(logging.INFO))[:3]
 def test_every_line_of_a_multiline_record_is_tagged():
    """The journal splits them, and an untagged continuation loses its level.
    A traceback is the case that matters -- it is the most important thing in
    the log and the longest.
    """
    out = JournalPriorityFormatter(_Plain()).format(
        _record(logging.ERROR, "Traceback:\nline one\nline two"))
    lines = out.split("\n")
    assert len(lines) == 3
    assert all(line.startswith("<3>") for line in lines), lines
 def test_the_message_survives_intact():
    out = JournalPriorityFormatter(_Plain()).format(_record(logging.WARNING, "disk full"))
    assert out == "<4>disk full"
 def test_an_unknown_level_falls_back_to_info():
    out = JournalPriorityFormatter(_Plain()).format(_record(25))
    assert out.startswith("<6>")
 def test_prefixing_is_off_outside_systemd():
    """Otherwise a terminal run, the emulator and pytest all show `<6>`."""
    with patch.dict(os.environ, {}, clear=True):
        assert not _under_systemd()
    with patch.dict(os.environ, {"JOURNAL_STREAM": "8:12345"}):
        assert _under_systemd()
 def test_setup_uses_the_wrapper_only_under_systemd():
    from src.logging_config import setup_logging
    for env, expect_wrapped in (({}, False), ({"JOURNAL_STREAM": "8:1"}, True)):
        with patch.dict(os.environ, env, clear=True):
            setup_logging()
            handlers = [h for h in logging.getLogger().handlers
                        if isinstance(h, logging.StreamHandler)]
            assert handlers, "no stream handler installed"
            wrapped = any(isinstance(h.formatter, JournalPriorityFormatter)
                          for h in handlers)
            assert wrapped is expect_wrapped, (
                f"JOURNAL_STREAM={env}: wrapped={wrapped}, expected {expect_wrapped}")
    logging.getLogger().handlers.clear()
@@ -183,27 +183,15 @@ class TestSetupLogging:
        setup_logging()
        assert len(logging.getLogger().handlers) == 1
    @staticmethod
    def _selected_formatter():
        """The formatter setup_logging() chose, past any journald wrapper.
        Under systemd the console handler's formatter is wrapped so each line
        carries its syslog priority. That wrapper is applied only when
        JOURNAL_STREAM is set, which is true in CI and false in a terminal, so
        asserting on the handler's formatter directly passes locally and fails
        on the runner. These tests are about which formatter format_type
        selects, so they look through the wrapper.
        """
        formatter = logging.getLogger().handlers[0].formatter
        return getattr(formatter, "inner", formatter)
    def test_json_format_selects_structured_formatter(self):
        setup_logging(format_type="json")
-        assert isinstance(self._selected_formatter(), StructuredFormatter)
+        assert isinstance(
            logging.getLogger().handlers[0].formatter, StructuredFormatter)
    def test_readable_format_selects_contextual_formatter(self):
        setup_logging(format_type="readable")
-        assert isinstance(self._selected_formatter(), ContextualFormatter)
+        assert isinstance(
            logging.getLogger().handlers[0].formatter, ContextualFormatter)
    def test_log_file_adds_file_handler(self, tmp_path):
        log_file = tmp_path / "test.log"
@@ -1,100 +0,0 @@
 """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)
@@ -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"