Merge remote-tracking branch 'origin/main' into check-412-conflict

# Conflicts:
#	src/image_utils.py

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"]))