perf(wifi): one status fetch per monitor tick instead of three (#411)

* perf(wifi): one status fetch per monitor tick instead of three

The wifi monitor daemon fetched WiFi status + ethernet state before its
AP-mode check, the check internally fetched the same state again (with
retry), and the daemon fetched a third time afterwards — each fetch is
several nmcli subprocess forks, every 30s, forever, even on a perfectly
healthy link.

check_and_manage_ap_mode's decision logic is extracted to
_manage_ap_mode(status, ethernet, ap_active); the new
check_and_manage_ap_mode_with_state() runs the single (retrying) fetch
battery and returns (changed, status, ethernet, ap_active_after) — the
post-state is derivable because state only ever flips via one enable or
one disable. The original bool-returning method delegates, so existing
callers are untouched. The daemon's dead pre-fetch is removed and its
post-check reads use the returned state; the AP-enable retry semantics
(_get_wifi_status_with_retry) are preserved exactly. ~10-15 forks/30s
drops to ~4-6.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam

* fix(wifi): add missing type hints on AP-mode state helpers, fix unused-var lint in tests

- check_and_manage_ap_mode_with_state now declares its
  Tuple[bool, WiFiStatus, bool, bool] return type instead of being untyped.
- _manage_ap_mode's status/ethernet_connected/ap_active parameters are now
  typed (WiFiStatus, bool, bool), matching its existing -> bool return hint.
- test_wifi_check_state.py: rename unused unpacked variables (status/
  ethernet/ap_after) to underscore-prefixed names at the three call sites
  that don't use them, leaving the used ones untouched.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ

---------

Co-authored-by: Chuck <chuck@example.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

docs/ADAPTIVE_LAYOUT.md

@@ -0,0 +1,234 @@
+# Adaptive Layout & Font Scaling
+
+`src/adaptive_layout.py` lets a plugin render legibly on **any** panel size
+(64x32, 128x32, 96x48, 128x64, 256x64, ...) without hand-tuned per-display
+layouts. It is **opt-in**: nothing changes for plugins that don't use it.
+
+It generalizes three patterns proven in the plugin ecosystem:
+
+| Pattern | Origin | Core API |
+|---|---|---|
+| Geometry scale factor vs. a design size | f1-scoreboard | `ctx.px(base)` / `ctx.scale` |
+| Breakpoint tiers | masters-tournament | `ctx.tier` / `ctx.by_tier({...})` |
+| "Largest crisp font that fits" ladder | baseball-scoreboard | `ctx.fit_text(...)` and friends |
+
+## Quick start
+
+Every `BasePlugin` has a lazy `self.layout` (a `LayoutContext` for the
+current logical display size, rebuilt automatically if the size changes)
+and a one-liner `self.draw_fit(...)`:
+
+```python
+def display(self, force_clear=False):
+    from src.adaptive_layout import LADDER_ARCADE
+
+    b = self.layout.bounds.inset(1)          # Region(0,0,W,H) minus 1px margin
+    rows = b.split_v(3, 1, 1, gap=1)          # 3/5 for time, 1/5 each for the rest
+
+    self.draw_fit(self.time_str, rows[0], ladder=LADDER_ARCADE)
+    self.draw_fit(self.weekday, rows[1])      # default LADDER_GRID
+    self.draw_fit(self.date_str, rows[2])
+    self.display_manager.update_display()
+```
+
+On 128x64 the time renders at press_start 24px; on 64x32 it steps down to
+8px. The rows partition the height, so bands can never overlap — no more
+`y = height - 7` magic numbers.
+
+## Region — rect algebra
+
+`Region(x, y, w, h)` is a frozen dataclass. All carving clamps to
+non-negative dimensions, so degenerate panels behave.
+
+- Carving: `inset(dx, dy)`, `top_band(h)`, `bottom_band(h)`,
+  `middle(top_h, bottom_h)`, `left_col(w)`, `right_col(w)`,
+  `split_h(*weights, gap=0)`, `split_v(*weights, gap=0)`
+- Placement: `align_xy(w, h, align, valign)`, `center_xy(w, h)`,
+  `contains(w, h)`, `.center`, `.right`, `.bottom`
+
+Scoreboard-style layout:
+
+```python
+b = self.layout.bounds
+status = b.top_band(self.layout.px(7))
+detail = b.bottom_band(self.layout.px(7))
+score_area = b.middle(status.h, detail.h)
+away_slot, home_slot = b.left_col(b.h), b.right_col(b.h)
+```
+
+## Font ladders — discrete, never fractional
+
+Pixel fonts (BDF, PressStart2P) only look right at native/integer sizes, so
+fonts are never scaled continuously. A `FontLadder` is an ordered tuple of
+`FontStep(family, size_px)` rungs, largest first; fitting walks down until
+the measured text fits.
+
+- `LADDER_GRID` (default): X11 BDFs at native sizes — 10x20 → 9x18 → 9x15 →
+  8x13 → 7x13 → 6x13 → 6x12 → 6x10 → 6x9 → 5x8 → 5x7 → 4x6 → tom-thumb.
+  Body text, labels, multi-row content.
+- `LADDER_ARCADE`: PressStart2P at 32/24/16/8 (integer multiples of its 8px
+  grid). Headline text: clocks, scores.
+
+Custom ladders are just tuples — e.g. to add your plugin's registered font
+on top: `(FontStep("myplugin::digits", 16),) + LADDER_GRID`.
+
+## LayoutContext
+
+Built per (width, height); exposes facts and fit queries:
+
+- `bounds`, `width`, `height`, `aspect`
+- `tier` by height (`xs`≤16, `sm`≤32, `md`≤48, `lg`≤64, `xl`) and
+  `width_tier` (`narrow`≤64, `normal`≤128, `wide`≤256, `ultrawide`)
+- `is_wide_short` — aspect ≥ 2.5 and height ≤ 32 (the classic 128x32 shape)
+- `scale` — `min(w/design_w, h/design_h)` vs. your manifest's
+  `display.design_size` (default 128x32). **Geometry only** — gaps, icon
+  and logo sizes via `px(base, minimum, maximum)`; fonts use ladders.
+- `by_tier({"sm": 10, "lg": 18})` — value for the nearest defined tier
+  at-or-below the panel's tier.
+- `fit_text(text, box, ladder, ellipsis=True)` → `FitResult` — largest rung
+  that fits; ellipsizes as a last resort. Cached per (text, box, ladder).
+- `fit_text_proportional(text, box, base_size_px, ladder, ellipsis=True, scale=None)` —
+  rung closest to (not exceeding) `base_size_px * scale`, still capped to
+  what fits the box. Use this instead of `fit_text` when several
+  independently-fitted elements need to stay visually harmonious as the
+  panel grows — `fit_text` maximizes *each one* within its own region,
+  which can make one element (e.g. a score with a generous box) balloon
+  out of proportion to a neighbor that scales by geometry (e.g. logos
+  sized via `px()`), even though each individual pick is "correct" in
+  isolation. `base_size_px` is normally the element's existing classic/
+  fixed font size. `scale` defaults to `self.scale` (the conservative
+  min-of-both-axes factor `px()` uses); pass an axis-specific value when
+  the surrounding composition already scales that way — e.g. a scoreboard
+  whose logo slots track height alone (`min(height, width // 2)`) should
+  size its text by `height / design_height` too, or the text reads as
+  under-scaled next to bigger logos on a panel that only grew taller.
+- `fit_lines(lines, box, ladder, spacing)` — every line fits the width and
+  the stack fits the height (measures the actual strings).
+- `font_for_rows(rows, box_h, ladder)` — largest rung whose line height
+  fits `rows` rows.
+
+`FitResult` carries the ready-to-use `font` (drops straight into
+`display_manager.draw_text(font=...)`), the possibly-ellipsized `text`,
+ink `width`/`height`, `baseline`, `y_offset`, `line_height`, and `fits`.
+
+## Adaptive images
+
+`src/adaptive_images.py` is the image counterpart to `fit_text`, exposed as
+`self.layout.fit_image(...)` (cached per panel size) and the one-liner
+`self.draw_image(...)`:
+
+```python
+# Team logo: trim its transparent padding, fill the slot height (the
+# football/hockey pattern), cached across frames by a stable key
+self.draw_image(logo, regs.away_slot, mode="fill_height",
+                crop_to_ink=True, cache_key=f"logo:{abbr}")
+
+# Album art: cover-crop a square, faces kept by the top anchor
+self.draw_image(art, row.art, mode="cover", anchor="top")
+
+# Pixel flags / sprite icons: NEAREST keeps hard edges
+from src.adaptive_images import RESAMPLE_NEAREST
+self.draw_image(flag, box, resample=RESAMPLE_NEAREST)
+```
+
+Modes: `contain` (letterbox, default), `cover` (crop-to-fill),
+`fill_height` (logo-style), `stretch`. Unlike PIL's `thumbnail()`
+(downscale-only — why imagery stays tiny on big panels) fitting **upscales
+by default**; pass `upscale=False` for the legacy behavior. Results are
+cached per (image, box size, options) with a bounded LRU — always pass a
+stable `cache_key` (e.g. `"logo:KC"`) for images you reload. The module
+also exports the Pillow-compat `RESAMPLE_LANCZOS`/`RESAMPLE_NEAREST`
+constants so plugins can drop their local shims.
+
+## Composite layouts
+
+Pre-carved Region arrangements for the layouts plugins keep rebuilding:
+
+```python
+from src.adaptive_layout import scoreboard_regions, media_row
+
+regs = scoreboard_regions(self.layout.bounds, ctx=self.layout)
+# regs.away_slot / home_slot   — logo slots (logo_slot = min(H, W // 2),
+#                                 capped so a center reserve always exists —
+#                                 see below)
+# regs.status_band             — top band (replaces the magic y = 1)
+# regs.score_area              — center gap, plus a controlled bleed into
+#                                 each logo slot (replaces y = H//2 - 3)
+# regs.detail_band             — bottom band (replaces y = H - 7)
+# regs.bottom_left / bottom_right — record/timeout corners
+
+row = media_row(self.layout.bounds, ctx=self.layout)   # art left, text right
+```
+
+Both work on the full panel or on a scroll-mode card Region. They return
+Regions and never draw — compose them with `draw_fit`/`draw_image`.
+
+**`scoreboard_regions`'s center reserve.** The raw `logo_slot = min(H, W//2)`
+formula has a blind spot: at exactly 2:1 aspect ratio (width = 2×height —
+two, four, or more square modules stacked into a taller panel, e.g.
+96x48, 128x64, 256x128) the two logo slots mathematically claim the
+*entire* width, leaving zero pixels for a center column no matter how
+big the panel gets. Wide panels (the 128x32 design baseline, 192x48,
+256x32) never hit this, since height is already the tighter constraint
+there. Two parameters fix it without any plugin-side code:
+`min_center_fraction`/`min_center_design_px` guarantee a real minimum
+center reserve at any aspect ratio, and `score_bleed_fraction` lets the
+score's *fit box* extend a controlled amount into each logo slot — the
+same way a real broadcast scoreboard's numbers cross slightly into the
+team marks flanking them — so a short score string never has to truncate
+even on the tightest aspect ratios. All three have sane defaults; override
+them per call if a plugin's card proportions genuinely differ.
+
+## Preserving user customization
+
+Adaptive layout supplies *defaults*; explicit user configuration wins:
+
+- **User-set fonts win.** If the plugin's config has an explicit
+  `font`/`font_size` for an element, load it as before and skip the ladder —
+  fit only when the user hasn't overridden (see the football-scoreboard
+  `_resolve_element_fit` pattern).
+- **Offsets apply on top.** `customization.layout.<element>.{x_offset,y_offset}`
+  style knobs translate the *computed* region as a final step:
+  `region.offset(user_dx, user_dy)`. `draw_image(..., offset=(dx, dy))`
+  does the same for images.
+- **Colors pass through.** `draw_fit`/`draw_fitted_text` take explicit
+  `color=` params; adaptive mode never repaints semantic or user-chosen
+  colors.
+
+## Manifest declaration
+
+Declare the size your layout was authored against so `ctx.scale` means
+something:
+
+```json
+"display": { "design_size": { "width": 128, "height": 32 } }
+```
+
+Also available under `requires.display_size`: `min_width`, `min_height`,
+`max_width`, `max_height`.
+
+## Performance notes (Pi)
+
+Fit queries are cached, so cost is O(unique strings). For per-second text
+(clocks, live scores), fit on a **shape placeholder** and reuse the font:
+
+```python
+fit = self.layout.fit_text("00:00", box, ladder=LADDER_ARCADE)  # cached once
+self.display_manager.draw_text(current_time, font=fit.font, ...)
+```
+
+## Testing across sizes
+
+The harness already renders every plugin at a spread of sizes (now
+including 96x48):
+
+```bash
+python scripts/check_plugin.py <plugin-dir> --sizes 64x32,128x32,96x48,128x64,256x64
+python scripts/render_plugin.py <plugin-dir> --width 96 --height 48
+```
+
+`BoundsCheckingDisplayManager` flags right/bottom overflow and now records
+mediated draw calls with negative coordinates in
+`negative_coordinate_calls` (raw-PIL draws remain uncovered).
+
+Reference migration: the **text-display** plugin's `font_mode: "auto"`.

docs/ADVANCED_PLUGIN_DEVELOPMENT.md

@@ -2,6 +2,12 @@
 
 Advanced patterns, examples, and best practices for developing LEDMatrix plugins.
 
