Merge remote-tracking branch 'origin/main' into perf/dirty-tracking

# Conflicts:
#	src/display_manager.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/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_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_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

@@ -14,6 +14,7 @@ The invariants:
 
 import os
 import sys
+import time
 
 os.environ["EMULATOR"] = "true"
 
@@ -96,22 +97,32 @@ class TestDirtyTracking:
         dm.set_brightness(90)
 
     def test_snapshot_still_written_on_skip(self, dm, tmp_path):
-        """The web preview path must keep working through skipped pushes."""
+        """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
+        dm.update_display()   # push + snapshot write (first frame)
         assert os.path.exists(dm._snapshot_path)
+        first_mtime = os.path.getmtime(dm._snapshot_path)
 
-        # Re-open the snapshot's own throttle and remove the file, then push
-        # the identical frame again: dirty tracking must skip the panel
-        # write but the snapshot must still be (re-)written on that path.
-        os.remove(dm._snapshot_path)
-        dm._last_snapshot_ts = 0.0
+        # 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.exists(dm._snapshot_path)
+        assert os.path.getmtime(dm._snapshot_path) > first_mtime
 
 
 class TestKillSwitch:

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_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_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