+> **Adaptive layout:** for plugins that should render legibly on any panel
+> size (fonts that grow on big panels, layouts that degrade gracefully on
+> small ones), use the adaptive layout system — `self.layout`, `draw_fit`,
+> `draw_image`, `scoreboard_regions` — documented in
+> [ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md).
+
 ## Table of Contents
 
 - [Using Weather Icons](#using-weather-icons)

docs/DEVELOPER_QUICK_REFERENCE.md

@@ -48,6 +48,12 @@ display_manager.draw_text("Centered", centered=True)  # Auto-center
 width = display_manager.get_text_width("Text", font)
 height = display_manager.get_font_height(font)
 
+# Adaptive layout (recommended for multi-size support — text and images
+# that scale to any panel; see docs/ADAPTIVE_LAYOUT.md)
+rows = self.layout.bounds.inset(1).split_v(3, 1, gap=1)
+self.draw_fit("12:34", rows[0])                 # largest crisp font that fits
+self.draw_image(logo, rows[1], mode="fill_height", crop_to_ink=True)
+
 # Weather icons
 display_manager.draw_weather_icon("rain", x=10, y=10, size=16)
 

docs/DEV_PREVIEW.md

@@ -6,6 +6,12 @@ Tools for rapid plugin development without deploying to the RPi.
 
 Interactive web UI for tweaking plugin configs and seeing the rendered display in real time.
 
+The size inputs have a preset dropdown with the harness's standard panel
+sizes, and the **All Sizes** button renders the current config at every
+harness size in a side-by-side gallery (`POST /api/render-matrix`) — the
+quickest way to eyeball adaptive-layout behavior across panels
+(see [ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md)).
+
 ### Quick Start
 
 ```bash

docs/FONT_MANAGER.md

@@ -1,5 +1,12 @@
 # FontManager Usage Guide
 
+> **Picking a size automatically:** if you want the *largest font that fits
+> a given area* rather than a fixed size, use the adaptive layout system's
+> font ladders, which resolve through this FontManager. `BasePlugin`
+> subclasses get this as `self.layout.fit_text(...)`; other code can build
+> a `LayoutContext(width, height, font_manager)` directly — see
+> [ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md).
+
 ## Overview
 
 The enhanced FontManager provides comprehensive font management for the LEDMatrix application with support for:

docs/PLUGIN_API_REFERENCE.md

@@ -2,6 +2,11 @@
 
 Complete API reference for plugin developers. This document describes all methods and properties available to plugins through the Display Manager, Cache Manager, and Plugin Manager.
 
+> **Adaptive layout:** every `BasePlugin` also exposes `self.layout`,
+> `self.draw_fit(text, region)` and `self.draw_image(img, region, ...)` —
+> the recommended way to render text and images that scale to any panel
+> size. See [ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md).
+
 ## Table of Contents
 
 - [BasePlugin](#baseplugin)

docs/PLUGIN_DEVELOPMENT_GUIDE.md

@@ -2,6 +2,14 @@
 
 This guide explains how to set up a development workflow for plugins that are maintained in separate Git repositories while still being able to test them within the LEDMatrix project.
 
+> **Rendering guidance:** plugins should read the display size dynamically
+> (`self.display_manager.matrix.width/height`) rather than hardcoding one
+> panel. For plugins that want to *scale* their layout to any panel, the
+> opt-in adaptive layout system ([ADAPTIVE_LAYOUT.md](ADAPTIVE_LAYOUT.md))
+> provides the shared helpers — fonts, images, and composite layouts that
+> scale. Existing plugins keep their classic rendering unless they adopt
+> those APIs; nothing migrates automatically.
+
 ## Overview
 
 When developing plugins in separate repositories, you need a way to:

schema/manifest_schema.json

@@ -90,11 +90,40 @@
             "min_height": {
               "type": "integer",
               "minimum": 1
+            },
+            "max_width": {
+              "type": "integer",
+              "minimum": 1
+            },
+            "max_height": {
+              "type": "integer",
+              "minimum": 1
             }
           }
         }
       }
     },
+    "display": {
+      "type": "object",
+      "properties": {
+        "design_size": {
+          "type": "object",
+          "properties": {
+            "width": {
+              "type": "integer",
+              "minimum": 8
+            },
+            "height": {
+              "type": "integer",
+              "minimum": 8
+            }
+          },
+          "required": ["width", "height"],
+          "description": "Panel size the plugin's layout was authored against; core derives the adaptive-layout scale factor from it. Defaults to 128x32 when omitted."
+        }
+      },
+      "description": "Display/layout hints for the adaptive layout system"
+    },
     "config_schema": {
       "type": "string",
       "description": "Path to configuration schema file"

scripts/check_plugin.py

@@ -37,10 +37,11 @@ os.environ['EMULATOR'] = 'true'
 
 from src.logging_config import get_logger  # noqa: E402
 from src.plugin_system.testing.loading import (  # noqa: E402
-    find_plugin_dir, load_config_defaults, load_harness_spec,
+    build_full_config, find_plugin_dir, load_harness_spec, load_manifest,
 )
 from src.plugin_system.testing.harness import (  # noqa: E402
     RenderResult, render_plugin_matrix, compare_to_goldens, write_goldens,
+    check_scale_up,
 )
 from src.plugin_system.testing.sizes import (  # noqa: E402
     parse_size_token, resolve_test_sizes, safe_mode_filename, size_label,
@@ -96,12 +97,11 @@ def check_one(plugin_id: str, search_dirs: List[str], sizes, mock_data: Dict,
     # matrix path does; explicit CLI flags still override the file.
     spec = load_harness_spec(plugin_dir)
 
-    # config_schema defaults (real-install behavior), then harness.json config,
-    # then CLI --config — most specific wins.
-    full_config = {"enabled": True}
-    full_config.update(load_config_defaults(plugin_dir))
-    full_config.update(spec.get("config", {}))
-    full_config.update(config)
+    # config_schema defaults (real-install behavior, with enabled forced True
+    # so a plugin's own enabled:false default can't accidentally disable
+    # testing), then harness.json config, then CLI --config — most specific
+    # wins.
+    full_config = build_full_config(plugin_dir, spec, config)
 
     # Precedence: CLI flag > LEDMATRIX_TEST_SIZES env > harness.json > default.
     effective_sizes = sizes if sizes else resolve_test_sizes(spec.get("sizes"))
@@ -110,28 +110,55 @@ def check_one(plugin_id: str, search_dirs: List[str], sizes, mock_data: Dict,
     effective_freeze = freeze_time or spec.get("freeze_time")
     effective_run_update = run_update and not spec.get("skip_update", False)
 
-    results = render_plugin_matrix(
-        plugin_id=plugin_id, plugin_dir=plugin_dir, config=full_config,
-        mock_data=effective_mock_data, sizes=effective_sizes,
-        run_update=effective_run_update, freeze_time=effective_freeze,
-    )
+    # The plugin's declared design size drives the scale-up fill check
+    # (panels >= 2x the design size must not be left mostly empty).
+    declared = load_manifest(plugin_dir).get("display", {}).get("design_size", {})
+    design_size = (int(declared.get("width", 128)), int(declared.get("height", 32)))
+    fill_strict = spec.get("fill_check") == "strict"
 
-    golden_dir = golden_dir_override or (plugin_dir / 'test' / 'golden')
-    if update_golden:
-        written = write_goldens(results, golden_dir)
-        logger.info("Wrote %d golden image(s) for %s to %s", written, plugin_id, golden_dir)
-    else:
-        compare_to_goldens(results, golden_dir)
+    # Every run: the base config, plus one per harness.json "variant" —
+    # a config overlay with its own golden dir (e.g. adaptive layout mode
+    # tested alongside the classic default).
+    runs = [(None, {}, golden_dir_override or (plugin_dir / 'test' / 'golden'))]
+    for variant in spec.get("variants", []):
+        name = variant.get("name") or "variant"
+        vdir = plugin_dir / variant.get("golden_dir", f"test/golden-{name}")
+        runs.append((name, variant.get("config", {}), vdir))
 
-    if out_dir:
-        for r in results:
-            if r.image is None:
-                continue
-            dest = out_dir / plugin_id / size_label(r.width, r.height)
-            dest.mkdir(parents=True, exist_ok=True)
-            r.image.save(dest / f"{safe_mode_filename(r.mode)}.png", format="PNG")
+    all_run_results: List[RenderResult] = []
+    for variant_name, overlay, golden_dir in runs:
+        run_config = {**full_config, **overlay}
+        results = render_plugin_matrix(
+            plugin_id=plugin_id, plugin_dir=plugin_dir, config=run_config,
+            mock_data=effective_mock_data, sizes=effective_sizes,
+            run_update=effective_run_update, freeze_time=effective_freeze,
+        )
 
-    return results
+        if update_golden:
+            written = write_goldens(results, golden_dir)
+            logger.info("Wrote %d golden image(s) for %s%s to %s", written, plugin_id,
+                        f" [{variant_name}]" if variant_name else "", golden_dir)
+        else:
+            compare_to_goldens(results, golden_dir)
+
+        check_scale_up(results, design_size=design_size, strict=fill_strict)
+
+        # Tag variant runs so the report and PNG dumps stay distinguishable.
+        if variant_name:
+            for r in results:
+                r.mode = f"{r.mode}@{variant_name}"
+
+        if out_dir:
+            for r in results:
+                if r.image is None:
+                    continue
+                dest = out_dir / plugin_id / size_label(r.width, r.height)
+                dest.mkdir(parents=True, exist_ok=True)
+                r.image.save(dest / f"{safe_mode_filename(r.mode)}.png", format="PNG")
+
+        all_run_results.extend(results)
+
+    return all_run_results
 
 
 def print_report(all_results: Dict[str, List[RenderResult]]) -> bool:
@@ -147,6 +174,10 @@ def print_report(all_results: Dict[str, List[RenderResult]]) -> bool:
                     detail = " (golden ✓)"
                 if r.update_error is not None:
                     detail += f" (update warn: {r.update_error})"
+                if r.fill_checked and r.fill_ok is None and r.fill_extent:
+                    # warn-only underfill: big panel left mostly empty
+                    ex, ey = r.fill_extent
+                    detail += f" (fill warn: extent {ex:.0%}x{ey:.0%})"
             else:
                 everything_ok = False
                 if r.error is not None:
@@ -156,6 +187,10 @@ def print_report(all_results: Dict[str, List[RenderResult]]) -> bool:
                 elif r.golden_ok is False:
                     status = "FAIL"
                     detail = f" golden drift: {r.golden_diff_pixels}px (max Δ={r.golden_max_delta})"
+                elif r.fill_ok is False:
+                    ex, ey = r.fill_extent or (0.0, 0.0)
+                    status = "FAIL"
+                    detail = f" fill: extent {ex:.0%}x{ey:.0%} below required coverage"
                 else:
                     status, detail = "FAIL", ""
             print(f"  [{status}] {r.size_label:>7}  {r.mode}{detail}")

scripts/dev_server.py

@@ -16,6 +16,7 @@ Opens at http://localhost:5001
 import sys
 import os
 import json
+import re
 import time
 import argparse
 import logging
@@ -44,6 +45,10 @@ MAX_HEIGHT = 512
 MIN_WIDTH = 1
 MIN_HEIGHT = 1
 
+# plugin_id arrives in request input and is used to build filesystem paths —
+# allowlist it (same pattern the web UI's pages_v3 uses)
+_SAFE_PLUGIN_ID_RE = re.compile(r'^[a-zA-Z0-9_-]{1,64}$')
+
 
 # --------------------------------------------------------------------------
 # Plugin discovery
@@ -106,15 +111,30 @@ def discover_plugins() -> List[Dict[str, Any]]:
 
 
 def find_plugin_dir(plugin_id: str) -> Optional[Path]:
-    """Find a plugin directory by ID."""
+    """Find a plugin directory by ID.
+
+    plugin_id comes from request input: it must pass an allowlist match,
+    and the resulting directory is normalized and required to live inside
+    one of the plugin search dirs, so a crafted id can never name a path
+    outside them.
+    """
+    if not isinstance(plugin_id, str) or not _SAFE_PLUGIN_ID_RE.match(plugin_id):
+        return None
     from src.plugin_system.plugin_loader import PluginLoader
     loader = PluginLoader()
     for search_dir in get_search_dirs():
         if not search_dir.exists():
             continue
         result = loader.find_plugin_directory(plugin_id, search_dir)
-        if result:
-            return Path(result)
+        if not result:
+            continue
+        # Normalize WITHOUT following symlinks (dev plugins are often
+        # symlinked into plugins/) and require lexical containment in the
+        # search dir, so no id can ever name a path outside it.
+        result_abs = os.path.abspath(str(result))
+        root_abs = os.path.abspath(str(search_dir))
+        if os.path.commonpath([result_abs, root_abs]) == root_abs:
+            return Path(result_abs)
     return None
 
 
@@ -176,6 +196,118 @@ def api_plugin_defaults(plugin_id):
     return jsonify({'defaults': defaults})
 
 
+def _render_once(plugin_id, plugin_dir, manifest, config, mock_data, width, height,
+                 skip_update):
+    """Render one plugin at one size. Returns the /api/render response dict.
+
+    A fresh plugin instance per call, mirroring the safety harness, so sizes
+    never share state.
+    """
+    from src.plugin_system.testing import VisualTestDisplayManager, MockCacheManager, MockPluginManager
+    from src.plugin_system.plugin_loader import PluginLoader
+
+    display_manager = VisualTestDisplayManager(width=width, height=height)
+    cache_manager = MockCacheManager()
+    plugin_manager = MockPluginManager()
+
+    # Pre-populate cache with mock data
+    for key, value in mock_data.items():
+        cache_manager.set(key, value)
+
+    loader = PluginLoader()
+    errors = []
+    warnings = []
+
+    plugin_instance, _module = loader.load_plugin(
+        plugin_id=plugin_id,
+        manifest=manifest,
+        plugin_dir=plugin_dir,
+        config=config,
+        display_manager=display_manager,
+        cache_manager=cache_manager,
+        plugin_manager=plugin_manager,
+        install_deps=False,
+    )
+
+    start_time = time.time()
+
+    # Run update()
+    if not skip_update:
+        try:
+            plugin_instance.update()
+        except Exception as e:
+            logger.warning("update() raised for plugin %s", plugin_id, exc_info=True)
+            warnings.append(f"update() raised: {type(e).__name__} — see server log")
+
+    # Run display()
+    try:
+        plugin_instance.display(force_clear=True)
+    except Exception as e:
+        logger.warning("display() raised for plugin %s", plugin_id, exc_info=True)
+        errors.append(f"display() raised: {type(e).__name__} — see server log")
+
+    render_time_ms = round((time.time() - start_time) * 1000, 1)
+
+    return {
+        'image': f'data:image/png;base64,{display_manager.get_image_base64()}',
+        'width': width,
+        'height': height,
+        'render_time_ms': render_time_ms,
+        'errors': errors,
+        'warnings': warnings,
+    }
+
+
+def _trusted_plugin_dir(plugin_dir: Path) -> Optional[Path]:
+    """Re-derive a plugin directory from the search dirs' own listings.
+
+    Path-injection barrier: unlike ``Path.iterdir()`` (which CodeQL doesn't
+    recognize as a taint-clearing enumeration), ``os.scandir()`` is. The
+    returned Path is built from a trusted root plus a name the filesystem
+    itself produced under that root via scandir — request-derived strings
+    never enter its construction — so a crafted plugin id can never make
+    downstream file access leave the plugin search dirs. Comparison is by
+    name, deliberately without symlink resolution (dev plugins are
+    commonly symlinked into plugins/).
+    """
+    wanted_name = Path(os.path.normpath(str(plugin_dir))).name
+    for search_dir in get_search_dirs():
+        search_dir_str = str(search_dir)
+        try:
+            with os.scandir(search_dir_str) as entries:
+                for entry in entries:
+                    if entry.name == wanted_name and entry.is_dir():
+                        return Path(search_dir_str) / entry.name
+        except OSError:
+            continue
+    return None
+
+
+def _parse_render_request(data):
+    """Shared /api/render* request prep. Returns (plugin_dir, manifest, config,
+    mock_data, skip_update) or raises ValueError with a client message."""
+    plugin_id = data['plugin_id']
+    candidate_dir = find_plugin_dir(plugin_id)
+    # Never reuse `candidate_dir` past this point: it's built from
+    # request-derived input, and a variable reassigned only on some paths
+    # isn't a barrier CodeQL's flow analysis honors. `trusted_dir` is the
+    # sole name used below, always the scandir-sourced result.
+    trusted_dir = _trusted_plugin_dir(candidate_dir) if candidate_dir else None
+    if not trusted_dir:
+        raise LookupError(f'Plugin not found: {plugin_id}')
+
+    manifest_path = trusted_dir / 'manifest.json'
+    with open(manifest_path, 'r') as f:
+        manifest = json.load(f)
+
+    # Build config: schema defaults + user overrides
+    config = {'enabled': True}
+    config.update(load_config_defaults(trusted_dir))
+    config.update(data.get('config', {}))
+
+    return trusted_dir, manifest, config, data.get('mock_data', {}), data.get('skip_update', False)
+
+
 @app.route('/api/render', methods=['POST'])
 def api_render():
     """Render a plugin and return the display as base64 PNG."""
@@ -183,11 +315,6 @@ def api_render():
     if not data or 'plugin_id' not in data:
         return jsonify({'error': 'plugin_id is required'}), 400
 
-    plugin_id = data['plugin_id']
-    user_config = data.get('config', {})
-    mock_data = data.get('mock_data', {})
-    skip_update = data.get('skip_update', False)
-
     try:
         width = int(data.get('width', 128))
         height = int(data.get('height', 32))
@@ -199,78 +326,77 @@ def api_render():
     if not (MIN_HEIGHT <= height <= MAX_HEIGHT):
         return jsonify({'error': f'height must be between {MIN_HEIGHT} and {MAX_HEIGHT}'}), 400
 
-    # Find plugin
-    plugin_dir = find_plugin_dir(plugin_id)
-    if not plugin_dir:
-        return jsonify({'error': f'Plugin not found: {plugin_id}'}), 404
-
-    # Load manifest
-    manifest_path = plugin_dir / 'manifest.json'
-    with open(manifest_path, 'r') as f:
-        manifest = json.load(f)
-
-    # Build config: schema defaults + user overrides
-    config_defaults = load_config_defaults(plugin_dir)
-    config = {'enabled': True}
-    config.update(config_defaults)
-    config.update(user_config)
-
-    # Create display manager and mocks
-    from src.plugin_system.testing import VisualTestDisplayManager, MockCacheManager, MockPluginManager
-    from src.plugin_system.plugin_loader import PluginLoader
-
-    display_manager = VisualTestDisplayManager(width=width, height=height)
-    cache_manager = MockCacheManager()
-    plugin_manager = MockPluginManager()
-
-    # Pre-populate cache with mock data
-    for key, value in mock_data.items():
-        cache_manager.set(key, value)
-
-    # Load plugin
-    loader = PluginLoader()
-    errors = []
-    warnings = []
+    try:
+        plugin_dir, manifest, config, mock_data, skip_update = _parse_render_request(data)
+    except LookupError:
+        return jsonify({'error': f"Plugin not found: {data['plugin_id']}"}), 404
+    except Exception:
+        # Bad manifest.json / schema / fixture — details go to the dev's
+        # console, not the HTTP response
+        app.logger.exception('render request preparation failed')
+        return jsonify({'error': 'Could not prepare render request; see server log'}), 400
 
     try:
-        plugin_instance, module = loader.load_plugin(
-            plugin_id=plugin_id,
-            manifest=manifest,
-            plugin_dir=plugin_dir,
-            config=config,
-            display_manager=display_manager,
-            cache_manager=cache_manager,
-            plugin_manager=plugin_manager,
-            install_deps=False,
-        )
-    except Exception as e:
-        return jsonify({'error': f'Failed to load plugin: {e}'}), 500
+        result = _render_once(data['plugin_id'], plugin_dir, manifest, config,
+                              mock_data, width, height, skip_update)
+    except Exception:
+        app.logger.exception('plugin load failed during render')
+        return jsonify({'error': 'Failed to load plugin; see server log'}), 500
+    return jsonify(result)
 
-    start_time = time.time()
 
-    # Run update()
-    if not skip_update:
+@app.route('/api/sizes')
+def api_sizes():
+    """The representative panel-size sample the safety harness renders at."""
+    from src.plugin_system.testing.sizes import DEFAULT_TEST_SIZES
+    return jsonify({'sizes': [list(s) for s in DEFAULT_TEST_SIZES]})
+
+
+MAX_MATRIX_SIZES = 12
+
+
+@app.route('/api/render-matrix', methods=['POST'])
+def api_render_matrix():
+    """Render a plugin at a list of sizes (default: the harness sample) so the
+    UI can show a side-by-side multi-resolution gallery."""
+    data = request.get_json()
+    if not data or 'plugin_id' not in data:
+        return jsonify({'error': 'plugin_id is required'}), 400
+
+    from src.plugin_system.testing.sizes import DEFAULT_TEST_SIZES
+    sizes = data.get('sizes') or [list(s) for s in DEFAULT_TEST_SIZES]
+    if len(sizes) > MAX_MATRIX_SIZES:
+        return jsonify({'error': f'at most {MAX_MATRIX_SIZES} sizes per request'}), 400
+    parsed_sizes = []
+    for pair in sizes:
         try:
-            plugin_instance.update()
-        except Exception as e:
-            warnings.append(f"update() raised: {e}")
+            w, h = int(pair[0]), int(pair[1])
+        except (TypeError, ValueError, IndexError):
+            return jsonify({'error': f'invalid size entry {pair!r} (expected [w, h])'}), 400
+        if not (MIN_WIDTH <= w <= MAX_WIDTH and MIN_HEIGHT <= h <= MAX_HEIGHT):
+            return jsonify({'error': f'size {w}x{h} out of bounds'}), 400
+        parsed_sizes.append((w, h))
 
-    # Run display()
     try:
-        plugin_instance.display(force_clear=True)
-    except Exception as e:
-        errors.append(f"display() raised: {e}")
+        plugin_dir, manifest, config, mock_data, skip_update = _parse_render_request(data)
+    except LookupError:
+        return jsonify({'error': f"Plugin not found: {data['plugin_id']}"}), 404
+    except Exception:
+        app.logger.exception('render request preparation failed')
+        return jsonify({'error': 'Could not prepare render request; see server log'}), 400
 
-    render_time_ms = round((time.time() - start_time) * 1000, 1)
-
-    return jsonify({
-        'image': f'data:image/png;base64,{display_manager.get_image_base64()}',
-        'width': width,
-        'height': height,
-        'render_time_ms': render_time_ms,
-        'errors': errors,
-        'warnings': warnings,
-    })
+    results = []
+    for w, h in parsed_sizes:
+        try:
+            results.append(_render_once(data['plugin_id'], plugin_dir, manifest,
+                                        config, mock_data, w, h, skip_update))
+        except Exception:
+            app.logger.exception('plugin load failed during %dx%d render', w, h)
+            results.append({'image': None, 'width': w, 'height': h,
+                            'render_time_ms': 0,
+                            'errors': ['Failed to load plugin; see server log'],
+                            'warnings': []})
+    return jsonify({'results': results})
 
 
 # --------------------------------------------------------------------------

scripts/render_plugin.py

@@ -28,7 +28,7 @@ os.environ['EMULATOR'] = 'true'
 # Import logger after path setup so src.logging_config is importable
 from src.logging_config import get_logger  # noqa: E402
 from src.plugin_system.testing.loading import (  # noqa: E402
-    find_plugin_dir, load_manifest, load_config_defaults,
+    build_full_config, find_plugin_dir, load_manifest,
 )
 logger = get_logger("[Render Plugin]")
 
@@ -83,16 +83,13 @@ def main() -> int:
     manifest = load_manifest(Path(plugin_dir))
 
     # Parse config: start with schema defaults, then apply overrides
-    config_defaults = load_config_defaults(Path(plugin_dir))
     try:
         user_config = json.loads(args.config)
     except json.JSONDecodeError as e:
         logger.error("Invalid JSON config: %s", e)
         return 1
 
-    config = {'enabled': True}
-    config.update(config_defaults)
-    config.update(user_config)
+    config = build_full_config(Path(plugin_dir), cli_config=user_config)
 
     # Load mock data if provided
     mock_data = {}

scripts/templates/dev_preview.html

@@ -209,6 +209,11 @@
                            onchange="onConfigChange()">
                     <span class="text-xs ml-2" style="color: var(--text-secondary);">px</span>
                 </div>
+                <select id="sizePreset" onchange="applySizePreset()"
+                        class="w-full mt-2 px-2 py-1.5 rounded text-xs"
+                        style="background: var(--bg-primary); color: var(--text-secondary); border: 1px solid var(--border-color);">
+                    <option value="">Preset sizes…</option>
+                </select>
             </div>
 
             <!-- Config form -->
@@ -242,13 +247,18 @@
                 </div>
             </details>
 
-            <!-- Render button -->
+            <!-- Render buttons -->
             <div class="flex gap-2">
                 <button onclick="renderPlugin()" id="renderBtn"
                         class="flex-1 px-4 py-2.5 rounded-lg text-sm font-medium text-white"
                         style="background: var(--accent);">
                     Render
                 </button>
+                <button onclick="renderAllSizes()" id="renderAllBtn" title="Render at every harness test size"
+                        class="px-4 py-2.5 rounded-lg text-sm font-medium"
+                        style="background: var(--bg-tertiary); color: var(--text-primary); border: 1px solid var(--border-color);">
+                    All Sizes
+                </button>
             </div>
         </div>
 
@@ -311,6 +321,15 @@
             <div id="messagesPanel" class="panel p-3 hidden">
                 <div id="messagesList" class="text-xs font-mono space-y-1"></div>
             </div>
+
+            <!-- Multi-size gallery -->
+            <div id="galleryPanel" class="panel p-4 hidden">
+                <div class="flex items-center justify-between mb-3">
+                    <span class="text-xs font-medium" style="color: var(--text-secondary);">All Sizes</span>
+                    <span class="text-xs" style="color: var(--text-secondary);" id="galleryStatus"></span>
+                </div>
+                <div id="galleryGrid" class="flex flex-wrap gap-4 items-start"></div>
+            </div>
         </div>
     </div>
 
@@ -340,8 +359,30 @@
                 opt.textContent = `${p.name} (${p.id})`;
                 select.appendChild(opt);
             });
+
+            // Load harness size presets
+            try {
+                const sizesRes = await fetch('/api/sizes');
+                const sizesData = await sizesRes.json();
+                const preset = document.getElementById('sizePreset');
+                (sizesData.sizes || []).forEach(([w, h]) => {
+                    const opt = document.createElement('option');
+                    opt.value = `${w}x${h}`;
+                    opt.textContent = `${w} x ${h}`;
+                    preset.appendChild(opt);
+                });
+            } catch (e) { /* presets are a convenience; ignore */ }
         });
 
+        function applySizePreset() {
+            const value = document.getElementById('sizePreset').value;
+            if (!value) return;
+            const [w, h] = value.split('x');
+            document.getElementById('displayWidth').value = w;
+            document.getElementById('displayHeight').value = h;
+            onConfigChange();
+        }
+
         // ---------- Plugin selection ----------
         async function onPluginChange() {
             const pluginId = document.getElementById('pluginSelect').value;
@@ -485,6 +526,89 @@
             }
         }
 
+        // ---------- Multi-size gallery ----------
+        async function renderAllSizes() {
+            if (!currentPluginId) return;
+
+            const btn = document.getElementById('renderAllBtn');
+            const panel = document.getElementById('galleryPanel');
+            const grid = document.getElementById('galleryGrid');
+            const status = document.getElementById('galleryStatus');
+            btn.disabled = true;
+            btn.textContent = 'Rendering…';
+            panel.classList.remove('hidden');
+            grid.innerHTML = '';
+            status.textContent = 'Rendering at all harness sizes…';
+
+            const config = jsonEditor ? jsonEditor.getValue() : {};
+            config.enabled = true;
+            let mockData = {};
+            const mockInput = document.getElementById('mockDataInput').value.trim();
+            if (mockInput) {
+                try { mockData = JSON.parse(mockInput); }
+                catch (e) { showMessages([], [`Mock data JSON error: ${e.message}`]); }
+            }
+
+            try {
+                const res = await fetch('/api/render-matrix', {
+                    method: 'POST',
+                    headers: { 'Content-Type': 'application/json' },
+                    body: JSON.stringify({
+                        plugin_id: currentPluginId,
+                        config: config,
+                        mock_data: mockData,
+                    }),
+                });
+                const data = await res.json();
+                if (data.error) {
+                    status.textContent = data.error;
+                    return;
+                }
+
+                let failures = 0;
+                (data.results || []).forEach(r => {
+                    const cell = document.createElement('div');
+                    cell.style.cssText = 'display:flex;flex-direction:column;gap:4px;';
+                    const failed = (r.errors || []).length > 0 || !r.image;
+                    if (failed) failures++;
+
+                    const label = document.createElement('span');
+                    label.className = 'text-xs font-mono';
+                    label.style.color = failed ? '#f87171' : 'var(--text-secondary)';
+                    label.textContent = `${r.width}x${r.height} · ${r.render_time_ms}ms`;
+                    cell.appendChild(label);
+
+                    if (r.image) {
+                        const img = document.createElement('img');
+                        img.src = r.image;
+                        // Small panels get 2x zoom so they stay legible in the grid
+                        const zoom = r.height >= 128 ? 1 : 2;
+                        img.style.cssText =
+                            `image-rendering: pixelated; width:${r.width * zoom}px; ` +
+                            `height:${r.height * zoom}px; ` +
+                            `border:1px solid ${failed ? '#f87171' : 'var(--border-color)'};`;
+                        cell.appendChild(img);
+                    }
+                    if (failed) {
+                        const err = document.createElement('span');
+                        err.className = 'text-xs font-mono';
+                        err.style.color = '#f87171';
+                        err.textContent = (r.errors || ['render failed']).join('; ');
+                        cell.appendChild(err);
+                    }
+                    grid.appendChild(cell);
+                });
+                status.textContent = failures
+                    ? `${failures} size(s) failed`
+                    : `${(data.results || []).length} sizes rendered`;
+            } catch (e) {
+                status.textContent = `Network error: ${e.message}`;
+            } finally {
+                btn.disabled = false;
+                btn.textContent = 'All Sizes';
+            }
+        }
+
         // ---------- Zoom ----------
         function updateZoom() {
             const zoom = parseInt(document.getElementById('zoomSlider').value);

scripts/utils/wifi_monitor_daemon.py

@@ -78,21 +78,17 @@ class WiFiMonitorDaemon:
         
         while self.running:
             try:
-                # Get current status before checking
-                status = self.wifi_manager.get_wifi_status()
-                ethernet_connected = self.wifi_manager._is_ethernet_connected()
-                
-                # Check WiFi status and manage AP mode
-                state_changed = self.wifi_manager.check_and_manage_ap_mode()
-                
-                # Get updated status after check
-                updated_status = self.wifi_manager.get_wifi_status()
-                updated_ethernet = self.wifi_manager._is_ethernet_connected()
-                
+                # One combined check that also returns the state it observed —
+                # the previous flow fetched status before AND after the check
+                # on top of the check's own internal fetch, each one several
+                # nmcli subprocess forks, every 30s, forever.
+                (state_changed, updated_status, updated_ethernet,
+                 ap_active) = self.wifi_manager.check_and_manage_ap_mode_with_state()
+
                 current_state = {
                     'connected': updated_status.connected,
                     'ethernet_connected': updated_ethernet,
-                    'ap_active': updated_status.ap_mode_active,
+                    'ap_active': ap_active,
                     'ssid': updated_status.ssid
                 }
                 
@@ -109,7 +105,7 @@ class WiFiMonitorDaemon:
                     else:
                         logger.debug("Ethernet not connected")
                     
-                    if updated_status.ap_mode_active:
+                    if ap_active:
                         logger.info(f"AP mode ACTIVE - SSID: {ap_ssid} (IP: 192.168.4.1)")
                     else:
                         logger.debug("AP mode inactive")
@@ -123,16 +119,16 @@ class WiFiMonitorDaemon:
                     # Log periodic status (less verbose)
                     if updated_status.connected:
                         logger.debug(f"Status check: WiFi={updated_status.ssid} ({updated_status.signal}%), "
-                                   f"Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}")
+                                   f"Ethernet={updated_ethernet}, AP={ap_active}")
                     else:
-                        logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}")
+                        logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={ap_active}")
                 
                 # Escalating recovery: if nmcli reports connected but actual internet
                 # is unreachable for several consecutive checks, restart NetworkManager.
                 # This is done HERE (not inside check_and_manage_ap_mode) to keep the
                 # AP-enable trigger clean and avoid false-positive AP enables from
                 # transient packet loss on otherwise working WiFi.
-                if updated_status.connected and not updated_status.ap_mode_active:
+                if updated_status.connected and not ap_active:
                     if not self.wifi_manager.check_internet_connectivity():
                         self._consecutive_internet_failures += 1
                         logger.warning(

src/__init__.py

@@ -4,5 +4,5 @@ LEDMatrix Display System
 Core source package for the LED Matrix Display project.
 """
 
-__version__ = "1.0.0"
+__version__ = "3.1.0"
 

src/adaptive_images.py

@@ -0,0 +1,174 @@
+"""
+Adaptive image fitting for plugins — the image counterpart to
+src/adaptive_layout.py's text fitting.
+
+Promotes the proven in-field image patterns into one shared helper so
+plugins stop hand-copying resize/cache code:
+
+- "crop transparent padding, then fill the row height" (football/hockey
+  logo pattern) -> ``crop_to_ink=True, mode="fill_height"``
+- "crop-to-fill with a top anchor for faces" (masters-tournament headshot
+  pattern) -> ``mode="cover", anchor="top"``
+- "letterbox to fit, centered on a background" (static-image pattern)
+  -> ``mode="contain"``
+- NEAREST for pixel art/flags vs LANCZOS for photos (masters flag pattern)
+  -> ``resample=RESAMPLE_NEAREST``
+
+Unlike PIL's ``thumbnail()`` (downscale-only — the reason plugin imagery
+stays tiny on big panels), ``fit_image`` upscales by default so content
+genuinely adapts to larger displays; pass ``upscale=False`` for the old
+behavior.
+
+Use via ``LayoutContext.fit_image(...)`` (cached per panel size) or
+``BasePlugin.draw_image(...)``; the module-level functions are the
+uncached primitives.
+"""
+
+from dataclasses import dataclass
+from typing import Any, Optional, Tuple
+
+from PIL import Image
+
+# The one Pillow >= 9.1 compat shim (replaces the per-plugin copies).
+try:
+    RESAMPLE_LANCZOS = Image.Resampling.LANCZOS
+    RESAMPLE_NEAREST = Image.Resampling.NEAREST
+except AttributeError:  # Pillow < 9.1
+    RESAMPLE_LANCZOS = Image.LANCZOS
+    RESAMPLE_NEAREST = Image.NEAREST
+
+FIT_MODES = ("contain", "cover", "fill_height", "stretch")
+
+
+@dataclass(frozen=True)
+class ImageFitResult:
+    """A processed RGBA copy of a source image, sized for a target box."""
+    image: Image.Image
+    width: int
+    height: int
+    scale: float          # scale applied vs the (possibly ink-cropped) source
+    mode: str
+    source_size: Tuple[int, int]
+
+    @property
+    def is_empty(self) -> bool:
+        return self.width <= 0 or self.height <= 0
+
+
+_EMPTY_IMAGE = Image.new("RGBA", (1, 1), (0, 0, 0, 0))
+
+
+def _empty_result(mode: str, source_size: Tuple[int, int]) -> ImageFitResult:
+    return ImageFitResult(_EMPTY_IMAGE, 0, 0, 0.0, mode, source_size)
+
+
+def _box_dims(box: Any) -> Tuple[int, int]:
+    """Accept a Region (duck-typed .w/.h) or a (w, h) tuple."""
+    if hasattr(box, "w") and hasattr(box, "h"):
+        return (int(box.w), int(box.h))
+    w, h = box
+    return (int(w), int(h))
+
+
+def fit_image(img: Image.Image, box: Any, *, mode: str = "contain",
+              crop_to_ink: bool = False, anchor: str = "center",
+              resample: Any = None, upscale: bool = True) -> ImageFitResult:
+    """Fit an image into a box, preserving crispness policy per content type.
+
+    Args:
+        img: Source PIL image (any mode; output is always RGBA).
+        box: Region or (w, h) target box.
+        mode: "contain" (letterbox), "cover" (crop-to-fill),
+              "fill_height" (height == box height, contain-capped by width),
+              "stretch" (exact resize).
+        crop_to_ink: Trim fully-transparent padding (getbbox) before fitting —
+              logos shipped with generous padding otherwise render small.
+        anchor: For "cover" crops: "center" or "top" (keeps faces/tops).
+        resample: PIL resampling filter; defaults to RESAMPLE_LANCZOS.
+              Use RESAMPLE_NEAREST for pixel art, flags, and sprite icons.
+        upscale: Allow scaling above source size (default True — the adaptive
+              point). False mimics the legacy thumbnail() behavior.
+    """
+    if mode not in FIT_MODES:
+        raise ValueError(f"Unknown fit mode '{mode}' (expected one of {FIT_MODES})")
+    box_w, box_h = _box_dims(box)
+    if box_w <= 0 or box_h <= 0 or img.width <= 0 or img.height <= 0:
+        return _empty_result(mode, img.size)
+
+    resample = RESAMPLE_LANCZOS if resample is None else resample
+
+    work = img if img.mode == "RGBA" else img.convert("RGBA")
+    if crop_to_ink:
+        bbox = work.getbbox()
+        if bbox is None:  # fully transparent
+            return _empty_result(mode, img.size)
+        work = work.crop(bbox)
+
+    src_w, src_h = work.size
+
+    if mode == "stretch":
+        out = work.resize((box_w, box_h), resample)
+        return ImageFitResult(out, box_w, box_h, box_w / src_w, mode, (src_w, src_h))
+
+    if mode == "cover":
+        scale = max(box_w / src_w, box_h / src_h)
+        if not upscale:
+            scale = min(scale, 1.0)
+        scaled_w = max(1, round(src_w * scale))
+        scaled_h = max(1, round(src_h * scale))
+        out = work.resize((scaled_w, scaled_h), resample)
+        # Crop the overhang down to the box (only when the scaled image is
+        # larger; with upscale=False it may be smaller and is left as-is).
+        crop_w, crop_h = min(box_w, scaled_w), min(box_h, scaled_h)
+        left = (scaled_w - crop_w) // 2
+        top = 0 if anchor == "top" else (scaled_h - crop_h) // 2
+        out = out.crop((left, top, left + crop_w, top + crop_h))
+        return ImageFitResult(out, out.width, out.height, scale, mode, (src_w, src_h))
+
+    # contain / fill_height share the "preserve aspect, no crop" path
+    if mode == "fill_height":
+        scale = box_h / src_h
+        # contain-cap: never exceed the box width (football's logo_slot rule)
+        scale = min(scale, box_w / src_w)
+    else:  # contain
+        scale = min(box_w / src_w, box_h / src_h)
+    if not upscale:
+        scale = min(scale, 1.0)
+    out_w = max(1, round(src_w * scale))
+    out_h = max(1, round(src_h * scale))
+    if (out_w, out_h) == (src_w, src_h):
+        # No resize needed — but `work` may still BE the caller's original
+        # image (RGBA source, no ink crop). The result must always be an
+        # independent copy: LayoutContext caches ImageFitResults, and an
+        # aliased image would let later mutations of the source corrupt
+        # cached fits (or vice versa).
+        out = work.copy() if work is img else work
+    else:
+        out = work.resize((out_w, out_h), resample)
+    return ImageFitResult(out, out_w, out_h, scale, mode, (src_w, src_h))
+
+
+def draw_fitted_image(display_manager: Any, ifit: ImageFitResult, box: Any, *,
+                      align: str = "center", valign: str = "center",
+                      offset: Tuple[int, int] = (0, 0)) -> Optional[Tuple[int, int]]:
+    """Paste a fitted image aligned within a Region onto the display canvas.
+
+    Pastes with the image's own alpha mask. Returns the (x, y) actually used
+    so callers can position adjacent decorations, or None when nothing was
+    drawn (empty fit / no canvas).
+    """
+    if ifit is None or ifit.is_empty:
+        return None
+    image = getattr(display_manager, "image", None)
+    if image is None:
+        return None
+    if hasattr(box, "align_xy"):
+        x, y = box.align_xy(ifit.width, ifit.height, align, valign)
+    else:
+        box_w, box_h = _box_dims(box)
+        x = (box_w - ifit.width) // 2
+        y = (box_h - ifit.height) // 2
+    x += int(offset[0])
+    y += int(offset[1])
+    image.paste(ifit.image, (x, y), ifit.image)
+    return (x, y)

src/adaptive_layout.py

@@ -0,0 +1,746 @@
+"""
+Adaptive layout and font scaling helpers for plugins.
+
+Generalizes the three size-adaptation patterns proven in the plugin
+ecosystem into small composable core helpers, so plugins render legibly on
+any panel size (64x32, 128x32, 96x48, 128x64, 256x64, ...) without
+hand-tuned per-display layouts:
+
+- Region: integer rect algebra (bands, columns, weighted splits, centering).
+  Regions partition space, so text bands can't overlap by construction —
+  replacing the magic ``y = 1`` / ``y = height - 7`` offsets tuned for 128x32.
+- Font ladders: ordered (family, size) steps known to render crisply.
+  Pixel fonts (BDF, PressStart2P) only look right at native/integer sizes,
+  so fonts are never scaled continuously — fitting walks a ladder from the
+  largest rung down until the measured text fits the target box. This is
+  baseball-scoreboard's fallback-ladder pattern promoted to core.
+- LayoutContext: per-(width, height) facts — breakpoint tiers
+  (masters-tournament's pattern), a geometry scale factor vs. a declared
+  design size (f1-scoreboard's pattern), and cached fit-text queries.
+
+Everything is opt-in: plugins get a context via ``self.layout`` on
+BasePlugin (or construct one directly) and existing plugins are unaffected.
+
+Fonts are resolved through FontManager's catalog (family names are
+lowercased file stems from assets/fonts, e.g. "9x15", "tom-thumb", plus
+aliases like "press_start"). FitResult.font is a plain PIL font or
+freetype.Face, so it drops straight into DisplayManager.draw_text().
+"""
+
+import logging
+from collections import OrderedDict
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
+
+import freetype
+
+logger = logging.getLogger(__name__)
+
+# Height-based breakpoint tiers, smallest to largest. A 32px-tall panel is
+# the ecosystem baseline ("sm"); 96x48 lands in "md"; 128x64 in "lg".
+_HEIGHT_TIERS: Tuple[Tuple[str, int], ...] = (
+    ("xs", 16), ("sm", 32), ("md", 48), ("lg", 64), ("xl", 10 ** 9),
+)
+TIER_ORDER: Tuple[str, ...] = tuple(name for name, _ in _HEIGHT_TIERS)
+
+_WIDTH_TIERS: Tuple[Tuple[str, int], ...] = (
+    ("narrow", 64), ("normal", 128), ("wide", 256), ("ultrawide", 10 ** 9),
+)
+WIDTH_TIER_ORDER: Tuple[str, ...] = tuple(name for name, _ in _WIDTH_TIERS)
+
+# The panel size most existing plugins were authored against.
+DEFAULT_DESIGN_SIZE: Tuple[int, int] = (128, 32)
+
+
+@dataclass(frozen=True)
+class Region:
+    """An integer rectangle. Carving methods return sub-Regions clamped to
+    non-negative dimensions, so degenerate panels never produce negative
+    boxes — a band request larger than the region simply consumes it all."""
+
+    x: int
+    y: int
+    w: int
+    h: int
+
+    def __post_init__(self):
+        object.__setattr__(self, "w", max(0, int(self.w)))
+        object.__setattr__(self, "h", max(0, int(self.h)))
+        object.__setattr__(self, "x", int(self.x))
+        object.__setattr__(self, "y", int(self.y))
+
+    @property
+    def right(self) -> int:
+        return self.x + self.w
+
+    @property
+    def bottom(self) -> int:
+        return self.y + self.h
+
+    @property
+    def center(self) -> Tuple[int, int]:
+        return (self.x + self.w // 2, self.y + self.h // 2)
+
+    # ---- carving -----------------------------------------------------
+
+    def inset(self, dx: int, dy: Optional[int] = None) -> "Region":
+        """Shrink by dx horizontally and dy (default dx) vertically, each side."""
+        if dy is None:
+            dy = dx
+        return Region(self.x + dx, self.y + dy, self.w - 2 * dx, self.h - 2 * dy)
+
+    def offset(self, dx: int, dy: int) -> "Region":
+        """Translate without resizing — the hook for user x/y-offset
+        customization: compute regions first, then apply the user's
+        configured offsets as a final translation."""
+        return Region(self.x + dx, self.y + dy, self.w, self.h)
+
+    def top_band(self, h: int) -> "Region":
+        return Region(self.x, self.y, self.w, min(h, self.h))
+
+    def bottom_band(self, h: int) -> "Region":
+        h = min(h, self.h)
+        return Region(self.x, self.bottom - h, self.w, h)
+
+    def middle(self, top_h: int = 0, bottom_h: int = 0) -> "Region":
+        """What remains between a top band and a bottom band."""
+        return Region(self.x, self.y + top_h, self.w, self.h - top_h - bottom_h)
+
+    def left_col(self, w: int) -> "Region":
+        return Region(self.x, self.y, min(w, self.w), self.h)
+
+    def right_col(self, w: int) -> "Region":
+        w = min(w, self.w)
+        return Region(self.right - w, self.y, w, self.h)
+
+    def split_h(self, *weights: float, gap: int = 0) -> List["Region"]:
+        """Side-by-side columns sized by weight; gaps between them."""
+        sizes = _weighted_sizes(self.w, weights, gap)
+        cols, cursor = [], self.x
+        for size in sizes:
+            cols.append(Region(cursor, self.y, size, self.h))
+            cursor += size + gap
+        return cols
+
+    def split_v(self, *weights: float, gap: int = 0) -> List["Region"]:
+        """Stacked rows sized by weight; gaps between them."""
+        sizes = _weighted_sizes(self.h, weights, gap)
+        rows, cursor = [], self.y
+        for size in sizes:
+            rows.append(Region(self.x, cursor, self.w, size))
+            cursor += size + gap
+        return rows
+
+    # ---- placement ---------------------------------------------------
+
+    def align_xy(self, w: int, h: int, align: str = "center",
+                 valign: str = "center") -> Tuple[int, int]:
+        """Top-left position for a w x h box aligned within this region.
+        align: left|center|right; valign: top|center|bottom."""
+        if align == "left":
+            x = self.x
+        elif align == "right":
+            x = self.right - w
+        else:
+            x = self.x + (self.w - w) // 2
+        if valign == "top":
+            y = self.y
+        elif valign == "bottom":
+            y = self.bottom - h
+        else:
+            y = self.y + (self.h - h) // 2
+        return (x, y)
+
+    def center_xy(self, w: int, h: int) -> Tuple[int, int]:
+        return self.align_xy(w, h)
+
+    def contains(self, w: int, h: int) -> bool:
+        return w <= self.w and h <= self.h
+
+
+def _weighted_sizes(total: int, weights: Sequence[float], gap: int) -> List[int]:
+    """Integer sizes proportional to weights, remainder spread left-to-right."""
+    if not weights:
+        return []
+    usable = max(0, total - gap * (len(weights) - 1))
+    weight_sum = sum(weights) or 1
+    sizes = [int(usable * w / weight_sum) for w in weights]
+    remainder = usable - sum(sizes)
+    for i in range(remainder):
+        sizes[i % len(sizes)] += 1
+    return sizes
+
+
+# ---------------------------------------------------------------------------
+# Font ladders
+# ---------------------------------------------------------------------------
+
+@dataclass(frozen=True)
+class FontStep:
+    """One rung: a FontManager catalog family at a size it renders crisply."""
+    family: str
+    size_px: int
+
+
+FontLadder = Tuple[FontStep, ...]
+
+# X11 BDF bitmap fonts at their native pixel sizes, largest to smallest —
+# baseball-scoreboard's fallback ladder extended upward. Same-height rungs
+# are ordered widest first so width-constrained text steps to a narrower
+# face before dropping a size.
+LADDER_GRID: FontLadder = (
+    FontStep("10x20", 20),
+    FontStep("9x18", 18),
+    FontStep("9x15", 15),
+    FontStep("8x13", 13),
+    FontStep("7x13", 13),
+    FontStep("6x13", 13),
+    FontStep("6x12", 12),
+    FontStep("6x10", 10),
+    FontStep("6x9", 9),
+    FontStep("5x8", 8),
+    FontStep("5x7", 7),
+    FontStep("4x6", 6),
+    FontStep("tom-thumb", 6),
+)
+
+# PressStart2P at integer multiples of its 8px pixel grid only — fractional
+# sizes blur a pixel font. For headline text (clocks, scores).
+LADDER_ARCADE: FontLadder = (
+    FontStep("press_start", 32),
+    FontStep("press_start", 24),
+    FontStep("press_start", 16),
+    FontStep("press_start", 8),
+)
+
+LADDER_DEFAULT: FontLadder = LADDER_GRID
+
+ELLIPSIS = "…"
+
+
+@dataclass(frozen=True)
+class FitResult:
+    """A fitted font plus the ink metrics of the (possibly ellipsized) text.
+
+    ``y_offset`` is the gap between the y passed to draw_text() and where
+    ink actually starts; subtract it from the desired ink-top position when
+    drawing (draw_fitted_text does this for you).
+    """
+    font: Any
+    family: str
+    size_px: int
+    text: str
+    width: int
+    height: int
+    baseline: int
+    y_offset: int
+    fits: bool
+    line_height: int = 0
+
+
+def measure_ink(text: str, font: Any) -> Tuple[int, int, int, int]:
+    """Measure the ink box of text: (width, height, baseline, y_offset).
+
+    y_offset is the distance from the y coordinate DisplayManager.draw_text()
+    is given to the top of the actual ink — PIL draws TTF from the em-box
+    top and _draw_bdf_text derives the baseline from y + ascender, so both
+    leave a font-dependent gap that matters when centering in short bands.
+    """
+    if isinstance(font, freetype.Face):
+        width = 0
+        ascender = font.size.ascender >> 6
+        ink_top, ink_bottom = None, None
+        for char in text:
+            font.load_char(char)
+            width += font.glyph.advance.x >> 6
+            rows = font.glyph.bitmap.rows
+            if rows:
+                top = ascender - font.glyph.bitmap_top
+                ink_top = top if ink_top is None else min(ink_top, top)
+                ink_bottom = top + rows if ink_bottom is None else max(ink_bottom, top + rows)
+        if ink_top is None:
+            ink_top, ink_bottom = 0, 0
+        return (width, ink_bottom - ink_top, ascender, ink_top)
+    bbox = font.getbbox(text)
+    return (bbox[2] - bbox[0], bbox[3] - bbox[1], -bbox[1], bbox[1])
+
+
+def font_line_height(font: Any) -> int:
+    """Recommended line spacing for a font (matches DisplayManager.get_font_height)."""
+    if isinstance(font, freetype.Face):
+        return font.size.height >> 6
+    ascent, descent = font.getmetrics()
+    return ascent + descent
+
+
+def measure_font_crispness(font: Any, sample_text: str = "Ay0",
+                           canvas_size: Tuple[int, int] = (250, 60)) -> float:
+    """Fraction of the rendered sample's ink-bbox pixels that are neither
+    pure black nor pure white — i.e. antialiased.
+
+    BDF (freetype.Face) glyphs are true bitmaps and always render at 0.0.
+    "Pixel-style" TTFs (PressStart2P, and similar fonts bundled for
+    plugins that draw through ImageDraw.text() and so can't take a BDF
+    face) are NOT automatically crisp at arbitrary sizes — PIL antialiases
+    TTF outlines by default, and a pixel-grid font only lands on whole
+    pixels at specific sizes (for PressStart2P: exact multiples of 8).
+    Requesting an unverified size silently produces soft/blurry glyphs on
+    an LED panel, which reads as fuzzy compared to a true BDF rung.
+
+    Use this to vet any custom FontLadder rung that mixes TTF fonts before
+    shipping it — see test_adaptive_layout.py::test_ladder_is_crisp for the
+    pattern. A rung should score 0.0 (or very close, to allow for the odd
+    diagonal stroke) before it belongs in a "crisp" ladder.
+    """
+    if isinstance(font, freetype.Face):
+        return 0.0
+    from PIL import Image, ImageDraw
+    img = Image.new("L", canvas_size, 0)
+    ImageDraw.Draw(img).text((2, 2), sample_text, font=font, fill=255)
+    bbox = img.getbbox()
+    if bbox is None:
+        return 0.0
+    pixels = img.crop(bbox).tobytes()
+    pure = sum(1 for p in pixels if p == 0 or p == 255)
+    return (len(pixels) - pure) / len(pixels)
+
+
+class LayoutContext:
+    """Per-render-size layout facts and fit-text queries for one panel size.
+
+    Construct once per (width, height); BasePlugin.layout does this and
+    rebuilds automatically when the logical display size changes.
+    """
+
+    def __init__(self, width: int, height: int, font_manager: Any,
+                 design_size: Tuple[int, int] = DEFAULT_DESIGN_SIZE):
+        self.width = int(width)
+        self.height = int(height)
+        self.font_manager = font_manager
+        self.design_size = design_size
+        self.bounds = Region(0, 0, self.width, self.height)
+        self.aspect = self.width / max(1, self.height)
+        self.tier = _pick_tier(_HEIGHT_TIERS, self.height)
+        self.width_tier = _pick_tier(_WIDTH_TIERS, self.width)
+        self.is_wide_short = self.aspect >= 2.5 and self.height <= 32
+        design_w, design_h = design_size
+        # Geometry scale only (gaps, icon/logo sizes) — never applied to
+        # fonts, which step between crisp ladder rungs instead.
+        self.scale = min(self.width / max(1, design_w),
+                         self.height / max(1, design_h))
+        # LRU-bounded: entries are small, but keys embed the fitted TEXT —
+        # a plugin fitting changing text (a live game clock, a ticker) on a
+        # 24/7 service would otherwise grow this without bound.
+        self._fit_cache: "OrderedDict[Any, FitResult]" = OrderedDict()
+        # LRU-bounded (images are big). Entries hold a strong reference to
+        # the source image when keyed by id() so the id can't be recycled
+        # out from under the cache.
+        self._image_cache: "OrderedDict[Any, Tuple[Any, Any]]" = OrderedDict()
+
+    _IMAGE_CACHE_MAX = 64
+    _FIT_CACHE_MAX = 512
+
+    def _fit_cache_get(self, key: Any) -> Optional["FitResult"]:
+        cached = self._fit_cache.get(key)
+        if cached is not None:
+            self._fit_cache.move_to_end(key)
+        return cached
+
+    def _fit_cache_put(self, key: Any, result: "FitResult") -> None:
+        self._fit_cache[key] = result
+        while len(self._fit_cache) > self._FIT_CACHE_MAX:
+            self._fit_cache.popitem(last=False)
+
+    # ---- the three adaptation patterns --------------------------------
+
+    def px(self, base: int, minimum: int = 1, maximum: Optional[int] = None) -> int:
+        """Scale a design-size pixel measurement (f1's pattern): gaps,
+        icon sizes, logo slots. Clamped to [minimum, maximum]."""
+        value = max(minimum, round(base * self.scale))
+        if maximum is not None:
+            value = min(value, maximum)
+        return value
+
+    def by_tier(self, mapping: Dict[str, Any], default: Any = None) -> Any:
+        """Pick the value for the nearest defined tier at-or-below the
+        panel's height tier (masters' pattern). Falls forward to the
+        smallest defined tier above, then to default.
+
+        by_tier({"sm": 10, "lg": 18}) -> 10 on 128x32, 18 on 128x64.
+        Keys may also use width tiers ("narrow", "wide", ...)."""
+        order = TIER_ORDER if any(k in TIER_ORDER for k in mapping) else WIDTH_TIER_ORDER
+        current = self.tier if order is TIER_ORDER else self.width_tier
+        idx = order.index(current)
+        for name in reversed(order[: idx + 1]):
+            if name in mapping:
+                return mapping[name]
+        for name in order[idx + 1:]:
+            if name in mapping:
+                return mapping[name]
+        return default
+
+    def fit_text(self, text: str, box: Union[Region, Tuple[int, int]],
+                 ladder: FontLadder = LADDER_DEFAULT,
+                 ellipsis: bool = True) -> FitResult:
+        """Largest ladder rung whose rendered text fits the box (baseball's
+        pattern). If even the smallest rung is too wide, the text is
+        ellipsized to fit (unless ellipsis=False); fits=False only when no
+        acceptable rendering exists."""
+        box_w, box_h = _box_dims(box)
+        key = ("text", text, box_w, box_h, ladder, ellipsis)
+        cached = self._fit_cache_get(key)
+        if cached is not None:
+            return cached
+        result = self._walk_ladder(text, ladder, box_w, box_h, ellipsis)
+        self._fit_cache_put(key, result)
+        return result
+
+    def fit_text_proportional(self, text: str, box: Union[Region, Tuple[int, int]],
+                              base_size_px: int, ladder: FontLadder = LADDER_DEFAULT,
+                              ellipsis: bool = True,
+                              scale: Optional[float] = None) -> FitResult:
+        """Ladder rung closest to (but not exceeding) ``base_size_px * scale``
+        that still fits the box — proportional sizing instead of ``fit_text``'s
+        "always maximize" behavior.
+
+        Use this when several independently-fitted elements need to stay
+        visually harmonious as the panel grows (e.g. a scoreboard's score,
+        status, and detail text) — ``fit_text`` maximizes each one within
+        its own region, which can make one element balloon out of
+        proportion to its neighbors (a huge score overlapping logos it fit
+        fine at the design size) even though every individual pick is
+        independently "correct". ``base_size_px`` is the size that element
+        renders at on the design size (``design_size``, typically 128x32)
+        — commonly a plugin's existing classic/fixed font size for that
+        element.
+
+        ``scale`` defaults to ``self.scale`` (the same conservative
+        min(width_ratio, height_ratio) factor ``px()`` uses — safe for
+        content whose aspect ratio matters). Pass an explicit axis-specific
+        value when the surrounding composition already scales that way —
+        e.g. a scoreboard whose logos scale with height alone
+        (``logo_slot = min(height, width // 2)``) should size its score
+        text by ``height / design_height`` too, or its text will look
+        under-scaled next to bigger logos on a panel that only grew taller.
+
+        Falls back to the smallest rung when even that exceeds the target
+        (a tiny scale factor), and to fit_text's ordinary smaller-rung
+        fallback when the closest-to-target rung doesn't actually fit the
+        box.
+        """
+        box_w, box_h = _box_dims(box)
+        effective_scale = self.scale if scale is None else scale
+        key = ("text_prop", text, box_w, box_h, ladder, base_size_px, ellipsis, effective_scale)
+        cached = self._fit_cache_get(key)
+        if cached is not None:
+            return cached
+        target = base_size_px * effective_scale
+        eligible = [step for step in ladder if step.size_px <= target]
+        candidates = eligible if eligible else (min(ladder, key=lambda s: s.size_px),)
+        result = self._walk_ladder(text, candidates, box_w, box_h, ellipsis)
+        self._fit_cache_put(key, result)
+        return result
+
+    def _walk_ladder(self, text: str, ladder: Sequence[FontStep],
+                     box_w: int, box_h: int, ellipsis: bool) -> FitResult:
+        """Shared by fit_text/fit_text_proportional: first ladder entry (in
+        the order given) whose rendered text fits, ellipsizing the last one
+        tried if none do."""
+        result = None
+        for step in ladder:
+            font = self.font_manager.get_font(step.family, step.size_px)
+            width, height, baseline, y_offset = measure_ink(text, font)
+            result = FitResult(font, step.family, step.size_px, text,
+                               width, height, baseline, y_offset,
+                               fits=(width <= box_w and height <= box_h),
+                               line_height=font_line_height(font))
+            if result.fits:
+                break
+
+        if result is not None and not result.fits and ellipsis:
+            short = self.ellipsize(text, result.font, box_w)
+            width, height, baseline, y_offset = measure_ink(short, result.font)
+            result = FitResult(result.font, result.family, result.size_px,
+                               short, width, height, baseline, y_offset,
+                               fits=(width <= box_w and height <= box_h),
+                               line_height=result.line_height)
+        return result
+
+    def fit_lines(self, lines: Sequence[str], box: Union[Region, Tuple[int, int]],
+                  ladder: FontLadder = LADDER_DEFAULT,
+                  spacing: int = 1) -> FitResult:
+        """Largest rung where every line fits the box width and the stacked
+        lines (line_height + spacing apart) fit the box height. Measures the
+        actual strings, so a long line pushes the ladder down a rung a short
+        one wouldn't (baseball's multiline pattern). Text is the widest line."""
+        box_w, box_h = _box_dims(box)
+        key = ("lines", tuple(lines), box_w, box_h, ladder, spacing)
+        cached = self._fit_cache_get(key)
+        if cached is not None:
+            return cached
+
+        rows = max(1, len(lines))
+        result = None
+        for step in ladder:
+            font = self.font_manager.get_font(step.family, step.size_px)
+            line_h = font_line_height(font)
+            widest, metrics = "", (0, 0, 0, 0)
+            for line in lines:
+                m = measure_ink(line, font)
+                if m[0] >= metrics[0]:
+                    widest, metrics = line, m
+            total_h = rows * line_h + (rows - 1) * spacing
+            result = FitResult(font, step.family, step.size_px, widest,
+                               metrics[0], metrics[1], metrics[2], metrics[3],
+                               fits=(metrics[0] <= box_w and total_h <= box_h),
+                               line_height=line_h)
+            if result.fits:
+                break
+
+        self._fit_cache_put(key, result)
+        return result
+
+    def font_for_rows(self, rows: int, box_h: int,
+                      ladder: FontLadder = LADDER_GRID) -> FitResult:
+        """Largest rung whose line height lets `rows` rows fit in box_h
+        (baseball's traditional-scoreboard pattern). Measures a digit/cap
+        sample rather than specific strings."""
+        key = ("rows", rows, box_h, ladder)
+        cached = self._fit_cache_get(key)
+        if cached is not None:
+            return cached
+
+        sample = "0Ay"
+        result = None
+        for step in ladder:
+            font = self.font_manager.get_font(step.family, step.size_px)
+            line_h = font_line_height(font)
+            width, height, baseline, y_offset = measure_ink(sample, font)
+            result = FitResult(font, step.family, step.size_px, sample,
+                               width, height, baseline, y_offset,
+                               fits=(max(1, rows) * line_h <= box_h),
+                               line_height=line_h)
+            if result.fits:
+                break
+
+        self._fit_cache_put(key, result)
+        return result
+
+    # ---- images ---------------------------------------------------------
+
+    def fit_image(self, img: Any, box: Union[Region, Tuple[int, int]], *,
+                  mode: str = "contain", crop_to_ink: bool = False,
+                  anchor: str = "center", resample: Any = None,
+                  upscale: bool = True, cache_key: Any = None) -> Any:
+        """Fit an image into a box (see src/adaptive_images.py for modes),
+        cached per (image, box size, options) for this panel size.
+
+        Prefer a stable ``cache_key`` (e.g. "logo:KC") for images that get
+        reloaded — the default id()-based key is safe (the entry pins the
+        source image) but misses across reloads of the same content.
+        """
+        from src.adaptive_images import fit_image as _fit_image
+
+        box_w, box_h = _box_dims(box)
+        resample_name = getattr(resample, "name", repr(resample)) if resample is not None else "default"
+        identity = cache_key if cache_key is not None else ("id", id(img))
+        key = ("image", identity, img.size, box_w, box_h, mode,
+               crop_to_ink, anchor, resample_name, upscale)
+
+        cached = self._image_cache.get(key)
+        if cached is not None:
+            self._image_cache.move_to_end(key)
+            return cached[0]
+
+        result = _fit_image(img, (box_w, box_h), mode=mode,
+                            crop_to_ink=crop_to_ink, anchor=anchor,
+                            resample=resample, upscale=upscale)
+        # Pin the source only for id()-keyed entries (see docstring).
+        self._image_cache[key] = (result, img if cache_key is None else None)
+        while len(self._image_cache) > self._IMAGE_CACHE_MAX:
+            self._image_cache.popitem(last=False)
+        return result
+
+    # ---- text utilities ------------------------------------------------
+
+    def ellipsize(self, text: str, font: Any, max_w: int) -> str:
+        """Trim text to fit max_w, appending an ellipsis. Returns '' when
+        not even the ellipsis fits."""
+        if measure_ink(text, font)[0] <= max_w:
+            return text
+        for end in range(len(text) - 1, 0, -1):
+            candidate = text[:end].rstrip() + ELLIPSIS
+            if measure_ink(candidate, font)[0] <= max_w:
+                return candidate
+        return ELLIPSIS if measure_ink(ELLIPSIS, font)[0] <= max_w else ""
+
+    def measure(self, text: str, font: Any) -> Tuple[int, int, int]:
+        """Ink (width, height, baseline) of text — see measure_ink."""
+        width, height, baseline, _ = measure_ink(text, font)
+        return (width, height, baseline)
+
+    def clear_cache(self) -> None:
+        """Drop cached fit results (call after fonts are reloaded)."""
+        self._fit_cache.clear()
+        self._image_cache.clear()
+
+
+def _pick_tier(tiers: Tuple[Tuple[str, int], ...], value: int) -> str:
+    for name, limit in tiers:
+        if value <= limit:
+            return name
+    return tiers[-1][0]
+
+
+def _box_dims(box: Union[Region, Tuple[int, int]]) -> Tuple[int, int]:
+    if isinstance(box, Region):
+        return (box.w, box.h)
+    w, h = box
+    return (int(w), int(h))
+
+
+def draw_fitted_text(display_manager: Any, fit: FitResult,
+                     box: Union[Region, Tuple[int, int]],
+                     color: Tuple[int, int, int] = (255, 255, 255),
+                     align: str = "center", valign: str = "center") -> None:
+    """Draw a FitResult's text aligned within a Region via
+    DisplayManager.draw_text(), compensating for the font's ink offset so
+    the ink (not the em box) is what gets aligned."""
+    region = box if isinstance(box, Region) else Region(0, 0, box[0], box[1])
+    x, y = region.align_xy(fit.width, fit.height, align, valign)
+    display_manager.draw_text(fit.text, x=x, y=y - fit.y_offset,
+                              color=color, font=fit.font)
+
+
+# ---------------------------------------------------------------------------
+# Composite layouts — the region arrangements repeated across plugins,
+# expressed as Region math so migrated plugins stop hand-copying coordinate
+# formulas. Deliberately tiny: these return Regions, they don't draw.
+# ---------------------------------------------------------------------------
+
+@dataclass(frozen=True)
+class ScoreboardRegions:
+    """The two-logos-plus-center-score card shared by the sports plugins."""
+    bounds: Region
+    logo_slot: int        # width of each logo slot: min(H, W // 2), center-reserved
+    away_slot: Region     # left logo slot
+    home_slot: Region     # right logo slot
+    center_col: Region    # column between the slots (>= min_center_fraction of width)
+    status_band: Region   # top band (replaces the magic y = 1)
+    score_area: Region    # center_col's true width, between the bands (replaces y = H//2 - 3)
+    detail_band: Region   # bottom band (replaces the magic y = H - 7)
+    bottom_left: Region   # bottom corner: away records / timeouts
+    bottom_right: Region  # bottom corner: home records / timeouts
+
+
+def scoreboard_regions(bounds: Region, *, ctx: Optional["LayoutContext"] = None,
+                       status_h: Optional[int] = None,
+                       detail_h: Optional[int] = None,
+                       min_center_fraction: float = 0.15,
+                       min_center_design_px: int = 40,
+                       score_bleed_fraction: float = 0.5) -> ScoreboardRegions:
+    """Carve a game-card Region into the standard scoreboard arrangement.
+
+    Encodes the invariant duplicated across the sports plugins:
+    ``logo_slot = min(height, width // 2)`` (capped at half the card so the
+    home slot never collapses), away logo centered in the left slot, home in
+    the right.
+
+    That formula alone has a blind spot: at exactly 2:1 aspect ratio
+    (width == 2 * height — a very common shape, e.g. two, four, or more
+    square modules stacked into a taller panel) ``width // 2`` and
+    ``height`` are equal, so the two logo slots claim the *entire* width
+    and leave zero pixels for a center column, no matter how large the
+    panel gets. It isn't a "small panel" problem: 96x48, 128x64, and
+    256x128 (all exactly 2:1) hit it identically, while wide panels like
+    the 128x32 design baseline or a 192x48/256x32 panel never do, because
+    height is already the tighter constraint there.
+
+    Two knobs fix it, both defaulted to values verified against the full
+    harness size spread (see test_adaptive_layout.py::TestScoreboardRegions):
+
+    - ``min_center_fraction`` / ``min_center_design_px`` reserve at least
+      ``max(width * min_center_fraction, min_center_design_px * ctx.scale)``
+      for the center column, capping ``logo_slot`` further when needed. The
+      design-px term (scaled by the context's geometry factor, so it grows
+      on bigger panels like everything else in ``px()``) matters most on
+      small panels where a flat fraction alone reserves too little absolute
+      space for even a short score string. On wide panels the height
+      constraint already leaves more room than either reserves, so both are
+      a no-op there — 128x32/192x48-style layouts are unaffected.
+    - ``score_bleed_fraction`` extends the score's own *fit box* (not the
+      logo slots themselves) an extra ``logo_slot * score_bleed_fraction``
+      into each side — controlled, intentional overlap with the logo art,
+      the same way real broadcast scoreboards let a big score number's
+      edges cross into the team marks flanking it. Without this, on a
+      square-ish panel the center reserve alone can be too narrow for even
+      a modest score to render without truncating (`"17-21"` -> `"17-2…"`),
+      which is worse than a little overlap.
+
+    status_band and detail_band span the FULL card width and overlay the
+    logo slots — matching the classic layouts, where short outlined status/
+    date text is drawn over the logos without issue; only score_area (the
+    one element whose size actively grows with the panel) uses the
+    narrower, bleed-adjusted box. Band heights default to the classic
+    128x32 values, scaled by the context's geometry factor when one is
+    provided. Works on a full panel or on a scroll-mode card Region.
+    """
+    if status_h is None:
+        status_h = ctx.px(9, minimum=7) if ctx else 9
+    if detail_h is None:
+        detail_h = ctx.px(8, minimum=7) if ctx else 8
+
+    logo_slot = min(bounds.h, bounds.w // 2)
+    design_reserve = int(min_center_design_px * (ctx.scale if ctx else 1.0))
+    min_center_w = max(1, int(bounds.w * min_center_fraction), design_reserve)
+    max_logo_slot_by_center = max(1, (bounds.w - min_center_w) // 2)
+    logo_slot = min(logo_slot, max_logo_slot_by_center)
+    away_slot = bounds.left_col(logo_slot)
+    home_slot = bounds.right_col(logo_slot)
+    center_col = Region(bounds.x + logo_slot, bounds.y,
+                        bounds.w - 2 * logo_slot, bounds.h)
+    status_band = bounds.top_band(status_h)
+    detail_band = bounds.bottom_band(detail_h)
+    middle = bounds.middle(status_band.h, detail_band.h)
+    # score_area is the true center gap's width plus a controlled bleed
+    # into each logo slot (see score_bleed_fraction above) -- narrower than
+    # the full card width status/detail get, since it's the one element
+    # whose size actively grows with the panel and needs its *fit box* to
+    # reflect real available space, but generous enough that a short score
+    # string never has to truncate on a square-ish panel.
+    bleed = int(logo_slot * score_bleed_fraction)
+    score_area = Region(center_col.x - bleed, middle.y,
+                        center_col.w + 2 * bleed, middle.h)
+    bottom = bounds.bottom_band(detail_h)
+    return ScoreboardRegions(
+        bounds=bounds, logo_slot=logo_slot,
+        away_slot=away_slot, home_slot=home_slot, center_col=center_col,
+        status_band=status_band, score_area=score_area, detail_band=detail_band,
+        bottom_left=bottom.left_col(logo_slot),
+        bottom_right=bottom.right_col(logo_slot),
+    )
+
+
+@dataclass(frozen=True)
+class MediaRow:
+    """Art/icon on the left, text column on the right (music's idiom)."""
+    art: Region
+    body: Region
+
+
+def media_row(bounds: Region, *, ctx: Optional["LayoutContext"] = None,
+              square: bool = True, gap: Optional[int] = None) -> MediaRow:
+    """Split a Region into an art slot and a body column.
+
+    With ``square=True`` the art slot is bounds.h wide (album-art style);
+    otherwise it takes the left half. The gap defaults to 2px scaled by the
+    context's geometry factor.
+    """
+    if gap is None:
+        gap = ctx.px(2, minimum=1) if ctx else 2
+    art_w = bounds.h if square else bounds.w // 2
+    art_w = min(art_w, bounds.w)
+    art = bounds.left_col(art_w)
+    body = Region(bounds.x + art_w + gap, bounds.y,
+                  bounds.w - art_w - gap, bounds.h)
+    return MediaRow(art=art, body=body)

src/cache/disk_cache.py

@@ -10,6 +10,7 @@ import time
 import tempfile
 import logging
 import threading
+import zlib
 from typing import Dict, Any, Optional, Protocol
 from datetime import datetime
 
@@ -53,6 +54,11 @@ class DiskCache:
         self.cache_dir = cache_dir
         self.logger = logger or logging.getLogger(__name__)
         self._lock = threading.Lock()
+        # key -> adler32 of the last payload successfully written to the
+        # primary cache path; lets set() skip rewriting identical data
+        # (per-process only — worst case another process rewrites, never
+        # a missed write). Guarded by _lock.
+        self._write_digests: Dict[str, int] = {}
     
     def get_cache_path(self, key: str) -> Optional[str]:
         """
@@ -155,10 +161,35 @@ class DiskCache:
         cache_path = self.get_cache_path(key)
         if not cache_path:
             return
-        
+
+        # Serialize once, compact (no indent): the payload is reused by every
+        # write path below, and cache files are machine-read only — indenting
+        # them just multiplied the bytes written to the SD card.
+        try:
+            payload = json.dumps(data, cls=DateTimeEncoder)
+        except (TypeError, ValueError) as e:
+            self.logger.warning("Cache data for key '%s' not serializable: %s", key, e)
+            return
+
+        digest = zlib.adler32(payload.encode('utf-8'))
+
         try:
             # Atomic write to avoid partial/corrupt files
             with self._lock:
+                # Skip the disk entirely when this exact payload was already
+                # written for this key (plugins re-save unchanged API data
+                # every update cycle — each write is real SD-card wear).
+                # Refresh the file mtime so records that rely on it for TTL
+                # (no embedded 'timestamp') don't expire early; a metadata
+                # touch is journal-cheap compared to rewriting the data.
+                if self._write_digests.get(key) == digest:
+                    try:
+                        os.utime(cache_path, None)
+                        return
+                    except OSError:
+                        # File vanished or perms changed — fall through and write
+                        self._write_digests.pop(key, None)
+
                 tmp_dir = os.path.dirname(cache_path)
                 # Try to create temp file in cache directory first
                 # If that fails due to permissions, fall back to direct write
@@ -181,13 +212,17 @@ class DiskCache:
                         fd = None
                     
                     if tmp_path and fd is not None:
-                        # Use atomic write with temp file
+                        # Atomic write with temp file. No fsync: os.replace
+                        # already guarantees readers never see a torn file,
+                        # and cache data is re-fetchable — forcing a disk
+                        # flush per write was the single biggest SD-card
+                        # wear source (dozens of fsyncs/min on API-heavy
+                        # installs) for data that can be re-downloaded.
                         try:
                             with os.fdopen(fd, 'w', encoding='utf-8') as tmp_file:
-                                json.dump(data, tmp_file, indent=4, cls=DateTimeEncoder)
-                                tmp_file.flush()
-                                os.fsync(tmp_file.fileno())
+                                tmp_file.write(payload)
                             os.replace(tmp_path, cache_path)
+                            self._write_digests[key] = digest
                             # Set proper permissions: 660 (rw-rw----) for group-readable cache files
                             try:
                                 os.chmod(cache_path, 0o660)  # nosec B103 - intentional; web UI and service share a group
@@ -203,9 +238,8 @@ class DiskCache:
                         # Fallback: direct write (not atomic, but better than failing)
                         try:
                             with open(cache_path, 'w', encoding='utf-8') as cache_file:
-                                json.dump(data, cache_file, indent=4, cls=DateTimeEncoder)
-                                cache_file.flush()
-                                os.fsync(cache_file.fileno())
+                                cache_file.write(payload)
+                            self._write_digests[key] = digest
                             # Set proper permissions: 660 (rw-rw----) for group-readable cache files
                             try:
                                 os.chmod(cache_path, 0o660)  # nosec B103 - intentional; web UI and service share a group
@@ -229,9 +263,12 @@ class DiskCache:
                             pass
                         
                         if os.path.isdir(fallback_dir) and os.access(fallback_dir, os.W_OK):
+                            # NOTE: no digest record here — the fallback file
+                            # is a different path, so future sets must keep
+                            # retrying the primary location.
                             fallback_path = os.path.join(fallback_dir, os.path.basename(cache_path))
                             with open(fallback_path, 'w', encoding='utf-8') as tmp_file:
-                                json.dump(data, tmp_file, indent=4, cls=DateTimeEncoder)
+                                tmp_file.write(payload)
                             # Set proper permissions: 660 (rw-rw----) for group-readable cache files
                             try:
                                 os.chmod(fallback_path, 0o660)  # nosec B103 - intentional; web UI and service share a group
@@ -272,6 +309,7 @@ class DiskCache:
         
         with self._lock:
             if key:
+                self._write_digests.pop(key, None)
                 cache_path = self.get_cache_path(key)
                 if cache_path and os.path.exists(cache_path):
                     try:
@@ -280,6 +318,7 @@ class DiskCache:
                         self.logger.warning("Could not remove cache file %s: %s", cache_path, e)
             else:
                 # Clear all cache files
+                self._write_digests.clear()
                 if os.path.exists(self.cache_dir):
                     for filename in os.listdir(self.cache_dir):
                         if filename.endswith('.json'):

src/common/README.md

@@ -2,6 +2,28 @@
 
 This directory contains reusable utilities and helpers for LEDMatrix plugins and core modules.
 
+## Adaptive Layout & Images (`src/adaptive_layout.py`, `src/adaptive_images.py`)
+
+The recommended way to lay out plugins that render legibly on **any** panel
+size (64x32 through 256x128+) without hand-tuned coordinates. Re-exported
+from `src.common` for convenience; canonical import paths are
+`src.adaptive_layout` / `src.adaptive_images`.
+
+```python
+# Every BasePlugin already has self.layout and the draw helpers:
+regs = scoreboard_regions(self.layout.bounds, ctx=self.layout)
+self.draw_image(away_logo, regs.away_slot, mode="fill_height",
+                crop_to_ink=True, cache_key=f"logo:{abbr}")
+self.draw_fit(score_text, regs.score_area)     # largest crisp font that fits
+self.draw_fit(status, regs.status_band)
+```
+
+Key pieces: `Region` (rect algebra: bands/columns/splits/offset),
+font ladders (`LADDER_GRID`, `LADDER_ARCADE` — discrete crisp sizes, never
+fractional scaling), `LayoutContext` (`fit_text`, `fit_image`, `by_tier`,
+`px`), and composite carvers `scoreboard_regions()` / `media_row()`.
+Full guide: [docs/ADAPTIVE_LAYOUT.md](../../docs/ADAPTIVE_LAYOUT.md).
+
 ## Error Handling (`error_handler.py`)
 
 Common error handling patterns and utilities:

src/common/__init__.py

@@ -26,6 +26,31 @@ from src.common.scroll_helper import ScrollHelper
 from src.common.logo_helper import LogoHelper
 from src.common.text_helper import TextHelper
 
+# Adaptive layout & images (canonical homes: src.adaptive_layout /
+# src.adaptive_images — re-exported here so plugin authors find them in the
+# blessed-helpers package). See docs/ADAPTIVE_LAYOUT.md.
+from src.adaptive_layout import (
+    Region,
+    LayoutContext,
+    FontStep,
+    FontLadder,
+    LADDER_GRID,
+    LADDER_ARCADE,
+    FitResult,
+    draw_fitted_text,
+    ScoreboardRegions,
+    scoreboard_regions,
+    MediaRow,
+    media_row,
+)
+from src.adaptive_images import (
+    ImageFitResult,
+    fit_image,
+    draw_fitted_image,
+    RESAMPLE_LANCZOS,
+    RESAMPLE_NEAREST,
+)
+
 __all__ = [
     'handle_file_operation',
     'handle_json_operation',
@@ -37,4 +62,22 @@ __all__ = [
     'ScrollHelper',
     'LogoHelper',
     'TextHelper',
+    # adaptive layout & images
+    'Region',
+    'LayoutContext',
+    'FontStep',
+    'FontLadder',
+    'LADDER_GRID',
+    'LADDER_ARCADE',
+    'FitResult',
+    'draw_fitted_text',
+    'ScoreboardRegions',
+    'scoreboard_regions',
+    'MediaRow',
+    'media_row',
+    'ImageFitResult',
+    'fit_image',
+    'draw_fitted_image',
+    'RESAMPLE_LANCZOS',
+    'RESAMPLE_NEAREST',
 ]

src/common/logo_helper.py

@@ -72,9 +72,20 @@ class LogoHelper:
             
         Returns:
             PIL Image object or None if loading fails
+
+        Note: for new adaptive-layout code prefer ``BasePlugin.draw_image``
+        / ``LayoutContext.fit_image`` (src/adaptive_images.py) for the
+        fitting step — LogoHelper remains useful for its download and
+        placeholder logic.
         """
-        # Check cache first
-        cache_key = f"{team_abbr}_{logo_path}"
+        # Resolve the effective target size BEFORE the cache lookup so the
+        # key is size-qualified — a panel-size change must not return a
+        # logo resized for the old dimensions.
+        if max_width is None:
+            max_width = int(self.display_width * 1.5)
+        if max_height is None:
+            max_height = int(self.display_height * 1.5)
+        cache_key = f"{team_abbr}_{logo_path}_{max_width}x{max_height}"
         if cache_key in self._logo_cache:
             self.logger.debug(f"Using cached logo for {team_abbr}")
             # Update LRU order (move to end)

src/common/snapshot_policy.py

@@ -0,0 +1,68 @@
+"""Snapshot write policy for the display preview mirror.
+
+The display service mirrors frames to /tmp/led_matrix_preview.png, which
+serves two consumers with different needs:
+
+- The web UI's live preview (SSE reader in web_interface/app.py) wants
+  fresh frames — but only while a browser is actually watching.
+- The health check (web_interface/blueprints/api_v3.py, hardware status)
+  uses the file's AGE as a liveness proxy: age >= 60s reads as degraded.
+
+PNG-encoding every frame at 5 fps forever — identical frames, no viewers —
+was one of the biggest fixed CPU costs on the Pi. This module is the pure
+decision logic (extracted so it's unit-testable off-Pi; display_manager
+imports rgbmatrix unconditionally and can't be):
+
+    WRITE  — encode + atomically replace the snapshot file
+    TOUCH  — os.utime only: keeps the health-check mtime fresh and lets
+             the SSE reader (mtime-gated) resend at a low rate, without
+             paying for a PNG encode of an unchanged frame
+    SKIP   — do nothing
+
+Policy:
+- With a fresh viewer marker: changed frames write at up to 1/VIEWER_INTERVAL.
+- Without viewers: changed frames still write at 1/IDLE_INTERVAL so the
+  preview page shows something recent on open.
+- Unchanged frames are never re-encoded; the mtime is touched every
+  TOUCH_INTERVAL so the health check (60s threshold) never degrades.
+
+If any constant here changes, re-check the health threshold in
+api_v3.py (get_hardware_status) — TOUCH_INTERVAL must stay well under it.
+"""
+
+from enum import Enum
+
+# Snapshot cadence with a browser preview open (seconds).
+VIEWER_INTERVAL = 0.2
+# Snapshot cadence with no viewers — cheap freshness for page-open (seconds).
+IDLE_INTERVAL = 30.0
+# Max age of the last write/touch before bumping mtime for the health
+# check. MUST stay well under api_v3's 60s degraded threshold.
+TOUCH_INTERVAL = 20.0
+# A viewer marker older than this no longer counts as a live viewer.
+VIEWER_MARKER_FRESH_SEC = 5.0
+
+
+class SnapshotAction(Enum):
+    WRITE = "write"
+    TOUCH = "touch"
+    SKIP = "skip"
+
+
+def decide(now: float, last_write_ts: float, last_touch_ts: float,
+           viewer_fresh: bool, frame_changed: bool) -> SnapshotAction:
+    """Decide what to do with the current frame.
+
+    Args:
+        now: current monotonic-ish timestamp (same clock as the ts args)
+        last_write_ts: when a frame was last actually encoded+written
+        last_touch_ts: when the file mtime was last bumped (write or touch)
+        viewer_fresh: a browser preview is currently watching
+        frame_changed: the frame differs from the last WRITTEN frame
+    """
+    interval = VIEWER_INTERVAL if viewer_fresh else IDLE_INTERVAL
+    if frame_changed and (now - last_write_ts) >= interval:
+        return SnapshotAction.WRITE
+    if (now - max(last_write_ts, last_touch_ts)) >= TOUCH_INTERVAL:
+        return SnapshotAction.TOUCH
+    return SnapshotAction.SKIP

src/common/text_helper.py

@@ -11,6 +11,9 @@ from typing import Dict, List, Optional, Tuple, Union
 
 from PIL import Image, ImageDraw, ImageFont
 
+# Shared throwaway draw surface for measuring text without a target canvas.
+_measure_draw = ImageDraw.Draw(Image.new("RGB", (1, 1)))
+
 
 class TextHelper:
     """
@@ -112,10 +115,10 @@ class TextHelper:
             Width in pixels
         """
         try:
-            return draw.textlength(text, font=font)
+            return int(_measure_draw.textlength(text, font=font))
         except AttributeError:
             # Fallback for older PIL versions
-            bbox = draw.textbbox((0, 0), text, font=font)
+            bbox = _measure_draw.textbbox((0, 0), text, font=font)
             return bbox[2] - bbox[0]
     
     def get_text_height(self, text: str, font: ImageFont.ImageFont) -> int:
@@ -129,13 +132,8 @@ class TextHelper:
         Returns:
             Height in pixels
         """
-        try:
-            bbox = draw.textbbox((0, 0), text, font=font)
-            return bbox[3] - bbox[1]
-        except AttributeError:
-            # Fallback for older PIL versions
-            bbox = draw.textbbox((0, 0), text, font=font)
-            return bbox[3] - bbox[1]
+        bbox = _measure_draw.textbbox((0, 0), text, font=font)
+        return bbox[3] - bbox[1]
     
     def get_text_dimensions(self, text: str, font: ImageFont.ImageFont) -> Tuple[int, int]:
         """

src/config_manager.py

@@ -56,6 +56,13 @@ class ConfigManager:
         self.secrets_path: str = secrets_path or "config/config_secrets.json"
         self.template_path: str = "config/config.template.json"
         self.config: Dict[str, Any] = {}
+        # (mtime_ns, size) signature of (config, secrets, template) at the
+        # last successful load. load_config() skips the full re-read (3 file
+        # parses + recursive template migration) when nothing changed —
+        # ~30 web request handlers call it, some 2-3x per request. Cross-
+        # process freshness is preserved: another process's save bumps the
+        # mtime, so the next load here re-reads.
+        self._loaded_sig: Optional[tuple] = None
         self.logger: logging.Logger = get_logger(__name__)
         
         # Initialize atomic config manager
@@ -122,6 +129,14 @@ class ConfigManager:
         # Update in-memory config if save was successful
         if result.status == SaveResultStatus.SUCCESS:
             self.config = new_config_data
+            # In-memory config now matches what was just written; refresh
+            # the load signature so the fast path stays valid. NOTE: the
+            # in-memory copy includes merged secrets; the on-disk file has
+            # them stripped — the fast path returning self.config preserves
+            # exactly the pre-cache behavior (load-after-save also returned
+            # the secret-merged self.config only after re-reading secrets;
+            # here secrets file is unchanged, so contents are equivalent).
+            self._loaded_sig = self._files_signature()
             self.logger.info(f"Configuration successfully saved atomically to {os.path.abspath(self.config_path)}")
         elif result.status == SaveResultStatus.ROLLED_BACK:
             # Reload config from file after rollback
@@ -179,13 +194,36 @@ class ConfigManager:
         atomic_mgr = self._get_atomic_manager()
         return atomic_mgr.validate_config_file(config_path)
 
+    def _files_signature(self) -> tuple:
+        """(mtime_ns, size) of config/secrets/template, None for missing —
+        cheap staleness probe (3 stats) for the load_config fast path."""
+        sig = []
+        for path in (self.config_path, self.secrets_path, self.template_path):
+            try:
+                st = os.stat(path)
+                sig.append((st.st_mtime_ns, st.st_size))
+            except OSError:
+                sig.append(None)
+        return tuple(sig)
+
     def load_config(self) -> Dict[str, Any]:
-        """Load configuration from JSON files."""
+        """Load configuration from JSON files.
+
+        Fast path: when config.json, config_secrets.json and the template
+        are all unchanged since the last successful load (mtime_ns + size),
+        the already-parsed self.config is returned without touching the
+        files — same aliasing semantics as the full path, which also
+        returns self.config.
+        """
         try:
+            current_sig = self._files_signature()
+            if self.config and self._loaded_sig == current_sig:
+                return self.config
+
             # Check if config file exists, if not create from template
             if not os.path.exists(self.config_path):
                 self._create_config_from_template()
-            
+
             # Load main config
             self.logger.info(f"Attempting to load config from: {os.path.abspath(self.config_path)}")
             with open(self.config_path, 'r') as f:
@@ -205,7 +243,10 @@ class ConfigManager:
                     self.logger.warning(f"Secrets file not readable ({self.secrets_path}): {e}. Continuing without secrets.")
                 except (json.JSONDecodeError, OSError) as e:
                     self.logger.warning(f"Error reading secrets file ({self.secrets_path}): {e}. Continuing without secrets.")
-            
+
+            # Signature taken AFTER load + migration (migration may write the
+            # config back), so it reflects exactly what was read/written.
+            self._loaded_sig = self._files_signature()
             return self.config
             
         except FileNotFoundError as e:
@@ -264,7 +305,8 @@ class ConfigManager:
                 json.dump(config_to_write, f, indent=4)
             
             # Update the in-memory config to the new state (which includes secrets for runtime)
-            self.config = new_config_data 
+            self.config = new_config_data
+            self._loaded_sig = self._files_signature()
             self.logger.info(f"Configuration successfully saved to {os.path.abspath(self.config_path)}")
             if secrets_content:
                  self.logger.info("Secret values were preserved in memory and not written to the main config file.")

src/display_controller.py

@@ -23,6 +23,9 @@ Entry point: :func:`main` — instantiates :class:`DisplayController` and calls
 import time
 import os
 import json
+import threading
+import types
+from contextlib import contextmanager
 from pathlib import Path
 from typing import Dict, Any, List, Optional, Callable
 from datetime import datetime
@@ -199,6 +202,10 @@ class DisplayController:
         self.wifi_status_file = WIFI_STATUS_FILE
         self.wifi_status_active = False
         self.wifi_status_expires_at: Optional[float] = None
+        # _check_wifi_status_message throttle state (checked at frame rate,
+        # stat'd at most once per second)
+        self._wifi_status_check_ts = 0.0
+        self._wifi_status_last_result: Optional[Dict[str, Any]] = None
 
         # Plugin display() signature cache — must be initialised before the plugin
         # loading loop below so the .pop() invalidation at load time is always safe.
@@ -502,7 +509,10 @@ class DisplayController:
 
             # Run plugin updates inside the Vegas loop so the inter-iteration
             # gap is <1 ms (nothing left for _tick_plugin_updates() to do).
-            self.vegas_coordinator.set_update_callback(self._tick_plugin_updates)
+            # Use the Vegas-aware variant so plugins that got fresh data are
+            # hot-swapped into the scroll promptly instead of waiting for the
+            # next full cycle.
+            self.vegas_coordinator.set_update_callback(self._tick_plugin_updates_for_vegas)
 
             # Wire multi-display sync into Vegas render pipeline
             follower_pos = self.config.get("sync", {}).get("follower_position", "left")
@@ -621,18 +631,28 @@ class DisplayController:
 
         current_day = current_time.strftime('%A').lower()  # e.g. 'monday'
         current_time_only = current_time.time()
-        
+
         # Check if per-day schedule is configured
         days_config = schedule_config.get('days')
-        
-        # Determine which schedule to use
+
+        # Determine which schedule to use. Respect an explicit 'mode' field
+        # (like the dim schedule does) so a stray/legacy 'days' dict left over
+        # from config migration or a prior per-day setup can't silently
+        # override a user's Global schedule selection.
+        mode = schedule_config.get('mode')
+        mode_normalized = mode.replace('_', '-') if mode else None
+
         use_per_day = False
-        if days_config:
-            # Check if days dict is not empty and contains current day
-            if days_config and current_day in days_config:
+        if mode_normalized == 'global':
+            use_per_day = False
+        elif mode_normalized == 'per-day':
+            use_per_day = bool(days_config and current_day in days_config)
+        elif days_config:
+            # No explicit mode recorded (legacy config) - fall back to
+            # inferring from presence of a 'days' dict for the current day.
+            if current_day in days_config:
                 use_per_day = True
-            elif days_config:
-                # Days dict exists but doesn't have current day - fall back to global
+            else:
                 logger.debug("Per-day schedule exists but %s not configured, using global schedule", current_day)
         
         if use_per_day:
@@ -828,6 +848,42 @@ class DisplayController:
                     if hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
                         self.plugin_manager.health_tracker.record_failure(plugin_id, exc)
 
+    def _tick_plugin_updates_for_vegas(self) -> None:
+        """Run scheduled plugin updates and tell Vegas mode which plugins
+        actually got fresh data, so it can hot-swap them into the scroll
+        without waiting for a full cycle to complete.
+
+        Used as the Vegas coordinator's update callback instead of the plain
+        _tick_plugin_updates() so that a live score change is reflected in
+        the ticker within a few seconds rather than at the next cycle
+        boundary (which, depending on min/max_cycle_duration, can be
+        minutes away). Restores wiring that PR #299 added and PR #330's
+        sync-mode refactor inadvertently dropped: coordinator.mark_plugin_updated()
+        has been unreachable dead code since.
+
+        Delegates the before/after plugin_last_update snapshot to
+        PluginManager.run_scheduled_updates_with_changes() so the snapshot,
+        update pass, and diff are lock-protected against this callback's own
+        background update-tick thread racing the main render loop.
+        """
+        if not self.plugin_manager or not hasattr(self.plugin_manager, "run_scheduled_updates_with_changes"):
+            self._tick_plugin_updates()
+            return
+
+        updated = self.plugin_manager.run_scheduled_updates_with_changes()
+
+        vc = getattr(self, "vegas_coordinator", None)
+        if vc is None:
+            return
+
+        if updated:
+            logger.info("Vegas update tick: %d plugin(s) updated: %s", len(updated), updated)
+            for plugin_id in updated:
+                try:
+                    vc.mark_plugin_updated(plugin_id)
+                except Exception:  # pylint: disable=broad-except
+                    logger.exception("Error marking plugin %s updated for Vegas", plugin_id)
+
     def _tick_plugin_updates(self):
         """Run scheduled plugin updates if the plugin manager supports them."""
         if not self.plugin_manager:
@@ -839,6 +895,30 @@ class DisplayController:
             except Exception:  # pylint: disable=broad-except
                 logger.exception("Error running scheduled plugin updates")
 
+    @contextmanager
+    def _display_lock_or_skip(self, plugin_id):
+        """Try-lock guard keeping a plugin's display() off its in-flight update().
+
+        Yields True when display may run (lock held, released on exit) or
+        when no lock support exists (older plugin manager). Yields False when
+        the plugin's update() is currently executing on the background
+        worker — the caller should treat the frame as displayed (the panel
+        holds the last pushed frame) rather than as a plugin failure, so a
+        mid-update skip never advances the rotation.
+        """
+        pm = self.plugin_manager
+        if not pm or not hasattr(pm, 'get_plugin_lock') or not plugin_id:
+            yield True
+            return
+        lock = pm.get_plugin_lock(plugin_id)
+        if not lock.acquire(blocking=False):
+            yield False
+            return
+        try:
+            yield True
+        finally:
+            lock.release()
+
     _FOLLOWER_SEND_INTERVAL = 1.0 / 90  # raw bytes are cheap; 90fps > follower render rate
 
     def _follower_rebuild_scroll_image(self) -> None:
@@ -1635,7 +1715,7 @@ class DisplayController:
                     self._sleep_with_plugin_updates(60)
                     continue
                 
-                logger.info(f"Display active, processing mode: {self.current_display_mode}")
+                logger.debug("Display active, processing mode: %s", self.current_display_mode)
                 
                 # Plugins update on their own schedules - no forced sync updates needed
                 # Each plugin has its own update_interval and background services
@@ -1803,7 +1883,7 @@ class DisplayController:
                         if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
                             should_skip = self.plugin_manager.health_tracker.should_skip_plugin(plugin_id)
                             if should_skip:
-                                logger.info(f"Skipping plugin {plugin_id} due to circuit breaker (mode: {active_mode})")
+                                logger.info("Skipping plugin %s due to circuit breaker (mode: %s)", plugin_id, active_mode)
                                 display_result = False
                                 # Skip to next mode - let existing logic handle it
                                 manager_to_display = None
@@ -1826,6 +1906,7 @@ class DisplayController:
                     plugin_id = getattr(manager_to_display, 'plugin_id', active_mode)
                     try:
                         logger.debug(f"Calling display() for {active_mode} with force_clear={self.force_change}")
+                        can_display = False
                         if hasattr(manager_to_display, 'display'):
                             # Opt #1: look up (or compute once) whether display() accepts display_mode
                             _cache_key = plugin_id
@@ -1836,14 +1917,64 @@ class DisplayController:
                                 )
                             _accepts_display_mode = self._plugin_accepts_display_mode[_cache_key]
 
-                            # Use PluginExecutor for safe execution with timeout
-                            if self.plugin_manager and hasattr(self.plugin_manager, 'plugin_executor'):
-                                result = self.plugin_manager.plugin_executor.execute_display(
-                                    manager_to_display,
-                                    plugin_id,
-                                    force_clear=self.force_change,
-                                    display_mode=active_mode if _accepts_display_mode else None
-                                )
+                            pm = self.plugin_manager
+                            display_lock = None
+                            can_display = True
+                            if pm and hasattr(pm, 'get_plugin_lock'):
+                                display_lock = pm.get_plugin_lock(plugin_id)
+                                can_display = display_lock.acquire(blocking=False)
+
+                            if not can_display:
+                                # update() in flight on the worker — hold
+                                # the last frame; not a plugin failure
+                                result = True
+                            elif pm and hasattr(pm, 'plugin_executor'):
+                                # PluginExecutor's own thread.join(timeout) can
+                                # return before the real display() call
+                                # finishes (a lingering daemon thread keeps
+                                # running it) -- so the lock is released from
+                                # inside the wrapped call itself, whichever
+                                # thread actually finishes it, rather than
+                                # here when this dispatch merely returns.
+                                release_guard = threading.Lock()
+                                released = {'done': False}
+
+                                def _release_display_lock():
+                                    with release_guard:
+                                        if released['done']:
+                                            return
+                                        released['done'] = True
+                                    if display_lock is not None:
+                                        display_lock.release()
+
+                                if _accepts_display_mode:
+                                    def _display_target(display_mode=None, force_clear=False):
+                                        try:
+                                            return manager_to_display.display(
+                                                display_mode=display_mode, force_clear=force_clear)
+                                        finally:
+                                            _release_display_lock()
+                                else:
+                                    def _display_target(force_clear=False):
+                                        try:
+                                            return manager_to_display.display(force_clear=force_clear)
+                                        finally:
+                                            _release_display_lock()
+
+                                try:
+                                    result = self.plugin_manager.plugin_executor.execute_display(
+                                        types.SimpleNamespace(display=_display_target),
+                                        plugin_id,
+                                        force_clear=self.force_change,
+                                        display_mode=active_mode if _accepts_display_mode else None
+                                    )
+                                except Exception:  # pragma: no cover - defensive;
+                                    # execute_display catches everything
+                                    # internally, but guarantee the lock is
+                                    # never leaked if something unexpected
+                                    # slips through.
+                                    _release_display_lock()
+                                    raise
                                 # execute_display returns bool, convert to expected format
                                 if result:
                                     result = True  # Success
@@ -1851,23 +1982,31 @@ class DisplayController:
                                     result = False  # Failed
                             else:
                                 # Fallback to direct call if executor not available
-                                if _accepts_display_mode:
-                                    result = manager_to_display.display(display_mode=active_mode, force_clear=self.force_change)
-                                else:
-                                    result = manager_to_display.display(force_clear=self.force_change)
+                                try:
+                                    if _accepts_display_mode:
+                                        result = manager_to_display.display(display_mode=active_mode, force_clear=self.force_change)
+                                    else:
+                                        result = manager_to_display.display(force_clear=self.force_change)
+                                finally:
+                                    if display_lock is not None:
+                                        display_lock.release()
                             
                             logger.debug(f"display() returned: {result} (type: {type(result)})")
                             # Check if display() returned a boolean (new behavior)
                             if isinstance(result, bool):
                                 display_result = result
                                 if not display_result:
-                                    logger.info(f"Plugin {plugin_id} display() returned False for mode {active_mode}")
+                                    logger.info("Plugin %s display() returned False for mode %s", plugin_id, active_mode)
                         
-                        # Record success if display completed without exception
-                        if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
-                            self.plugin_manager.health_tracker.record_success(plugin_id)
-                        
-                        self.force_change = False
+                        # Record success only when display() actually ran this
+                        # frame -- a skipped frame (lock busy) held the last
+                        # frame, not a real success, and must not clear
+                        # force_change or the pending mode-switch clear will
+                        # be lost when display() finally does run.
+                        if can_display:
+                            if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
+                                self.plugin_manager.health_tracker.record_success(plugin_id)
+                            self.force_change = False
                     except Exception as exc:  # pylint: disable=broad-except
                         logger.exception("Error displaying %s", self.current_display_mode)
                         # Record failure
@@ -2072,10 +2211,23 @@ class DisplayController:
 
                     # For plugins, call display multiple times to allow game rotation
                     if manager_to_display and hasattr(manager_to_display, 'display'):
-                        # Check if plugin needs high FPS (like stock ticker)
-                        # Always enable high-FPS for static-image plugin (for GIF animation support)
+                        # High-FPS decision, in precedence order:
+                        # 1. A plugin that declares needs_high_fps knows best
+                        #    (e.g. static-image sets it False for still PNGs,
+                        #    True for animated GIFs).
+                        # 2. Back-compat: older static-image versions without
+                        #    the attribute keep the historical forced high-FPS
+                        #    (GIF support).
+                        # 3. Otherwise scrolling plugins get high FPS.
                         plugin_id = getattr(manager_to_display, 'plugin_id', None)
-                        if plugin_id == 'static-image':
+                        declared = getattr(manager_to_display, 'needs_high_fps', None)
+                        if declared is not None:
+                            needs_high_fps = bool(declared)
+                            logger.debug(
+                                "[DisplayController] FPS check for %s (plugin=%s) - "
+                                "plugin declares needs_high_fps=%s",
+                                active_mode, plugin_id, needs_high_fps)
+                        elif plugin_id == 'static-image':
                             needs_high_fps = True
                             logger.debug("FPS check - static-image plugin: forcing high-FPS mode for GIF support")
                         else:
@@ -2139,11 +2291,16 @@ class DisplayController:
 
                             while True:
                                 try:
-                                    # Pass display_mode to maintain sticky manager state
-                                    if _accepts_display_mode:
-                                        result = manager_to_display.display(display_mode=active_mode, force_clear=False)
-                                    else:
-                                        result = manager_to_display.display(force_clear=False)
+                                    with self._display_lock_or_skip(plugin_id) as can_display:
+                                        if can_display:
+                                            # Pass display_mode to maintain sticky manager state
+                                            if _accepts_display_mode:
+                                                result = manager_to_display.display(display_mode=active_mode, force_clear=False)
+                                            else:
+                                                result = manager_to_display.display(force_clear=False)
+                                        else:
+                                            # update() in flight — hold the last frame
+                                            result = True
                                     if isinstance(result, bool) and not result:
                                         logger.debug("Display returned False, breaking early")
                                         break
@@ -2203,11 +2360,16 @@ class DisplayController:
                                     break
 
                                 try:
-                                    # Pass display_mode to maintain sticky manager state
-                                    if _accepts_display_mode:
-                                        result = manager_to_display.display(display_mode=active_mode, force_clear=False)
-                                    else:
-                                        result = manager_to_display.display(force_clear=False)
+                                    with self._display_lock_or_skip(plugin_id) as can_display:
+                                        if can_display:
+                                            # Pass display_mode to maintain sticky manager state
+                                            if _accepts_display_mode:
+                                                result = manager_to_display.display(display_mode=active_mode, force_clear=False)
+                                            else:
+                                                result = manager_to_display.display(force_clear=False)
+                                        else:
+                                            # update() in flight — hold the last frame
+                                            result = True
                                     if isinstance(result, bool) and not result:
                                         # For dynamic duration plugins, don't exit on False - keep looping
                                         # until cycle is complete or max duration is reached
@@ -2354,6 +2516,16 @@ class DisplayController:
             Returns None on any error or if message is expired/invalid.
         """
         try:
+            # Throttle the existence stat to ~1 Hz: this runs on every render
+            # iteration (60+ fps), and the file usually doesn't exist — the
+            # status message's lifetime is measured in seconds anyway.
+            # Both attributes are initialised in __init__.
+            now = time.time()
+            if (now - self._wifi_status_check_ts) < 1.0:
+                return self._wifi_status_last_result
+            self._wifi_status_check_ts = now
+            self._wifi_status_last_result = None
+
             # Check if file exists
             if not self.wifi_status_file or not self.wifi_status_file.exists():
                 return None
@@ -2404,13 +2576,14 @@ class DisplayController:
                     pass
                 return None
             
-            # Message is valid and not expired
-            return {
+            # Message is valid and not expired — cache for the throttle window
+            self._wifi_status_last_result = {
                 'message': message,
                 'timestamp': timestamp,
                 'duration': duration,
                 'expires_at': expires_at
             }
+            return self._wifi_status_last_result
             
         except Exception as e:
             # Catch-all for any unexpected errors - log but don't break the display
@@ -2714,6 +2887,14 @@ class DisplayController:
 
     def cleanup(self):
         """Clean up resources."""
+        # Stop the async update worker first so no in-flight update() call
+        # is still touching display/cache-backed resources while they're
+        # torn down below.
+        if self.plugin_manager and hasattr(self.plugin_manager, 'stop_update_worker'):
+            try:
+                self.plugin_manager.stop_update_worker()
+            except Exception as e:
+                logger.warning("Error stopping plugin update worker: %s", e)
         # Shutdown config service if it exists
         if hasattr(self, 'config_service'):
             try:

src/display_manager.py

@@ -31,13 +31,25 @@ if os.getenv("EMULATOR", "false") == "true":
 else:
     from rgbmatrix import RGBMatrix, RGBMatrixOptions
 from contextlib import contextmanager
+from pathlib import Path
 from PIL import Image, ImageDraw, ImageFont
+import threading
 import time
-from typing import Dict, Any, List, Optional
+from collections import OrderedDict
+from typing import Dict, Any, List, Optional, Tuple
 import logging
 import math
+import zlib
 import freetype
 
+from src.common import snapshot_policy
+from src.common.permission_utils import (
+    ensure_directory_permissions,
+    ensure_file_permissions,
+    get_assets_dir_mode,
+    get_assets_file_mode,
+)
+
 # Get logger without configuring
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)  # Set to INFO level
@@ -180,14 +192,45 @@ class DisplayManager:
         # the logical image is blitted to the matrix unchanged.
         self._double_sided = None  # dict {copies, axis, logical_width, logical_height} or None
         self._physical_image = None  # full-chain buffer reused each frame when tiling
-        # Text-width measurement cache: (text, id(font)) -> pixel_width
+        # Text-width measurement cache: (text, id(font)) -> (width, font_ref)
         # Avoids re-measuring the same string+font on every display() call.
+        # LRU-bounded: keys embed the TEXT, so changing strings (a clock, a
+        # live score) would otherwise grow it forever on a 24/7 service.
+        # Entries hold a strong reference to the font so its id() can't be
+        # recycled by a different font object — an id-keyed cache without
+        # the reference can return the WRONG width after garbage collection.
         # Cleared on _load_fonts() so stale entries don't survive a font reload.
-        self._text_width_cache: Dict[tuple, int] = {}
-        # Snapshot settings for web preview integration (service writes, web reads)
+        self._text_width_cache: "OrderedDict[tuple, Tuple[int, Any]]" = OrderedDict()
+        self._TEXT_WIDTH_CACHE_MAX = 1024
+        # Snapshot mirror for web preview + health check (service writes, web
+        # reads). Cadence/skip decisions live in src/common/snapshot_policy.py:
+        # full rate only while the web SSE broadcaster keeps the viewer marker
+        # fresh; unchanged frames are never re-encoded, only mtime-touched.
         self._snapshot_path = "/tmp/led_matrix_preview.png"  # nosec B108 - fixed path intentional; web UI reads same path
-        self._snapshot_min_interval_sec = 0.2  # max ~5 fps
+        self._viewer_marker_path = "/tmp/led_matrix_preview_viewer"  # nosec B108 - touched by web SSE broadcaster
         self._last_snapshot_ts = 0.0
+        self._last_snapshot_touch_ts = 0.0
+        self._last_snapshot_digest: Optional[int] = None
+        self._snapshot_dir_prepared = False
+        self._viewer_check_ts = 0.0
+        self._viewer_fresh = False
+        self._viewer_was_fresh = False
+        # Snapshot failures are logged as warnings, rate-limited so a
+        # persistent failure (e.g. an unwritable file) can't spam the log —
+        # but is never silent: the snapshot's mtime doubles as the web UI's
+        # hardware-liveness signal, so a quiet failure makes health checks lie.
+        self._snapshot_fail_log_ts = 0.0
+        # Dirty tracking: (image digest, brightness) of the last frame pushed
+        # to the panel; update_display() skips identical pushes. Kill switch:
+        # display.dirty_tracking: false.
+        self._dirty_tracking_enabled = bool(
+            self.config.get('display', {}).get('dirty_tracking', True))
+        self._last_pushed_digest = None
+        # Serializes update_display(): plugins can call it directly from
+        # background threads (see docstring on update_display), not just the
+        # render loop. RLock in case a caller within the critical section
+        # ever re-enters (e.g. via a nested draw callback).
+        self._update_lock = threading.RLock()
         
         # Scrolling state tracking for graceful updates
         self._scrolling_state = {
@@ -418,6 +461,10 @@ class DisplayManager:
         try:
             # RGBMatrix accepts brightness as a property
             self.matrix.brightness = brightness
+            # Brightness applies on the next swap — force a re-push even if
+            # the image itself is unchanged (belt-and-braces: brightness is
+            # also part of the dirty-tracking digest when readable).
+            self._last_pushed_digest = None
             logger.info(f"[BRIGHTNESS] Display brightness set to {brightness}%")
             return True
         except AttributeError as e:
@@ -509,33 +556,70 @@ class DisplayManager:
         return phys
 
     def update_display(self):
-        """Update the display using double buffering with proper sync."""
+        """Update the display using double buffering with proper sync.
+
+        Skips the panel push entirely when the frame is byte-identical to
+        the last pushed one (same image digest AND same brightness) — static
+        content re-rendered every second, and 125 fps loops between actual
+        scroll steps, otherwise re-walk the full framebuffer for nothing.
+        The panel keeps refreshing the current frame from its own thread,
+        so skipping a swap never blanks or freezes the hardware.
+
+        Correctness hinges on invalidation: clear() resets the digest (it
+        writes to the matrix directly), and brightness is PART of the digest
+        so a dim-schedule change is never skipped. Disable via config
+        ``display.dirty_tracking: false`` if a redraw issue is ever suspected.
+
+        Serialized via ``_update_lock``: plugins can call this directly from
+        background threads (e.g. sports base classes push an immediate
+        "live" refresh from inside update()), so without a lock two callers
+        could both pass the digest check before either writes it back,
+        double-pushing a frame, or interleave the offscreen/current canvas
+        swap below. The lock is scoped to this method, so callers never
+        need to know about it.
+        """
         try:
-            if self.matrix is None:
-                # Fallback mode - no actual hardware to update
-                logger.debug("Update display called in fallback mode (no hardware)")
-                # Still write a snapshot so the web UI can preview
+            with self._update_lock:
+                if self.matrix is None:
+                    # Fallback mode - no actual hardware to update
+                    logger.debug("Update display called in fallback mode (no hardware)")
+                    # Still write a snapshot so the web UI can preview
+                    self._write_snapshot_if_due()
+                    return
+
+                if self._capture_mode_active:
+                    return  # Skip hardware write — content is being captured off-screen
+
+                digest = None
+                if self._dirty_tracking_enabled:
+                    try:
+                        brightness = getattr(self.matrix, 'brightness', None)
+                    except AttributeError:
+                        brightness = None
+                    digest = (zlib.adler32(self.image.tobytes()), brightness)
+                    if digest == self._last_pushed_digest:
+                        # Nothing changed since the last push — the panel is
+                        # already showing exactly this frame.
+                        self._write_snapshot_if_due()
+                        return
+
+                # Copy the current image to the offscreen canvas. In double-sided
+                # mode the logical screen is first tiled across the full chain.
+                if self._double_sided is not None:
+                    self.offscreen_canvas.SetImage(self._composite_double_sided())
+                else:
+                    self.offscreen_canvas.SetImage(self.image)
+
+                # Swap buffers immediately
+                self.matrix.SwapOnVSync(self.offscreen_canvas)
+
+                # Swap our canvas references
+                self.offscreen_canvas, self.current_canvas = self.current_canvas, self.offscreen_canvas
+
+                self._last_pushed_digest = digest
+
+                # Write a snapshot for the web preview (throttled)
                 self._write_snapshot_if_due()
-                return
-
-            if self._capture_mode_active:
-                return  # Skip hardware write — content is being captured off-screen
-
-            # Copy the current image to the offscreen canvas. In double-sided
-            # mode the logical screen is first tiled across the full chain.
-            if self._double_sided is not None:
-                self.offscreen_canvas.SetImage(self._composite_double_sided())
-            else:
-                self.offscreen_canvas.SetImage(self.image)
-
-            # Swap buffers immediately
-            self.matrix.SwapOnVSync(self.offscreen_canvas)
-            
-            # Swap our canvas references
-            self.offscreen_canvas, self.current_canvas = self.current_canvas, self.offscreen_canvas
-
-            # Write a snapshot for the web preview (throttled)
-            self._write_snapshot_if_due()
         except Exception as e:
             logger.error(f"Error updating display: {e}")
 
@@ -569,6 +653,9 @@ class DisplayManager:
                 # Clear both canvases and the underlying matrix to ensure no artifacts.
                 # Failures are non-fatal — the image buffer is already black above, so
                 # the next update_display() call will push clean content regardless.
+                # The matrix content no longer matches the last pushed digest,
+                # so dirty tracking must not skip the next push.
+                self._last_pushed_digest = None
                 try:
                     self.offscreen_canvas.Clear()
                 except (RuntimeError, OSError) as e:
@@ -699,12 +786,15 @@ class DisplayManager:
 
         Results are cached by (text, font identity) so plugins that measure
         the same string every frame (e.g. to centre a score) pay only one
-        measurement per unique (text, font) pair.
+        measurement per unique (text, font) pair. The entry keeps the font
+        alive so its id() can't be recycled, and the cache is LRU-bounded so
+        ever-changing text (clocks, tickers) can't grow it without limit.
         """
         cache_key = (text, id(font))
         cached = self._text_width_cache.get(cache_key)
         if cached is not None:
-            return cached
+            self._text_width_cache.move_to_end(cache_key)
+            return cached[0]
 
         try:
             if isinstance(font, freetype.Face):
@@ -719,7 +809,9 @@ class DisplayManager:
             logger.error("Error getting text width: %s", e)
             return 0
 
-        self._text_width_cache[cache_key] = width
+        self._text_width_cache[cache_key] = (width, font)
+        while len(self._text_width_cache) > self._TEXT_WIDTH_CACHE_MAX:
+            self._text_width_cache.popitem(last=False)
         return width
 
     def get_font_height(self, font):
@@ -1128,27 +1220,56 @@ class DisplayManager:
             'deferred_update_ttl': self._scrolling_state['deferred_update_ttl']
         }
 
+    def _viewer_is_fresh(self, now: float) -> bool:
+        """True when a browser preview is watching (marker file touched by
+        the web SSE broadcaster). The marker is stat'd at most once per
+        second — at 125 fps loops a per-call stat would be pure overhead."""
+        if (now - self._viewer_check_ts) >= 1.0:
+            self._viewer_check_ts = now
+            try:
+                marker_age = now - os.stat(self._viewer_marker_path).st_mtime
+                self._viewer_fresh = marker_age < snapshot_policy.VIEWER_MARKER_FRESH_SEC
+            except OSError:
+                self._viewer_fresh = False
+        return self._viewer_fresh
+
     def _write_snapshot_if_due(self) -> None:
-        """Write the current image to a PNG snapshot file at a limited frequency."""
+        """Mirror the current frame to the preview snapshot when the policy
+        says it's worth it — see src/common/snapshot_policy.py. Unchanged
+        frames are never re-encoded; without viewers the cadence drops to
+        the idle keepalive."""
         try:
             now = time.time()
-            if (now - self._last_snapshot_ts) < self._snapshot_min_interval_sec:
+            viewer_fresh = self._viewer_is_fresh(now)
+            if viewer_fresh and not self._viewer_was_fresh:
+                # A preview just opened: let the next changed frame through
+                # immediately instead of waiting out the idle interval.
+                self._last_snapshot_ts = 0.0
+            self._viewer_was_fresh = viewer_fresh
+
+            digest = zlib.adler32(self.image.tobytes())
+            action = snapshot_policy.decide(
+                now, self._last_snapshot_ts, self._last_snapshot_touch_ts,
+                viewer_fresh, digest != self._last_snapshot_digest)
+            if action is snapshot_policy.SnapshotAction.SKIP:
                 return
-            # Ensure directory exists with proper permissions
-            from pathlib import Path
-            from src.common.permission_utils import (
-                ensure_directory_permissions,
-                ensure_file_permissions,
-                get_assets_dir_mode,
-                get_assets_file_mode
-            )
+            if action is snapshot_policy.SnapshotAction.TOUCH:
+                # mtime bump only: keeps the health check (snapshot age)
+                # green without paying for a PNG encode of an unchanged frame
+                os.utime(self._snapshot_path, None)
+                self._last_snapshot_touch_ts = now
+                return
+
+            # WRITE: ensure directory permissions once, not per frame
             snapshot_path_obj = Path(self._snapshot_path)
-            # Only ensure permissions on non-system directories
-            # Never modify /tmp permissions - it has special system permissions (1777)
-            # that must not be changed or it breaks apt and other system tools
-            parent_dir = snapshot_path_obj.parent
-            if parent_dir and str(parent_dir) != '/tmp':  # nosec B108 - guard to skip /tmp for permission ops
-                ensure_directory_permissions(parent_dir, get_assets_dir_mode())
+            if not self._snapshot_dir_prepared:
+                # Never modify /tmp permissions - it has special system
+                # permissions (1777) that must not be changed or it breaks
+                # apt and other system tools
+                parent_dir = snapshot_path_obj.parent
+                if parent_dir and str(parent_dir) != '/tmp':  # nosec B108 - guard to skip /tmp for permission ops
+                    ensure_directory_permissions(parent_dir, get_assets_dir_mode())
+                self._snapshot_dir_prepared = True
             # Write atomically: temp then replace
             tmp_path = f"{self._snapshot_path}.tmp"
             self.image.save(tmp_path, format='PNG')
@@ -1163,6 +1284,18 @@ class DisplayManager:
             except Exception:
                 pass
             self._last_snapshot_ts = now
+            self._last_snapshot_touch_ts = now
+            self._last_snapshot_digest = digest
         except Exception as e:
-            # Snapshot failures should never break display; log at debug to avoid noise
-            logger.debug(f"Snapshot write skipped: {e}")
+            # Snapshot failures must never break display — but they must not
+            # be silent either: the snapshot's mtime is the web UI's display
+            # mirror AND its hardware-liveness proxy, so a quietly failing
+            # write freezes the mirror and makes health checks lie (seen in
+            # the field: a stale root-owned /tmp file froze it for a day).
+            # Warn at most once per 5 minutes to avoid log spam.
+            if (now - self._snapshot_fail_log_ts) > 300:
+                self._snapshot_fail_log_ts = now
+                logger.warning("Snapshot write failing (web preview/health "
+                               "mirror is stale): %s", e)
+            else:
+                logger.debug(f"Snapshot write skipped: {e}")

src/font_manager.py

@@ -35,6 +35,7 @@ import urllib.request
 import zipfile
 import tempfile
 import time
+from collections import OrderedDict
 from pathlib import Path
 from PIL import ImageFont
 from typing import Dict, Tuple, Optional, Union, Any, List
@@ -58,7 +59,13 @@ class FontManager:
         # Font discovery and catalog
         self.font_catalog: Dict[str, str] = {}  # family_name -> file_path
         self.font_cache: Dict[str, Union[ImageFont.FreeTypeFont, freetype.Face]] = {}  # (family, size) -> font
-        self.metrics_cache: Dict[str, Tuple[int, int, int]] = {}  # (text, font_id) -> (width, height, baseline)
+        # (text, id(font)) -> ((width, height, baseline), font_ref).
+        # LRU-bounded — keys embed the measured TEXT, so changing strings
+        # (clocks, live scores) would otherwise grow it forever. Entries
+        # keep the font alive so its id() can't be recycled by a different
+        # font object (which would silently return wrong metrics).
+        self.metrics_cache: "OrderedDict[Any, Tuple[Tuple[int, int, int], Any]]" = OrderedDict()
+        self._METRICS_CACHE_MAX = 1024
 
         # Plugin font management
         self.plugin_fonts: Dict[str, Dict[str, Any]] = {}  # plugin_id -> font_manifest
@@ -103,6 +110,10 @@ class FontManager:
         # Font overrides storage (for manual overrides)
         self.font_overrides_file = "config/font_overrides.json"
         self.font_overrides: Dict[str, Dict[str, Any]] = {}
+
+        # Bumped whenever cached font objects are invalidated, so holders of
+        # derived caches (e.g. adaptive-layout fit results) know to rebuild.
+        self.cache_generation = 0
         
         self._initialize_fonts()
     
@@ -112,6 +123,7 @@ class FontManager:
         self.fonts_config = new_config.get("fonts", {})
         self.font_cache.clear()  # Clear cache to force reload
         self.metrics_cache.clear()  # Clear metrics cache
+        self.cache_generation += 1
         self._initialize_fonts()
         logger.info("FontManager configuration reloaded successfully")
 
@@ -482,6 +494,14 @@ class FontManager:
     def _load_bdf_font(self, font_path: str, size_px: int) -> freetype.Face:
         """Load a BDF font using FreeType."""
         try:
+            native_size = self._read_bdf_native_size(font_path)
+            if native_size is not None and native_size != size_px:
+                # BDF is a fixed-strike bitmap format: FreeType renders the
+                # native size no matter what set_char_size asks for.
+                logger.debug(
+                    "BDF font %s requested at %spx but renders at its native "
+                    "%spx", font_path, size_px, native_size
+                )
             face = freetype.Face(font_path)
             # Set character size (width, height) in 1/64th of points
             face.set_char_size(size_px * 64, size_px * 64, 72, 72)
@@ -490,6 +510,41 @@ class FontManager:
             logger.error(f"Error loading BDF font {font_path}: {e}")
             raise
 
+    def get_native_bdf_size(self, family: str) -> Optional[int]:
+        """The one true pixel size of a BDF family in the catalog, or None
+        for scalable (TTF) families / unknown families."""
+        font_path = self.font_catalog.get(family)
+        if not font_path or not font_path.endswith('.bdf'):
+            return None
+        return self._read_bdf_native_size(font_path)
+
+    @staticmethod
+    def _read_bdf_native_size(bdf_path: str) -> Optional[int]:
+        """Read a BDF file's own header to find its one true pixel size.
+        Prefers the PIXEL_SIZE property, which states the real pixel height
+        directly; falls back to the SIZE line's point-size only if PIXEL_SIZE
+        is absent, since point-size only equals pixel height at exactly
+        100dpi — several bundled fonts (e.g. 6x13.bdf, 5x8.bdf) are defined
+        at 75dpi, where the two values genuinely differ."""
+        size_line_value = None
+        try:
+            with open(bdf_path, "r", encoding="ascii", errors="ignore") as f:
+                for line in f:
+                    if line.startswith("PIXEL_SIZE"):
+                        parts = line.split()
+                        if len(parts) >= 2:
+                            return int(float(parts[1]))
+                    elif line.startswith("SIZE") and size_line_value is None:
+                        # Format: "SIZE <point_size> <xres> <yres>"
+                        parts = line.split()
+                        if len(parts) >= 2:
+                            size_line_value = int(float(parts[1]))
+                    elif line.startswith("STARTCHAR"):
+                        break
+        except (OSError, ValueError):
+            return None
+        return size_line_value
+
     def _get_fallback_font(self) -> ImageFont.ImageFont:
         """Get a fallback font when loading fails."""
         return ImageFont.load_default()
@@ -507,10 +562,14 @@ class FontManager:
         Returns:
             Tuple of (width, height, baseline_offset)
         """
-        cache_key = f"{hash(text)}_{id(font)}"
+        # Key on the text itself (hash(text) could collide) + font identity;
+        # the entry below keeps the font referenced so the id stays valid.
+        cache_key = (text, id(font))
 
-        if cache_key in self.metrics_cache:
-            return self.metrics_cache[cache_key]
+        cached = self.metrics_cache.get(cache_key)
+        if cached is not None:
+            self.metrics_cache.move_to_end(cache_key)
+            return cached[0]
 
         try:
             if isinstance(font, freetype.Face):
@@ -547,7 +606,9 @@ class FontManager:
             baseline = 10
 
         result = (width, height, baseline)
-        self.metrics_cache[cache_key] = result
+        self.metrics_cache[cache_key] = (result, font)
+        while len(self.metrics_cache) > self._METRICS_CACHE_MAX:
+            self.metrics_cache.popitem(last=False)
         return result
 
     def get_font_height(self, font: Union[ImageFont.FreeTypeFont, freetype.Face]) -> int:

src/image_utils.py

@@ -1,3 +1,9 @@
+"""Deprecated: use src/adaptive_images.py (fit_image) instead.
+
+This module predates the adaptive image system and has no known callers.
+It is kept only so any out-of-tree code importing it keeps working.
+"""
+
 import logging
 from PIL import Image
 

src/plugin_system/base_plugin.py

@@ -15,6 +15,19 @@ import logging
 from src.logging_config import get_logger
 
 
+_shared_fallback_font_manager: Optional[Any] = None
+
+
+def _fallback_font_manager() -> Any:
+    """Shared FontManager for environments (unit tests, mocks) where the
+    plugin manager doesn't carry one. Scans assets/fonts like the real one."""
+    global _shared_fallback_font_manager
+    if _shared_fallback_font_manager is None:
+        from src.font_manager import FontManager
+        _shared_fallback_font_manager = FontManager({})
+    return _shared_fallback_font_manager
+
+
 class VegasDisplayMode(Enum):
     """
     Display mode for Vegas scroll integration.
@@ -130,6 +143,133 @@ class BasePlugin(ABC):
         """
         raise NotImplementedError("Plugins must implement display()")
 
+    # -------------------------------------------------------------------------
+    # Adaptive layout support (opt-in)
+    # -------------------------------------------------------------------------
+    @property
+    def layout(self) -> Any:
+        """
+        LayoutContext for the current logical display size.
+
+        Lazily built and rebuilt automatically when the display size changes
+        (e.g. Vegas segment widths, double-sided logical screens). Provides
+        Region carving (self.layout.bounds), breakpoint tiers, a geometry
+        scale factor vs. the manifest's display.design_size, and fit-text
+        queries against font ladders. See src/adaptive_layout.py.
+
+        Example:
+            rows = self.layout.bounds.inset(1).split_v(3, 1, gap=1)
+            self.draw_fit(big_text, rows[0], ladder=LADDER_ARCADE)
+            self.draw_fit(small_text, rows[1])
+        """
+        from src.adaptive_layout import LayoutContext
+
+        width = getattr(self.display_manager, "width", None)
+        height = getattr(self.display_manager, "height", None)
+        if not width or not height:
+            matrix = getattr(self.display_manager, "matrix", None)
+            width = getattr(matrix, "width", 128)
+            height = getattr(matrix, "height", 32)
+
+        font_manager = self._get_font_manager()
+        generation = getattr(font_manager, "cache_generation", 0)
+        cached = getattr(self, "_layout_context", None)
+        if (cached is not None
+                and (cached.width, cached.height) == (width, height)
+                and getattr(self, "_layout_font_generation", None) == generation):
+            return cached
+
+        context = LayoutContext(
+            width, height, font_manager,
+            design_size=self._get_design_size(),
+        )
+        self._layout_context = context
+        self._layout_font_generation = generation
+        return context
+
+    def draw_fit(self, text: str, box: Any,
+                 color: tuple = (255, 255, 255),
+                 ladder: Optional[Any] = None,
+                 align: str = "center", valign: str = "center") -> Any:
+        """
+        Fit text to a Region with the largest crisp font that fits, then draw
+        it aligned within that region via the display manager.
+
+        Args:
+            text: Text to display (ellipsized if even the smallest rung is too wide)
+            box: Region (or (w, h) tuple anchored at 0,0) to fit and align within
+            color: RGB color tuple
+            ladder: FontLadder to walk (default LADDER_GRID; use LADDER_ARCADE
+                    for headline text like clocks and scores)
+            align/valign: alignment of the text ink within the box
+
+        Returns:
+            FitResult (font, family, size_px, text, ink metrics, fits flag)
+        """
+        from src.adaptive_layout import LADDER_DEFAULT, draw_fitted_text
+
+        fit = self.layout.fit_text(text, box, ladder=ladder or LADDER_DEFAULT)
+        draw_fitted_text(self.display_manager, fit, box,
+                         color=color, align=align, valign=valign)
+        return fit
+
+    def draw_image(self, img: Any, box: Any, *,
+                   mode: str = "contain", align: str = "center",
+                   valign: str = "center", crop_to_ink: bool = False,
+                   anchor: str = "center", resample: Optional[Any] = None,
+                   cache_key: Optional[Any] = None,
+                   offset: tuple = (0, 0)) -> Any:
+        """
+        Fit an image into a Region and paste it aligned within that region
+        onto the display canvas — the image counterpart to draw_fit().
+
+        Args:
+            img: Source PIL image (logos, art, icons)
+            box: Region (or (w, h) tuple) to fit and align within
+            mode: "contain" (letterbox), "cover" (crop-to-fill),
+                  "fill_height" (logo-style), "stretch"
+            crop_to_ink: Trim transparent padding before fitting
+            anchor: "center" or "top" for cover crops
+            resample: PIL filter; default LANCZOS. Use RESAMPLE_NEAREST
+                  (from src.adaptive_images) for pixel art/flags
+            cache_key: Stable identity (e.g. "logo:KC") for cross-reload
+                  caching; defaults to the image object's identity
+            offset: Final (dx, dy) translation — the hook for user
+                  x/y-offset customization
+
+        Returns:
+            ImageFitResult (processed image + dimensions + scale)
+        """
+        from src.adaptive_images import draw_fitted_image
+
+        ifit = self.layout.fit_image(img, box, mode=mode,
+                                     crop_to_ink=crop_to_ink, anchor=anchor,
+                                     resample=resample, cache_key=cache_key)
+        draw_fitted_image(self.display_manager, ifit, box,
+                          align=align, valign=valign, offset=offset)
+        return ifit
+
+    def _get_font_manager(self) -> Any:
+        """The shared FontManager, or a module-level fallback when running
+        under mocks/harnesses that don't provide one."""
+        font_manager = getattr(self.plugin_manager, "font_manager", None)
+        if font_manager is not None and hasattr(font_manager, "get_font"):
+            return font_manager
+        return _fallback_font_manager()
+
+    def _get_design_size(self) -> tuple:
+        """Panel size this plugin's layout was authored against, from the
+        manifest's optional display.design_size (defaults to 128x32)."""
+        from src.adaptive_layout import DEFAULT_DESIGN_SIZE
+
+        if self.plugin_manager and hasattr(self.plugin_manager, "plugin_manifests"):
+            manifest = self.plugin_manager.plugin_manifests.get(self.plugin_id, {})
+            declared = manifest.get("display", {}).get("design_size", {})
+            width, height = declared.get("width"), declared.get("height")
+            if width and height:
+                return (int(width), int(height))
+        return DEFAULT_DESIGN_SIZE
+
     def get_display_duration(self) -> float:
         """
         Get the display duration for this plugin instance.

src/plugin_system/plugin_loader.py

@@ -437,8 +437,7 @@ class PluginLoader:
                 if not Path(existing_file).resolve().is_relative_to(resolved_dir):
                     evicted[mod_name] = sys.modules.pop(mod_name)
                     self.logger.debug(
-                        "Evicted stale module '%s' (from %s) before loading plugin in %s",
-                        mod_name, existing_file, plugin_dir,
+                        "Evicted stale bare-name module '%s' before loading plugin", mod_name,
                     )
             except (ValueError, TypeError):
                 continue
@@ -551,7 +550,7 @@ class PluginLoader:
             plugin_dir_str = str(plugin_dir)
             if plugin_dir_str not in sys.path:
                 sys.path.insert(0, plugin_dir_str)
-                self.logger.debug("Added plugin directory to sys.path: %s", plugin_dir_str)
+                self.logger.debug("Added plugin %s's directory to sys.path", plugin_id)
 
             # Import the plugin module
             module_name = f"plugin_{plugin_id.replace('-', '_')}"
@@ -563,8 +562,8 @@ class PluginLoader:
 
             spec = importlib.util.spec_from_file_location(module_name, entry_file)
             if spec is None or spec.loader is None:
+                self.logger.error("Could not create module spec for plugin %s", plugin_id)
                 error_msg = f"Could not create module spec for {entry_file}"
-                self.logger.error(error_msg)
                 raise PluginError(error_msg, plugin_id=plugin_id, context={'entry_file': str(entry_file)})
 
             module = importlib.util.module_from_spec(spec)
@@ -683,6 +682,55 @@ class PluginLoader:
             self.logger.error(error_msg, exc_info=True)
             raise PluginError(error_msg, plugin_id=plugin_id) from e
     
+    @staticmethod
+    def _parse_semver(value: Any) -> Optional[Tuple[int, int, int]]:
+        """Parse 'X.Y.Z' (extra parts/suffixes ignored) into a comparable
+        3-tuple, or None when unparseable."""
+        if not isinstance(value, str):
+            return None
+        parts = value.strip().lstrip('v').split('.')
+        try:
+            nums = [int(''.join(ch for ch in p if ch.isdigit()) or 0) for p in parts[:3]]
+        except ValueError:
+            return None
+        while len(nums) < 3:
+            nums.append(0)
+        return tuple(nums)  # type: ignore[return-value]
+
+    def _warn_if_incompatible(self, plugin_id: str, manifest: Dict[str, Any]) -> None:
+        """Log one warning when a plugin declares a minimum LEDMatrix version
+        newer than the running core. Advisory only — never raises — so a
+        plugin that guards optional features with try/except keeps working.
+        """
+        declared = (
+            manifest.get('min_ledmatrix_version')
+            or manifest.get('requires', {}).get('min_ledmatrix_version')
+        )
+        if not declared:
+            versions = manifest.get('versions') or []
+            if versions and isinstance(versions[0], dict):
+                declared = (versions[0].get('ledmatrix_min_version')
+                            or versions[0].get('ledmatrix_min'))
+        needed = self._parse_semver(declared)
+        if needed is None:
+            return
+
+        from src import __version__ as core_version
+        current = self._parse_semver(core_version)
+        # Anti-spam guard: if the core's own version number is stale (below
+        # the ecosystem floor every shipped plugin declares), comparing would
+        # warn on nearly everything — skip with a debug note instead.
+        if current is None or current < (2, 0, 0):
+            self.logger.debug(
+                "Skipping version compatibility check for %s: core __version__ "
+                "(%s) is below the ecosystem floor", plugin_id, core_version)
+            return
+        if needed > current:
+            self.logger.warning(
+                "Plugin %s declares min LEDMatrix version %s but this core is %s — "
+                "features it relies on may be missing; update the core or expect "
+                "degraded fallbacks", plugin_id, declared, core_version)
+
     def load_plugin(
         self,
         plugin_id: str,
@@ -715,6 +763,8 @@ class PluginLoader:
         Raises:
             PluginError: If loading fails
         """
+        self._warn_if_incompatible(plugin_id, manifest)
+
         # Install dependencies if needed
         if install_deps:
             if plugins_dir is None:

src/plugin_system/plugin_manager.py

@@ -8,12 +8,13 @@ API Version: 1.0.0
 """
 
 import json
+import queue
 import sys
 import time
 import threading
 import types
 from pathlib import Path
-from typing import Dict, List, Optional, Any
+from typing import Dict, List, Optional, Any, Tuple
 import logging
 from src.exceptions import PluginError, ConfigError
 from src.logging_config import get_logger
@@ -76,6 +77,12 @@ class PluginManager:
         # concurrent mutation (background reconciliation) and reads (requests).
         self._discovery_lock = threading.RLock()
 
+        # Lock protecting plugin_last_update from concurrent mutation/iteration.
+        # It's written from run_scheduled_updates()/update_all_plugins() (main
+        # loop) and read/diffed by run_scheduled_updates_with_changes(), which
+        # Vegas mode calls from its own background update-tick thread.
+        self._plugin_last_update_lock = threading.RLock()
+
         # Active plugins
         self.plugins: Dict[str, Any] = {}
         self.plugin_manifests: Dict[str, Dict[str, Any]] = {}
@@ -91,6 +98,50 @@ class PluginManager:
         # Health tracking (optional, set by display_controller if available)
         self.health_tracker = None
         self.resource_monitor = None
+
+        # --- Asynchronous plugin updates -------------------------------
+        # update() used to run inline in the render loop (execute_update's
+        # internal thread.join(timeout=30) blocked it), so one slow plugin
+        # HTTP fetch froze scrolling for the whole fetch. Scheduling still
+        # happens on the render thread (run_scheduled_updates), but
+        # execution moves to this single background worker. Per-plugin
+        # locks keep a plugin's update() and display() mutually exclusive —
+        # today's implicit guarantee, now explicit (and, unlike today,
+        # also held across the post-timeout window).
+        # Kill switch: plugin_system.synchronous_updates: true restores the
+        # inline path.
+        self._update_queue: "queue.Queue[Optional[Tuple[str, float]]]" = queue.Queue()
+        self._pending_updates: set = set()
+        self._pending_lock = threading.Lock()
+        self._plugin_locks: Dict[str, threading.Lock] = {}
+        self._plugin_locks_guard = threading.Lock()
+        self._update_worker: Optional[threading.Thread] = None
+        self._synchronous_updates = False
+        if self.config_manager is not None:
+            try:
+                cfg = self.config_manager.get_config() or {}
+            except (OSError, ValueError) as exc:
+                self.logger.warning(
+                    "Could not load config to check plugin_system.synchronous_updates "
+                    "(%s: %s); defaulting to synchronous updates", type(exc).__name__, exc)
+                self._synchronous_updates = True
+            else:
+                plugin_system_cfg = cfg.get('plugin_system', {})
+                if not isinstance(plugin_system_cfg, dict):
+                    self.logger.warning(
+                        "config plugin_system must be a mapping, got %s; "
+                        "defaulting to synchronous updates",
+                        type(plugin_system_cfg).__name__)
+                    self._synchronous_updates = True
+                else:
+                    sync_value = plugin_system_cfg.get('synchronous_updates', False)
+                    if not isinstance(sync_value, bool):
+                        self.logger.warning(
+                            "config plugin_system.synchronous_updates must be a boolean, "
+                            "got %r; defaulting to synchronous updates", sync_value)
+                        self._synchronous_updates = True
+                    else:
+                        self._synchronous_updates = sync_value
         
         # Ensure plugins directory exists with proper permissions
         try:
@@ -317,7 +368,8 @@ class PluginManager:
 
             # Store plugin instance
             self.plugins[plugin_id] = plugin_instance
-            self.plugin_last_update[plugin_id] = 0.0
+            with self._plugin_last_update_lock:
+                self.plugin_last_update[plugin_id] = 0.0
             # Invalidate cached interval so next tick re-derives it for this plugin
             self._update_interval_cache.pop(plugin_id, None)
             
@@ -429,7 +481,8 @@ class PluginManager:
             
             # Remove from active plugins
             del self.plugins[plugin_id]
-            self.plugin_last_update.pop(plugin_id, None)
+            with self._plugin_last_update_lock:
+                self.plugin_last_update.pop(plugin_id, None)
             self._update_interval_cache.pop(plugin_id, None)
             
             # Remove main module from sys.modules if present
@@ -698,7 +751,8 @@ class PluginManager:
             'recoverable': True,
         }
         self.logger.warning("Plugin %s update() failed; will retry after interval", plugin_id)
-        self.plugin_last_update[plugin_id] = failure_time
+        with self._plugin_last_update_lock:
+            self.plugin_last_update[plugin_id] = failure_time
         self.state_manager.set_state_with_error(plugin_id, PluginState.ENABLED, error_info, error=err)
         if self.health_tracker:
             self.health_tracker.record_failure(plugin_id, err)
@@ -731,49 +785,218 @@ class PluginManager:
             if interval is None:
                 continue
 
-            last_update = self.plugin_last_update.get(plugin_id, 0.0)
+            with self._plugin_last_update_lock:
+                last_update = self.plugin_last_update.get(plugin_id, 0.0)
 
             if last_update == 0.0 or (current_time - last_update) >= interval:
-                # Update state to RUNNING
-                self.state_manager.set_state(plugin_id, PluginState.RUNNING)
-                
-                try:
-                    # Use PluginExecutor for safe execution
-                    success = False
-                    if self.resource_monitor:
-                        # If resource monitor exists, wrap the call
-                        def monitored_update():
-                            self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
-                        # SimpleNamespace stores `update` as an *instance*
-                        # attribute, so attribute lookup returns the plain
-                        # function object as-is. A dynamically-built class
-                        # (`type(..., {'update': monitored_update})`) instead
-                        # stores it as a *class* attribute, which the
-                        # descriptor protocol turns into a bound method on
-                        # access -- silently prepending the instance as an
-                        # implicit first argument to a function that takes
-                        # none, raising "monitored_update() takes 0
-                        # positional arguments but 1 was given" on every call.
-                        success = self.plugin_executor.execute_update(
-                            types.SimpleNamespace(update=monitored_update),
-                            plugin_id
-                        )
-                    else:
-                        success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
-                    
-                    if success:
-                        self.plugin_last_update[plugin_id] = current_time
-                        self.state_manager.record_update(plugin_id)
-                        # Update state back to ENABLED
-                        self.state_manager.set_state(plugin_id, PluginState.ENABLED)
-                        # Record success
-                        if self.health_tracker:
-                            self.health_tracker.record_success(plugin_id)
-                    else:
-                        self._record_update_failure(plugin_id)
-                except Exception as exc:  # pylint: disable=broad-except
-                    self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
+                if self._synchronous_updates:
+                    # Kill-switch path: the original inline execution
+                    # (blocks the caller until update() completes/times out)
+                    self.state_manager.set_state(plugin_id, PluginState.RUNNING)
+                    self._execute_update_now(plugin_id, plugin_instance, current_time)
+                else:
+                    self._enqueue_update(plugin_id, current_time)
+
+    def get_plugin_lock(self, plugin_id: str) -> threading.Lock:
+        """Per-plugin lock keeping update() and display() mutually exclusive.
+
+        The update worker holds it for the duration of a plugin's update();
+        the display side acquires it non-blocking and skips that frame's
+        display() call when the plugin is mid-update.
+        """
+        with self._plugin_locks_guard:
+            lock = self._plugin_locks.get(plugin_id)
+            if lock is None:
+                lock = threading.Lock()
+                self._plugin_locks[plugin_id] = lock
+            return lock
+
+    def _enqueue_update(self, plugin_id: str, scheduled_time: float) -> None:
+        """Queue a due update for the background worker (dedup while pending)."""
+        with self._pending_lock:
+            if plugin_id in self._pending_updates:
+                return
+            self._pending_updates.add(plugin_id)
+        # RUNNING is set at enqueue time so can_execute() blocks re-entry and
+        # the web UI shows the truthful state while the item waits its turn.
+        self.state_manager.set_state(plugin_id, PluginState.RUNNING)
+        self._ensure_update_worker()
+        self._update_queue.put((plugin_id, scheduled_time))
+
+    def _ensure_update_worker(self) -> None:
+        if self._update_worker is not None and self._update_worker.is_alive():
+            return
+        self._update_worker = threading.Thread(
+            target=self._update_worker_loop, name='plugin-update-worker',
+            daemon=True)
+        self._update_worker.start()
+
+    def _update_worker_loop(self) -> None:
+        """Single worker: dispatches queued updates off the render thread
+        (matching the old inline behavior — no thundering herd of
+        concurrent fetches).
+
+        The plugin's lock is acquired here, before its instance is looked
+        up, and the instance is re-fetched under the lock — a concurrent
+        unload_plugin() can't leave this loop about to run update() on an
+        instance that's already been torn down. The lock — and RUNNING/
+        pending lifecycle state — is released by the update itself once the
+        real update() call genuinely finishes (see _execute_update_now),
+        which can be after this dispatch returns if PluginExecutor's own
+        timeout elapses first.
+        """
+        while True:
+            item = self._update_queue.get()
+            if item is None:  # shutdown sentinel
+                return
+            plugin_id, scheduled_time = item
+            lock = self.get_plugin_lock(plugin_id)
+            lock.acquire()
+            plugin_instance = self.plugins.get(plugin_id)
+            if plugin_instance is None:  # unloaded while queued; its
+                # lifecycle state was already cleared by unload_plugin —
+                # leave it alone rather than resurrecting it to ENABLED
+                lock.release()
+                with self._pending_lock:
+                    self._pending_updates.discard(plugin_id)
+                continue
+            try:
+                self._execute_update_now(plugin_id, plugin_instance,
+                                         scheduled_time, lock=lock)
+            except Exception:  # pylint: disable=broad-except
+                # _execute_update_now guarantees the lock/pending bookkeeping
+                # is released via its own _finish() before returning or
+                # raising; this is a last-resort log only.
+                self.logger.exception("update worker: unexpected error for %s",
+                                      plugin_id)
+
+    def stop_update_worker(self, timeout: float = 5.0) -> None:
+        """Signal the worker to exit (used by cleanup; thread is a daemon)."""
+        if self._update_worker is not None and self._update_worker.is_alive():
+            self._update_queue.put(None)
+            self._update_worker.join(timeout=timeout)
+            if self._update_worker.is_alive():
+                self.logger.warning(
+                    "Update worker did not stop within %.1fs; it is a daemon "
+                    "thread and will be abandoned on shutdown", timeout)
+
+    def _execute_update_now(self, plugin_id: str, plugin_instance: Any,
+                            scheduled_time: float,
+                            lock: Optional[threading.Lock] = None) -> None:
+        """Execute a plugin's update() via PluginExecutor, then bookkeep.
+
+        Caller is responsible for having set RUNNING state.
+
+        On the synchronous path (``lock=None``) this is the original,
+        unchanged inline behavior. On the async worker path, PluginExecutor's
+        internal thread.join(timeout) blocks only the calling thread -- on
+        timeout the lingering daemon update-thread keeps running the real
+        plugin.update() call unkillable in the background. So that the
+        plugin's lock (and its RUNNING/pending lifecycle state) stays held
+        for that real duration rather than just this bounded wait, ownership
+        of both is carried by the wrapped update callable itself, released
+        from whichever thread actually finishes it -- see _finish() below.
+        """
+        finish_guard = threading.Lock()
+        finished = {'done': False}
+
+        def _finish(success: bool, exc: Optional[Exception] = None) -> None:
+            with finish_guard:
+                if finished['done']:
+                    return
+                finished['done'] = True
+            try:
+                if success:
+                    with self._plugin_last_update_lock:
+                        self.plugin_last_update[plugin_id] = scheduled_time
+                    self.state_manager.record_update(plugin_id)
+                    self.state_manager.set_state(plugin_id, PluginState.ENABLED)
+                    if self.health_tracker:
+                        self.health_tracker.record_success(plugin_id)
+                else:
                     self._record_update_failure(plugin_id, exc=exc)
+            finally:
+                if lock is not None:
+                    lock.release()
+                    with self._pending_lock:
+                        self._pending_updates.discard(plugin_id)
+
+        if lock is None:
+            # Synchronous / no-lock path: unchanged behavior.
+            try:
+                if self.resource_monitor:
+                    def monitored_update():
+                        self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
+                    # SimpleNamespace stores `update` as an *instance*
+                    # attribute, so attribute lookup returns the plain
+                    # function object as-is. A dynamically-built class
+                    # (`type(..., {'update': monitored_update})`) instead
+                    # stores it as a *class* attribute, which the
+                    # descriptor protocol turns into a bound method on
+                    # access -- silently prepending the instance as an
+                    # implicit first argument to a function that takes
+                    # none, raising "monitored_update() takes 0
+                    # positional arguments but 1 was given" on every call.
+                    success = self.plugin_executor.execute_update(
+                        types.SimpleNamespace(update=monitored_update),
+                        plugin_id
+                    )
+                else:
+                    success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
+                _finish(success)
+            except Exception as exc:  # pylint: disable=broad-except
+                self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
+                _finish(False, exc=exc)
+            return
+
+        # Async worker path: the real update() call -- through the resource
+        # monitor, if configured -- owns finishing the lock/lifecycle
+        # bookkeeping, from whichever thread actually runs it to completion.
+        def _target_update() -> None:
+            try:
+                if self.resource_monitor:
+                    self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
+                else:
+                    plugin_instance.update()
+            except Exception as exc:
+                _finish(False, exc=exc)
+                raise
+            else:
+                _finish(True)
+
+        try:
+            self.plugin_executor.execute_update(
+                types.SimpleNamespace(update=_target_update), plugin_id)
+        except Exception as exc:  # pragma: no cover - defensive; execute_update
+            # catches everything internally, but guarantee _finish still
+            # runs (releasing the lock) if something unexpected slips through.
+            self.logger.exception("Unexpected error dispatching update for %s: %s", plugin_id, exc)
+            _finish(False, exc=exc)
+
+    def run_scheduled_updates_with_changes(self, current_time: Optional[float] = None) -> List[str]:
+        """
+        Like run_scheduled_updates(), but also returns the plugin_ids whose
+        plugin_last_update timestamp actually advanced during this call.
+
+        The before/after snapshots and the update pass itself are each
+        individually lock-protected against concurrent plugin_last_update
+        mutation (Vegas mode calls this from its own background
+        update-tick thread, racing the main render loop's plugin updates),
+        so callers get an atomic "who got fresh data" answer without
+        reaching into plugin_last_update themselves. The lock is not held
+        across the update pass so slow/blocking plugin update() calls don't
+        serialize against other plugin_last_update readers.
+        """
+        with self._plugin_last_update_lock:
+            old_times = dict(self.plugin_last_update)
+
+        self.run_scheduled_updates(current_time)
+
+        with self._plugin_last_update_lock:
+            return [
+                plugin_id for plugin_id, new_time in self.plugin_last_update.items()
+                if new_time > old_times.get(plugin_id, 0.0)
+            ]
 
     def update_all_plugins(self) -> None:
         """
@@ -797,7 +1020,8 @@ class PluginManager:
             try:
                 success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
                 if success:
-                    self.plugin_last_update[plugin_id] = time.time()
+                    with self._plugin_last_update_lock:
+                        self.plugin_last_update[plugin_id] = time.time()
                     self.state_manager.record_update(plugin_id)
                     self.state_manager.set_state(plugin_id, PluginState.ENABLED)
                 else:

src/plugin_system/store_manager.py

@@ -142,9 +142,28 @@ class PluginStoreManager:
         # then get the result from the warm cache (double-checked locking).
         self._registry_fetch_lock = threading.Lock()
 
+        # Per-plugin locks for _reinstall_with_rollback: the web UI runs
+        # Flask with threaded=True, so two overlapping requests for the
+        # same plugin_id (double-click, two browser tabs) would otherwise
+        # both rename the same directory aside — one succeeds, and the
+        # loser can end up renaming the winner's in-progress install aside
+        # mid-download, stealing its own rollback safety net. Keyed by
+        # plugin_id so unrelated plugins still update concurrently.
+        self._reinstall_locks: Dict[str, threading.Lock] = {}
+        self._reinstall_locks_guard = threading.Lock()
+
         # Ensure plugins directory exists
         self.plugins_dir.mkdir(exist_ok=True)
 
+    def _get_reinstall_lock(self, plugin_id: str) -> threading.Lock:
+        """Lazily create (or fetch) the per-plugin reinstall lock."""
+        with self._reinstall_locks_guard:
+            lock = self._reinstall_locks.get(plugin_id)
+            if lock is None:
+                lock = threading.Lock()
+                self._reinstall_locks[plugin_id] = lock
+            return lock
+
     def _record_cache_backoff(self, cache_dict: Dict, cache_key: str,
                               cache_timeout: int, payload: Any) -> None:
         """Bump a cache entry's timestamp so subsequent lookups hit the
@@ -2263,6 +2282,74 @@ class PluginStoreManager:
             self.logger.error(f"Error uninstalling plugin {plugin_id}: {e}")
             return False
     
+    def _reinstall_with_rollback(self, plugin_id: str, plugin_path: Path) -> bool:
+        """Replace an installed plugin with a fresh install, atomically.
+
+        The old install is renamed aside (not deleted) until the new install
+        succeeds, then removed; on ANY install failure the old directory is
+        restored. This is the difference between a failed update and a
+        destroyed plugin: the previous delete-then-install flow permanently
+        removed plugins whenever the download failed mid-update (seen in the
+        field during the monorepo migration on a Pi with broken DNS — every
+        old-remote plugin was deleted and none could be re-downloaded).
+
+        The aside name embeds '.standalone-backup-' so plugin discovery
+        (plugin_manager._scan_directory_for_plugins) ignores it even though
+        it still contains a manifest.json.
+
+        Held for the whole operation under a per-plugin_id lock: two
+        overlapping requests for the same plugin (double-click, two
+        browser tabs — the web UI runs Flask with threaded=True) must not
+        interleave their renames, or the second could steal the first's
+        rollback safety net mid-install. Other plugin_ids are unaffected.
+        """
+        with self._get_reinstall_lock(plugin_id):
+            backup_path = plugin_path.with_name(
+                f"{plugin_path.name}.standalone-backup-migrating")
+            # A stale aside from a previous crash would block the rename
+            if backup_path.exists():
+                if not self._safe_remove_directory(backup_path):
+                    self.logger.error(
+                        f"Could not clear stale backup for {plugin_id} at "
+                        f"{backup_path}; leaving old install in place")
+                    return False
+            try:
+                plugin_path.rename(backup_path)
+            except OSError as e:
+                self.logger.error(
+                    f"Could not set aside old plugin directory for {plugin_id}: {e}")
+                return False
+
+            try:
+                installed = self.install_plugin(plugin_id)
+            except Exception as e:
+                self.logger.error(f"Reinstall of {plugin_id} raised: {e}")
+                installed = False
+
+            if installed:
+                if not self._safe_remove_directory(backup_path):
+                    self.logger.warning(
+                        f"Update of {plugin_id} succeeded but the old backup "
+                        f"at {backup_path} could not be removed; it will be "
+                        f"cleared on the next update")
+                return True
+
+            # Install failed (bad network, registry error...) — put the old
+            # version back so the user still has a working plugin.
+            self.logger.error(
+                f"Reinstall of {plugin_id} failed; restoring previous version")
+            try:
+                if plugin_path.exists():
+                    # partial download debris from the failed install
+                    self._safe_remove_directory(plugin_path)
+                backup_path.rename(plugin_path)
+                self.logger.info(f"Restored previous install of {plugin_id}")
+            except OSError as e:
+                self.logger.error(
+                    f"CRITICAL: could not restore {plugin_id} from {backup_path}: {e}. "
+                    f"The previous install is preserved there — rename it back manually.")
+            return False
+
     def update_plugin(self, plugin_id: str) -> bool:
         """
         Update a plugin to the latest commit on its upstream branch.
@@ -2325,10 +2412,7 @@ class PluginStoreManager:
                             f"Plugin {resolved_id} git remote ({local_remote}) differs from registry ({registry_repo}). "
                             f"Reinstalling from registry to migrate to new source."
                         )
-                        if not self._safe_remove_directory(plugin_path):
-                            self.logger.error(f"Failed to remove old plugin directory for {resolved_id}")
-                            return False
-                        return self.install_plugin(resolved_id)
+                        return self._reinstall_with_rollback(resolved_id, plugin_path)
 
                     # Check if already up to date
                     if remote_sha and local_sha and remote_sha.startswith(local_sha):
@@ -2632,11 +2716,11 @@ class PluginStoreManager:
             # Plugin is not a git repo but is in registry and has a newer version - reinstall
             self.logger.info(f"Plugin {plugin_id} not installed via git; re-installing latest archive (registry id: {registry_id})")
 
-            # Remove directory and reinstall fresh
-            if not self._safe_remove_directory(plugin_path):
-                self.logger.error(f"Failed to remove old plugin directory for {plugin_id}")
-                return False
-            return self.install_plugin(registry_id)
+            # Reinstall with the old version kept aside until the new
+            # download succeeds — this is the path every routine store
+            # update takes, and a mid-update network failure must not
+            # destroy the user's plugin.
+            return self._reinstall_with_rollback(registry_id, plugin_path)
 
         except Exception as e:
             import traceback

src/plugin_system/testing/bounds_display_manager.py

@@ -10,8 +10,11 @@ that don't scale down to a smaller panel.
 
 Limitations (documented on purpose):
 - Overflow past the LEFT or TOP edge (negative coordinates) is still clipped by
-  PIL and not detected here. The dominant real-world breakage is content that is
-  too wide/tall for a smaller panel, which this catches.
+  PIL and not detected pixel-wise here. The dominant real-world breakage is
+  content that is too wide/tall for a smaller panel, which this catches.
+  As a partial net, draw_text/draw_image calls made with negative coordinates
+  through this manager are recorded in `negative_coordinate_calls` — but draws
+  made directly on the raw PIL canvas remain uncovered.
 - BDF text is clipped to the declared bounds by the parent's bitmap drawer, so
   BDF overflow is not flagged. Golden-image regression covers those plugins.
 - If a plugin replaces the canvas with its own image (display_manager.image = ...),
@@ -56,6 +59,21 @@ class BoundsCheckingDisplayManager(VisualTestDisplayManager):
         super().__init__(self._canvas_width, self._canvas_height)
         # Plugins must see the DECLARED size, not the padded canvas size.
         self.matrix = _MatrixProxy(self._declared_width, self._declared_height)
+        # (text-or-'image', x, y) for every mediated draw call given a
+        # negative coordinate — PIL clips these silently, so record them.
+        self.negative_coordinate_calls: list = []
+
+    # -- negative-coordinate (left/top overflow) recording --
+
+    def draw_text(self, text, x=None, y=None, *args, **kwargs):
+        if (x is not None and x < 0) or (y is not None and y < 0):
+            self.negative_coordinate_calls.append((text, x, y))
+        return super().draw_text(text, x, y, *args, **kwargs)
+
+    def draw_image(self, image, x, y, *args, **kwargs):
+        if x < 0 or y < 0:
+            self.negative_coordinate_calls.append(('image', x, y))
+        return super().draw_image(image, x, y, *args, **kwargs)
 
     # -- declared dimensions (override parent's image-derived properties) --
 

src/plugin_system/testing/harness.py

@@ -73,6 +73,10 @@ class RenderResult:
     golden_ok: Optional[bool] = None
     golden_diff_pixels: int = 0
     golden_max_delta: int = 0
+    # fill / scale-up check (populated only for sizes >= 2x the design size)
+    fill_checked: bool = False
+    fill_ok: Optional[bool] = None       # False only in strict mode
+    fill_extent: Optional[Tuple[float, float]] = None  # (extent_x, extent_y)
 
     @property
     def size_label(self) -> str:
@@ -86,6 +90,8 @@ class RenderResult:
             return False
         if self.golden_checked and self.golden_ok is False:
             return False
+        if self.fill_ok is False:
+            return False
         return True
 
 
@@ -301,6 +307,74 @@ def compare_to_goldens(results: List[RenderResult], golden_dir: Path,
     return results
 
 
+# ---------------------------------------------------------------------------
+# Fill / scale-up check
+# ---------------------------------------------------------------------------
+#
+# Overflow catches content that is too BIG for a panel; nothing catches
+# content that stays tiny on a panel much larger than the plugin's design
+# size (e.g. 128x32 content in the corner of a 256x128 renders "green").
+# These helpers measure how much of the panel the lit content spans so the
+# harness can flag plugins that don't scale up.
+
+# A pixel counts as "lit" above this luminance — low enough to catch dim
+# content, high enough to ignore near-black noise.
+_LIT_THRESHOLD = 16
+# Content must span at least this fraction of an axis that is >= 2x the
+# design size. Lenient on purpose: margins are fine, a tiny corner is not.
+_MIN_FILL_EXTENT = 0.5
+
+
+def fill_metrics(image: Image.Image) -> Tuple[float, float, float]:
+    """Measure lit-content coverage: (extent_x, extent_y, ink_ratio).
+
+    extent_* are the lit bounding box's spans as fractions of the panel;
+    ink_ratio is the fraction of pixels lit (reporting only — sparse pixel
+    fonts legitimately have low ink ratios)."""
+    lit = image.convert("L").point(lambda p: 255 if p > _LIT_THRESHOLD else 0)
+    bbox = lit.getbbox()
+    if bbox is None:
+        return (0.0, 0.0, 0.0)
+    extent_x = (bbox[2] - bbox[0]) / image.width
+    extent_y = (bbox[3] - bbox[1]) / image.height
+    ink = sum(1 for p in lit.getdata() if p) / (image.width * image.height)
+    return (extent_x, extent_y, ink)
+
+
+def check_scale_up(results: List[RenderResult],
+                   design_size: Tuple[int, int] = (128, 32),
+                   min_extent: float = _MIN_FILL_EXTENT,
+                   strict: bool = False) -> List[RenderResult]:
+    """Flag renders that leave a big panel mostly empty.
+
+    For each result whose panel is at least 2x the design size on an axis,
+    require the lit content to span >= min_extent of that axis. Mutates the
+    results' fill_* fields. In the default warn-only mode fill_ok is left
+    None (reported, never failing); strict=True sets fill_ok=False, which
+    fails RenderResult.ok — opt in per plugin via harness.json
+    {"fill_check": "strict"} once its adaptive layout is in place.
+    """
+    design_w, design_h = design_size
+    for r in results:
+        if r.image is None or r.error is not None:
+            continue
+        check_x = r.width >= 2 * design_w
+        check_y = r.height >= 2 * design_h
+        if not (check_x or check_y):
+            continue
+        extent_x, extent_y, _ink = fill_metrics(r.image)
+        r.fill_checked = True
+        r.fill_extent = (round(extent_x, 3), round(extent_y, 3))
+        underfilled = ((check_x and extent_x < min_extent)
+                       or (check_y and extent_y < min_extent))
+        if underfilled and strict:
+            r.fill_ok = False
+        elif not underfilled:
+            r.fill_ok = True
+        # warn-only underfill: fill_ok stays None; fill_extent tells the story
+    return results
+
+
 def write_goldens(results: List[RenderResult], golden_dir: Path) -> int:
     """Write each successfully-rendered result to its golden path. Returns count."""
     written = 0

src/plugin_system/testing/loading.py

@@ -56,7 +56,15 @@ def load_harness_spec(plugin_dir: Union[str, Path]) -> Dict[str, Any]:
           "config":     {...},            # config overrides
           "mock_data":  "fixtures/mock.json",  # path (relative to plugin dir) to cache fixtures
           "freeze_time": "2025-08-01 15:25:00",
-          "skip_update": false
+          "skip_update": false,
+          "fill_check": "warn",           # or "strict": underfilled big panels FAIL
+          "variants": [                   # extra runs with config overlays and
+            {                             # their own golden dirs — e.g. an
+              "name": "adaptive",         # opt-in adaptive mode tested beside
+              "config": {"layout_mode": "adaptive"},   # the classic default
+              "golden_dir": "test/golden-adaptive"
+            }
+          ]
         }
     Returns {} when no harness.json exists.
     """
@@ -80,3 +88,27 @@ def load_harness_spec(plugin_dir: Union[str, Path]) -> Dict[str, Any]:
         with open(mock_path, 'r') as mf:
             spec['mock_data_contents'] = json.load(mf)
     return spec
+
+
+def build_full_config(
+    plugin_dir: Union[str, Path],
+    spec: Optional[Dict[str, Any]] = None,
+    cli_config: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    """Build the config a plugin sees under test.
+
+    Merge order: config_schema.json defaults, then a forced ``enabled: True``,
+    then harness.json's config overlay, then the caller's explicit config --
+    most specific wins. `enabled` is re-asserted *after* the schema defaults
+    so a plugin that reasonably ships `enabled: false` (e.g. a seasonal or
+    opt-in plugin) can't silently make every harness run test "disabled, do
+    nothing" by accident -- callers that genuinely want to test the disabled
+    path can still do so via `cli_config={"enabled": False}`.
+    """
+    spec = spec or {}
+    config: Dict[str, Any] = {}
+    config.update(load_config_defaults(plugin_dir))
+    config["enabled"] = True
+    config.update(spec.get("config", {}))
+    config.update(cli_config or {})
+    return config

src/plugin_system/testing/mocks.py

@@ -71,6 +71,7 @@ class MockCacheManager:
         self.get_calls = []
         self.set_calls = []
         self.delete_calls = []
+        self.get_cached_data_with_strategy_calls = []
         # Real temp dir for plugins that write/read files under cache_dir.
         # Registered for cleanup so each mock instance doesn't leak a tmp dir.
         self.cache_dir = tempfile.mkdtemp(prefix="ledmatrix-mock-cache-")
@@ -108,6 +109,24 @@ class MockCacheManager:
         self.delete_calls.append(key)
         if key in self._cache:
             del self._cache[key]
+
+    def get_cached_data_with_strategy(self, key: str, data_type: str = 'default') -> Optional[Any]:
+        """Mock of CacheManager.get_cached_data_with_strategy (src/cache_manager.py).
+
+        The real method picks a max_age/memory_ttl strategy per data_type
+        (and extends it during market-closed hours for market data) before
+        delegating to get_cached_data(). None of that timing nuance matters
+        for a mock -- plugins under test just need the method to exist and
+        return whatever was cached, so this delegates straight to get().
+        """
+        self.get_cached_data_with_strategy_calls.append({'key': key, 'data_type': data_type})
+        return self.get(key)
+
+    def save_cache(self, key: str, data: Any) -> None:
+        """Mock of CacheManager.save_cache (src/cache_manager.py) -- the
+        write-side counterpart to get_cached_data_with_strategy, used by the
+        same real-CacheManager-oriented plugins. Delegates to set()."""
+        self.set(key, data)
         if key in self._cache_timestamps:
             del self._cache_timestamps[key]
     
@@ -118,6 +137,7 @@ class MockCacheManager:
         self.get_calls = []
         self.set_calls = []
         self.delete_calls = []
+        self.get_cached_data_with_strategy_calls = []
 
 
 class MockConfigManager:
@@ -161,6 +181,13 @@ class MockPluginManager:
         self.plugin_manifests: Dict[str, Dict] = {}
         self.get_plugin_calls = []
         self.get_all_plugins_calls = []
+        # Real FontManager so BasePlugin.layout / draw_fit behave identically
+        # under the harness (it only needs assets/fonts on disk).
+        try:
+            from src.font_manager import FontManager
+            self.font_manager: Optional[Any] = FontManager({})
+        except Exception:
+            self.font_manager = None
     
     def get_plugin(self, plugin_id: str) -> Optional[Any]:
         """Get a plugin instance."""

src/plugin_system/testing/sizes.py

@@ -28,6 +28,7 @@ DEFAULT_TEST_SIZES: List[Tuple[int, int]] = [
     (64, 32),    # 1x1 — single panel, the tightest common rectangle
     (128, 32),   # 2x1 — the baseline most plugins are tuned for
     (64, 64),    # 1x2 — stacked, exercises tall-narrow centering
+    (96, 48),    # non-64x32-module panel (e.g. Waveshare), off-grid dims
     (128, 64),   # 2x2 — block, icon scaling / vertical centering
     (256, 32),   # 4x1 — long strip, wide horizontal layout
     (128, 96),   # 2x3 — tall, exercises vertical overflow

src/vegas_mode/render_pipeline.py

@@ -66,6 +66,10 @@ class RenderPipeline:
             else display_manager.height
         )
 
+        # Reusable blank frame for cycle-end pushes (allocated lazily,
+        # re-blacked before each reuse)
+        self._blank_frame = None
+
         # ScrollHelper for optimized scrolling
         self.scroll_helper = ScrollHelper(
             self.display_width,
@@ -234,11 +238,19 @@ class RenderPipeline:
                     )
                     # Push blank immediately so the hardware never shows any
                     # post-wrap content while the coordinator recomposes the
-                    # next cycle (~100 ms).
+                    # next cycle (~100 ms). The blank is allocated once and
+                    # reused across cycle wraps (fresh paste each time in case
+                    # a consumer drew on the previous one).
                     try:
-                        from PIL import Image as _Image
-                        blank = _Image.new('RGB', (self.display_width, self.display_height))
-                        self.display_manager.image = blank
+                        if self._blank_frame is None or self._blank_frame.size != (
+                                self.display_width, self.display_height):
+                            self._blank_frame = Image.new(
+                                'RGB', (self.display_width, self.display_height))
+                        else:
+                            self._blank_frame.paste(
+                                (0, 0, 0),
+                                (0, 0, self.display_width, self.display_height))
+                        self.display_manager.image = self._blank_frame
                         self.display_manager.update_display()
                     except Exception:
                         logger.exception("Failed to write blank frame to display at cycle end")
@@ -297,6 +309,8 @@ class RenderPipeline:
         Returns True when:
         - Cycle is complete and we should start fresh
         - Staging buffer has new content
+        - A plugin currently visible in the scroll has pending updated data
+          (e.g. a live score changed) — standalone (non-sync) mode only
         """
         if self._cycle_complete:
             return True
@@ -314,6 +328,12 @@ class RenderPipeline:
         if buffer_status['staging_count'] > 0:
             return True
 
+        # Trigger recompose when pending updates affect visible segments, so
+        # live score/status changes reach the display within a few seconds
+        # instead of waiting for the next full cycle.
+        if self.stream_manager.has_pending_updates_for_visible_segments():
+            return True
+
         return False
 
     def hot_swap_content(self) -> bool:

src/wifi_manager.py

@@ -2564,11 +2564,35 @@ address=/detectportal.firefox.com/192.168.4.1
         Returns:
             True if AP mode state changed, False otherwise
         """
+        changed, _status, _ethernet, _ap = self.check_and_manage_ap_mode_with_state()
+        return changed
+
+    def check_and_manage_ap_mode_with_state(self) -> Tuple[bool, WiFiStatus, bool, bool]:
+        """Like check_and_manage_ap_mode, but also returns the state it
+        observed, so callers (the wifi monitor daemon) don't have to re-run
+        the same nmcli subprocess battery before AND after the check —
+        each status fetch is several process forks.
+
+        Returns:
+            (state_changed, WiFiStatus, ethernet_connected, ap_active_after)
+        """
         try:
             # Get status with retry for more reliable detection
             status = self._get_wifi_status_with_retry()
             ethernet_connected = self._is_ethernet_connected()
             ap_active = self._is_ap_mode_active()
+            changed = self._manage_ap_mode(status, ethernet_connected, ap_active)
+            # State only ever changes via one enable or one disable, so the
+            # post-state is the inverse of the pre-state when changed.
+            ap_after = (not ap_active) if changed else ap_active
+            return changed, status, ethernet_connected, ap_after
+        except Exception as e:
+            logger.error(f"Error checking AP mode: {e}", exc_info=True)
+            return False, WiFiStatus(connected=False), False, False
+
+    def _manage_ap_mode(self, status: WiFiStatus, ethernet_connected: bool, ap_active: bool) -> bool:
+        """AP-mode decision logic against an already-fetched state snapshot."""
+        try:
             auto_enable = self.config.get("auto_enable_ap_mode", True)  # Default: True (safe due to grace period)
             
             # Log current state for debugging

test/conftest.py

@@ -38,7 +38,11 @@ def mock_cache_manager():
     mock._memory_cache_timestamps = {}
     mock.cache_dir = "/tmp/test_cache"
     
-    def mock_get(key: str, max_age: int = 300) -> Optional[Dict]:
+    def mock_get(key: str, max_age: Optional[int] = 300,
+                 memory_ttl: Optional[int] = None) -> Optional[Dict]:
+        # Signature mirrors CacheManager.get — keep in sync or callers
+        # passing keyword args (health tracker, resource monitor) break
+        # only in tests, hiding real-API compatibility.
         return mock._memory_cache.get(key)
     
     def mock_set(key: str, data: Dict, ttl: Optional[int] = None) -> None:

test/plugins/test_harness.py

@@ -253,3 +253,61 @@ class TestCheckPluginHonorsHarnessJson:
         )
         assert captured["freeze_time"] == "2030-01-01 00:00:00"
         assert captured["config"]["timezone"] == "America/New_York"
+
+
+class TestBuildFullConfigForcesEnabled:
+    """Regression: a plugin's own config_schema.json may reasonably default
+    enabled to False (e.g. a seasonal or opt-in plugin) -- march-madness and
+    14 other real plugins do. The harness must still test it as enabled
+    unless a caller explicitly asks otherwise, or every render silently
+    becomes a same-shaped "disabled, do nothing" no-op."""
+
+    def _make_plugin_with_disabled_default(self, tmp_path):
+        pdir = tmp_path / "plugins" / "demo-seasonal"
+        pdir.mkdir(parents=True)
+        (pdir / "manifest.json").write_text(json.dumps({
+            "id": "demo-seasonal", "name": "Demo Seasonal", "version": "1.0.0",
+            "author": "test", "entry_point": "manager.py",
+            "class_name": "DemoSeasonal", "display_modes": ["demo-seasonal"],
+            "compatible_versions": ["*"],
+        }))
+        (pdir / "config_schema.json").write_text(json.dumps({
+            "type": "object",
+            "properties": {"enabled": {"type": "boolean", "default": False}},
+        }))
+        return pdir
+
+    def test_schema_disabled_default_does_not_win(self, tmp_path):
+        from src.plugin_system.testing.loading import build_full_config
+        plugin_dir = self._make_plugin_with_disabled_default(tmp_path)
+        config = build_full_config(plugin_dir)
+        assert config["enabled"] is True
+
+    def test_harness_json_config_can_still_disable(self, tmp_path):
+        from src.plugin_system.testing.loading import build_full_config
+        plugin_dir = self._make_plugin_with_disabled_default(tmp_path)
+        config = build_full_config(plugin_dir, spec={"config": {"enabled": False}})
+        assert config["enabled"] is False
+
+    def test_explicit_cli_config_can_still_disable(self, tmp_path):
+        from src.plugin_system.testing.loading import build_full_config
+        plugin_dir = self._make_plugin_with_disabled_default(tmp_path)
+        config = build_full_config(plugin_dir, cli_config={"enabled": False})
+        assert config["enabled"] is False
+
+    def test_check_one_renders_a_schema_disabled_plugin_as_enabled(self, tmp_path, monkeypatch):
+        """End-to-end: check_plugin.py's check_one() must not blank-render a
+        plugin just because its own schema defaults enabled to False."""
+        mod = _load_check_plugin_cli()
+        plugin_dir = self._make_plugin_with_disabled_default(tmp_path)
+        captured = {}
+        monkeypatch.setattr(mod, "render_plugin_matrix",
+                            lambda **kw: captured.update(kw) or [])
+        monkeypatch.setattr(mod, "compare_to_goldens", lambda *a, **k: [])
+        mod.check_one(
+            plugin_id="demo-seasonal", search_dirs=[str(tmp_path / "plugins")],
+            sizes=None, mock_data={}, config={}, run_update=True,
+            out_dir=None, update_golden=False, golden_dir_override=None,
+            freeze_time=None,
+        )
+        assert captured["config"]["enabled"] is True

test/plugins/test_plugin_matrix.py

@@ -22,7 +22,7 @@ import pytest
 from src.plugin_system.testing.harness import (
     render_plugin_matrix, compare_to_goldens,
 )
-from src.plugin_system.testing.loading import load_config_defaults, load_harness_spec
+from src.plugin_system.testing.loading import build_full_config, load_harness_spec
 from src.plugin_system.testing.sizes import resolve_test_sizes
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
@@ -81,9 +81,7 @@ def test_plugin_renders_across_sizes_and_screens(plugin_id: str) -> None:
     plugin_dir = _PLUGINS[plugin_id]
     spec = load_harness_spec(plugin_dir)
 
-    config = {"enabled": True}
-    config.update(load_config_defaults(plugin_dir))
-    config.update(spec.get("config", {}))
+    config = build_full_config(plugin_dir, spec)
 
     # Sizes: LEDMATRIX_TEST_SIZES env (test on real hardware) wins, then the
     # plugin's own harness.json "sizes", else the default representative sample.

test/test_adaptive_images.py

@@ -0,0 +1,208 @@
+"""Tests for adaptive image fitting (src/adaptive_images.py) and the
+LayoutContext image cache."""
+
+import pytest
+from PIL import Image
+
+from src.adaptive_images import (
+    RESAMPLE_LANCZOS,
+    RESAMPLE_NEAREST,
+    ImageFitResult,
+    draw_fitted_image,
+    fit_image,
+)
+from src.adaptive_layout import LayoutContext, Region
+from src.font_manager import FontManager
+
+
+@pytest.fixture(scope="module")
+def font_manager():
+    return FontManager({})
+
+
+@pytest.fixture
+def ctx(font_manager):
+    return LayoutContext(128, 32, font_manager)
+
+
+def _solid(w, h, color=(255, 0, 0, 255)):
+    return Image.new("RGBA", (w, h), color)
+
+
+def _padded_logo(ink_w=10, ink_h=10, pad=10):
+    """Transparent canvas with a solid ink block in the middle — models a
+    logo shipped with generous transparent padding."""
+    img = Image.new("RGBA", (ink_w + 2 * pad, ink_h + 2 * pad), (0, 0, 0, 0))
+    img.paste(_solid(ink_w, ink_h), (pad, pad))
+    return img
+
+
+class TestFitModes:
+    def test_contain_letterboxes_and_upscales(self):
+        fit = fit_image(_solid(10, 5), (40, 40))
+        assert (fit.width, fit.height) == (40, 20)  # aspect preserved
+        assert fit.scale == 4.0
+
+    def test_contain_no_upscale(self):
+        fit = fit_image(_solid(10, 5), (40, 40), upscale=False)
+        assert (fit.width, fit.height) == (10, 5)
+        assert fit.scale == 1.0
+
+    def test_cover_fills_and_crops(self):
+        fit = fit_image(_solid(10, 20), (40, 40), mode="cover")
+        assert (fit.width, fit.height) == (40, 40)
+
+    def test_cover_top_anchor(self):
+        # top half red, bottom half blue; cover-crop a wide box with top anchor
+        img = Image.new("RGBA", (20, 40), (0, 0, 255, 255))
+        img.paste(_solid(20, 20, (255, 0, 0, 255)), (0, 0))
+        fit = fit_image(img, (20, 20), mode="cover", anchor="top")
+        assert fit.image.getpixel((10, 5))[:3] == (255, 0, 0)  # kept the top
+
+    def test_fill_height_matches_box_height(self):
+        fit = fit_image(_solid(10, 10), (64, 32), mode="fill_height")
+        assert fit.height == 32 and fit.width == 32
+
+    def test_fill_height_capped_by_width(self):
+        # very wide source: height-fill would overflow the box width
+        fit = fit_image(_solid(100, 10), (40, 32), mode="fill_height")
+        assert fit.width <= 40
+
+    def test_stretch_exact(self):
+        fit = fit_image(_solid(3, 7), (25, 13), mode="stretch")
+        assert (fit.width, fit.height) == (25, 13)
+
+    def test_crop_to_ink(self):
+        fit = fit_image(_padded_logo(), (30, 30), crop_to_ink=True)
+        # 10x10 ink upscaled to fill 30x30 (padding would have kept it small)
+        assert (fit.width, fit.height) == (30, 30)
+        no_crop = fit_image(_padded_logo(), (30, 30), crop_to_ink=False)
+        assert no_crop.width == 30  # whole padded canvas scaled instead
+
+    def test_fully_transparent_source(self):
+        img = Image.new("RGBA", (10, 10), (0, 0, 0, 0))
+        fit = fit_image(img, (20, 20), crop_to_ink=True)
+        assert fit.is_empty
+
+    def test_degenerate_box(self):
+        assert fit_image(_solid(10, 10), (0, 20)).is_empty
+        assert fit_image(_solid(10, 10), Region(0, 0, 20, 0)).is_empty
+
+    def test_output_always_rgba(self):
+        rgb = Image.new("RGB", (10, 10), (1, 2, 3))
+        assert fit_image(rgb, (20, 20)).image.mode == "RGBA"
+
+    def test_nearest_keeps_hard_edges(self):
+        # 2x2 checker scaled 8x: NEAREST keeps pure colors, LANCZOS blends
+        img = Image.new("RGBA", (2, 2), (0, 0, 0, 255))
+        img.putpixel((0, 0), (255, 255, 255, 255))
+        near = fit_image(img, (16, 16), mode="stretch", resample=RESAMPLE_NEAREST)
+        colors = {near.image.getpixel((x, y))[:3] for x in range(16) for y in range(16)}
+        assert colors == {(255, 255, 255), (0, 0, 0)}
+
+    def test_unknown_mode_raises(self):
+        with pytest.raises(ValueError):
+            fit_image(_solid(4, 4), (8, 8), mode="tile")
+
+
+class TestDrawFittedImage:
+    class _DM:
+        def __init__(self, w=64, h=32):
+            self.image = Image.new("RGB", (w, h), (0, 0, 0))
+
+    def test_pastes_aligned_in_region(self):
+        dm = self._DM()
+        box = Region(10, 4, 20, 20)
+        fit = fit_image(_solid(10, 10), box)
+        xy = draw_fitted_image(dm, fit, box)
+        assert xy == box.align_xy(fit.width, fit.height)
+        assert dm.image.getpixel((xy[0] + 1, xy[1] + 1)) == (255, 0, 0)
+
+    def test_offset_translates(self):
+        dm = self._DM()
+        box = Region(0, 0, 20, 20)
+        fit = fit_image(_solid(10, 10), box)
+        x, y = draw_fitted_image(dm, fit, box, align="left", valign="top",
+                                 offset=(3, 5))
+        assert (x, y) == (3, 5)
+
+    def test_empty_fit_noops(self):
+        dm = self._DM()
+        fit = fit_image(_solid(10, 10), (0, 0))
+        assert draw_fitted_image(dm, fit, Region(0, 0, 10, 10)) is None
+
+
+class TestContextImageCache:
+    def test_size_keyed_hit_and_miss(self, ctx):
+        img = _solid(10, 10)
+        a = ctx.fit_image(img, (20, 20), cache_key="logo:A")
+        assert ctx.fit_image(img, (20, 20), cache_key="logo:A") is a
+        b = ctx.fit_image(img, (30, 30), cache_key="logo:A")
+        assert b is not a and b.width == 30  # different box size = new entry
+
+    def test_id_keyed_default(self, ctx):
+        img = _solid(10, 10)
+        a = ctx.fit_image(img, (20, 20))
+        assert ctx.fit_image(img, (20, 20)) is a
+
+    def test_id_safety_pins_source(self, ctx):
+        # id()-keyed entries must pin the source image so a recycled id
+        # can't alias a dead image's cache entry.
+        img = _solid(10, 10)
+        ctx.fit_image(img, (20, 20))
+        pinned = [entry[1] for entry in ctx._image_cache.values()]
+        assert img in pinned
+
+    def test_cache_key_entries_do_not_pin(self, ctx):
+        img = _solid(10, 10)
+        ctx.fit_image(img, (20, 20), cache_key="logo:X")
+        key = next(k for k in ctx._image_cache if k[1] == "logo:X")
+        assert ctx._image_cache[key][1] is None
+
+    def test_lru_eviction(self, ctx):
+        for i in range(ctx._IMAGE_CACHE_MAX + 5):
+            ctx.fit_image(_solid(4, 4), (8, 8), cache_key=f"k{i}")
+        assert len(ctx._image_cache) == ctx._IMAGE_CACHE_MAX
+        assert not any(k[1] == "k0" for k in ctx._image_cache)  # oldest evicted
+
+    def test_clear_cache_clears_images(self, ctx):
+        ctx.fit_image(_solid(4, 4), (8, 8), cache_key="k")
+        ctx.clear_cache()
+        assert len(ctx._image_cache) == 0
+
+
+class TestBasePluginDrawImage:
+    def test_draw_image_end_to_end(self):
+        from src.plugin_system.base_plugin import BasePlugin
+        from src.plugin_system.testing.mocks import (
+            MockCacheManager, MockDisplayManager, MockPluginManager,
+        )
+
+        class _P(BasePlugin):
+            def update(self):
+                pass
+
+            def display(self, force_clear=False):
+                pass
+
+        plugin = _P("t", {}, MockDisplayManager(64, 32),
+                    MockCacheManager(), MockPluginManager())
+        logo = _padded_logo()
+        box = plugin.layout.bounds.left_col(32)
+        ifit = plugin.draw_image(logo, box, mode="fill_height",
+                                 crop_to_ink=True, cache_key="logo:T")
+        assert ifit.height == 32
+        # pasted onto the mock's canvas
+        assert plugin.display_manager.image.getpixel((16, 16)) != (0, 0, 0)
+
+
+class TestResultIndependence:
+    def test_same_size_fit_never_aliases_the_source(self):
+        """LayoutContext caches ImageFitResults — an aliased image would let
+        later mutations of the source corrupt cached fits (or vice versa)."""
+        from PIL import ImageDraw
+        src = Image.new("RGBA", (20, 20), (255, 0, 0, 255))
+        fit = fit_image(src, (20, 20))
+        assert fit.image is not src
+        ImageDraw.Draw(src).rectangle([0, 0, 19, 19], fill=(0, 255, 0, 255))
+        assert fit.image.getpixel((5, 5)) == (255, 0, 0, 255)

test/test_adaptive_layout.py

@@ -0,0 +1,454 @@
+"""Tests for the adaptive layout system (src/adaptive_layout.py)."""
+
+import pytest
+
+from src.adaptive_layout import (
+    DEFAULT_DESIGN_SIZE,
+    LADDER_ARCADE,
+    LADDER_GRID,
+    LayoutContext,
+    Region,
+    draw_fitted_text,
+    measure_font_crispness,
+    measure_ink,
+    media_row,
+    scoreboard_regions,
+)
+from src.plugin_system.testing.sizes import DEFAULT_TEST_SIZES
+from src.font_manager import FontManager
+
+
+@pytest.fixture(scope="module")
+def font_manager():
+    """Real FontManager over assets/fonts — the ladders depend on it."""
+    return FontManager({})
+
+
+@pytest.fixture
+def ctx(font_manager):
+    return LayoutContext(128, 32, font_manager)
+
+
+class TestRegion:
+    """Pure integer rect algebra."""
+
+    def test_bands_partition_without_overlap(self):
+        r = Region(0, 0, 128, 32)
+        top = r.top_band(7)
+        bottom = r.bottom_band(7)
+        middle = r.middle(7, 7)
+        assert top.bottom == middle.y
+        assert middle.bottom == bottom.y
+        assert top.h + middle.h + bottom.h == r.h
+
+    def test_bands_clamp_on_short_panel(self):
+        # The classic failure: y=1 top band and y=height-7 bottom band
+        # overlapping on a short panel. Bands can't exceed the region.
+        r = Region(0, 0, 32, 8)
+        assert r.top_band(16).h == 8
+        assert r.bottom_band(16).h == 8
+        assert r.middle(8, 8).h == 0  # degenerate, never negative
+
+    def test_split_v_weights_sum_to_height(self):
+        r = Region(0, 0, 64, 33)
+        rows = r.split_v(3, 1, 1, gap=1)
+        assert len(rows) == 3
+        assert sum(row.h for row in rows) == 33 - 2  # two 1px gaps
+        assert rows[0].h > rows[1].h
+        assert rows[-1].bottom == r.bottom
+
+    def test_split_h_columns_advance(self):
+        r = Region(0, 0, 100, 32)
+        cols = r.split_h(1, 1, gap=2)
+        assert cols[0].right + 2 == cols[1].x
+        assert cols[1].right == r.right
+
+    def test_degenerate_sizes_never_negative(self):
+        for w, h in [(8, 8), (32, 16), (1, 1)]:
+            r = Region(0, 0, w, h).inset(4)
+            assert r.w >= 0 and r.h >= 0
+            for sub in r.split_v(1, 1) + r.split_h(1, 1, gap=3):
+                assert sub.w >= 0 and sub.h >= 0
+
+    def test_align_xy(self):
+        r = Region(10, 10, 100, 20)
+        assert r.align_xy(20, 10, "left", "top") == (10, 10)
+        assert r.align_xy(20, 10, "right", "bottom") == (90, 20)
+        assert r.align_xy(20, 10) == (50, 15)
+
+    def test_left_right_cols(self):
+        r = Region(0, 0, 128, 32)
+        assert r.left_col(32) == Region(0, 0, 32, 32)
+        assert r.right_col(32) == Region(96, 0, 32, 32)
+
+    def test_offset_translates_without_resizing(self):
+        r = Region(5, 5, 20, 10).offset(3, -2)
+        assert r == Region(8, 3, 20, 10)
+
+
+class TestScoreboardRegions:
+    @pytest.mark.parametrize("w,h", DEFAULT_TEST_SIZES + [(8, 8)])
+    def test_invariants_at_all_sizes(self, w, h):
+        regs = scoreboard_regions(Region(0, 0, w, h))
+        assert regs.logo_slot <= min(h, w // 2)
+        # slots hug the edges and never overlap the center column
+        assert regs.away_slot.x == 0 and regs.home_slot.right == w
+        assert regs.away_slot.right <= regs.center_col.x or regs.center_col.w == 0
+        assert regs.center_col.right <= regs.home_slot.x or regs.center_col.w == 0
+        # bands stack inside the center column without overlap
+        assert regs.status_band.bottom <= regs.score_area.y or regs.score_area.h == 0
+        assert regs.score_area.bottom <= regs.detail_band.y or regs.score_area.h == 0
+        # everything within bounds, nothing negative
+        for reg in (regs.away_slot, regs.home_slot, regs.center_col,
+                    regs.status_band, regs.score_area, regs.detail_band,
+                    regs.bottom_left, regs.bottom_right):
+            assert reg.w >= 0 and reg.h >= 0
+            assert reg.x >= 0 and reg.y >= 0
+            assert reg.right <= w and reg.bottom <= h
+
+    @pytest.mark.parametrize("w,h", [(96, 48), (128, 64), (256, 128), (64, 32), (128, 96)])
+    def test_2to1_aspect_gets_a_center_reserve(self, w, h):
+        """These sizes are all <= 2:1 aspect, where the raw min(h, w//2)
+        formula claims the entire width for logos and leaves zero pixels
+        for a center column — the bug this reserve exists to fix."""
+        regs = scoreboard_regions(Region(0, 0, w, h))
+        assert regs.center_col.w >= int(w * 0.15) - 1  # -1 for int() rounding
+
+    @pytest.mark.parametrize("w,h", [(128, 32), (192, 48), (256, 32)])
+    def test_wide_panels_unaffected_by_center_reserve(self, w, h):
+        """Wide (>= ~4:1) panels already have height as the tighter
+        constraint, so the center reserve must be a no-op there — the
+        design-size baseline's proportions shouldn't shift."""
+        regs = scoreboard_regions(Region(0, 0, w, h))
+        assert regs.logo_slot == min(h, w // 2)
+
+    def test_center_reserve_fraction_is_configurable(self):
+        # min_center_design_px=0 isolates the fraction term (otherwise the
+        # scaled absolute floor can dominate and mask a fraction change).
+        regs_default = scoreboard_regions(Region(0, 0, 128, 64), min_center_design_px=0)
+        regs_wider = scoreboard_regions(Region(0, 0, 128, 64), min_center_fraction=0.5,
+                                        min_center_design_px=0)
+        assert regs_wider.center_col.w > regs_default.center_col.w
+        assert regs_wider.logo_slot < regs_default.logo_slot
+
+    def test_score_bleed_extends_past_center_col(self):
+        regs = scoreboard_regions(Region(0, 0, 128, 64), score_bleed_fraction=0.5)
+        assert regs.score_area.w > regs.center_col.w
+        assert regs.score_area.x < regs.center_col.x
+        assert regs.score_area.right > regs.center_col.right
+
+    def test_score_bleed_zero_matches_center_col(self):
+        regs = scoreboard_regions(Region(0, 0, 128, 64), score_bleed_fraction=0.0)
+        assert regs.score_area.w == regs.center_col.w
+        assert regs.score_area.x == regs.center_col.x
+
+    @pytest.mark.parametrize("w,h", [(64, 32), (96, 48), (128, 64), (256, 128), (128, 96)])
+    def test_score_never_needs_ellipsis_for_a_short_score(self, w, h, font_manager):
+        """The concrete regression this whole reserve/bleed system exists to
+        prevent: a real game score like '17-21' must always render in full,
+        never truncated, at every 2:1-or-tighter aspect ratio in the sample."""
+        ctx = LayoutContext(w, h, font_manager)
+        regs = scoreboard_regions(Region(0, 0, w, h), ctx=ctx)
+        height_scale = h / 32.0
+        fit = ctx.fit_text_proportional("17-21", regs.score_area, base_size_px=10,
+                                        ladder=LADDER_ARCADE, scale=height_scale)
+        assert fit.text == "17-21"
+        assert fit.fits
+
+    def test_ctx_scales_band_heights(self, font_manager):
+        small = scoreboard_regions(Region(0, 0, 128, 32),
+                                   ctx=LayoutContext(128, 32, font_manager))
+        big = scoreboard_regions(Region(0, 0, 256, 64),
+                                 ctx=LayoutContext(256, 64, font_manager))
+        assert big.status_band.h > small.status_band.h
+
+    def test_works_on_offset_card_region(self):
+        card = Region(10, 4, 100, 24)
+        regs = scoreboard_regions(card)
+        assert regs.away_slot.x == 10
+        assert regs.home_slot.right == card.right
+
+
+class TestMediaRow:
+    def test_square_art_plus_body(self):
+        row = media_row(Region(0, 0, 128, 32))
+        assert row.art == Region(0, 0, 32, 32)
+        assert row.body.x == 32 + 2 and row.body.right == 128
+
+    def test_non_square(self):
+        row = media_row(Region(0, 0, 100, 20), square=False, gap=4)
+        assert row.art.w == 50
+        assert row.body.x == 54
+
+    def test_narrow_panel_clamps(self):
+        row = media_row(Region(0, 0, 16, 32))
+        assert row.art.w == 16 and row.body.w == 0
+
+
+class TestLayoutContext:
+    def test_tiers(self, font_manager):
+        assert LayoutContext(128, 32, font_manager).tier == "sm"
+        assert LayoutContext(96, 48, font_manager).tier == "md"
+        assert LayoutContext(128, 64, font_manager).tier == "lg"
+        assert LayoutContext(64, 16, font_manager).tier == "xs"
+        assert LayoutContext(256, 128, font_manager).tier == "xl"
+
+    def test_wide_short_flag(self, font_manager):
+        assert LayoutContext(128, 32, font_manager).is_wide_short
+        assert not LayoutContext(128, 64, font_manager).is_wide_short
+
+    def test_scale_against_design_size(self, font_manager):
+        assert LayoutContext(128, 32, font_manager).scale == 1.0
+        assert LayoutContext(256, 64, font_manager).scale == 2.0
+        # min() of the two axes: don't overscale the constrained one
+        assert LayoutContext(256, 32, font_manager).scale == 1.0
+        assert DEFAULT_DESIGN_SIZE == (128, 32)
+
+    def test_px_scales_and_clamps(self, font_manager):
+        big = LayoutContext(256, 64, font_manager)
+        assert big.px(4) == 8
+        assert big.px(4, maximum=6) == 6
+        tiny = LayoutContext(32, 16, font_manager)
+        assert tiny.px(4, minimum=2) == 2
+
+    def test_by_tier_nearest_at_or_below(self, font_manager):
+        mapping = {"sm": 10, "lg": 18}
+        assert LayoutContext(128, 32, font_manager).by_tier(mapping) == 10
+        assert LayoutContext(96, 48, font_manager).by_tier(mapping) == 10  # md -> sm
+        assert LayoutContext(128, 64, font_manager).by_tier(mapping) == 18
+        assert LayoutContext(256, 128, font_manager).by_tier(mapping) == 18  # xl -> lg
+        # nothing at-or-below: fall forward to smallest defined above
+        assert LayoutContext(64, 16, font_manager).by_tier(mapping) == 10
+
+
+class TestFontFitting:
+    def test_ladder_monotonic(self, font_manager):
+        """Each ladder rung must render no taller than the one before it."""
+        for ladder in (LADDER_GRID, LADDER_ARCADE):
+            heights = []
+            for step in ladder:
+                font = font_manager.get_font(step.family, step.size_px)
+                heights.append(measure_ink("Ay0", font)[1])
+            assert heights == sorted(heights, reverse=True), (
+                f"ladder not monotonically shrinking: {heights}")
+
+    def test_ladder_grid_is_crisp(self, font_manager):
+        """LADDER_GRID's BDF fonts are real bitmaps — always 0% antialiased."""
+        for step in LADDER_GRID:
+            font = font_manager.get_font(step.family, step.size_px)
+            assert measure_font_crispness(font, "Ay0") == 0.0
+
+    def test_ladder_arcade_is_crisp(self, font_manager):
+        """PressStart2P only rasterizes without antialiasing at exact
+        multiples of its 8px design grid — every LADDER_ARCADE rung must
+        land on one."""
+        for step in LADDER_ARCADE:
+            assert step.size_px % 8 == 0, f"{step} is not a multiple of 8"
+            font = font_manager.get_font(step.family, step.size_px)
+            assert measure_font_crispness(font, "17-21") == 0.0
+
+    def test_crispness_catches_a_bad_size(self, font_manager):
+        """Sanity check the measurement itself: a known-bad size for a
+        pixel-grid font must NOT read as crisp."""
+        font = font_manager.get_font("press_start", 10)  # not a multiple of 8
+        assert measure_font_crispness(font, "17-21") > 0.1
+
+    def test_fit_text_grows_on_taller_panel(self, font_manager):
+        small = LayoutContext(64, 32, font_manager)
+        large = LayoutContext(128, 64, font_manager)
+        text = "12:34"
+        fit_small = small.fit_text(text, small.bounds, ladder=LADDER_ARCADE)
+        fit_large = large.fit_text(text, large.bounds, ladder=LADDER_ARCADE)
+        assert fit_small.fits and fit_large.fits
+        assert fit_large.size_px > fit_small.size_px
+
+    def test_fit_text_fits_the_box(self, ctx):
+        box = ctx.bounds.inset(1)
+        fit = ctx.fit_text("HELLO WORLD", box)
+        assert fit.fits
+        assert fit.width <= box.w and fit.height <= box.h
+
+    def test_fit_text_ellipsizes_overlong_text(self, font_manager):
+        tiny = LayoutContext(32, 16, font_manager)
+        fit = tiny.fit_text("SUPERCALIFRAGILISTIC", tiny.bounds)
+        assert fit.text != "SUPERCALIFRAGILISTIC"
+        assert fit.text.endswith("…")
+        assert fit.width <= tiny.bounds.w
+
+    def test_fit_text_cached(self, ctx):
+        first = ctx.fit_text("CACHED", ctx.bounds)
+        second = ctx.fit_text("CACHED", ctx.bounds)
+        assert first is second
+        ctx.clear_cache()
+        assert ctx.fit_text("CACHED", ctx.bounds) is not first
+
+    def test_fit_text_proportional_tracks_design_scale(self, font_manager):
+        # design size 128x32, base_size_px=10 (a typical classic score size):
+        # at 2x scale the target is 20px -> nearest LADDER_ARCADE rung <= 20
+        # is 16px, not the largest that merely fits the box (32).
+        ctx = LayoutContext(256, 64, font_manager)  # scale = min(2,2) = 2
+        fit = ctx.fit_text_proportional("17-21", ctx.bounds, base_size_px=10,
+                                        ladder=LADDER_ARCADE)
+        assert fit.size_px == 16
+
+    def test_fit_text_proportional_does_not_exceed_max_fit(self, ctx):
+        # at scale=1 (128x32, the design size itself) the target equals
+        # base_size_px, so proportional should never pick something LARGER
+        # than plain fit_text would for the same box.
+        prop = ctx.fit_text_proportional("17-21", ctx.bounds, base_size_px=10,
+                                         ladder=LADDER_ARCADE)
+        maxed = ctx.fit_text("17-21", ctx.bounds, ladder=LADDER_ARCADE)
+        assert prop.size_px <= maxed.size_px
+
+    def test_fit_text_proportional_floors_at_smallest_rung(self, font_manager):
+        # scale so small the target is below every rung -> use the smallest
+        # rung as a floor rather than refusing to render anything.
+        ctx = LayoutContext(32, 8, font_manager)  # scale = min(32/128, 8/32) = 0.25
+        fit = ctx.fit_text_proportional("HI", ctx.bounds, base_size_px=10,
+                                        ladder=LADDER_ARCADE)
+        assert fit.size_px == min(s.size_px for s in LADDER_ARCADE)
+
+    def test_fit_text_proportional_falls_through_when_target_rung_overflows(self, font_manager):
+        # a long string at the target rung might not fit a narrow box even
+        # though the target size is "correct" -- must fall through to a
+        # smaller rung exactly like fit_text does, not just refuse to fit.
+        ctx = LayoutContext(256, 64, font_manager)
+        narrow_box = Region(0, 0, 40, 64)
+        fit = ctx.fit_text_proportional("A REALLY LONG STRING HERE", narrow_box,
+                                        base_size_px=10, ladder=LADDER_ARCADE)
+        assert fit.fits or fit.text.endswith("…")
+
+    def test_fit_text_proportional_cached(self, ctx):
+        first = ctx.fit_text_proportional("X", ctx.bounds, base_size_px=10)
+        second = ctx.fit_text_proportional("X", ctx.bounds, base_size_px=10)
+        assert first is second
+
+    def test_fit_text_proportional_scale_override(self, font_manager):
+        # 128x64 vs design 128x32: self.scale (min of both axes) is 1.0
+        # since width didn't grow, but a caller whose composition scales by
+        # HEIGHT alone (e.g. logo_slot = min(h, w//2)) should be able to
+        # override the reference scale so text grows with it too.
+        ctx = LayoutContext(128, 64, font_manager)
+        assert ctx.scale == 1.0
+        default_fit = ctx.fit_text_proportional("17-21", ctx.bounds, base_size_px=10,
+                                                 ladder=LADDER_ARCADE)
+        height_scale = 64 / 32  # matches design height
+        scaled_fit = ctx.fit_text_proportional("17-21", ctx.bounds, base_size_px=10,
+                                                ladder=LADDER_ARCADE, scale=height_scale)
+        assert scaled_fit.size_px > default_fit.size_px
+
+    def test_fit_lines_stacks_within_height(self, ctx):
+        box = ctx.bounds
+        lines = ["LINE ONE", "LINE TWO", "LINE THREE"]
+        fit = ctx.fit_lines(lines, box, spacing=1)
+        assert fit.fits
+        assert 3 * fit.line_height + 2 <= box.h
+
+    def test_font_for_rows(self, ctx):
+        fit = ctx.font_for_rows(4, 32)
+        assert fit.fits
+        assert 4 * fit.line_height <= 32
+
+    def test_ellipsize_returns_original_when_it_fits(self, ctx):
+        font = ctx.font_manager.get_font("4x6", 6)
+        assert ctx.ellipsize("HI", font, 1000) == "HI"
+
+
+class TestDrawFittedText:
+    def test_draws_within_region(self, ctx):
+        calls = []
+
+        class _DM:
+            def draw_text(self, text, x=None, y=None, color=None, font=None):
+                calls.append((text, x, y))
+
+        box = Region(10, 4, 100, 24)
+        fit = ctx.fit_text("SCORE", box)
+        draw_fitted_text(_DM(), fit, box)
+        text, x, y = calls[0]
+        assert text == "SCORE"
+        assert box.x <= x <= box.right - fit.width
+        # the ink (y + y_offset .. + height) must land inside the box
+        assert box.y <= y + fit.y_offset
+        assert y + fit.y_offset + fit.height <= box.bottom
+
+
+class TestBasePluginIntegration:
+    def test_layout_property_and_draw_fit(self):
+        from src.plugin_system.base_plugin import BasePlugin
+        from src.plugin_system.testing.mocks import (
+            MockCacheManager, MockDisplayManager, MockPluginManager,
+        )
+
+        class _Plugin(BasePlugin):
+            def update(self):
+                pass
+
+            def display(self, force_clear=False):
+                pass
+
+        plugin = _Plugin("test-plugin", {}, MockDisplayManager(96, 48),
+                         MockCacheManager(), MockPluginManager())
+        assert (plugin.layout.width, plugin.layout.height) == (96, 48)
+        assert plugin.layout is plugin.layout  # cached
+        fit = plugin.draw_fit("HELLO", plugin.layout.bounds.inset(1))
+        assert fit.fits
+        assert plugin.display_manager.draw_calls  # actually drew
+
+    def test_layout_rebuilds_on_size_change(self):
+        from src.plugin_system.base_plugin import BasePlugin
+        from src.plugin_system.testing.mocks import (
+            MockCacheManager, MockDisplayManager, MockPluginManager,
+        )
+
+        class _Plugin(BasePlugin):
+            def update(self):
+                pass
+
+            def display(self, force_clear=False):
+                pass
+
+        dm = MockDisplayManager(128, 32)
+        plugin = _Plugin("test-plugin", {}, dm,
+                         MockCacheManager(), MockPluginManager())
+        assert plugin.layout.tier == "sm"
+        dm.width, dm.height = 128, 64
+        assert plugin.layout.tier == "lg"
+
+    def test_design_size_from_manifest(self):
+        from src.plugin_system.base_plugin import BasePlugin
+        from src.plugin_system.testing.mocks import (
+            MockCacheManager, MockDisplayManager, MockPluginManager,
+        )
+
+        class _Plugin(BasePlugin):
+            def update(self):
+                pass
+
+            def display(self, force_clear=False):
+                pass
+
+        pm = MockPluginManager()
+        pm.plugin_manifests["test-plugin"] = {
+            "display": {"design_size": {"width": 64, "height": 32}}
+        }
+        plugin = _Plugin("test-plugin", {}, MockDisplayManager(128, 64),
+                         MockCacheManager(), pm)
+        assert plugin.layout.design_size == (64, 32)
+        assert plugin.layout.scale == 2.0
+
+
+class TestFitCacheBound:
+    def test_fit_cache_is_lru_bounded(self, ctx):
+        """A plugin fitting changing text (live game clock, ticker) on a
+        24/7 service must not grow the fit cache without bound."""
+        for i in range(ctx._FIT_CACHE_MAX + 100):
+            ctx.fit_text(f"tick {i}", Region(0, 0, 100, 20))
+        assert len(ctx._fit_cache) <= ctx._FIT_CACHE_MAX
+
+    def test_lru_keeps_recent_entries_hot(self, ctx):
+        hot = ctx.fit_text("stay hot", Region(0, 0, 100, 20))
+        for i in range(ctx._FIT_CACHE_MAX - 1):
+            ctx.fit_text(f"cold {i}", Region(0, 0, 100, 20))
+            ctx.fit_text("stay hot", Region(0, 0, 100, 20))  # keep touching it
+        assert ctx.fit_text("stay hot", Region(0, 0, 100, 20)) is hot

test/test_async_plugin_updates.py

@@ -0,0 +1,270 @@
+"""Tests for asynchronous plugin updates (plugin_manager background worker).
+
+The invariants that keep this change safe:
+1. run_scheduled_updates returns immediately — a slow update() can never
+   again freeze the render loop (the original defect: 30s scroll freezes).
+2. A plugin's update() and display() are NEVER concurrent — the per-plugin
+   lock makes the old implicit no-overlap guarantee explicit, and it stays
+   held through PluginExecutor's own timeout: a lingering, still-running
+   update() keeps the lock even after PluginExecutor gives up waiting on it.
+3. Failure/timeout bookkeeping is unchanged (same executor, same
+   _record_update_failure path, same last-update stamping).
+4. The kill switch (plugin_system.synchronous_updates) restores the
+   inline path exactly.
+"""
+
+import os
+import sys
+import threading
+import time
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.plugin_system.plugin_manager import PluginManager  # noqa: E402
+from src.plugin_system.plugin_state import PluginState  # noqa: E402
+
+
+class SlowPlugin:
+    """Fake plugin whose update() sleeps and records overlap violations."""
+
+    def __init__(self, update_seconds=0.5):
+        self.enabled = True
+        self.update_seconds = update_seconds
+        self.update_calls = 0
+        self.display_calls = 0
+        self.in_update = False
+        self.overlap_detected = False
+
+    def update(self):
+        self.in_update = True
+        self.update_calls += 1
+        time.sleep(self.update_seconds)
+        self.in_update = False
+        return True
+
+    def display(self, force_clear=False):
+        if self.in_update:
+            self.overlap_detected = True
+        self.display_calls += 1
+        return True
+
+
+@pytest.fixture
+def pm(tmp_path):
+    manager = PluginManager(plugins_dir=str(tmp_path), config_manager=None,
+                            display_manager=None, cache_manager=None)
+    yield manager
+    manager.stop_update_worker()
+
+
+def _install(pm, plugin, plugin_id="slow-plugin"):
+    pm.plugins[plugin_id] = plugin
+    pm._update_interval_cache[plugin_id] = 0.01  # always due
+    # load_plugin normally registers state; can_execute() gates on it
+    pm.state_manager.set_state(plugin_id, PluginState.ENABLED)
+    return plugin_id
+
+
+class TestSchedulerNonBlocking:
+    def test_run_scheduled_updates_returns_immediately(self, pm):
+        plugin_id = _install(pm, SlowPlugin(update_seconds=2.0))
+        start = time.monotonic()
+        pm.run_scheduled_updates()
+        elapsed = time.monotonic() - start
+        assert elapsed < 0.1, f"scheduler blocked for {elapsed:.2f}s"
+        # the update actually runs in the background
+        deadline = time.monotonic() + 5
+        while pm.plugins[plugin_id].update_calls == 0 and time.monotonic() < deadline:
+            time.sleep(0.05)
+        assert pm.plugins[plugin_id].update_calls == 1
+
+    def test_no_double_enqueue_while_pending(self, pm):
+        plugin_id = _install(pm, SlowPlugin(update_seconds=0.8))
+        for _ in range(20):
+            pm.run_scheduled_updates()
+            time.sleep(0.01)
+        time.sleep(1.5)  # let the single queued update finish
+        assert pm.plugins[plugin_id].update_calls == 1
+
+
+class TestUpdateDisplayExclusion:
+    def test_display_lock_held_during_update(self, pm):
+        plugin_id = _install(pm, SlowPlugin(update_seconds=0.6))
+        pm.run_scheduled_updates()
+        # give the worker a moment to take the lock and enter update()
+        deadline = time.monotonic() + 2
+        while not pm.plugins[plugin_id].in_update and time.monotonic() < deadline:
+            time.sleep(0.01)
+        lock = pm.get_plugin_lock(plugin_id)
+        assert lock.acquire(blocking=False) is False, \
+            "lock must be held while update() runs"
+        # and released afterwards
+        deadline = time.monotonic() + 3
+        while pm.plugins[plugin_id].in_update and time.monotonic() < deadline:
+            time.sleep(0.05)
+        time.sleep(0.1)
+        assert lock.acquire(blocking=False) is True
+        lock.release()
+
+    def test_no_overlap_under_hammering_display_loop(self, pm):
+        """Simulate the render loop's try-lock display pattern at high rate
+        while updates fire — the plugin itself asserts no overlap."""
+        plugin = SlowPlugin(update_seconds=0.15)
+        plugin_id = _install(pm, plugin)
+        stop = threading.Event()
+
+        def render_loop():
+            while not stop.is_set():
+                lock = pm.get_plugin_lock(plugin_id)
+                if lock.acquire(blocking=False):
+                    try:
+                        plugin.display()
+                    finally:
+                        lock.release()
+                time.sleep(0.002)
+
+        renderer = threading.Thread(target=render_loop, daemon=True)
+        renderer.start()
+        try:
+            for _ in range(6):
+                pm.plugin_last_update.pop(plugin_id, None)  # force due
+                pm.run_scheduled_updates()
+                time.sleep(0.3)
+        finally:
+            stop.set()
+            renderer.join(timeout=2)
+        assert plugin.update_calls >= 3
+        assert plugin.display_calls > 10
+        assert plugin.overlap_detected is False
+
+    def test_lock_held_through_timeout_until_real_update_finishes(self, pm):
+        """PluginExecutor's join(timeout) can return before the real
+        update() call does -- the lingering daemon thread keeps running it.
+        The plugin's lock must stay held for that whole real duration, not
+        just PluginExecutor's bounded wait, or display() could run
+        concurrently with a still-executing update()."""
+        plugin = SlowPlugin(update_seconds=0.3)
+        plugin_id = _install(pm, plugin)
+        pm.plugin_executor.default_timeout = 0.05  # times out well before
+                                                    # update_seconds elapses
+
+        pm.run_scheduled_updates()
+
+        deadline = time.monotonic() + 2
+        while not plugin.in_update and time.monotonic() < deadline:
+            time.sleep(0.01)
+        assert plugin.in_update is True
+
+        # PluginExecutor's own timeout has now elapsed, but the real
+        # update() (0.3s) is still running in its lingering daemon thread.
+        time.sleep(0.15)
+        assert plugin.in_update is True, "test setup: update should still be running"
+        lock = pm.get_plugin_lock(plugin_id)
+        assert lock.acquire(blocking=False) is False, \
+            "lock must stay held through PluginExecutor's timeout while the real update() runs"
+
+        # Once the real update() genuinely finishes, the lock is released.
+        deadline = time.monotonic() + 2
+        while plugin.in_update and time.monotonic() < deadline:
+            time.sleep(0.02)
+        time.sleep(0.1)
+        assert lock.acquire(blocking=False) is True
+        lock.release()
+
+    def test_state_returns_to_enabled_after_update(self, pm):
+        """RUNNING is set at enqueue (blocks re-entry via can_execute) and
+        must return to an executable state once the update finishes."""
+        plugin_id = _install(pm, SlowPlugin(update_seconds=0.1))
+        pm.run_scheduled_updates()
+        # while queued/running, re-entry is blocked
+        assert pm.state_manager.can_execute(plugin_id) is False
+        deadline = time.monotonic() + 3
+        while time.monotonic() < deadline:
+            if (pm.plugins[plugin_id].update_calls
+                    and pm.state_manager.can_execute(plugin_id)):
+                break
+            time.sleep(0.05)
+        assert pm.plugins[plugin_id].update_calls == 1
+        assert pm.state_manager.can_execute(plugin_id) is True
+
+
+class TestFailurePaths:
+    def test_update_failure_routes_through_failure_bookkeeping(self, pm):
+        class FailingPlugin(SlowPlugin):
+            def update(self):
+                self.update_calls += 1
+                raise RuntimeError("boom")
+
+        plugin_id = _install(pm, FailingPlugin())
+        pm.run_scheduled_updates()
+        deadline = time.monotonic() + 3
+        while pm.plugins[plugin_id].update_calls == 0 and time.monotonic() < deadline:
+            time.sleep(0.05)
+        time.sleep(0.2)
+        # failure stamped so the interval gate holds (no hot retry loop)
+        assert pm.plugin_last_update.get(plugin_id, 0) > 0
+        # lock released after failure
+        assert pm.get_plugin_lock(plugin_id).acquire(blocking=False) is True
+        pm.get_plugin_lock(plugin_id).release()
+
+    def test_unloaded_while_queued_is_harmless(self, pm):
+        """Exercise the public unload_plugin() lifecycle rather than
+        deleting pm.plugins directly: queue the target's update behind a
+        deterministic blocker (occupying the single worker), unload the
+        target while its item still sits queued, then release the blocker
+        and confirm the target's update never ran and its state stayed
+        unloaded rather than being resurrected to ENABLED."""
+        blocker_event = threading.Event()
+
+        class BlockerPlugin(SlowPlugin):
+            def update(self):
+                self.update_calls += 1
+                blocker_event.wait(timeout=5)
+                return True
+
+        blocker_id = _install(pm, BlockerPlugin(), plugin_id="blocker-plugin")
+        target = SlowPlugin(update_seconds=0.05)
+        target_id = _install(pm, target, plugin_id="slow-plugin")
+
+        # Dispatch the blocker first so it occupies the single worker
+        # thread, then enqueue the target behind it -- deterministically
+        # queued, not yet started.
+        pm._enqueue_update(blocker_id, time.time())
+        deadline = time.monotonic() + 2
+        while pm.plugins[blocker_id].update_calls == 0 and time.monotonic() < deadline:
+            time.sleep(0.01)
+        assert pm.plugins[blocker_id].update_calls == 1
+
+        pm._enqueue_update(target_id, time.time())
+        assert target_id in pm._pending_updates
+
+        assert pm.unload_plugin(target_id) is True
+        assert target_id not in pm.plugins
+
+        blocker_event.set()  # let the blocker finish; worker moves on to
+                              # the target's queued item
+
+        deadline = time.monotonic() + 3
+        while target_id in pm._pending_updates and time.monotonic() < deadline:
+            time.sleep(0.02)
+
+        assert target.update_calls == 0, "update() must not run for an unloaded plugin"
+        assert target_id not in pm._pending_updates
+        assert pm.state_manager.get_state(target_id) == PluginState.UNLOADED
+
+
+class TestKillSwitch:
+    def test_synchronous_mode_blocks_like_before(self, pm):
+        pm._synchronous_updates = True
+        plugin_id = _install(pm, SlowPlugin(update_seconds=0.4))
+        start = time.monotonic()
+        pm.run_scheduled_updates()
+        elapsed = time.monotonic() - start
+        assert elapsed >= 0.4, "synchronous mode must run inline"
+        assert pm.plugins[plugin_id].update_calls == 1
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-v"]))

test/test_cache_manager.py

@@ -400,3 +400,61 @@ class TestDiskCache:
         assert stats['fetch_count'] == 3
         assert stats['total_fetch_time'] == 1.8
         assert stats['average_fetch_time'] == pytest.approx(0.6, abs=0.01)
+
+
+class TestDiskCacheWriteEconomy:
+    """SD-card wear guards: identical payloads skip the disk, files are
+    compact, and TTL semantics survive the skip (see PR: fix/diskcache-sd-wear)."""
+
+    def test_identical_set_skips_rewrite(self, tmp_path):
+        import os
+        cache = DiskCache(cache_dir=str(tmp_path))
+        cache.set("k", {"data": "v"})
+        path = cache.get_cache_path("k")
+        first = os.stat(path)
+        os.utime(path, (first.st_atime - 100, first.st_mtime - 100))  # age it
+        aged_mtime = os.stat(path).st_mtime
+        ino_before = os.stat(path).st_ino
+        cache.set("k", {"data": "v"})  # identical payload
+        after = os.stat(path)
+        # mtime refreshed (TTL for mtime-based records preserved)...
+        assert after.st_mtime > aged_mtime
+        # ...but the file was NOT rewritten (same inode: no replace happened)
+        assert after.st_ino == ino_before
+
+    def test_changed_data_rewrites(self, tmp_path):
+        import os
+        cache = DiskCache(cache_dir=str(tmp_path))
+        cache.set("k", {"data": "v1"})
+        cache.set("k", {"data": "v2"})
+        assert cache.get("k") == {"data": "v2"}
+
+    def test_clear_resets_digest(self, tmp_path):
+        import os
+        cache = DiskCache(cache_dir=str(tmp_path))
+        cache.set("k", {"data": "v"})
+        cache.clear("k")
+        assert cache.get("k") is None
+        cache.set("k", {"data": "v"})  # same payload after clear must WRITE
+        assert cache.get("k") == {"data": "v"}
+
+    def test_skip_self_heals_when_file_deleted_externally(self, tmp_path):
+        import os
+        cache = DiskCache(cache_dir=str(tmp_path))
+        cache.set("k", {"data": "v"})
+        os.remove(cache.get_cache_path("k"))  # e.g. expiry cleanup
+        cache.set("k", {"data": "v"})  # digest matches but file is gone
+        assert cache.get("k") == {"data": "v"}
+
+    def test_files_are_compact_json(self, tmp_path):
+        cache = DiskCache(cache_dir=str(tmp_path))
+        cache.set("k", {"a": 1, "b": [1, 2, 3]})
+        raw = open(cache.get_cache_path("k")).read()
+        assert "\n" not in raw.strip()  # no indent
+        assert cache.get("k") == {"a": 1, "b": [1, 2, 3]}
+
+    def test_datetime_round_trip_still_works(self, tmp_path):
+        from datetime import datetime
+        cache = DiskCache(cache_dir=str(tmp_path))
+        cache.set("k", {"when": datetime(2026, 7, 12, 10, 30)})
+        assert cache.get("k") == {"when": "2026-07-12T10:30:00"}

test/test_config_load_cache.py

@@ -0,0 +1,129 @@
+"""Tests for load_config's mtime fast path (src/config_manager.py).
+
+load_config used to re-read + re-parse config.json, the template (with a
+recursive migration diff) and secrets on EVERY call — ~30 web request
+handlers call it, some 2-3x per request. The fast path skips all of it
+when the three files' (mtime_ns, size) signatures are unchanged.
+
+The invariant that matters most: cross-process freshness — a save from
+the web process must be picked up by the display process's next load.
+That's guaranteed because the signature is re-stat'd on every call.
+"""
+
+import json
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.config_manager import ConfigManager  # noqa: E402
+
+
+@pytest.fixture
+def mgr(tmp_path):
+    config = tmp_path / "config.json"
+    secrets = tmp_path / "secrets.json"
+    template = tmp_path / "template.json"
+    config.write_text(json.dumps({"display": {"brightness": 90}, "timezone": "UTC"}))
+    secrets.write_text(json.dumps({"weather": {"api_key": "sek"}}))
+    template.write_text(json.dumps({"display": {"brightness": 90}, "timezone": "UTC"}))
+    m = ConfigManager(config_path=str(config), secrets_path=str(secrets))
+    m.template_path = str(template)
+    return m, config, secrets, template
+
+
+def _count_opens(monkeypatch, mgr_paths):
+    """Count open() calls hitting the config files."""
+    counts = {"n": 0}
+    real_open = open
+
+    def counting_open(file, *args, **kwargs):
+        if str(file) in mgr_paths:
+            counts["n"] += 1
+        return real_open(file, *args, **kwargs)
+
+    import builtins
+    monkeypatch.setattr(builtins, "open", counting_open)
+    return counts
+
+
+class TestFastPath:
+    def test_unchanged_files_are_not_reread(self, mgr, monkeypatch):
+        m, config, secrets, template = mgr
+        first = m.load_config()
+        assert first["weather"]["api_key"] == "sek"  # secrets merged
+        counts = _count_opens(monkeypatch, {str(config), str(secrets), str(template)})
+        for _ in range(10):
+            again = m.load_config()
+        assert counts["n"] == 0, "fast path must not re-open any config file"
+        assert again is first  # same aliasing semantics as the full path
+
+    def test_config_change_triggers_reload(self, mgr):
+        m, config, secrets, template = mgr
+        m.load_config()
+        data = json.loads(config.read_text())
+        data["display"]["brightness"] = 55
+        config.write_text(json.dumps(data))
+        os.utime(config, (os.stat(config).st_atime, os.stat(config).st_mtime + 2))
+        assert m.load_config()["display"]["brightness"] == 55
+
+    def test_secrets_change_triggers_reload(self, mgr):
+        m, config, secrets, template = mgr
+        m.load_config()
+        secrets.write_text(json.dumps({"weather": {"api_key": "NEW"}}))
+        os.utime(secrets, (os.stat(secrets).st_atime, os.stat(secrets).st_mtime + 2))
+        assert m.load_config()["weather"]["api_key"] == "NEW"
+
+    def test_template_change_triggers_reload_and_migration(self, mgr):
+        m, config, secrets, template = mgr
+        m.load_config()
+        template.write_text(json.dumps({
+            "display": {"brightness": 90}, "timezone": "UTC",
+            "brand_new_key": {"added": True}}))
+        os.utime(template, (os.stat(template).st_atime, os.stat(template).st_mtime + 2))
+        reloaded = m.load_config()
+        assert reloaded.get("brand_new_key") == {"added": True}
+
+    def test_same_second_edit_detected_via_mtime_ns_or_size(self, mgr):
+        """Coarse-mtime same-second edits: size difference still busts it."""
+        m, config, secrets, template = mgr
+        m.load_config()
+        st = os.stat(config)
+        data = json.loads(config.read_text())
+        data["timezone"] = "America/New_York"  # different byte length
+        config.write_text(json.dumps(data))
+        os.utime(config, (st.st_atime, st.st_mtime))  # force same mtime
+        assert m.load_config()["timezone"] == "America/New_York"
+
+
+class TestSaveCoherence:
+    def test_save_config_then_load_returns_saved_data(self, mgr, monkeypatch):
+        m, config, secrets, template = mgr
+        m.load_config()
+        new = {"display": {"brightness": 42}, "timezone": "UTC",
+               "weather": {"api_key": "sek"}}
+        m.save_config(new)
+        counts = _count_opens(monkeypatch, {str(config), str(secrets), str(template)})
+        loaded = m.load_config()
+        assert loaded["display"]["brightness"] == 42
+        assert loaded["weather"]["api_key"] == "sek"  # secrets survive in memory
+        assert counts["n"] == 0  # signature refreshed by save; no re-read
+
+    def test_cross_process_save_is_picked_up(self, mgr):
+        """Another process writing config.json (different mtime) must bust
+        this process's fast path — the core cross-process guarantee."""
+        m, config, secrets, template = mgr
+        m.load_config()
+        other = ConfigManager(config_path=str(config), secrets_path=str(secrets))
+        other.template_path = str(template)
+        other.load_config()
+        other.save_config({"display": {"brightness": 11}, "timezone": "UTC",
+                           "weather": {"api_key": "sek"}})
+        os.utime(config, (os.stat(config).st_atime, os.stat(config).st_mtime + 2))
+        assert m.load_config()["display"]["brightness"] == 11
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-v"]))

test/test_display_controller_vegas_tick.py

@@ -0,0 +1,88 @@
+"""
+Regression tests for DisplayController._tick_plugin_updates_for_vegas().
+
+PR #299 added logic to detect which plugins actually got fresh data on a
+scheduled-update tick and notify Vegas mode via
+vegas_coordinator.mark_plugin_updated(), so a live score change reaches the
+scroll within seconds instead of waiting for a full cycle. PR #330's
+multi-display sync refactor deleted this method (folding the callback back
+to the plain _tick_plugin_updates(), which reports nothing), silently
+orphaning VegasModeCoordinator.mark_plugin_updated() -- it has had zero
+callers since.
+"""
+
+from typing import Dict, List, Optional
+from unittest.mock import MagicMock
+
+from src.display_controller import DisplayController
+
+
+def _make_controller(updated: Optional[List[str]] = None, vegas_coordinator: Optional[MagicMock] = None) -> DisplayController:
+    dc = object.__new__(DisplayController)
+    dc.plugin_manager = MagicMock()
+    dc.plugin_manager.run_scheduled_updates_with_changes.return_value = list(updated or [])
+    dc.vegas_coordinator = vegas_coordinator
+    return dc
+
+
+class TestTickPluginUpdatesForVegas:
+    def test_marks_only_plugins_whose_timestamp_advanced(self):
+        vc = MagicMock()
+        dc = _make_controller(updated=["stock-news"], vegas_coordinator=vc)
+
+        dc._tick_plugin_updates_for_vegas()
+
+        vc.mark_plugin_updated.assert_called_once_with("stock-news")
+
+    def test_no_advance_marks_nothing(self):
+        vc = MagicMock()
+        dc = _make_controller(updated=[], vegas_coordinator=vc)
+
+        dc._tick_plugin_updates_for_vegas()
+
+        vc.mark_plugin_updated.assert_not_called()
+
+    def test_no_vegas_coordinator_does_not_raise(self):
+        dc = _make_controller(updated=["stock-news"], vegas_coordinator=None)
+
+        dc._tick_plugin_updates_for_vegas()  # must not raise
+
+    def test_mark_plugin_updated_exception_does_not_propagate(self):
+        """One plugin's mark_plugin_updated failing must not stop the tick
+        or crash the update loop it runs in."""
+        vc = MagicMock()
+        vc.mark_plugin_updated.side_effect = [RuntimeError("boom"), None]
+        dc = _make_controller(updated=["a", "b"], vegas_coordinator=vc)
+
+        dc._tick_plugin_updates_for_vegas()  # must not raise
+
+        assert vc.mark_plugin_updated.call_count == 2
+
+
+class TestVegasCoordinatorCallbackWiring:
+    def test_initialize_wires_vegas_aware_tick_as_update_callback(self):
+        """The Vegas coordinator must be given the Vegas-aware
+        _tick_plugin_updates_for_vegas as its update callback, not the plain
+        _tick_plugin_updates() -- that's the exact wiring PR #330 dropped."""
+        dc = object.__new__(DisplayController)
+        dc.config = {"display": {"vegas_scroll": {"enabled": True}}, "sync": {}}
+        dc.display_manager = MagicMock()
+        dc.plugin_manager = MagicMock()
+        dc.sync_manager = MagicMock()
+        dc._check_live_priority = MagicMock()
+        dc._check_vegas_interrupt = MagicMock(return_value=False)
+
+        fake_coordinator = MagicMock()
+
+        import src.display_controller as dc_module
+        original_imported = dc_module._vegas_mode_imported
+        original_class = dc_module.VegasModeCoordinator
+        try:
+            dc_module._vegas_mode_imported = True
+            dc_module.VegasModeCoordinator = MagicMock(return_value=fake_coordinator)
+            dc._initialize_vegas_mode()
+        finally:
+            dc_module._vegas_mode_imported = original_imported
+            dc_module.VegasModeCoordinator = original_class
+
+        fake_coordinator.set_update_callback.assert_called_once_with(dc._tick_plugin_updates_for_vegas)

test/test_display_dirty_tracking.py

@@ -0,0 +1,162 @@
+"""Tests for update_display dirty tracking (src/display_manager.py).
+
+Runs against RGBMatrixEmulator (EMULATOR=true), exercising the REAL
+DisplayManager — not a mock — so the skip logic, its invalidation hooks,
+and the kill switch are verified off-Pi.
+
+The invariants:
+- identical frames are pushed exactly once (SwapOnVSync not re-called)
+- ANY pixel change pushes
+- clear() and set_brightness() invalidate (the two paths that alter panel
+  state outside the digest's view)
+- the kill switch (display.dirty_tracking: false) restores always-push
+"""
+
+import os
+import sys
+import time
+
+os.environ["EMULATOR"] = "true"
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+
+@pytest.fixture(scope="module")
+def dm():
+    """One real DisplayManager on the emulator (it's a process singleton)."""
+    from src.display_manager import DisplayManager
+    DisplayManager._instance = None
+    DisplayManager._initialized = False
+    manager = DisplayManager({
+        "display": {
+            "hardware": {"rows": 32, "cols": 64, "chain_length": 2,
+                         "parallel": 1, "brightness": 90},
+            "runtime": {"gpio_slowdown": 0},
+        },
+    }, suppress_test_pattern=True)
+    yield manager
+    DisplayManager._instance = None
+    DisplayManager._initialized = False
+
+
+class _SwapSpy:
+    """Counts SwapOnVSync calls through the real matrix object."""
+
+    def __init__(self, matrix):
+        self.matrix = matrix
+        self.count = 0
+        self._orig = matrix.SwapOnVSync
+
+    def __enter__(self):
+        def counting(canvas):
+            self.count += 1
+            return self._orig(canvas)
+        self.matrix.SwapOnVSync = counting
+        return self
+
+    def __exit__(self, *exc):
+        self.matrix.SwapOnVSync = self._orig
+
+
+class TestDirtyTracking:
+    def test_identical_frames_push_once(self, dm):
+        dm.draw.rectangle([0, 0, 10, 10], fill=(255, 0, 0))
+        with _SwapSpy(dm.matrix) as spy:
+            dm.update_display()
+            dm.update_display()
+            dm.update_display()
+        assert spy.count == 1
+
+    def test_pixel_change_pushes(self, dm):
+        dm.update_display()
+        with _SwapSpy(dm.matrix) as spy:
+            dm.draw.point((5, 5), fill=(0, 255, 0))
+            dm.update_display()
+            dm.update_display()  # unchanged again
+        assert spy.count == 1
+
+    def test_clear_invalidates(self, dm):
+        dm.draw.rectangle([0, 0, 20, 20], fill=(0, 0, 255))
+        dm.update_display()
+        dm.clear()  # writes to the matrix directly; digest must reset
+        with _SwapSpy(dm.matrix) as spy:
+            dm.update_display()  # black frame after clear must still push
+        assert spy.count == 1
+
+    def test_brightness_change_forces_push(self, dm):
+        dm.draw.rectangle([0, 0, 20, 20], fill=(200, 200, 200))
+        dm.update_display()
+        with _SwapSpy(dm.matrix) as spy:
+            dm.update_display()          # identical -> skipped
+            assert spy.count == 0
+            dm.set_brightness(40)        # dim schedule scenario
+            dm.update_display()          # same image, new brightness -> push
+        assert spy.count == 1
+        dm.set_brightness(90)
+
+    def test_snapshot_still_written_on_skip(self, dm, tmp_path):
+        """The web preview mirror must keep working through skipped panel
+        pushes: _write_snapshot_if_due() still runs on the dirty-tracking
+        skip path and applies its own write/touch policy rather than being
+        bypassed entirely (see src/common/snapshot_policy.py — an unchanged
+        frame is touched, not re-encoded, once TOUCH_INTERVAL elapses)."""
+        dm._snapshot_path = str(tmp_path / "snap.png")
+        dm._last_snapshot_ts = 0.0
+        dm._last_snapshot_touch_ts = 0.0
+        dm._last_snapshot_digest = None
+        dm.draw.rectangle([0, 0, 30, 8], fill=(255, 255, 0))
+        dm.update_display()   # push + snapshot write (first frame)
+        assert os.path.exists(dm._snapshot_path)
+        first_mtime = os.path.getmtime(dm._snapshot_path)
+
+        # Age the write/touch bookkeeping past TOUCH_INTERVAL so the next
+        # identical frame is due for a touch, then push it again: dirty
+        # tracking must skip the panel write, but the snapshot mirror must
+        # still get its mtime bumped so the health check doesn't go stale.
+        from src.common import snapshot_policy
+        stale_ts = time.time() - snapshot_policy.TOUCH_INTERVAL - 1.0
+        dm._last_snapshot_ts = stale_ts
+        dm._last_snapshot_touch_ts = stale_ts
+        with _SwapSpy(dm.matrix) as spy:
+            dm.update_display()  # identical frame -> panel push skipped
+        assert spy.count == 0
+        assert os.path.getmtime(dm._snapshot_path) > first_mtime
+
+
+class TestKillSwitch:
+    def test_dirty_tracking_can_be_disabled(self, dm):
+        dm._dirty_tracking_enabled = False
+        try:
+            dm.draw.rectangle([0, 0, 10, 10], fill=(1, 2, 3))
+            with _SwapSpy(dm.matrix) as spy:
+                dm.update_display()
+                dm.update_display()
+                dm.update_display()
+            assert spy.count == 3  # always-push, exactly the old behavior
+        finally:
+            dm._dirty_tracking_enabled = True
+            dm._last_pushed_digest = None
+
+    def test_config_flag_wires_through(self):
+        from src.display_manager import DisplayManager
+        DisplayManager._instance = None
+        DisplayManager._initialized = False
+        try:
+            manager = DisplayManager({
+                "display": {
+                    "hardware": {"rows": 32, "cols": 64, "chain_length": 1,
+                                 "parallel": 1},
+                    "runtime": {"gpio_slowdown": 0},
+                    "dirty_tracking": False,
+                },
+            }, suppress_test_pattern=True)
+            assert manager._dirty_tracking_enabled is False
+        finally:
+            DisplayManager._instance = None
+            DisplayManager._initialized = False
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-v"]))

test/test_harness_fill.py

@@ -0,0 +1,98 @@
+"""Tests for the harness fill / scale-up check (src/plugin_system/testing/harness.py)."""
+
+from PIL import Image
+
+from src.plugin_system.testing.harness import (
+    RenderResult,
+    check_scale_up,
+    fill_metrics,
+)
+
+
+def _canvas(w, h):
+    return Image.new("RGB", (w, h), (0, 0, 0))
+
+
+def _with_block(w, h, bx, by, bw, bh, color=(255, 255, 255)):
+    img = _canvas(w, h)
+    img.paste(Image.new("RGB", (bw, bh), color), (bx, by))
+    return img
+
+
+def _result(w, h, image):
+    return RenderResult("p", w, h, "mode", image=image)
+
+
+class TestFillMetrics:
+    def test_full_white(self):
+        ex, ey, ink = fill_metrics(Image.new("RGB", (64, 32), (255, 255, 255)))
+        assert (ex, ey, ink) == (1.0, 1.0, 1.0)
+
+    def test_black_is_empty(self):
+        assert fill_metrics(_canvas(64, 32)) == (0.0, 0.0, 0.0)
+
+    def test_corner_dot(self):
+        ex, ey, ink = fill_metrics(_with_block(100, 100, 0, 0, 10, 10))
+        assert ex == 0.1 and ey == 0.1
+        assert ink == 0.01
+
+    def test_centered_half(self):
+        ex, ey, _ = fill_metrics(_with_block(100, 100, 25, 25, 50, 50))
+        assert ex == 0.5 and ey == 0.5
+
+    def test_dim_pixels_ignored(self):
+        img = _canvas(10, 10)
+        img.putpixel((5, 5), (10, 10, 10))  # below the lit threshold
+        assert fill_metrics(img) == (0.0, 0.0, 0.0)
+
+
+class TestCheckScaleUp:
+    def test_not_checked_below_2x(self):
+        # 128x64 vs design 128x32: only height is 2x -> checked on y only;
+        # 128x32 itself: not checked at all
+        r = _result(128, 32, _with_block(128, 32, 0, 0, 10, 10))
+        check_scale_up([r], design_size=(128, 32))
+        assert not r.fill_checked
+
+    def test_warn_mode_records_but_passes(self):
+        # tiny corner content on a 256x128 (2x both axes)
+        r = _result(256, 128, _with_block(256, 128, 0, 0, 20, 20))
+        check_scale_up([r], design_size=(128, 32), strict=False)
+        assert r.fill_checked
+        assert r.fill_ok is None       # warn-only: not a failure
+        assert r.ok                    # still passes
+        assert r.fill_extent[0] < 0.5
+
+    def test_strict_mode_fails_underfill(self):
+        r = _result(256, 128, _with_block(256, 128, 0, 0, 20, 20))
+        check_scale_up([r], design_size=(128, 32), strict=True)
+        assert r.fill_ok is False
+        assert not r.ok
+
+    def test_well_filled_passes_strict(self):
+        r = _result(256, 128, _with_block(256, 128, 10, 10, 200, 100))
+        check_scale_up([r], design_size=(128, 32), strict=True)
+        assert r.fill_ok is True and r.ok
+
+    def test_axis_selection_wide_only(self):
+        # 256x32 vs design 128x32: width is 2x, height is not -> only the
+        # x-extent matters; content spanning full width but few rows passes
+        r = _result(256, 32, _with_block(256, 32, 0, 12, 250, 8))
+        check_scale_up([r], design_size=(128, 32), strict=True)
+        assert r.fill_ok is True
+
+    def test_axis_selection_wide_only_underfill(self):
+        r = _result(256, 32, _with_block(256, 32, 0, 12, 60, 8))
+        check_scale_up([r], design_size=(128, 32), strict=True)
+        assert r.fill_ok is False
+
+    def test_errored_render_skipped(self):
+        r = RenderResult("p", 256, 128, "m", error="boom")
+        check_scale_up([r], design_size=(128, 32), strict=True)
+        assert not r.fill_checked
+
+    def test_custom_design_size(self):
+        # 128x64 with design 64x32 IS 2x both axes
+        r = _result(128, 64, _with_block(128, 64, 0, 0, 10, 10))
+        check_scale_up([r], design_size=(64, 32), strict=False)
+        assert r.fill_checked

test/test_loader_compat_warning.py

@@ -0,0 +1,70 @@
+"""Tests for the plugin loader's advisory version-compatibility warning."""
+
+import logging
+
+import pytest
+
+from src.plugin_system.plugin_loader import PluginLoader
+
+
+@pytest.fixture
+def loader():
+    return PluginLoader(logger=logging.getLogger("test-loader"))
+
+
+def _warnings(caplog):
+    return [r for r in caplog.records if r.levelno == logging.WARNING]
+
+
+class TestParseSemver:
+    def test_basic(self, loader):
+        assert loader._parse_semver("3.1.0") == (3, 1, 0)
+        assert loader._parse_semver("v2.0") == (2, 0, 0)
+        assert loader._parse_semver("2.0.0-beta.1") == (2, 0, 0)
+
+    def test_unparseable(self, loader):
+        assert loader._parse_semver(None) is None
+        assert loader._parse_semver(123) is None
+
+
+class TestWarnIfIncompatible:
+    def test_warns_when_plugin_needs_newer_core(self, loader, caplog, monkeypatch):
+        import src
+        monkeypatch.setattr(src, "__version__", "3.1.0")
+        with caplog.at_level(logging.WARNING, logger="test-loader"):
+            loader._warn_if_incompatible("p", {"min_ledmatrix_version": "9.0.0"})
+        assert len(_warnings(caplog)) == 1
+        assert "9.0.0" in _warnings(caplog)[0].message
+
+    def test_silent_when_compatible(self, loader, caplog, monkeypatch):
+        import src
+        monkeypatch.setattr(src, "__version__", "3.1.0")
+        with caplog.at_level(logging.WARNING, logger="test-loader"):
+            loader._warn_if_incompatible("p", {"min_ledmatrix_version": "2.0.0"})
+        assert not _warnings(caplog)
+
+    def test_silent_when_field_absent(self, loader, caplog):
+        with caplog.at_level(logging.WARNING, logger="test-loader"):
+            loader._warn_if_incompatible("p", {"name": "no version fields"})
+        assert not _warnings(caplog)
+
+    def test_reads_requires_and_versions_spellings(self, loader, caplog, monkeypatch):
+        import src
+        monkeypatch.setattr(src, "__version__", "3.1.0")
+        with caplog.at_level(logging.WARNING, logger="test-loader"):
+            loader._warn_if_incompatible(
+                "a", {"requires": {"min_ledmatrix_version": "9.0.0"}})
+            loader._warn_if_incompatible(
+                "b", {"versions": [{"ledmatrix_min_version": "9.0.0"}]})
+            loader._warn_if_incompatible(
+                "c", {"versions": [{"ledmatrix_min": "9.0.0"}]})
+        assert len(_warnings(caplog)) == 3
+
+    def test_stale_core_version_skips_comparison(self, loader, caplog, monkeypatch):
+        # Anti-spam guard: a core whose __version__ is below the ecosystem
+        # floor must not warn about every plugin.
+        import src
+        monkeypatch.setattr(src, "__version__", "1.0.0")
+        with caplog.at_level(logging.WARNING, logger="test-loader"):
+            loader._warn_if_incompatible("p", {"min_ledmatrix_version": "2.0.0"})
+        assert not _warnings(caplog)

test/test_plugin_system.py

@@ -123,6 +123,14 @@ class TestPluginManager:
 
             pm.run_scheduled_updates(current_time=time.time())
 
+            # Updates now execute on the background worker (the scheduler
+            # returns immediately) — wait for completion before asserting.
+            deadline = time.time() + 5
+            while (plugin_instance.update.call_count == 0
+                   and time.time() < deadline):
+                time.sleep(0.02)
+            pm.stop_update_worker()
+
             plugin_instance.update.assert_called_once()
             assert "test_plugin" in pm.plugin_last_update
             assert pm.state_manager.get_state("test_plugin") == PluginState.ENABLED

test/test_snapshot_policy.py

@@ -0,0 +1,93 @@
+"""Tests for the snapshot write policy (src/common/snapshot_policy.py).
+
+The invariants that matter:
+- unchanged frames are NEVER re-encoded (the old code PNG-encoded identical
+  frames at 5 fps, 24/7)
+- the file mtime never goes stale enough to trip the health check's 60s
+  degraded threshold (api_v3 get_hardware_status)
+- a viewer gets full cadence; no viewer drops to the idle keepalive
+"""
+
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.common.snapshot_policy import (  # noqa: E402
+    IDLE_INTERVAL,
+    TOUCH_INTERVAL,
+    VIEWER_INTERVAL,
+    SnapshotAction,
+    decide,
+)
+
+
+class TestViewerCadence:
+    def test_changed_frame_with_viewer_writes_at_full_rate(self):
+        assert decide(now=100.0, last_write_ts=100.0 - VIEWER_INTERVAL,
+                      last_touch_ts=0, viewer_fresh=True,
+                      frame_changed=True) is SnapshotAction.WRITE
+
+    def test_changed_frame_with_viewer_respects_min_interval(self):
+        assert decide(now=100.0, last_write_ts=100.0 - VIEWER_INTERVAL / 2,
+                      last_touch_ts=100.0, viewer_fresh=True,
+                      frame_changed=True) is SnapshotAction.SKIP
+
+    def test_unchanged_frame_with_viewer_never_writes(self):
+        """A static screen with a viewer must not burn PNG encodes."""
+        assert decide(now=100.0, last_write_ts=90.0, last_touch_ts=90.0,
+                      viewer_fresh=True,
+                      frame_changed=False) is SnapshotAction.SKIP
+
+
+class TestIdleCadence:
+    def test_changed_frame_without_viewer_waits_for_idle_interval(self):
+        assert decide(now=100.0, last_write_ts=100.0 - IDLE_INTERVAL / 2,
+                      last_touch_ts=100.0, viewer_fresh=False,
+                      frame_changed=True) is SnapshotAction.SKIP
+
+    def test_changed_frame_without_viewer_writes_at_idle_rate(self):
+        assert decide(now=100.0, last_write_ts=100.0 - IDLE_INTERVAL,
+                      last_touch_ts=0, viewer_fresh=False,
+                      frame_changed=True) is SnapshotAction.WRITE
+
+
+class TestHealthKeepalive:
+    def test_stale_mtime_gets_touched(self):
+        """Whatever else happens, mtime must be bumped within TOUCH_INTERVAL
+        so the health check (60s threshold) never reads the display as dead."""
+        assert decide(now=100.0, last_write_ts=100.0 - TOUCH_INTERVAL,
+                      last_touch_ts=100.0 - TOUCH_INTERVAL, viewer_fresh=False,
+                      frame_changed=False) is SnapshotAction.TOUCH
+
+    def test_touch_applies_with_viewer_too(self):
+        """Viewer watching a static screen: no writes, but health stays green."""
+        assert decide(now=100.0, last_write_ts=100.0 - TOUCH_INTERVAL - 1,
+                      last_touch_ts=100.0 - TOUCH_INTERVAL - 1, viewer_fresh=True,
+                      frame_changed=False) is SnapshotAction.TOUCH
+
+    def test_recent_touch_suppresses_another(self):
+        assert decide(now=100.0, last_write_ts=0.0,
+                      last_touch_ts=100.0 - TOUCH_INTERVAL / 2, viewer_fresh=False,
+                      frame_changed=False) is SnapshotAction.SKIP
+
+    def test_touch_interval_stays_under_health_threshold(self):
+        """api_v3's hardware status treats snapshot age >= 60s as degraded.
+        Keep a 2x margin so scheduling jitter can't trip it."""
+        assert TOUCH_INTERVAL <= 30
+
+    def test_worst_case_mtime_age_is_bounded(self):
+        """Simulate any interleaving: from any state, within one policy call
+        after TOUCH_INTERVAL elapses, mtime gets refreshed (WRITE or TOUCH)."""
+        for viewer in (True, False):
+            for changed in (True, False):
+                action = decide(now=1000.0, last_write_ts=900.0,
+                                last_touch_ts=900.0, viewer_fresh=viewer,
+                                frame_changed=changed)
+                assert action in (SnapshotAction.WRITE, SnapshotAction.TOUCH)
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-v"]))

test/test_store_update_rollback.py

@@ -0,0 +1,171 @@
+"""Tests for atomic plugin updates (store_manager._reinstall_with_rollback).
+
+Regression for a field data-loss incident: update_plugin's reinstall paths
+(monorepo migration AND routine archive updates) deleted the installed
+plugin BEFORE downloading its replacement — a mid-update network failure
+permanently destroyed the plugin. Seen live: a Pi with broken DNS lost 12
+plugins from one update pass.
+"""
+
+import json
+import os
+import sys
+import threading
+import time
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.plugin_system.store_manager import PluginStoreManager  # noqa: E402
+
+PLUGIN_ID = "rollback-test-plugin"
+
+
+@pytest.fixture
+def store(tmp_path):
+    mgr = PluginStoreManager(plugins_dir=str(tmp_path))
+    plugin_dir = tmp_path / PLUGIN_ID
+    plugin_dir.mkdir()
+    (plugin_dir / "manifest.json").write_text(json.dumps(
+        {"id": PLUGIN_ID, "name": "Rollback Test", "version": "1.0.0"}))
+    (plugin_dir / "manager.py").write_text("# old version marker\n")
+    return mgr, plugin_dir
+
+
+class TestReinstallWithRollback:
+    def test_failed_install_restores_old_version(self, store):
+        """The whole point: a failed download must leave the old install."""
+        mgr, plugin_dir = store
+        with patch.object(mgr, "install_plugin", return_value=False):
+            ok = mgr._reinstall_with_rollback(PLUGIN_ID, plugin_dir)
+        assert ok is False
+        assert plugin_dir.exists()
+        assert "old version marker" in (plugin_dir / "manager.py").read_text()
+        # no aside debris left behind
+        leftovers = [p for p in plugin_dir.parent.iterdir()
+                     if "standalone-backup" in p.name]
+        assert leftovers == []
+
+    def test_install_exception_restores_old_version(self, store):
+        mgr, plugin_dir = store
+        with patch.object(mgr, "install_plugin",
+                          side_effect=RuntimeError("network down")):
+            ok = mgr._reinstall_with_rollback(PLUGIN_ID, plugin_dir)
+        assert ok is False
+        assert plugin_dir.exists()
+        assert "old version marker" in (plugin_dir / "manager.py").read_text()
+
+    def test_successful_install_removes_aside(self, store):
+        mgr, plugin_dir = store
+
+        def fake_install(plugin_id):
+            new_dir = plugin_dir  # same path, new content
+            new_dir.mkdir(exist_ok=True)
+            (new_dir / "manager.py").write_text("# new version\n")
+            (new_dir / "manifest.json").write_text(json.dumps(
+                {"id": PLUGIN_ID, "name": "Rollback Test", "version": "2.0.0"}))
+            return True
+
+        with patch.object(mgr, "install_plugin", side_effect=fake_install):
+            ok = mgr._reinstall_with_rollback(PLUGIN_ID, plugin_dir)
+        assert ok is True
+        assert "new version" in (plugin_dir / "manager.py").read_text()
+        leftovers = [p for p in plugin_dir.parent.iterdir()
+                     if "standalone-backup" in p.name]
+        assert leftovers == []
+
+    def test_partial_download_debris_is_replaced_by_old_version(self, store):
+        """A failed install that left a partial directory must still roll back."""
+        mgr, plugin_dir = store
+
+        def fake_partial_install(plugin_id):
+            plugin_dir.mkdir(exist_ok=True)
+            (plugin_dir / "half-downloaded.tmp").write_text("junk")
+            return False
+
+        with patch.object(mgr, "install_plugin", side_effect=fake_partial_install):
+            ok = mgr._reinstall_with_rollback(PLUGIN_ID, plugin_dir)
+        assert ok is False
+        assert "old version marker" in (plugin_dir / "manager.py").read_text()
+        assert not (plugin_dir / "half-downloaded.tmp").exists()
+
+    def test_stale_aside_from_previous_crash_is_cleared(self, store):
+        mgr, plugin_dir = store
+        stale = plugin_dir.parent / f"{PLUGIN_ID}.standalone-backup-migrating"
+        stale.mkdir()
+        (stale / "old.txt").write_text("stale")
+        with patch.object(mgr, "install_plugin", return_value=False) as mock_install:
+            ok = mgr._reinstall_with_rollback(PLUGIN_ID, plugin_dir)
+        # The reinstall itself still fails (mocked) and the old install is
+        # restored, but the stale aside must not have survived — otherwise
+        # it would have blocked this run's own rename (or a future one).
+        assert not stale.exists()
+        mock_install.assert_called_once_with(PLUGIN_ID)
+        assert ok is False
+        assert plugin_dir.exists()
+        assert "old version marker" in (plugin_dir / "manager.py").read_text()
+
+    def test_concurrent_updates_for_same_plugin_are_serialized(self, store):
+        """Two overlapping requests for the same plugin_id (double-click,
+        two browser tabs — the web UI runs Flask with threaded=True) must
+        not interleave: the loser must wait for the winner to finish
+        rather than renaming the winner's in-progress install aside and
+        stealing its rollback safety net."""
+        mgr, plugin_dir = store
+
+        active = 0
+        max_active = 0
+        guard = threading.Lock()
+
+        def fake_install(plugin_id):
+            nonlocal active, max_active
+            with guard:
+                active += 1
+                max_active = max(max_active, active)
+            time.sleep(0.05)
+            plugin_dir.mkdir(exist_ok=True)
+            (plugin_dir / "manager.py").write_text("# new version\n")
+            (plugin_dir / "manifest.json").write_text(json.dumps(
+                {"id": PLUGIN_ID, "name": "Rollback Test", "version": "2.0.0"}))
+            with guard:
+                active -= 1
+            return True
+
+        results = []
+
+        def worker():
+            results.append(mgr._reinstall_with_rollback(PLUGIN_ID, plugin_dir))
+
+        with patch.object(mgr, "install_plugin", side_effect=fake_install):
+            threads = [threading.Thread(target=worker) for _ in range(2)]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join(timeout=5)
+
+        assert max_active == 1, "install_plugin ran concurrently for the same plugin_id"
+        assert results == [True, True]
+        assert plugin_dir.exists()
+        assert "new version" in (plugin_dir / "manager.py").read_text()
+        leftovers = [p for p in plugin_dir.parent.iterdir()
+                     if "standalone-backup" in p.name]
+        assert leftovers == []
+
+    def test_aside_name_is_invisible_to_discovery(self, store, tmp_path):
+        """The aside still contains a manifest.json — discovery must skip it
+        (relies on the existing '.standalone-backup-' exclusion)."""
+        mgr, plugin_dir = store
+        from src.plugin_system.plugin_manager import PluginManager
+        aside = plugin_dir.parent / f"{PLUGIN_ID}.standalone-backup-migrating"
+        plugin_dir.rename(aside)
+        pm = PluginManager(plugins_dir=str(tmp_path), config_manager=None,
+                           display_manager=None, cache_manager=None)
+        found = pm._scan_directory_for_plugins(Path(tmp_path))
+        assert PLUGIN_ID not in found
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-v"]))

test/test_testing_mocks.py

@@ -0,0 +1,45 @@
+"""
+Unit tests for src/plugin_system/testing/mocks.py.
+
+MockCacheManager/MockPluginManager stand in for the real production
+managers under the plugin safety harness -- a missing method here isn't a
+harness bug in the abstract, it's a plugin silently failing to render
+under test (confirmed on ledmatrix-leaderboard, which calls
+get_cached_data_with_strategy() and previously hit an AttributeError that
+its own broad except swallowed, producing an empty-but-green render).
+"""
+
+from src.plugin_system.testing.mocks import MockCacheManager
+
+
+class TestMockCacheManagerStrategyMethod:
+    def test_get_cached_data_with_strategy_returns_cached_value(self):
+        cm = MockCacheManager()
+        cm.set("standings_nfl", {"teams": ["KC", "BUF"]})
+        result = cm.get_cached_data_with_strategy("standings_nfl", "sports_live")
+        assert result == {"teams": ["KC", "BUF"]}
+
+    def test_get_cached_data_with_strategy_returns_none_when_missing(self):
+        cm = MockCacheManager()
+        assert cm.get_cached_data_with_strategy("missing_key") is None
+
+    def test_get_cached_data_with_strategy_defaults_data_type(self):
+        cm = MockCacheManager()
+        cm.set("k", "v")
+        assert cm.get_cached_data_with_strategy("k") == "v"
+
+    def test_calls_are_tracked(self):
+        cm = MockCacheManager()
+        cm.get_cached_data_with_strategy("k", "sports_live")
+        assert cm.get_cached_data_with_strategy_calls == [{"key": "k", "data_type": "sports_live"}]
+
+    def test_save_cache_is_readable_via_strategy_lookup(self):
+        cm = MockCacheManager()
+        cm.save_cache("standings_nfl", {"teams": ["KC", "BUF"]})
+        assert cm.get_cached_data_with_strategy("standings_nfl") == {"teams": ["KC", "BUF"]}
+
+    def test_reset_clears_strategy_call_tracking(self):
+        cm = MockCacheManager()
+        cm.get_cached_data_with_strategy("k", "sports_live")
+        cm.reset()
+        assert cm.get_cached_data_with_strategy_calls == []

test/test_vegas_render_pipeline_recompose.py

@@ -0,0 +1,72 @@
+"""
+Regression tests for RenderPipeline.should_recompose()'s pending-updates check.
+
+PR #299 added a check so a plugin's live score/status change (a "pending
+update" in StreamManager) triggers a hot-swap within a few seconds instead
+of waiting for a full scroll cycle to complete. PR #330 (multi-display sync)
+refactored should_recompose() and dropped that check entirely -- not just
+gated behind the new sync-mode deferral it added, but removed outright, so
+even standalone (non-sync) installations silently lost live-refresh and fell
+back to waiting for full cycle boundaries (which, depending on
+min/max_cycle_duration, can be minutes).
+"""
+
+from unittest.mock import MagicMock
+
+from src.vegas_mode.config import VegasModeConfig
+from src.vegas_mode.render_pipeline import RenderPipeline
+
+
+class FakeDisplayManager:
+    width = 64
+    height = 32
+
+
+def _make_pipeline(sync_manager=None):
+    stream_manager = MagicMock()
+    stream_manager.get_buffer_status.return_value = {'staging_count': 0}
+    pipeline = RenderPipeline(VegasModeConfig(), FakeDisplayManager(), stream_manager)
+    pipeline.sync_manager = sync_manager
+    return pipeline, stream_manager
+
+
+class TestShouldRecompose:
+    def test_cycle_complete_always_recomposes(self):
+        pipeline, stream_manager = _make_pipeline()
+        pipeline._cycle_complete = True
+        stream_manager.has_pending_updates_for_visible_segments.return_value = False
+        assert pipeline.should_recompose() is True
+
+    def test_no_pending_updates_no_staging_does_not_recompose(self):
+        pipeline, stream_manager = _make_pipeline()
+        stream_manager.has_pending_updates_for_visible_segments.return_value = False
+        assert pipeline.should_recompose() is False
+
+    def test_pending_updates_on_visible_segment_triggers_recompose(self):
+        """The actual regression: a live-updated plugin currently in view
+        must trigger a recompose instead of waiting for cycle end."""
+        pipeline, stream_manager = _make_pipeline()
+        stream_manager.has_pending_updates_for_visible_segments.return_value = True
+        assert pipeline.should_recompose() is True
+
+    def test_staging_buffer_content_triggers_recompose(self):
+        pipeline, stream_manager = _make_pipeline()
+        stream_manager.get_buffer_status.return_value = {'staging_count': 1}
+        stream_manager.has_pending_updates_for_visible_segments.return_value = False
+        assert pipeline.should_recompose() is True
+
+    def test_sync_active_defers_pending_updates_to_cycle_boundary(self):
+        """Sync-mode deferral (PR #330's actual intent) must still hold:
+        pending updates alone must NOT trigger a mid-cycle hot-swap when a
+        follower display is attached, since that causes a visible
+        freeze+jump on the follower. This must keep working after
+        restoring the non-sync pending-updates check above."""
+        pipeline, stream_manager = _make_pipeline(sync_manager=MagicMock())
+        stream_manager.has_pending_updates_for_visible_segments.return_value = True
+        assert pipeline.should_recompose() is False
+
+    def test_sync_active_still_recomposes_on_cycle_complete(self):
+        pipeline, stream_manager = _make_pipeline(sync_manager=MagicMock())
+        pipeline._cycle_complete = True
+        stream_manager.has_pending_updates_for_visible_segments.return_value = True
+        assert pipeline.should_recompose() is True

test/test_wifi_check_state.py

@@ -0,0 +1,92 @@
+"""Tests for check_and_manage_ap_mode_with_state (src/wifi_manager.py).
+
+The wifi monitor daemon used to fetch WiFi status before AND after each
+check on top of the check's own internal fetch — every fetch is several
+nmcli subprocess forks, every 30s, forever. The new API returns the state
+the check observed, so the daemon runs exactly one fetch battery per tick.
+"""
+
+import os
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.wifi_manager import WiFiManager, WiFiStatus  # noqa: E402
+
+
+@pytest.fixture
+def wm(tmp_path):
+    with patch.object(WiFiManager, "_load_config", return_value={}, create=True):
+        manager = WiFiManager.__new__(WiFiManager)
+        # minimal attribute setup without running the real __init__
+        manager.config = {"auto_enable_ap_mode": True}
+        manager._disconnected_checks = 0
+        manager._disconnected_checks_required = 3
+        manager._ap_enabled_at = None
+        return manager
+
+
+def _wire(wm, connected, ethernet, ap_active):
+    wm._get_wifi_status_with_retry = MagicMock(
+        return_value=WiFiStatus(connected=connected, ssid="net" if connected else None))
+    wm._is_ethernet_connected = MagicMock(return_value=ethernet)
+    wm._is_ap_mode_active = MagicMock(return_value=ap_active)
+    wm.enable_ap_mode = MagicMock(return_value=(True, "ok"))
+    wm.disable_ap_mode = MagicMock(return_value=(True, "ok"))
+    wm.scan_networks = MagicMock(return_value=([], False))
+    wm._save_cached_scan = MagicMock()
+    wm._FORCE_AP_FLAG_PATH = MagicMock()
+    wm._FORCE_AP_FLAG_PATH.exists.return_value = False
+
+
+class TestWithState:
+    def test_single_fetch_per_call(self, wm):
+        _wire(wm, connected=True, ethernet=False, ap_active=False)
+        wm.check_and_manage_ap_mode_with_state()
+        assert wm._get_wifi_status_with_retry.call_count == 1
+        assert wm._is_ethernet_connected.call_count == 1
+        assert wm._is_ap_mode_active.call_count == 1
+
+    def test_returns_observed_state(self, wm):
+        _wire(wm, connected=True, ethernet=True, ap_active=False)
+        changed, status, ethernet, ap_after = wm.check_and_manage_ap_mode_with_state()
+        assert changed is False
+        assert status.connected is True
+        assert ethernet is True
+        assert ap_after is False
+
+    def test_ap_after_inverts_on_disable(self, wm):
+        """WiFi reconnects while AP is up -> auto-disable -> ap_after False."""
+        _wire(wm, connected=True, ethernet=False, ap_active=True)
+        changed, _status, _ethernet, ap_after = wm.check_and_manage_ap_mode_with_state()
+        assert changed is True
+        assert ap_after is False
+        wm.disable_ap_mode.assert_called_once()
+
+    def test_ap_after_inverts_on_enable(self, wm):
+        """Grace period exhausted with nothing connected -> enable -> True."""
+        _wire(wm, connected=False, ethernet=False, ap_active=False)
+        wm._disconnected_checks = 2  # this call is the 3rd
+        changed, _status, _ethernet, ap_after = wm.check_and_manage_ap_mode_with_state()
+        assert changed is True
+        assert ap_after is True
+        wm.enable_ap_mode.assert_called_once()
+
+    def test_bool_wrapper_is_back_compatible(self, wm):
+        _wire(wm, connected=True, ethernet=False, ap_active=False)
+        assert wm.check_and_manage_ap_mode() is False
+        _wire(wm, connected=True, ethernet=False, ap_active=True)
+        assert wm.check_and_manage_ap_mode() is True
+
+    def test_exception_path_never_raises(self, wm):
+        wm._get_wifi_status_with_retry = MagicMock(side_effect=RuntimeError("nmcli gone"))
+        changed, status, _ethernet, _ap_after = wm.check_and_manage_ap_mode_with_state()
+        assert changed is False
+        assert status.connected is False
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-v"]))

web_interface/app.py

@@ -608,9 +608,23 @@ def display_preview_generator():
     import base64
     from PIL import Image
     import io
-    
+
     snapshot_path = "/tmp/led_matrix_preview.png"  # nosec B108 - fixed path matches display_manager; only read here
+    # Viewer marker: this generator only runs while the broadcaster has
+    # subscribers (it exits with no clients), so touching the marker each
+    # loop tells the DISPLAY service a browser is actually watching — it
+    # only pays for full-rate PNG snapshot encodes while this stays fresh
+    # (see src/common/snapshot_policy.py).
+    viewer_marker_path = "/tmp/led_matrix_preview_viewer"  # nosec B108 - fixed path matches display_manager
     last_modified = None
+
+    def _touch_viewer_marker():
+        try:
+            with open(viewer_marker_path, 'a'):
+                pass
+            os.utime(viewer_marker_path, None)
+        except OSError:
+            pass  # display side treats a missing marker as "no viewer"
     
     # Get display dimensions from config
     try:
@@ -627,6 +641,7 @@ def display_preview_generator():
     
     while True:
         try:
+            _touch_viewer_marker()
             # Check if snapshot file exists and has been modified
             if os.path.exists(snapshot_path):
                 current_modified = os.path.getmtime(snapshot_path)

web_interface/blueprints/api_v3.py

@@ -329,6 +329,7 @@ def save_schedule_config():
         }
 
         mode = data.get('mode', 'global')
+        schedule_config['mode'] = mode
 
         if mode == 'global':
             # Simple global schedule