a plugin API for content that changes while it scrolls (#696)

* perf(timing): say which render-thread work a late frame followed

The soak already says how often a moving frame reached the panel late, but
not what the render thread was doing just before it. Vegas does two kinds of
work there between frames -- building its strip (compose, extend) and, with
live elements, patching changed pixels into it -- and deciding whether either
is affordable needs their own numbers.

- FrameTimingRecorder.note_op(kind, nbytes) tags the next presented frame.
  Totals gain op_frames, late_op_frames, op_freezes and op_bytes per kind;
  aggregate() still takes frames without ops. The file schema is unchanged.
- Vegas tags compose and every strip extension (with the bytes it copied).
- frame_soak prints an "after work" table: frames, late %, freezes and MB
  moved per kind, only when something tagged its work.
- render_bench gains --strip-screens (Vegas-sized strips), --patch-bytes /
  --patch-every / --patch-where (in-place column writes, as a live element
  update does) and --extend-every-screens / --extend-width (append + trim on
  a fixed cadence that holds the strip's width).

No runtime behaviour changes: this is the measurement gate for live Vegas
elements.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* docs(changelog): note the frame-op attribution and bench modes

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* perf(scroll): build the strip's PIL image only when something reads it

Every Vegas strip extension rebuilt ScrollHelper.cached_image from
cached_array in full, twice (append, then trim), on the render thread:
Image.fromarray is 1.7ms for an 8,000px strip and 3.8ms for 20,000px on a
Pi 4 (measured on ledpi), about two thirds of an extension's render-thread
cost. Nothing on the frame path reads the image's pixels; every frame is cut
from the array.

cached_image is now a property. append_content and drop_scrolled_prefix
defer it; the first read builds it from the array it started with and keeps
it only if the strip has not changed meanwhile, so a sync push racing an
extension cannot leave a stale image cached. Assigning cached_image stores
exactly what was assigned, as before. has_strip() says whether there is a
strip without building its image; the helper's frame path, Vegas and the
adapter's scroll-cache invalidation use it. The strip is also no longer held
in memory twice.

In Vegas the image is now built only by a multi-display sync push.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat(vegas): live elements -- a plugin API for content that changes while it scrolls

Vegas bakes each plugin's pictures into one strip, so a card already on its
way across the panel keeps what it showed when it was drawn. This adds the
API and bookkeeping for content that can be updated in place; the worker
that redraws and swaps it follows separately. No shipped plugin implements
the hook yet, so nothing changes for users.

Plugin API (core 3.8.0), all no-ops by default:
- BasePlugin.get_vegas_elements() -> [VegasElement(key, image, version,
  live, refresh_hz)]: named, fixed-width pieces of Vegas content.
- BasePlugin.redraw_vegas_element(key, width, height, at): a lock-free
  redraw for content that changes with time.
- BasePlugin.notify_vegas_data_changed(): data that lands outside update().
- src/plugin_system/vegas_elements.py (VegasElement, re-exported from
  base_plugin).

Core:
- PluginAdapter asks a plugin that implements the hook for elements on the
  background fetch only (under its lock, on its own canvas); every other
  path keeps get_vegas_content(). Live elements are pinned (padded with
  content_padding, never trimmed), tagged with their key, digest and data
  epoch in Image.info so the existing cache and group plumbing carry them
  unchanged, and untagged if a width budget crops them.
- RenderPipeline records where each live element lands (ElementRecord), in
  absolute strip columns a trim does not move; the block-start arithmetic
  is shared with the STATIC markers.
- PluginManager update listeners (add/remove_update_listener,
  notify_data_changed): told the moment update() completes, not at the
  next ~4s Vegas poll. The coordinator uses one to move each plugin's data
  epoch on.
- vegas_scroll.live_refresh (kill switch), live_max_hz, live_min_interval,
  live_lead_screens; per-plugin core-owned vegas_live. Live elements are
  off under multi-display sync, in swap mode and with offscreen_prefetch off.
- scripts/check_plugin.py checks the element contract
  (src/plugin_system/testing/vegas.py); test/fixtures/plugins/vegas-live-stub
  is a working example.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -331,6 +331,43 @@ read any of them:
   lock stays busy past the same 5s bound the change is handed to the update
   worker, which applies the latest one as soon as the lock frees, and before
   the plugin's next update() at the latest. The plugin API is unchanged.
+- With a Vegas width budget set (`max_plugin_width_ratio` or a plugin's
+  `vegas_max_width_screens`), a single image over the budget with no gaps
+  between items -- a map, one long headline -- no longer takes a pass of its
+  own showing four blank columns. The cut landed in the middle of the blank
+  margin trimming leaves at the image's edge; margins are no longer cut
+  points, so such an image is cropped to the budget as intended.
+
+### Live Vegas elements (plugin API)
+
+- New plugin hooks for content that can change while it scrolls:
+  `BasePlugin.get_vegas_elements()` returns `VegasElement`s -- named,
+  fixed-width pieces of Vegas content -- instead of pictures;
+  `redraw_vegas_element(key, width, height, at)` redraws one without the
+  plugin lock for content that changes with time; and
+  `notify_vegas_data_changed()` reports data that arrived outside
+  `update()`. New module `src/plugin_system/vegas_elements.py`
+  (`VegasElement`, also re-exported from `base_plugin`). See "Live Vegas
+  elements" in `docs/PLUGIN_API_REFERENCE.md`.
+- The ticker asks a plugin that implements the hook for elements on its
+  background fetch (under the plugin's lock, on a canvas of its own) and
+  records where each one lands in the strip, in absolute columns a trim does
+  not move (`src/vegas_mode/elements.py`). Live elements are never trimmed to
+  their ink: each is padded with `content_padding` black columns either side.
+  Every other path -- the first strip, the render-thread fallback, plugins
+  without the hook -- is unchanged. Swapping redraws into the strip builds
+  on this.
+- `PluginManager.add_update_listener()` / `remove_update_listener()` /
+  `notify_data_changed()`: a listener hears a plugin id the moment its
+  `update()` completes, rather than at the next ~4s Vegas poll.
+- New `display.vegas_scroll` settings: `live_refresh` (default `true`; the
+  kill switch), `live_max_hz`, `live_min_interval`, `live_lead_screens`, and
+  a per-plugin core-owned `vegas_live`. Live elements are off whatever these
+  say under multi-display sync, in swap mode and with `offscreen_prefetch`
+  off.
+- `scripts/check_plugin.py` checks the element contract for any plugin that
+  implements it (`src/plugin_system/testing/vegas.py`), and
+  `test/fixtures/plugins/vegas-live-stub` is a working example.
 
 ### Scrolling
 

docs/CONFIG_REFERENCE.md

@@ -132,6 +132,10 @@ Read by `src/vegas_mode/config.py` (`VegasScrollConfig.from_config`). See
 | `offscreen_prefetch` | bool, `true` — render every plugin's ticker content on the background thread, each on its own canvas. `false` restores handing canvas-bound plugins to the render thread, one pause at a time. Temporary; see [OFFSCREEN_RENDERING.md](OFFSCREEN_RENDERING.md) |
 | `prefetch_gate` | bool, `true` — let that background thread run Python only while the render thread is waiting for the panel, so the render thread never waits for the GIL when a refresh comes round. Only takes effect with the rebuilt rgbmatrix binding (`scripts/build_rgbmatrix_nogil.sh`). See [OFFSCREEN_RENDERING.md](OFFSCREEN_RENDERING.md) |
 | `switch_interval_ms` | float, `0` — experimental: shorten Python's GIL switch interval to this many ms while Vegas runs. `0` leaves the default (5 ms) alone |
+| `live_refresh` | bool, `true` — live elements: a plugin that supports them (scores, the flight map) has what is already scrolling updated when its data changes, instead of freezing each card as it was drawn. Always off under multi-display sync, in swap mode and with `offscreen_prefetch` off. `false` restores the frozen behaviour exactly. Per plugin: `vegas_live` in the plugin's section |
+| `live_max_hz` | float, `5` (0–10) — ceiling on how often an animated live element (a moving aircraft) is redrawn; `0` keeps data updates and turns animation off. Capped at 1 Hz without the rebuilt rgbmatrix binding |
+| `live_min_interval` | float, `2` (0.5–60) — shortest time between two data redraws of one plugin; a faster plugin is redrawn at this rate, never skipped |
+| `live_lead_screens` | float, `1` (0–5) — how far ahead of the screen, in screen widths, an animated element starts being redrawn |
 | `smooth_scroll` | bool, `true` — move a whole number of pixels per panel refresh, locked to vsync. `scroll_speed` is snapped to the nearest speed the panel can show that way (at 95Hz: 95, 47.5, 31.7 px/s…), measured against the panel's real refresh rate once scrolling starts |
 | `sub_pixel_blend` | bool, `false` — the older smoothing: advance by elapsed time and blend neighbouring pixel columns. Looks anti-aliased in the web preview but shimmers on the panel and is not locked to the refresh. Overrides `smooth_scroll` when on |
 | `extend_threshold_screens` | float, `2.0` |

docs/PLUGIN_API_REFERENCE.md

@@ -387,6 +387,79 @@ The width Vegas wants this plugin's content to occupy, from the plugin's
 `display_manager` while it asks for content, so a plugin that sizes itself
 from `display_manager.width` does not need to read this.
 
+#### Live Vegas elements
+
+*New in core 3.8.0.* Content from `get_vegas_content()` is baked into the
+ticker's strip when the plugin's turn is prefetched, so a score drawn then
+scrolls past with that score however many goals are scored while it crosses
+the panel. A plugin that returns **live elements** instead gets them updated
+in place: after its `update()` the ticker asks again, compares each element
+with what the strip holds, and swaps the changed ones in between two frames
+-- on screen included -- without anything next to them moving.
+
+```python
+try:
+    from src.plugin_system.vegas_elements import VegasElement
+except ImportError:          # core older than 3.8.0: the hook is never called
+    VegasElement = None
+
+class MyScoreboard(BasePlugin):
+    def get_vegas_elements(self):
+        if VegasElement is None:
+            return None
+        return [VegasElement(key=f"game:{g['id']}",
+                             image=self._card(g),           # cache by fingerprint
+                             version=self._fingerprint(g))  # changes iff pixels would
+                for g in self.games]
+```
+
+`VegasElement(key, image, version=None, live=True, refresh_hz=0.0)`:
+
+| Field | Meaning |
+|---|---|
+| `key` | Names the element across redraws; unique in the list, stable for the same logical item (`"game:nfl:401547417"`, `"map"`). |
+| `image` | The element now, at the display's height. A live element's **width must not depend on its data**: a redraw at another width is never swapped in (it appears the next time the plugin comes round), because nothing on screen may move. |
+| `version` | Anything hashable that changes exactly when the pixels would; lets the ticker skip unchanged elements. `None` means "compare pixels". |
+| `live` | `False` places it as plain content (trimmed, never refreshed): separators, decoration. |
+| `refresh_hz` | For content that changes with **time** rather than data (an aircraft moving between position reports): the ticker calls `redraw_vegas_element()` about this often while the element is on or near the screen, capped by `vegas_scroll.live_max_hz` and at 1 Hz without the rebuilt rgbmatrix binding. |
+
+**`get_vegas_elements() -> Optional[List[VegasElement]]`** — called on the
+ticker's background thread under the plugin's lock (never while `update()`
+runs), on a canvas of its own and told its render width, exactly like
+`get_vegas_content()`. It is called after every `update()` while any of the
+plugin's elements is on or ahead of the screen, so it must be cheap when
+nothing changed (cache images by version), idempotent, and must not fetch.
+Return `None` to use `get_vegas_content()`, which a plugin must keep working
+for older cores and for the paths that do not ask for elements (the ticker's
+first strip, multi-display sync, the `live_refresh` switch).
+
+**`redraw_vegas_element(key, width, height, at) -> Optional[PIL.Image]`** —
+only for elements with `refresh_hz`. Called **without** the plugin's lock,
+possibly while `update()` runs, so read only state `update()` replaces in one
+assignment (an immutable snapshot), never state it mutates in place. `at` is
+the `time.monotonic()` the pixels are expected on the panel: draw the element
+as it should look then. Return exactly `width` x `height`, or `None` to skip
+the tick.
+
+**`notify_vegas_data_changed()`** — data that arrives outside `update()` (a
+background thread, a push callback) calls this so the ticker redraws without
+waiting for the next `update()`. Safe from any thread.
+
+Live elements are never trimmed to their ink: the ticker pads each with
+`content_padding` black columns either side, the margin trimming would have
+left. A single element wider than the plugin's width budget
+(`vegas_max_width_screens`, not counting that padding) is cropped like any
+other content and scrolls by as plain, no longer live. The user can turn them off per plugin with `vegas_live: false` (a
+core-owned property) or for the whole ticker with
+`display.vegas_scroll.live_refresh: false`; they are always off under
+multi-display sync.
+
+`scripts/check_plugin.py` checks the contract for any plugin that implements
+the hook (unique keys, height, width stable with no new data, redraw size,
+slow calls) and prints a `vegas elements` row; the checks are in
+`src/plugin_system/testing/vegas.py`. `test/fixtures/plugins/vegas-live-stub`
+is a small working example.
+
 #### Legacy: `get_vegas_content_type()` and `get_vegas_display_mode()`
 
 Superseded by participation, and still read to derive it when neither the

docs/PLUGIN_CONFIG_CORE_PROPERTIES.md

@@ -47,6 +47,17 @@ is `CORE_PLUGIN_PROPERTIES` in `src/plugin_system/schema_manager.py`):
      `src/plugin_system/base_plugin.py`; see
      [PLUGIN_API_REFERENCE.md](PLUGIN_API_REFERENCE.md#vegas-participation)
 
+6. **`vegas_live`** (boolean; no default, unset means on)
+   - Description: for a plugin with live Vegas elements (it implements
+     `get_vegas_elements()`), whether the ticker changes what is already
+     scrolling when the plugin's data changes. `false` shows each card as it
+     was when drawn, as before live elements existed
+   - Ignored by plugins without live elements, and whenever live elements
+     are off for the whole ticker (`display.vegas_scroll.live_refresh`)
+   - Read by `PluginAdapter.is_live_capable()` in
+     `src/vegas_mode/plugin_adapter.py`; see
+     [PLUGIN_API_REFERENCE.md](PLUGIN_API_REFERENCE.md#live-vegas-elements)
+
 `skin` and `skin_options` were core properties until the skin system was
 removed. A plugin config saved with them still loads and saves; the keys are
 dropped on the next save (see `RETIRED_PLUGIN_KEYS` in `schema_manager.py`).

scripts/check_plugin.py

@@ -72,6 +72,7 @@ from src.plugin_system.testing.harness import (  # noqa: E402
 from src.plugin_system.testing.sizes import (  # noqa: E402
     parse_size_token, resolve_test_sizes, safe_mode_filename, size_label,
 )
+from src.plugin_system.testing.vegas import check_plugin_vegas_elements  # noqa: E402
 
 logger = get_logger("[Check Plugin]")
 
@@ -193,6 +194,24 @@ def check_one(plugin_id: str, search_dirs: List[str], sizes, mock_data: Dict,
 
         all_run_results.extend(results)
 
+    # Live Vegas elements, for a plugin that has them: checked once, at the
+    # first size, with the base config.
+    width, height = effective_sizes[0]
+    try:
+        vegas = check_plugin_vegas_elements(
+            plugin_id, plugin_dir, full_config, effective_mock_data, width, height,
+            run_update=effective_run_update)
+    except Exception as exc:  # noqa: BLE001 - one plugin must not end an --all run
+        all_run_results.append(RenderResult(
+            plugin_id, width, height, "vegas elements",
+            error=f"the element check itself failed: {exc!r}"))
+        return all_run_results
+    if vegas.implemented:
+        all_run_results.append(RenderResult(
+            plugin_id, width, height, "vegas elements",
+            error="; ".join(vegas.errors) or None,
+            notes=[f"{vegas.elements} element(s), {vegas.live} live"] + vegas.warnings))
+
     return all_run_results
 
 
@@ -236,6 +255,8 @@ def print_report(all_results: Dict[str, List[RenderResult]]) -> bool:
                               f" controller skips the mode")
                 else:
                     status, detail = "FAIL", ""
+            if r.notes:
+                detail += f" ({'; '.join(r.notes)})"
             print(f"  [{status}] {r.size_label:>7}  {r.mode}{detail}")
     print()
     return everything_ok

src/plugin_system/base_plugin.py

@@ -15,6 +15,8 @@ import os
 import sys
 from src.deprecation import deprecated, warn_deprecated
 from src.logging_config import get_logger
+# Re-exported: a plugin may import it from here beside VegasDisplayMode.
+from src.plugin_system.vegas_elements import VegasElement  # noqa: F401
 
 
 _shared_fallback_font_manager: Optional[Any] = None
@@ -986,6 +988,88 @@ class BasePlugin(ABC):
         """
         return None
 
+    def get_vegas_elements(self) -> Optional[List[Any]]:
+        """
+        Vegas content as live elements: named, fixed-width pieces the ticker
+        can swap in place while they are on screen.
+
+        get_vegas_content() hands the ticker pictures, and a picture already
+        in the scrolling strip keeps what it showed when it was drawn. Return
+        a list of ``VegasElement`` (src/plugin_system/vegas_elements.py)
+        instead and the ticker records where each one is; after your update()
+        it calls this again, compares each element's ``version`` (or pixels)
+        with what the strip holds, and swaps the changed ones in between two
+        frames -- a score changes on a card already crossing the panel, and
+        nothing next to it moves.
+
+        The contract:
+
+        - Called only on the ticker's background thread, under this plugin's
+          lock (never while update() runs), on a canvas of its own and told
+          its width (get_vegas_render_width()), like get_vegas_content().
+        - Called often -- after every update() while any of your elements is
+          on or ahead of the screen -- so it must be cheap when nothing
+          changed (cache images by version), idempotent, and must not fetch.
+        - A live element's width must not depend on its data: a redraw at a
+          different width is not swapped in (it shows the next time the
+          plugin comes round).
+        - Keys must be unique in the list and stable for the same logical
+          item.
+
+        Return None (the default) to use get_vegas_content(). A core older
+        than 3.8.0 never calls this, so keep get_vegas_content() working and
+        floor ``ledmatrix_min_version`` at 3.8.0 only if you rely on it.
+
+        Example (scoreboard)::
+
+            def get_vegas_elements(self):
+                return [VegasElement(key=f"game:{g['id']}",
+                                     image=self._card_for(g),   # cached by fingerprint
+                                     version=self._fingerprint(g))
+                        for g in self.games]
+
+        Returns:
+            A list of VegasElement, or None.
+        """
+        return None
+
+    def redraw_vegas_element(self, key: str, width: int, height: int,
+                             at: float) -> Optional[Any]:
+        """
+        Redraw one live element for a moment in time, without the plugin lock.
+
+        Only for elements returned with ``refresh_hz > 0``: content that
+        changes with time rather than with data, such as an aircraft moving
+        between position reports. The ticker calls it up to that often while
+        the element is on or near the screen.
+
+        - Called WITHOUT this plugin's lock, possibly while update() runs, so
+          read only state that update() replaces in a single assignment (an
+          immutable snapshot), never state it mutates in place.
+        - ``at`` is the time.monotonic() at which the pixels are expected to
+          reach the panel; draw the element as it should look then.
+        - Return an image of exactly ``width`` x ``height``, or None to skip
+          this tick.
+
+        Returns:
+            PIL Image of exactly (width, height), or None.
+        """
+        return None
+
+    def notify_vegas_data_changed(self) -> None:
+        """
+        Tell the Vegas ticker this plugin's data changed outside update().
+
+        The ticker redraws a plugin's live elements when its update()
+        completes. Data that lands some other way -- a background thread, a
+        push callback -- calls this so the change reaches the screen without
+        waiting for the next update(). Cheap and safe from any thread.
+        """
+        notify = getattr(getattr(self, 'plugin_manager', None),
+                         'notify_data_changed', None)
+        if callable(notify):
+            notify(self.plugin_id)
+
     def get_vegas_participation(self) -> str:
         """
         How this plugin takes part in Vegas mode: ``'scroll'``, ``'pause'`` or

src/plugin_system/plugin_manager.py

@@ -16,7 +16,7 @@ import time
 import threading
 import types
 from pathlib import Path
-from typing import Dict, List, NamedTuple, Optional, Any, Tuple, Union
+from typing import Callable, Dict, List, NamedTuple, Optional, Any, Tuple, Union
 import logging
 from src import display_watchdog
 from src.exceptions import PluginError, ConfigError
@@ -197,6 +197,11 @@ class PluginManager:
         # run_scheduled_updates_with_changes().
         self._completed_updates: set = set()
         self._completed_updates_lock = threading.Lock()
+        # Called with a plugin id the moment its data may have changed: its
+        # update() completed, or it called notify_vegas_data_changed(). See
+        # add_update_listener(). A tuple, replaced rather than mutated, so the
+        # worker can iterate it without a lock.
+        self._update_listeners: Tuple[Callable[[str], None], ...] = ()
         # Config changes that found the plugin's lock busy, latest per plugin,
         # with the instance they were meant for. See apply_config_change().
         self._deferred_config_changes: Dict[str, Tuple[Any, Dict[str, Any]]] = {}
@@ -589,8 +594,8 @@ class PluginManager:
     #: prefix rule would silently stop validating it.
     #:
     #: Read by: ``vegas_mode/plugin_adapter.py`` (``vegas_width_pct``,
-    #: ``vegas_overflow``) and ``base_plugin.py`` (``vegas_max_width_screens``,
-    #: ``vegas_participation``).
+    #: ``vegas_overflow``, ``vegas_live``) and ``base_plugin.py``
+    #: (``vegas_max_width_screens``, ``vegas_participation``).
     #:
     #: The list itself lives with the other core-owned per-plugin properties in
     #: ``schema_manager.CORE_PLUGIN_PROPERTIES``, which the web save path also
@@ -1723,9 +1728,53 @@ class PluginManager:
         return self.drain_completed_updates()
 
     def _note_update_completed(self, plugin_id: str) -> None:
-        """Record that a plugin's update() finished, for the next poll."""
+        """Record that a plugin's update() finished, for the next poll.
+
+        Also tells the update listeners at once, so Vegas live elements are
+        redrawn the moment new data lands instead of at the next ~4s poll.
+        This runs while the plugin's lock is still held (see _finish), which
+        is what makes the listeners' contract strict.
+        """
         with self._completed_updates_lock:
             self._completed_updates.add(plugin_id)
+        self._fire_update_listeners(plugin_id)
+
+    def add_update_listener(self, listener: Callable[[str], None]) -> None:
+        """Call ``listener(plugin_id)`` whenever a plugin's data may have changed.
+
+        That is: its update() completed successfully, or it called
+        notify_vegas_data_changed(). The listener runs on the thread that
+        noticed -- the update worker, with the plugin's lock still held, or
+        the plugin's own thread -- so it must return at once and take no lock
+        a plugin could hold: record the id and hand off (a dict store, a
+        queue put). An exception from it is logged and does not reach the
+        plugin. Adding the same listener twice has no effect.
+        """
+        # __dict__.get: tests build bare managers with PluginManager.__new__.
+        listeners = self.__dict__.get('_update_listeners', ())
+        if listener not in listeners:
+            self._update_listeners = listeners + (listener,)
+
+    def remove_update_listener(self, listener: Callable[[str], None]) -> None:
+        """Stop calling a listener added with add_update_listener()."""
+        self._update_listeners = tuple(
+            fn for fn in self.__dict__.get('_update_listeners', ()) if fn != listener)
+
+    def notify_data_changed(self, plugin_id: str) -> None:
+        """A plugin's data changed outside update(); tell the update listeners.
+
+        BasePlugin.notify_vegas_data_changed() lands here.
+        """
+        self._fire_update_listeners(plugin_id)
+
+    def _fire_update_listeners(self, plugin_id: str) -> None:
+        for listener in self.__dict__.get('_update_listeners', ()):
+            try:
+                listener(plugin_id)
+            except Exception as exc:  # pylint: disable=broad-except
+                self._warn_rate_limited(
+                    "update-listener",
+                    "An update listener failed for plugin %s: %r", plugin_id, exc)
 
     def drain_completed_updates(self) -> List[str]:
         """Return and clear the plugin ids whose update() has since finished."""

src/plugin_system/schema_manager.py

@@ -155,6 +155,18 @@ CORE_PLUGIN_PROPERTIES: Dict[str, Dict[str, Any]] = {
             "Leave unset to use the plugin's own default."
         ),
     },
+    # Read by vegas_mode/plugin_adapter.py (PluginAdapter.is_live_capable).
+    # No default, for the same reason: unset means on.
+    "vegas_live": {
+        "type": "boolean",
+        "title": "Update in the Vegas ticker",
+        "description": (
+            "Vegas mode: for a plugin with live elements (scores, the flight "
+            "map), change what is already scrolling when its data changes. "
+            "Off shows each card as it was when it was drawn, as before. "
+            "Leave unset for on."
+        ),
+    },
 }
 
 #: The keys of CORE_PLUGIN_PROPERTIES that are Vegas tuning rather than plugin
@@ -162,7 +174,7 @@ CORE_PLUGIN_PROPERTIES: Dict[str, Dict[str, Any]] = {
 #: PluginManager.CORE_OWNED_CONFIG_KEYS).
 CORE_VEGAS_TUNING_KEYS = frozenset({
     'vegas_width_pct', 'vegas_overflow', 'vegas_max_width_screens',
-    'vegas_participation',
+    'vegas_participation', 'vegas_live',
 })
 
 

src/plugin_system/testing/harness.py

@@ -19,7 +19,7 @@ from datetime import timedelta
 import socket
 import ssl
 import urllib.error
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -84,6 +84,8 @@ class RenderResult:
     fill_checked: bool = False
     fill_ok: Optional[bool] = None       # False only in strict mode
     fill_extent: Optional[Tuple[float, float]] = None  # (extent_x, extent_y)
+    # warnings worth printing that do not fail the result
+    notes: List[str] = field(default_factory=list)
 
     @property
     def size_label(self) -> str:

src/plugin_system/testing/vegas.py

@@ -0,0 +1,307 @@
+"""Offline checks for a plugin's live Vegas elements.
+
+A plugin that implements ``get_vegas_elements()`` promises the ticker a few
+things it cannot check for itself until they go wrong on a panel: unique,
+stable keys; images at the display's height; the same width for the same key
+until the data changes; the same result when nothing changed; and, for an
+element with ``refresh_hz``, a ``redraw_vegas_element()`` that returns exactly
+the size asked for, quickly. :func:`check_vegas_elements` exercises each of
+those the way the Vegas ticker calls the hooks -- on a canvas of the plugin's
+own, told its render width -- and says what failed.
+
+``scripts/check_plugin.py`` runs it for every plugin that implements the hook.
+See "Live Vegas elements" in docs/PLUGIN_API_REFERENCE.md.
+"""
+
+from __future__ import annotations
+
+import time
+from contextlib import contextmanager, nullcontext
+from dataclasses import dataclass, field
+from typing import Any, Iterator, List, Optional
+
+from PIL import Image
+
+#: A warm get_vegas_elements() slower than this holds the ticker's single
+#: background worker, and the plugin's lock, for longer than it should.
+SLOW_ELEMENTS_SECONDS = 0.2
+#: A redraw_vegas_element() slower than this cannot keep up with a few Hz.
+SLOW_REDRAW_SECONDS = 0.02
+#: The narrowed render width the check also tries, as a share of the panel.
+NARROW_PCT = 60
+
+
+@dataclass
+class VegasElementReport:
+    """What :func:`check_vegas_elements` found."""
+    implemented: bool
+    elements: int = 0
+    live: int = 0
+    errors: List[str] = field(default_factory=list)
+    warnings: List[str] = field(default_factory=list)
+
+    @property
+    def ok(self) -> bool:
+        return not self.errors
+
+
+def implements_vegas_elements(plugin: Any) -> bool:
+    """Whether the plugin's class overrides BasePlugin.get_vegas_elements."""
+    from src.plugin_system.base_plugin import BasePlugin
+
+    method = getattr(type(plugin), 'get_vegas_elements', None)
+    return method is not None and method is not getattr(
+        BasePlugin, 'get_vegas_elements', None)
+
+
+@contextmanager
+def _as_vegas_canvas(plugin: Any, display_manager: Any, width: int) -> Iterator[None]:
+    """Run a hook the way Vegas does: told its width, on a canvas of its own."""
+    plugin._vegas_render_width = width
+    try:
+        offscreen = getattr(display_manager, 'offscreen', None)
+        with offscreen(width) if offscreen is not None else nullcontext():
+            yield
+    finally:
+        plugin._vegas_render_width = None
+
+
+def render_vegas_elements(plugin: Any, display_manager: Any,
+                          width: Optional[int] = None) -> Optional[list]:
+    """Call ``plugin.get_vegas_elements()`` as the Vegas ticker does."""
+    render_width = int(width or display_manager.width)
+    with _as_vegas_canvas(plugin, display_manager, render_width):
+        return plugin.get_vegas_elements()
+
+
+def redraw_vegas_element(plugin: Any, display_manager: Any, key: str,
+                         width: int, height: int, at: Optional[float] = None,
+                         render_width: Optional[int] = None) -> Any:
+    """Call ``plugin.redraw_vegas_element()`` as the Vegas ticker does."""
+    with _as_vegas_canvas(plugin, display_manager,
+                          int(render_width or display_manager.width)):
+        return plugin.redraw_vegas_element(
+            key, width, height, time.monotonic() if at is None else at)
+
+
+def _refresh_hz(element: Any) -> Optional[float]:
+    """An element's refresh_hz as a number (None counts as 0), or None if it is not one."""
+    try:
+        return float(element.refresh_hz or 0.0)
+    except (TypeError, ValueError):
+        return None
+
+
+def _usable(element: Any) -> bool:
+    """A VegasElement the checks can read: a str key and an image."""
+    from src.plugin_system.vegas_elements import VegasElement
+    return (isinstance(element, VegasElement) and isinstance(element.key, str)
+            and bool(element.key) and isinstance(element.image, Image.Image))
+
+
+def check_vegas_elements(plugin: Any, display_manager: Any) -> VegasElementReport:
+    """Exercise a plugin's live-element hooks and report what breaks the contract.
+
+    Errors are what the ticker would refuse or show wrongly; warnings are what
+    it would cope with but should not have to (slow calls, an element wider
+    than the width the plugin was asked to render at).
+    """
+    report = VegasElementReport(implemented=implements_vegas_elements(plugin))
+    if not report.implemented:
+        return report
+    from src.plugin_system.vegas_elements import VegasElement
+
+    height = int(display_manager.height)
+    full = int(display_manager.width)
+
+    def fetch(width: int, label: str):
+        started = time.perf_counter()
+        try:
+            result = render_vegas_elements(plugin, display_manager, width)
+        except Exception as exc:  # noqa: BLE001 - a plugin hook can raise anything
+            report.errors.append(f"get_vegas_elements() raised {exc!r} ({label})")
+            return None, 0.0
+        return result, time.perf_counter() - started
+
+    first, _ = fetch(full, "full width")
+    if first is None:
+        if not report.errors:
+            report.warnings.append(
+                "get_vegas_elements() returned None: the ticker will use "
+                "get_vegas_content() instead")
+        return report
+    if not isinstance(first, (list, tuple)):
+        report.errors.append(
+            f"get_vegas_elements() returned {type(first).__name__}, expected a list")
+        return report
+
+    seen = set()
+    widths = {}
+    for index, element in enumerate(first):
+        where = f"element[{index}]"
+        if not isinstance(element, VegasElement):
+            report.errors.append(f"{where} is a {type(element).__name__}, not a VegasElement")
+            continue
+        key = element.key
+        if not isinstance(key, str) or not key:
+            report.errors.append(f"{where} has no key (a non-empty str is required)")
+            continue
+        where = f"element {key!r}"
+        if key in seen:
+            report.errors.append(f"{where} appears twice; keys must be unique")
+            continue
+        seen.add(key)
+        if not isinstance(element.image, Image.Image):
+            report.errors.append(f"{where} image is a {type(element.image).__name__}")
+            continue
+        if element.image.height != height:
+            report.errors.append(
+                f"{where} is {element.image.height}px tall; the display is {height}px")
+        if element.image.width <= 0 or element.image.height <= 0:
+            report.errors.append(f"{where} image is empty ({element.image.width}x"
+                                 f"{element.image.height})")
+            continue
+        if element.image.width > full:
+            report.warnings.append(
+                f"{where} is {element.image.width}px wide, wider than the "
+                f"{full}px it was asked to render at")
+        hz = _refresh_hz(element)
+        if hz is None:
+            report.errors.append(
+                f"{where} refresh_hz {element.refresh_hz!r} is not a number")
+        elif hz < 0:
+            report.errors.append(f"{where} has a negative refresh_hz")
+        report.elements += 1
+        if element.live:
+            report.live += 1
+            widths[key] = element.image.width
+
+    if report.errors:
+        return report
+
+    # The same data twice must give the same keys, widths and versions: the
+    # ticker redraws on every update and swaps in only what changed.
+    second, warm = fetch(full, "second call")
+    if second is not None and not isinstance(second, (list, tuple)):
+        report.errors.append(
+            f"get_vegas_elements() returned {type(second).__name__} on a second "
+            "call, expected a list")
+    elif isinstance(second, (list, tuple)):
+        again = {e.key: e for e in second if _usable(e)}
+        for element in first:
+            other = again.get(element.key)
+            if other is None:
+                report.errors.append(
+                    f"element {element.key!r} disappeared on a second call with "
+                    "no new data")
+                continue
+            if element.live and other.image.width != element.image.width:
+                report.errors.append(
+                    f"element {element.key!r} changed width with no new data "
+                    f"({element.image.width} -> {other.image.width}px); a live "
+                    "element's width must not depend on when it is drawn")
+            if element.version is not None and other.version != element.version \
+                    and not _refresh_hz(element):
+                report.warnings.append(
+                    f"element {element.key!r} changed version with no new data; "
+                    "every update will redraw it")
+    if warm > SLOW_ELEMENTS_SECONDS:
+        report.warnings.append(
+            f"get_vegas_elements() took {warm * 1000:.0f}ms with nothing new "
+            f"(over {SLOW_ELEMENTS_SECONDS * 1000:.0f}ms); cache what has not "
+            "changed")
+
+    narrow = max(1, full * NARROW_PCT // 100)
+    if narrow < full:
+        narrowed, _ = fetch(narrow, f"{NARROW_PCT}% width")
+        if isinstance(narrowed, (list, tuple)):
+            for element in narrowed:
+                if isinstance(element, VegasElement) and isinstance(element.image, Image.Image) \
+                        and element.image.width > narrow:
+                    report.warnings.append(
+                        f"element {element.key!r} is {element.image.width}px wide at "
+                        f"a {narrow}px render width; read get_vegas_render_width() "
+                        "or display_manager.width when sizing it")
+                    break
+
+    has_redraw = type(plugin).redraw_vegas_element is not _base_redraw()
+    for element in first:
+        hz = _refresh_hz(element) or 0.0
+        if not (element.live and hz > 0):
+            continue
+        if not has_redraw:
+            report.warnings.append(
+                f"element {element.key!r} asks for {hz:g}Hz but "
+                "redraw_vegas_element() is not implemented; the ticker re-runs "
+                "get_vegas_elements() under the plugin's lock instead")
+            continue
+        w, h = element.image.width, element.image.height
+        started = time.perf_counter()
+        try:
+            redrawn = redraw_vegas_element(plugin, display_manager, element.key, w, h)
+        except Exception as exc:  # noqa: BLE001 - a plugin hook can raise anything
+            report.errors.append(f"redraw_vegas_element({element.key!r}) raised {exc!r}")
+            continue
+        took = time.perf_counter() - started
+        if redrawn is not None:
+            if not isinstance(redrawn, Image.Image):
+                report.errors.append(
+                    f"redraw_vegas_element({element.key!r}) returned "
+                    f"{type(redrawn).__name__}, expected an Image or None")
+            elif redrawn.size != (w, h):
+                report.errors.append(
+                    f"redraw_vegas_element({element.key!r}) returned "
+                    f"{redrawn.width}x{redrawn.height}, asked for {w}x{h}")
+        if took > SLOW_REDRAW_SECONDS:
+            report.warnings.append(
+                f"redraw_vegas_element({element.key!r}) took {took * 1000:.1f}ms "
+                f"(over {SLOW_REDRAW_SECONDS * 1000:.0f}ms); the ticker will "
+                "slow its refresh")
+    return report
+
+
+def _base_redraw():
+    from src.plugin_system.base_plugin import BasePlugin
+    return BasePlugin.redraw_vegas_element
+
+
+def check_plugin_vegas_elements(plugin_id: str, plugin_dir: Any, config: dict,
+                                mock_data: dict, width: int, height: int,
+                                run_update: bool = True) -> VegasElementReport:
+    """Load a plugin from its directory at one panel size and check its elements.
+
+    What ``scripts/check_plugin.py`` runs: the plugin gets the same mocked
+    managers as the rendering harness, and its update() is run first (a
+    network error there is tolerated, as in the harness) so the elements are
+    drawn from data rather than from an empty start.
+    """
+    from pathlib import Path
+
+    from src.plugin_system.testing.harness import _TOLERATED_UPDATE_ERRORS, _instantiate
+    from src.plugin_system.testing.loading import load_manifest
+    from src.plugin_system.testing.visual_display_manager import VisualTestDisplayManager
+
+    plugin_dir = Path(plugin_dir)
+    display_manager = VisualTestDisplayManager(width=width, height=height)
+    try:
+        plugin = _instantiate(plugin_id, load_manifest(plugin_dir), plugin_dir,
+                              config, mock_data, display_manager)
+    except Exception as exc:  # noqa: BLE001 - the matrix run reports load errors
+        report = VegasElementReport(implemented=False)
+        report.warnings.append(f"not checked: the plugin did not load ({exc!r})")
+        return report
+    report = VegasElementReport(implemented=implements_vegas_elements(plugin))
+    if not report.implemented:
+        return report
+    if run_update:
+        try:
+            plugin.update()
+        except _TOLERATED_UPDATE_ERRORS as exc:
+            report.warnings.append(f"update() had no network ({exc!r}); checked "
+                                   "with whatever data the plugin starts with")
+        except Exception as exc:  # noqa: BLE001 - a plugin's update can raise anything
+            report.errors.append(f"update() raised {exc!r}")
+            return report
+    checked = check_vegas_elements(plugin, display_manager)
+    checked.warnings[:0] = report.warnings
+    return checked

src/plugin_system/vegas_elements.py

@@ -0,0 +1,73 @@
+"""Live elements: Vegas content that can change while it is on screen.
+
+A plugin's ``get_vegas_content()`` hands the Vegas ticker pictures, and the
+ticker bakes them into its strip: a score drawn when the plugin's turn was
+prefetched scrolls past with that score, however many goals are scored while
+it crosses the panel. A plugin that returns **elements** instead gives each
+picture a name and a fixed width. The ticker then keeps track of where each
+one is in the strip, and when the plugin's data changes it asks for just the
+changed elements and swaps their pixels in place -- on screen included,
+between two frames, without anything next to them moving.
+
+A plugin opts in by implementing ``BasePlugin.get_vegas_elements()``, and, for
+content that changes with time rather than with data (an aircraft moving
+between position reports), ``BasePlugin.redraw_vegas_element()``. See "Live
+Vegas elements" in docs/PLUGIN_API_REFERENCE.md.
+
+Added in LEDMatrix 3.8.0. Import it guarded, so the plugin still loads on an
+older core (which never calls the hooks)::
+
+    try:
+        from src.plugin_system.vegas_elements import VegasElement
+    except ImportError:          # core older than 3.8.0
+        VegasElement = None
+
+    def get_vegas_elements(self):
+        if VegasElement is None:
+            return None
+        return [VegasElement(key=f"game:{g['id']}", image=self._card(g),
+                             version=self._fingerprint(g))
+                for g in self._games]
+"""
+
+from dataclasses import dataclass
+from typing import Hashable, Optional
+
+from PIL import Image
+
+
+@dataclass(frozen=True, eq=False)
+class VegasElement:
+    """One named, fixed-width piece of a plugin's Vegas content.
+
+    Attributes:
+        key: Names the element across redraws, unique within one list the
+            plugin returns: ``"game:nfl:401547417"``, ``"sep:0:nfl"``,
+            ``"map"``. The ticker matches a redraw to the pixels already in
+            its strip by this key, so it must stay the same for the same
+            logical thing and must not be reused for a different one.
+        image: The element as drawn now, at the display's height. For a
+            ``live`` element its width is fixed for as long as the key is on
+            the strip: a redraw at a different width is never swapped in (it
+            appears the next time the plugin comes round instead), because
+            nothing on screen may move. Draw live elements at a width that
+            does not depend on the data -- a fixed card width, not the
+            width of the text.
+        version: Anything hashable that changes exactly when the pixels
+            would, such as the tuple of fields the element draws. The ticker
+            skips work for an unchanged version. ``None`` means "compare the
+            pixels", which is always correct and costs a checksum.
+        live: False places the element exactly as plain content is placed
+            (trimmed to its ink, never refreshed): separators, decoration.
+        refresh_hz: More than 0 asks for ``redraw_vegas_element()`` about
+            this often while the element is on or near the screen, for
+            content that changes with time rather than with data. The ticker
+            caps the rate (``vegas_scroll.live_max_hz``, 1 Hz on a display
+            without the rebuilt rgbmatrix binding) and slows it for an
+            element that is slow to draw.
+    """
+    key: str
+    image: Image.Image
+    version: Optional[Hashable] = None
+    live: bool = True
+    refresh_hz: float = 0.0

src/vegas_mode/config.py

@@ -99,6 +99,25 @@ class VegasModeConfig:
     # overall from 0.90% to 0.60%. See src/common/render_gate.py.
     prefetch_gate: bool = True
 
+    # Live elements (src/plugin_system/vegas_elements.py): a plugin that hands
+    # the ticker named, fixed-width elements has them redrawn when its data
+    # changes, and the changed pixels are swapped into the strip in place --
+    # on screen included -- instead of waiting for the plugin's next turn.
+    # False restores the frozen-segment behaviour exactly. Also off, whatever
+    # this says, under multi-display sync, in swap mode (continuous_scroll
+    # false) and with offscreen_prefetch false.
+    live_refresh: bool = True
+    # Ceiling on how often an element that animates (refresh_hz) is redrawn,
+    # in Hz. 0 turns animation off and keeps data-driven updates.
+    live_max_hz: float = 5.0
+    # Shortest time between two data redraws of one plugin, in seconds. A
+    # plugin updating faster is redrawn at this rate, never skipped: the
+    # latest data is always drawn eventually.
+    live_min_interval: float = 2.0
+    # How far ahead of the right edge, in screens, an animated element starts
+    # being redrawn, so it is already moving when it scrolls in.
+    live_lead_screens: float = 1.0
+
     # Keep one continuous strip, extending it with the next group of plugins as
     # the scroll approaches the end, instead of composing a fresh strip and
     # swapping it in. A swap stops the motion, substitutes every pixel at once
@@ -235,6 +254,10 @@ class VegasModeConfig:
             offscreen_prefetch=bool(get('offscreen_prefetch', d.offscreen_prefetch)),
             switch_interval_ms=float(get('switch_interval_ms', d.switch_interval_ms) or 0.0),
             prefetch_gate=bool(get('prefetch_gate', d.prefetch_gate)),
+            live_refresh=bool(get('live_refresh', d.live_refresh)),
+            live_max_hz=float(get('live_max_hz', d.live_max_hz)),
+            live_min_interval=float(get('live_min_interval', d.live_min_interval)),
+            live_lead_screens=float(get('live_lead_screens', d.live_lead_screens)),
             extend_threshold_screens=float(
                 get('extend_threshold_screens', d.extend_threshold_screens)),
             auto_trim=get('auto_trim', d.auto_trim),
@@ -281,6 +304,10 @@ class VegasModeConfig:
             'offscreen_prefetch': self.offscreen_prefetch,
             'switch_interval_ms': self.switch_interval_ms,
             'prefetch_gate': self.prefetch_gate,
+            'live_refresh': self.live_refresh,
+            'live_max_hz': self.live_max_hz,
+            'live_min_interval': self.live_min_interval,
+            'live_lead_screens': self.live_lead_screens,
             'extend_threshold_screens': self.extend_threshold_screens,
             'auto_trim': self.auto_trim,
             'trim_threshold': self.trim_threshold,
@@ -377,6 +404,18 @@ class VegasModeConfig:
                 "extend_threshold_screens must be between 1.0 and 10.0, "
                 f"got {self.extend_threshold_screens}")
 
+        if not 0.0 <= self.live_max_hz <= 10.0:
+            errors.append(
+                f"live_max_hz must be between 0 and 10, got {self.live_max_hz}")
+        if not 0.5 <= self.live_min_interval <= 60.0:
+            errors.append(
+                "live_min_interval must be between 0.5 and 60, "
+                f"got {self.live_min_interval}")
+        if not 0.0 <= self.live_lead_screens <= 5.0:
+            errors.append(
+                "live_lead_screens must be between 0 and 5, "
+                f"got {self.live_lead_screens}")
+
         if not 1 <= self.min_cut_gap <= 128:
             errors.append(
                 "min_cut_gap must be between 1 and 128, "

src/vegas_mode/coordinator.py

@@ -23,6 +23,7 @@ from typing import Optional, Dict, Any, List, Callable, TYPE_CHECKING
 from src import display_watchdog
 from src.common import render_gate
 from src.vegas_mode.config import VegasModeConfig
+from src.vegas_mode.elements import LiveEpochs
 from src.vegas_mode.plugin_adapter import PluginAdapter
 from src.vegas_mode.stream_manager import StreamManager
 from src.vegas_mode.render_pipeline import RenderPipeline
@@ -86,6 +87,9 @@ class VegasModeCoordinator:
 
     # Class-level so coordinators built without __init__ (tests) have it.
     _last_live_check: float = float('-inf')
+    #: Whether live elements are on for this run (see _apply_live_state).
+    live_active: bool = False
+    _live_reason: Optional[str] = None
     # Set only while Vegas has changed the GIL switch interval; read with getattr.
     _saved_switch_interval: Optional[float]
 
@@ -124,6 +128,12 @@ class VegasModeCoordinator:
             self.stream_manager
         )
 
+        # Live elements: one data epoch per plugin, shared with the adapter,
+        # which stamps every element it draws with it. Moved on by the plugin
+        # manager's update listener while Vegas runs. See _apply_live_state.
+        self.live_epochs = LiveEpochs()
+        self.plugin_adapter.live_epochs = self.live_epochs
+
         # State management
         self._is_active = False
         self._is_paused = False
@@ -293,6 +303,9 @@ class VegasModeCoordinator:
             self._fps_was_degraded = False
             self._apply_switch_interval()
             self._install_render_gate()
+            # Before the first background fetch below, which is the first
+            # that may ask a plugin for live elements.
+            self._apply_live_state()
 
         # Line up the next group immediately, so the first extension is already
         # warm rather than stalling the scroll to fetch it.
@@ -319,6 +332,7 @@ class VegasModeCoordinator:
 
         self._restore_switch_interval()
         self._remove_render_gate()
+        self._set_live(False, None)
 
         # Cleanup components
         self.render_pipeline.reset()
@@ -344,6 +358,57 @@ class VegasModeCoordinator:
             sys.setswitchinterval(saved)
             self._saved_switch_interval = None
 
+    # -- live elements ------------------------------------------------------
+
+    def _live_blocker(self) -> Optional[str]:
+        """Why live elements must stay off for this run, or None if they may run."""
+        cfg = self.vegas_config
+        if not getattr(cfg, 'live_refresh', False):
+            return "switched off (vegas_scroll.live_refresh)"
+        if getattr(self.render_pipeline, 'sync_manager', None) is not None:
+            # The follower mirrors whole strips only; a patch would not reach it.
+            return "multi-display sync is configured"
+        if not cfg.continuous_scroll:
+            return "swap mode (vegas_scroll.continuous_scroll is off)"
+        if not cfg.offscreen_prefetch:
+            return "vegas_scroll.offscreen_prefetch is off"
+        if not hasattr(self.display_manager, 'offscreen'):
+            return "the display manager has no off-screen canvas"
+        return None
+
+    def _apply_live_state(self) -> None:
+        """Switch live elements on or off for this run, as the config allows."""
+        blocker = self._live_blocker()
+        self._set_live(blocker is None, blocker)
+
+    def _set_live(self, active: bool, reason: Optional[str]) -> None:
+        was, self.live_active = self.live_active, active
+        # getattr: tests build coordinators without every component.
+        adapter = getattr(self, 'plugin_adapter', None)
+        if adapter is not None:
+            adapter.live_elements_enabled = active
+        plugin_manager = getattr(self, 'plugin_manager', None)
+        add = getattr(plugin_manager, 'add_update_listener', None)
+        remove = getattr(plugin_manager, 'remove_update_listener', None)
+        if active and callable(add):
+            add(self._on_plugin_data_changed)
+        elif not active and callable(remove):
+            remove(self._on_plugin_data_changed)
+        if active != was or (reason is not None and reason != self._live_reason):
+            if active:
+                logger.info("Vegas live elements on")
+            elif reason is not None:
+                logger.info("Vegas live elements off: %s", reason)
+        self._live_reason = reason
+
+    def _on_plugin_data_changed(self, plugin_id: str) -> None:
+        """Update listener: a plugin's data may have changed.
+
+        Runs on the update worker with the plugin's lock held, so it only
+        moves the plugin's epoch on; whatever redraws happen later read it.
+        """
+        self.live_epochs.bump(plugin_id)
+
     def _install_render_gate(self) -> None:
         """Gate the prefetch thread on the render thread's swaps; see VegasModeConfig."""
         if not self.vegas_config.prefetch_gate:
@@ -774,6 +839,8 @@ class VegasModeCoordinator:
             # Cached segments were trimmed under the old settings, so drop them
             # or a changed trim/padding value would not visibly take effect.
             self.plugin_adapter.invalidate_cache()
+            if self._is_active:
+                self._apply_live_state()
 
             # Force refresh of stream manager to pick up plugin_order/buffer changes
             self.stream_manager._last_refresh = 0

src/vegas_mode/elements.py

@@ -0,0 +1,138 @@
+"""Bookkeeping for live Vegas elements (see src/plugin_system/vegas_elements.py).
+
+A live element travels through the same plumbing as any other Vegas content --
+the adapter's cache, a prefetched group, the pipeline's join -- as a PIL image.
+What makes it live rides along in the image's ``info`` dict (:data:`INFO_KEY`),
+which Pillow copies through ``copy()``, ``crop()``, ``convert()`` and
+``resize()``, so none of that plumbing has to change shape. The pipeline reads
+the tag back when it places the image in the strip and keeps an
+:class:`ElementRecord` of where it went.
+
+Geometry is pinned: a live element is never trimmed to its ink. It is padded
+with ``content_padding`` black columns each side, the margin trimming would
+have left, so its width in the strip is its image width plus twice that, for
+as long as its key is there. That is what lets a redraw be swapped in place.
+"""
+
+from __future__ import annotations
+
+import itertools
+import threading
+import zlib
+from typing import Dict, NamedTuple, Optional, Tuple
+
+import numpy as np
+from PIL import Image
+
+#: Where a live element's :class:`ElementMeta` rides in ``Image.info``.
+INFO_KEY = "ledmatrix.vegas_element"
+
+
+class ElementMeta(NamedTuple):
+    """What the pipeline needs to know about one live element's pixels."""
+    plugin_id: str
+    key: str
+    #: The plugin's data epoch (LiveEpochs) the pixels were drawn from.
+    epoch: int
+    #: pixel_digest() of the pinned pixels.
+    digest: Tuple[Tuple[int, ...], int]
+    #: time.monotonic() when drawn.
+    rendered_at: float
+    refresh_hz: float
+    #: The plugin's own version for the pixels, or None.
+    version: object = None
+
+
+class ElementRecord(NamedTuple):
+    """Where one live element sits in the strip.
+
+    ``abs_x`` is in absolute strip columns: the strip's own column plus every
+    column trimmed off its front since it was composed (the pipeline's
+    ``_strip_origin``). Trimming therefore never moves a record.
+    """
+    seq: int
+    plugin_id: str
+    key: str
+    abs_x: int
+    width: int
+    epoch: int
+    digest: Tuple[Tuple[int, ...], int]
+    refresh_hz: float
+
+
+def tag(image: Image.Image, meta: ElementMeta) -> Image.Image:
+    """Mark ``image`` as the live element ``meta`` describes. Returns it."""
+    image.info[INFO_KEY] = meta
+    return image
+
+
+def meta_of(image: object) -> Optional[ElementMeta]:
+    """The live-element tag on ``image``, or None for plain content."""
+    info = getattr(image, 'info', None)
+    if not isinstance(info, dict):
+        return None
+    meta = info.get(INFO_KEY)
+    return meta if isinstance(meta, ElementMeta) else None
+
+
+def untag(image: Image.Image) -> Image.Image:
+    """Make ``image`` plain content again (e.g. after cropping it). Returns it."""
+    image.info.pop(INFO_KEY, None)
+    return image
+
+
+def pin_element(image: Image.Image, padding: int) -> Tuple[Image.Image, np.ndarray]:
+    """An element's pixels as they will sit in the strip, as image and array.
+
+    RGB, with ``padding`` black columns each side. The array is what a live
+    patch writes into the strip; it is read-only, so a patch in flight cannot
+    be changed under the render thread.
+    """
+    if image.mode != 'RGB':
+        image = image.convert('RGB')
+    pad = max(0, int(padding))
+    if pad:
+        pinned = Image.new('RGB', (image.width + 2 * pad, image.height), (0, 0, 0))
+        pinned.paste(image, (pad, 0))
+    else:
+        pinned = image.copy()
+    array = np.ascontiguousarray(np.asarray(pinned))
+    array.setflags(write=False)
+    return pinned, array
+
+
+def pixel_digest(array: np.ndarray) -> Tuple[Tuple[int, ...], int]:
+    """A cheap fingerprint of an element's pixels: its shape and a CRC.
+
+    Two redraws with the same digest are treated as the same pixels and the
+    second is not swapped in. CRC-32 rather than Adler-32: a changed digit is
+    a small, local change, which is exactly where Adler-32 is weakest.
+    """
+    data = np.ascontiguousarray(array)
+    return tuple(data.shape), zlib.crc32(memoryview(data).cast('B'))
+
+
+class LiveEpochs:
+    """A counter per plugin that moves on whenever its data may have changed.
+
+    Bumped when a plugin's update() completes (PluginManager's update
+    listener) or when it calls notify_vegas_data_changed(). Every live element
+    is tagged with the epoch it was drawn from; one drawn from an older epoch
+    than the plugin's current one is due a redraw. Epochs are the truth and
+    wake-ups only hints, so a missed wake-up delays a redraw but never loses
+    one.
+    """
+
+    def __init__(self) -> None:
+        self._counter = itertools.count(1)
+        self._epochs: Dict[str, int] = {}
+        self._lock = threading.Lock()
+
+    def bump(self, plugin_id: str) -> int:
+        with self._lock:
+            epoch = next(self._counter)
+            self._epochs[plugin_id] = epoch
+            return epoch
+
+    def get(self, plugin_id: str) -> int:
+        return self._epochs.get(plugin_id, 0)

src/vegas_mode/plugin_adapter.py

@@ -13,6 +13,17 @@ from typing import Optional, List, Any, Tuple, Union, TYPE_CHECKING
 from PIL import Image
 
 from src.common.scroll_helper import ScrollHelper
+from src.plugin_system.base_plugin import BasePlugin as _BasePlugin
+from src.plugin_system.vegas_elements import VegasElement
+from src.vegas_mode.elements import (
+    ElementMeta,
+    LiveEpochs,
+    meta_of,
+    pin_element,
+    pixel_digest,
+    tag,
+    untag,
+)
 from src.vegas_mode.geometry import (
     blank_runs,
     separation_gap,
@@ -88,6 +99,17 @@ class PluginAdapter:
         # into unrelated headlines once the strip refreshed to 9,505px.
         self._offset_shapes: dict = {}
 
+        # Live elements (src/vegas_mode/elements.py). Switched on by the
+        # coordinator for a run in which live updates are active; while off,
+        # no plugin is ever asked for elements and every path is as before.
+        self.live_elements_enabled = False
+        # Per-plugin data epochs, stamped on each element drawn. Set by the
+        # coordinator; without it every element is drawn "from epoch 0".
+        self.live_epochs: Optional[LiveEpochs] = None
+        # Element problems already reported, so a plugin with a bad hook logs
+        # once rather than on every fetch.
+        self._element_warnings: set = set()
+
         logger.debug(
             "PluginAdapter initialized: display=%dx%d",
             self.display_width, self.display_height
@@ -120,9 +142,24 @@ class PluginAdapter:
             plugin_id, plugin.__class__.__name__
         )
 
-        # Check cache first
+        # The old contract, kept behind the switch: background callers may
+        # not draw, so anything needing a canvas is left for the render thread.
+        restricted = offscreen_only and not getattr(
+            self.config, 'offscreen_prefetch', True)
+
+        # Live elements are asked for only on the background fetch, which
+        # holds the plugin's lock and draws on a canvas of its own. The render
+        # thread's fetches (the first compose, the inline fallback) take no
+        # lock, so they keep to get_vegas_content().
+        keyed = (offscreen_only and not restricted and self.live_elements_enabled
+                 and self.is_live_capable(plugin, plugin_id))
+
+        # Check cache first. A keyed fetch looks past legacy content cached
+        # by a render-thread fetch, or the plugin would not become live until
+        # that entry expired.
         cached = self._get_cached(plugin_id)
-        if cached is not None:
+        if cached is not None and not (
+                keyed and not any(meta_of(img) for img in cached)):
             total_width = sum(img.width for img in cached)
             logger.debug(
                 "[%s] Using cached content: %d images, %dpx total",
@@ -130,10 +167,6 @@ class PluginAdapter:
             )
             return cached
 
-        # The old contract, kept behind the switch: background callers may
-        # not draw, so anything needing a canvas is left for the render thread.
-        restricted = offscreen_only and not getattr(
-            self.config, 'offscreen_prefetch', True)
         if not offscreen_only or restricted:
             return self._fetch_content(plugin, plugin_id, restricted)
 
@@ -144,7 +177,25 @@ class PluginAdapter:
                     "round", plugin_id, self.PLUGIN_LOCK_TIMEOUT
                 )
                 return None
-            return self._fetch_content(plugin, plugin_id, restricted=False)
+            return self._fetch_content(plugin, plugin_id, restricted=False,
+                                       keyed=keyed)
+
+    def is_live_capable(self, plugin: 'BasePlugin', plugin_id: str) -> bool:
+        """Whether to ask this plugin for live elements rather than pictures.
+
+        It must implement get_vegas_elements() in its own class (a test double
+        or a plugin that only inherits BasePlugin's does not count), and its
+        config must not set ``vegas_live`` off.
+        """
+        method = getattr(type(plugin), 'get_vegas_elements', None)
+        if method is None or method is _BasePlugin.get_vegas_elements:
+            return False
+        raw = self._plugin_setting(plugin, 'vegas_live')
+        if raw is None:
+            return True
+        if isinstance(raw, str):
+            return raw.strip().lower() not in ('false', '0', 'off', 'no')
+        return bool(raw)
 
     @contextmanager
     def _plugin_lock(self, plugin_id: str):
@@ -189,13 +240,21 @@ class PluginAdapter:
             self.display_manager.image = original_image
 
     def _fetch_content(
-        self, plugin: 'BasePlugin', plugin_id: str, restricted: bool
+        self, plugin: 'BasePlugin', plugin_id: str, restricted: bool,
+        keyed: bool = False
     ) -> Optional[List[Image.Image]]:
-        """Every content path in order: native, scroll helper, display capture.
+        """Every content path in order: elements, native, scroll helper, capture.
 
         ``restricted`` is the pre-offscreen contract for background callers:
         skip every path that needs a canvas and return None instead.
+        ``keyed`` asks for live elements first (see get_content).
         """
+        if keyed:
+            content = self._get_keyed_content(plugin, plugin_id)
+            if content:
+                return self._finalize(content, plugin_id, 'elements', plugin)
+            logger.debug("[%s] No live elements; using its Vegas content", plugin_id)
+
         # Try native Vegas content method first
         has_native = hasattr(plugin, 'get_vegas_content')
         logger.debug("[%s] Has get_vegas_content: %s", plugin_id, has_native)
@@ -288,6 +347,12 @@ class PluginAdapter:
         dropped_blank = 0
 
         for img in images:
+            if meta_of(img) is not None:
+                # A live element is pinned, not trimmed: it already carries
+                # the margin trimming would leave, and its width must not
+                # follow its ink, or a redraw could never be swapped in place.
+                kept.append(img)
+                continue
             result = trim_to_content(
                 img,
                 threshold=self.config.trim_threshold,
@@ -609,7 +674,17 @@ class PluginAdapter:
             return images
 
         if len(images) == 1:
-            return [self._crop_to_budget(images[0], budget, plugin_id, mode)]
+            only = images[0]
+            pad = self.config.content_padding if self.config.auto_trim else 0
+            if meta_of(only) is not None and only.width - 2 * pad <= budget:
+                # A live element's pinned margins are not content. One whose
+                # drawing fits the budget is kept whole, and live, rather than
+                # cut for the sake of its own blank padding.
+                self._clear_offset(plugin_id)
+                return images
+            # A cropped live element is only part of itself, so it can no
+            # longer be swapped whole: it scrolls by as plain content.
+            return [untag(self._crop_to_budget(only, budget, plugin_id, mode))]
 
         shape = ('rows', len(images))
         if mode == 'truncate':
@@ -693,7 +768,12 @@ class PluginAdapter:
         # cycle — a lone "y" from "Wednesday" floating between two unrelated
         # plugins. Overshooting the budget is the lesser evil.
         min_run = max(2, self.config.min_cut_gap)
-        gaps = blank_runs(img, min_run, self.config.trim_threshold)
+        # A run touching either edge is the image's margin -- the
+        # content_padding trimming leaves, or a live element's pinned padding
+        # -- not a gap between items. Cutting mid-margin gave a window of a few
+        # blank columns, and a solid image with margins no continuous crop.
+        gaps = [(a, b) for a, b in blank_runs(img, min_run, self.config.trim_threshold)
+                if a > 0 and b < img.width]
 
         if not gaps:
             # No internal gaps means continuous content — a map, a chart, a
@@ -760,6 +840,114 @@ class PluginAdapter:
         )
         return img.crop((start, 0, end, img.height))
 
+    def _warn_element_once(self, plugin_id: str, problem: str, *args: Any) -> None:
+        """Report a plugin's element problem once per process, then quietly."""
+        key = (plugin_id, problem)
+        if key in self._element_warnings:
+            logger.debug("[%s] " + problem, plugin_id, *args)
+            return
+        self._element_warnings.add(key)
+        logger.warning("[%s] " + problem, plugin_id, *args)
+
+    def _get_keyed_content(
+        self, plugin: 'BasePlugin', plugin_id: str
+    ) -> Optional[List[Image.Image]]:
+        """The plugin's live elements, as tagged images, or None.
+
+        Called with the plugin's lock held (get_content), so update() is not
+        running and the plugin's data epoch cannot move while it draws. Drawn
+        on a canvas of the plugin's own at its render width, like
+        get_vegas_content(). Any failure returns None, and the caller falls
+        back to the plugin's ordinary Vegas content.
+        """
+        epochs = self.live_epochs
+        epoch = epochs.get(plugin_id) if epochs is not None else 0
+        render_width = self.resolve_render_width(plugin, plugin_id)
+        plugin._vegas_render_width = render_width
+        try:
+            with self._isolated_canvas(render_width):
+                result = plugin.get_vegas_elements()
+        except Exception as exc:  # pylint: disable=broad-except
+            # A plugin hook can raise anything; the legacy content still works.
+            self._warn_element_once(
+                plugin_id, "get_vegas_elements() raised %r; using its "
+                "get_vegas_content() instead", exc)
+            return None
+        finally:
+            plugin._vegas_render_width = None
+        try:
+            return self._images_from_elements(result, plugin_id, epoch)
+        except Exception as exc:  # pylint: disable=broad-except
+            # Converting is per element and guarded; this is the backstop, so
+            # nothing a plugin hands back can cost it its ordinary content.
+            self._warn_element_once(
+                plugin_id, "get_vegas_elements() returned elements that could not "
+                "be used (%r); using its get_vegas_content() instead", exc)
+            return None
+
+    def _images_from_elements(
+        self, result: Any, plugin_id: str, epoch: int
+    ) -> Optional[List[Image.Image]]:
+        """Turn get_vegas_elements()'s answer into images for the pipeline.
+
+        Live elements come out pinned (RGB, display height, content_padding
+        black each side, never trimmed afterwards) and tagged with their
+        ElementMeta; plain ones (``live=False``) come out as ordinary content.
+        Anything that is not a usable element is dropped with a warning; a
+        duplicate key keeps its first element.
+        """
+        if result is None:
+            return None
+        if not isinstance(result, (list, tuple)):
+            self._warn_element_once(
+                plugin_id, "get_vegas_elements() returned %s, expected a list "
+                "of VegasElement", type(result).__name__)
+            return None
+
+        padding = self.config.content_padding if self.config.auto_trim else 0
+        now = time.monotonic()
+        seen = set()
+        images: List[Image.Image] = []
+        for element in result:
+            if not (isinstance(element, VegasElement)
+                    and isinstance(element.key, str) and element.key
+                    and isinstance(element.image, Image.Image)):
+                self._warn_element_once(
+                    plugin_id, "get_vegas_elements() returned an item that is "
+                    "not a VegasElement with a key and an image (%s); skipping it",
+                    type(element).__name__)
+                continue
+            if element.key in seen:
+                self._warn_element_once(
+                    plugin_id, "get_vegas_elements() returned key %r twice; "
+                    "keeping the first", element.key)
+                continue
+            seen.add(element.key)
+
+            image = element.image
+            if image.height != self.display_height:
+                image = image.resize((image.width, self.display_height),
+                                     Image.Resampling.LANCZOS)
+            if image.mode != 'RGB':
+                image = image.convert('RGB')
+
+            if not element.live:
+                # Plain content; a tag copied from a reused image must not
+                # make it live by accident.
+                images.append(untag(image.copy()) if meta_of(image) else image)
+                continue
+
+            pinned, pixels = pin_element(image, padding)
+            try:
+                refresh_hz = max(0.0, float(element.refresh_hz or 0.0))
+            except (TypeError, ValueError):
+                refresh_hz = 0.0
+            images.append(tag(pinned, ElementMeta(
+                plugin_id=plugin_id, key=element.key, epoch=epoch,
+                digest=pixel_digest(pixels), rendered_at=now,
+                refresh_hz=refresh_hz, version=element.version)))
+        return images or None
+
     def _get_native_content(
         self, plugin: 'BasePlugin', plugin_id: str, restricted: bool = False
     ) -> Optional[List[Image.Image]]:

src/vegas_mode/render_pipeline.py

@@ -5,6 +5,7 @@ Composes plugin content into one wide strip and renders the visible window of
 it each frame, using ScrollHelper for the numpy-backed scroll.
 """
 
+import itertools
 import logging
 import os
 import time
@@ -18,6 +19,7 @@ from src.common.scroll_config import solve_crisp
 from src.common.scroll_helper import ScrollHelper
 from src.matrix_support import DEFAULT_REFRESH_LIMIT_HZ
 from src.vegas_mode.config import VegasModeConfig
+from src.vegas_mode.elements import ElementMeta, ElementRecord, meta_of
 from src.vegas_mode.geometry import separation_gap
 from src.vegas_mode.stream_manager import StreamManager
 
@@ -67,6 +69,15 @@ class RenderPipeline:
     # without __init__ (tests).
     _static_markers: Tuple[Tuple[int, str], ...] = ()
 
+    # Live elements in the strip (see "live element records" below). Replaced,
+    # never mutated, like _static_markers, and class-level for the same reason.
+    _elements: Tuple[ElementRecord, ...] = ()
+    # Columns trimmed off the strip's front since it was composed.
+    _strip_origin: int = 0
+    # Bumped whenever a new strip replaces the old one (compose, reset), so
+    # anything computed against the old strip can tell.
+    _strip_gen: int = 0
+
     def __init__(
         self,
         config: VegasModeConfig,
@@ -117,6 +128,7 @@ class RenderPipeline:
         # Render state
         self._cycle_complete = False
         self._segments_in_scroll: List[str] = []  # Plugin IDs in current scroll
+        self._record_by_seq: Dict[int, ElementRecord] = {}
 
         # The sub-pixel path's pacing; the crisp path solves its own (frame_interval).
         self._frame_interval = config.get_frame_interval()
@@ -292,6 +304,8 @@ class RenderPipeline:
             # plugin boundaries only.
             grouped = self.stream_manager.get_grouped_content_for_composition()
             self._static_markers = ()
+            # A compose replaces the strip, and every record with it.
+            self._reset_records()
 
             if not grouped:
                 logger.warning("No content available for composition")
@@ -304,10 +318,13 @@ class RenderPipeline:
             # row". Without this, a per-row ticker such as the F1 scoreboard got
             # the full separator between each of its ~116 rows.
             blocks = []
+            layouts = []
             total_rows = 0
-            for plugin_id, images in grouped:
+            for _plugin_id, images in grouped:
                 total_rows += len(images)
-                blocks.append(self._join_plugin_rows(images))
+                block, layout = self._join_plugin_rows_with_layout(images)
+                blocks.append(block)
+                layouts.append(layout)
 
             # Create scrolling image via ScrollHelper.
             #
@@ -328,6 +345,10 @@ class RenderPipeline:
                 return False
 
             self._static_markers = self._markers_for_composition(blocks)
+            self._register_elements(
+                self._block_starts([b.width for b in blocks], 0, False,
+                                   lead=self.config.lead_in_width),
+                layouts)
             self._note_op('compose', self._strip_nbytes())
 
             # Track which plugins are in this scroll (get safely via buffer status)
@@ -633,12 +654,18 @@ class RenderPipeline:
                 return bool(deferred)
 
             blocks = []
+            layouts = []
             total_rows = 0
             for _plugin_id, images in grouped:
                 total_rows += len(images)
-                blocks.append(self._join_plugin_rows(images))
+                block, layout = self._join_plugin_rows_with_layout(images)
+                blocks.append(block)
+                layouts.append(layout)
 
             had_strip = self.scroll_helper.has_strip()
+            if not had_strip:
+                # append_content is about to build a strip from scratch.
+                self._reset_records()
             appended = self.scroll_helper.append_content(
                 content_items=blocks,
                 item_gap=self.config.separator_width,
@@ -648,16 +675,13 @@ class RenderPipeline:
                 return False
             moved = self._strip_nbytes()
 
+            # Where each block starts, laid out as append_content does: a
+            # separator before every block, or -- when there was no strip to
+            # extend -- as create_scrolling_image does with no lead-in.
+            starts = self._block_starts([b.width for b in blocks], strip_end, had_strip)
+            self._register_elements(starts, layouts)
             if statics:
-                # Where each block ends, laid out as append_content does: a
-                # separator before every block, or -- when there was no strip
-                # to extend -- as create_scrolling_image does with no lead-in.
-                gap = max(0, self.config.separator_width)
-                ends = []
-                x = strip_end if had_strip else -gap
-                for block in blocks:
-                    x += gap + block.width
-                    ends.append(x)
+                ends = [start + block.width for start, block in zip(starts, blocks)]
                 self._add_static_markers([
                     (ends[n - 1] if n > 0 else strip_end, pid) for n, pid in statics
                 ])
@@ -667,6 +691,7 @@ class RenderPipeline:
             if cut and self._static_markers:
                 self._static_markers = tuple(
                     (max(0, x - cut), pid) for x, pid in self._static_markers)
+            self._forget_trimmed_records(cut)
             # The append built the whole strip anew, and a trim copies what is
             # left of it again: both land in the frame after this one.
             self._note_op('extend', moved + (self._strip_nbytes() if cut else 0))
@@ -703,8 +728,22 @@ class RenderPipeline:
             ``intra_plugin_gap``. Returned unchanged when there is only one row,
             which is the common case and avoids a pointless copy.
         """
+        return self._join_plugin_rows_with_layout(images)[0]
+
+    def _join_plugin_rows_with_layout(
+        self, images: List[Image.Image]
+    ) -> Tuple[Image.Image, List[Tuple[int, ElementMeta, int]]]:
+        """_join_plugin_rows, plus where each live element landed in the block.
+
+        Returns ``(block, layout)``, layout being ``(x, meta, width)`` for
+        every image tagged as a live element (src/vegas_mode/elements.py), x
+        measured from the block's left edge. The offsets were always computed
+        here; they used to be thrown away.
+        """
         if len(images) == 1:
-            return images[0]
+            meta = meta_of(images[0])
+            layout = [(0, meta, images[0].width)] if meta is not None else []
+            return images[0], layout
 
         floor = max(0, self.config.intra_plugin_gap)
         target = max(0, self.config.min_content_separation)
@@ -723,11 +762,100 @@ class RenderPipeline:
         height = max(img.height for img in images)
 
         block = Image.new('RGB', (width, height), (0, 0, 0))
+        layout: List[Tuple[int, ElementMeta, int]] = []
         x = 0
         for i, img in enumerate(images):
             block.paste(img, (x, 0))
+            meta = meta_of(img)
+            if meta is not None:
+                layout.append((x, meta, img.width))
             x += img.width + (gaps[i] if i < len(gaps) else 0)
-        return block
+        return block, layout
+
+    # -- live element records ---------------------------------------------
+    #
+    # Where each live element sits in the strip (ElementRecord), kept so a
+    # redraw can later be swapped into exactly its columns. Coordinates are
+    # absolute: a record's column in the strip is abs_x - _strip_origin, and
+    # a trim moves the origin instead of every record. Only the render thread
+    # changes any of this, at the points where it builds or trims the strip.
+
+    def _block_starts(self, widths: List[int], strip_end: int, had_strip: bool,
+                      lead: int = 0) -> List[int]:
+        """Strip columns where each of these blocks starts once placed.
+
+        Mirrors ScrollHelper exactly: append_content puts a separator before
+        every block after an existing strip; create_scrolling_image (a compose,
+        or an append with nothing to extend) puts ``lead`` columns first and a
+        separator between blocks.
+        """
+        gap = max(0, self.config.separator_width)
+        starts = []
+        if had_strip:
+            x = strip_end
+            for width in widths:
+                x += gap
+                starts.append(x)
+                x += width
+        else:
+            x = max(0, int(lead))
+            for width in widths:
+                starts.append(x)
+                x += width + gap
+        return starts
+
+    def _next_record_seq(self) -> int:
+        counter = self.__dict__.get('_record_counter')
+        if counter is None:
+            counter = self._record_counter = itertools.count(1)
+        return next(counter)
+
+    def _register_elements(
+        self, starts: List[int], layouts: List[List[Tuple[int, ElementMeta, int]]]
+    ) -> int:
+        """Record every live element in blocks just placed at ``starts``."""
+        new = []
+        for start, layout in zip(starts, layouts):
+            for offset, meta, width in layout:
+                new.append(ElementRecord(
+                    seq=self._next_record_seq(), plugin_id=meta.plugin_id,
+                    key=meta.key, abs_x=self._strip_origin + start + offset,
+                    width=width, epoch=meta.epoch, digest=meta.digest,
+                    refresh_hz=meta.refresh_hz))
+        if new:
+            self._elements = self._elements + tuple(new)
+            by_seq = self.__dict__.setdefault('_record_by_seq', {})
+            for record in new:
+                by_seq[record.seq] = record
+        return len(new)
+
+    def _forget_trimmed_records(self, cut: int) -> None:
+        """The strip lost ``cut`` columns off its front: move the origin on."""
+        if cut <= 0:
+            return
+        self._strip_origin += cut
+        origin = self._strip_origin
+        records = self._elements
+        if not records:
+            return
+        kept = tuple(r for r in records if r.abs_x + r.width > origin)
+        if len(kept) != len(records):
+            by_seq = self.__dict__.setdefault('_record_by_seq', {})
+            for record in records:
+                if record.abs_x + record.width <= origin:
+                    by_seq.pop(record.seq, None)
+            self._elements = kept
+
+    def _reset_records(self) -> None:
+        """A new strip: nothing recorded, coordinates from zero, a new generation."""
+        self._strip_gen += 1
+        self._strip_origin = 0
+        self._elements = ()
+        self._record_by_seq = {}
+
+    def live_records(self) -> Tuple[ElementRecord, ...]:
+        """The live elements in the strip, in the order they were placed."""
+        return self._elements
 
     def render_frame(self) -> bool:
         """
@@ -1059,6 +1187,7 @@ class RenderPipeline:
             self._prepared_group = None
             self._deferred_queue = []
         self._static_markers = ()
+        self._reset_records()
 
         self.display_manager.set_scrolling_state(False)
 

test/fixtures/plugins/vegas-live-stub/config_schema.json

@@ -0,0 +1,39 @@
+{
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "type": "object",
+    "properties": {
+        "enabled": {
+            "type": "boolean",
+            "default": false,
+            "description": "Enable the stub (test fixture only)."
+        },
+        "cards": {
+            "type": "integer",
+            "minimum": 0,
+            "maximum": 32,
+            "default": 6,
+            "description": "How many keyed cards to return."
+        },
+        "card_width": {
+            "type": "integer",
+            "minimum": 0,
+            "maximum": 512,
+            "default": 0,
+            "description": "Card width in px; 0 sizes cards from the render width."
+        },
+        "map_hz": {
+            "type": "number",
+            "minimum": 0,
+            "maximum": 30,
+            "default": 4,
+            "description": "refresh_hz of the full-width animated element; 0 leaves it out."
+        },
+        "dot_speed": {
+            "type": "number",
+            "minimum": 0,
+            "maximum": 200,
+            "default": 20,
+            "description": "How fast the animated element's dot moves, in px per second."
+        }
+    }
+}

test/fixtures/plugins/vegas-live-stub/manager.py

@@ -0,0 +1,134 @@
+"""
+Vegas live-element stub.
+
+A fixture, not a product: it exercises every part of the live-element contract
+(src/plugin_system/vegas_elements.py) with content whose changes are easy to
+see and to assert on, and nothing else -- no fonts, no network.
+
+- ``card:<n>`` -- fixed-width cards. Each draws the bits of ``tick + n`` as
+  lit bars, so every update() changes every card's pixels but never its width.
+- ``sep`` -- a separator, ``live=False``: placed and trimmed like plain content.
+- ``map`` -- one full-render-width element with ``refresh_hz``: a dot that
+  moves across it with time, drawn by redraw_vegas_element() from state
+  published in a single attribute store, so it is safe to call without the
+  plugin's lock.
+
+update() only advances the tick. display() draws the tick's bars full screen
+so the plugin also passes the ordinary rendering harness.
+"""
+
+import time
+from typing import List, Optional, Tuple
+
+from PIL import Image, ImageDraw
+
+from src.plugin_system.base_plugin import BasePlugin
+
+try:
+    from src.plugin_system.vegas_elements import VegasElement
+except ImportError:  # core older than 3.8.0: the hooks are never called
+    VegasElement = None
+
+_COLOURS = [(255, 64, 64), (64, 255, 64), (64, 128, 255), (255, 200, 0),
+            (255, 64, 255), (0, 220, 220)]
+
+
+class VegasLiveStub(BasePlugin):
+    """Keyed cards, a separator and an animated element for the Vegas ticker."""
+
+    def __init__(self, plugin_id, config, display_manager, cache_manager, plugin_manager):
+        super().__init__(plugin_id, config, display_manager, cache_manager, plugin_manager)
+        self.tick = 0
+        # Everything the lock-free redraw reads, published in one store.
+        self._snapshot: Tuple[int, float] = (0, time.monotonic())
+
+    # -- data -------------------------------------------------------------
+
+    def update(self) -> None:
+        self.tick += 1
+        self._snapshot = (self.tick, time.monotonic())
+
+    # -- drawing ----------------------------------------------------------
+
+    def _bars(self, image: Image.Image, value: int, colour, box) -> None:
+        x0, y0, x1, y1 = box
+        draw = ImageDraw.Draw(image)
+        draw.rectangle([x0, y0, x1, y1], outline=colour)
+        bits = 8
+        span = max(1, (x1 - x0 - 2) // bits)
+        for bit in range(bits):
+            if value >> bit & 1:
+                left = x0 + 1 + bit * span
+                draw.rectangle([left, y0 + 2, left + max(0, span - 2), y1 - 2],
+                               fill=colour)
+
+    def _card_width(self) -> int:
+        configured = int(self.config.get('card_width', 0) or 0)
+        if configured > 0:
+            return configured
+        return max(24, min(64, self.get_vegas_render_width() // 4))
+
+    def _card(self, index: int, tick: int) -> Image.Image:
+        width, height = self._card_width(), self.display_manager.height
+        image = Image.new('RGB', (width, height), (0, 0, 0))
+        self._bars(image, tick + index, _COLOURS[index % len(_COLOURS)],
+                   (0, 0, width - 1, height - 1))
+        return image
+
+    def _map(self, width: int, height: int, at: float) -> Image.Image:
+        tick, _published = self._snapshot
+        image = Image.new('RGB', (width, height), (0, 0, 16))
+        draw = ImageDraw.Draw(image)
+        draw.rectangle([0, 0, width - 1, height - 1], outline=(40, 40, 80))
+        speed = float(self.config.get('dot_speed', 20) or 0)
+        x = int(at * speed) % max(1, width - 4) + 2
+        y = 2 + tick % max(1, height - 4)
+        draw.rectangle([x - 1, y - 1, x + 1, y + 1], fill=(255, 255, 255))
+        return image
+
+    def _dot_column(self, width: int, at: float) -> int:
+        speed = float(self.config.get('dot_speed', 20) or 0)
+        return int(at * speed) % max(1, width - 4) + 2
+
+    def display(self, force_clear: bool = False) -> bool:
+        width, height = self.display_manager.width, self.display_manager.height
+        self.display_manager.clear()
+        self._bars(self.display_manager.image, self.tick, _COLOURS[0],
+                   (0, 0, width - 1, height - 1))
+        self.display_manager.update_display()
+        return True
+
+    # -- Vegas ------------------------------------------------------------
+
+    def get_vegas_content(self) -> Optional[List[Image.Image]]:
+        cards = int(self.config.get('cards', 6))
+        return [self._card(i, self.tick) for i in range(cards)] or None
+
+    def get_vegas_elements(self):
+        if VegasElement is None:
+            return None
+        tick = self.tick
+        elements = []
+        for i in range(int(self.config.get('cards', 6))):
+            elements.append(VegasElement(
+                key=f"card:{i}", image=self._card(i, tick),
+                version=(tick, i, self._card_width())))
+            if i == 0:
+                separator = Image.new('RGB', (4, self.display_manager.height), (0, 0, 0))
+                ImageDraw.Draw(separator).rectangle(
+                    [1, 0, 2, self.display_manager.height - 1], fill=(90, 90, 90))
+                elements.append(VegasElement(key="sep", image=separator, live=False))
+        hz = float(self.config.get('map_hz', 4) or 0)
+        if hz > 0:
+            width, height = self.get_vegas_render_width(), self.display_manager.height
+            now = time.monotonic()
+            elements.append(VegasElement(
+                key="map", image=self._map(width, height, now),
+                version=(tick, width, self._dot_column(width, now)),
+                refresh_hz=hz))
+        return elements
+
+    def redraw_vegas_element(self, key, width, height, at):
+        if key != "map":
+            return None
+        return self._map(width, height, at)

test/fixtures/plugins/vegas-live-stub/manifest.json

@@ -0,0 +1,13 @@
+{
+    "id": "vegas-live-stub",
+    "name": "Vegas Live Stub",
+    "version": "1.0.0",
+    "description": "Test fixture for live Vegas elements: keyed cards whose content changes on every update, a separator, and a full-width element that animates with time. Drives the live-element tests and the hardware soaks. Not installable from the store and never shipped to devices.",
+    "author": "LEDMatrix",
+    "entry_point": "manager.py",
+    "class_name": "VegasLiveStub",
+    "display_modes": ["vegas-live-stub"],
+    "update_interval": 2,
+    "min_ledmatrix_version": "2.0.0",
+    "compatible_versions": [">=2.0.0"]
+}

test/test_harness_vegas_elements.py

@@ -0,0 +1,180 @@
+"""The offline live-element checks (src/plugin_system/testing/vegas.py).
+
+Run against the stub fixture plugin, which honours the contract, and against
+small broken plugins, each breaking one clause of it.
+"""
+import sys
+from pathlib import Path
+
+from PIL import Image
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.plugin_system.base_plugin import BasePlugin  # noqa: E402
+from src.plugin_system.testing.harness import _instantiate  # noqa: E402
+from src.plugin_system.testing.loading import build_full_config, load_harness_spec, load_manifest  # noqa: E402
+from src.plugin_system.testing.vegas import (  # noqa: E402
+    check_vegas_elements, implements_vegas_elements, render_vegas_elements,
+)
+from src.plugin_system.testing.visual_display_manager import VisualTestDisplayManager  # noqa: E402
+from src.plugin_system.vegas_elements import VegasElement  # noqa: E402
+
+STUB = Path(__file__).resolve().parent / "fixtures" / "plugins" / "vegas-live-stub"
+W, H = 192, 48
+
+
+def _stub(**config):
+    dm = VisualTestDisplayManager(width=W, height=H)
+    full = {**build_full_config(STUB, load_harness_spec(STUB), {}), **config}
+    plugin = _instantiate("vegas-live-stub", load_manifest(STUB), STUB, full, {}, dm)
+    return plugin, dm
+
+
+class _Broken(BasePlugin):
+    """A plugin whose get_vegas_elements returns whatever it is given."""
+
+    def __init__(self, dm, result, redraw=None):
+        self.plugin_id = "broken"
+        self.config = {}
+        self.display_manager = dm
+        self.plugin_manager = None
+        self._result = result
+        self._redraw = redraw
+
+    def update(self):
+        pass
+
+    def display(self, force_clear=False):
+        pass
+
+    def get_vegas_elements(self):
+        return self._result() if callable(self._result) else self._result
+
+    def redraw_vegas_element(self, key, width, height, at):
+        return self._redraw(width, height) if self._redraw else None
+
+
+def _img(w, h=H):
+    return Image.new("RGB", (w, h), (255, 0, 0))
+
+
+def test_the_stub_passes_every_check():
+    plugin, dm = _stub()
+    report = check_vegas_elements(plugin, dm)
+    assert report.implemented and report.ok, report.errors
+    assert report.elements == 8 and report.live == 7
+    assert not report.warnings, report.warnings
+
+
+def test_the_stub_renders_at_the_width_it_is_given():
+    plugin, dm = _stub()
+    elements = render_vegas_elements(plugin, dm, width=96)
+    by_key = {e.key: e for e in elements}
+    assert by_key["map"].image.size == (96, H)
+    assert by_key["card:0"].image.width == 24
+    assert plugin.get_vegas_render_width() == W   # restored afterwards
+
+
+def test_the_stubs_cards_change_on_update_but_keep_their_width():
+    plugin, dm = _stub()
+    before = {e.key: e for e in render_vegas_elements(plugin, dm)}
+    plugin.update()
+    after = {e.key: e for e in render_vegas_elements(plugin, dm)}
+    for key in ("card:0", "card:3"):
+        assert after[key].version != before[key].version
+        assert after[key].image.size == before[key].image.size
+        assert after[key].image.tobytes() != before[key].image.tobytes()
+
+
+def test_a_plugin_without_the_hook_is_not_checked():
+    class Plain(BasePlugin):
+        def update(self):
+            pass
+
+        def display(self, force_clear=False):
+            pass
+
+    plugin = Plain.__new__(Plain)
+    dm = VisualTestDisplayManager(width=W, height=H)
+    report = check_vegas_elements(plugin, dm)
+    assert not implements_vegas_elements(plugin)
+    assert not report.implemented and report.ok
+
+
+def _errors(result, redraw=None):
+    dm = VisualTestDisplayManager(width=W, height=H)
+    return check_vegas_elements(_Broken(dm, result, redraw), dm)
+
+
+def test_each_broken_clause_is_an_error():
+    assert "expected a list" in _errors("x").errors[0]
+    assert "not a VegasElement" in _errors([object()]).errors[0]
+    assert "twice" in _errors([VegasElement("k", _img(10)),
+                               VegasElement("k", _img(10))]).errors[0]
+    assert "tall" in _errors([VegasElement("k", _img(10, H + 1))]).errors[0]
+    assert "no key" in _errors([VegasElement("", _img(10))]).errors[0]
+
+
+def test_a_width_that_changes_with_nothing_new_is_an_error():
+    widths = iter([10, 12, 10, 10])
+    report = _errors(lambda: [VegasElement("k", _img(next(widths)))])
+    assert any("changed width" in e for e in report.errors)
+
+
+def test_a_redraw_of_the_wrong_size_is_an_error():
+    report = _errors([VegasElement("m", _img(40), refresh_hz=2)],
+                     redraw=lambda w, h: _img(w + 1, h))
+    assert any("asked for 40x48" in e for e in report.errors)
+
+
+def test_animation_without_a_redraw_is_a_warning():
+    dm = VisualTestDisplayManager(width=W, height=H)
+
+    class NoRedraw(_Broken):
+        redraw_vegas_element = BasePlugin.redraw_vegas_element
+
+    report = check_vegas_elements(
+        NoRedraw(dm, [VegasElement("m", _img(40), refresh_hz=2)]), dm)
+    assert report.ok
+    assert any("not implemented" in w for w in report.warnings)
+
+
+def test_none_means_legacy_content_and_is_only_a_warning():
+    report = _errors(None)
+    assert report.ok and "get_vegas_content" in report.warnings[0]
+
+
+def test_a_refresh_rate_that_is_not_a_number_is_an_error_not_a_crash():
+    report = _errors([VegasElement("m", _img(40), refresh_hz="fast")])
+    assert any("not a number" in e for e in report.errors)
+    # None is what a plugin passing the dataclass default through gets.
+    assert _errors([VegasElement("m", _img(40), refresh_hz=None)]).ok
+
+
+def test_an_empty_image_is_an_error():
+    assert any("empty" in e for e in _errors([VegasElement("k", _img(0))]).errors)
+
+
+def test_a_second_call_that_breaks_the_contract_is_an_error_not_a_crash():
+    answers = iter([[VegasElement("k", _img(10))], "x", "x", "x"])
+    report = _errors(lambda: next(answers))
+    assert any("second call" in e for e in report.errors)
+    answers = iter([[VegasElement("k", _img(10))], [VegasElement("k", "not an image")],
+                    [], []])
+    report = _errors(lambda: next(answers))
+    assert any("disappeared" in e for e in report.errors)
+
+
+def test_check_plugin_reports_a_failing_element_check_and_carries_on(monkeypatch):
+    sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
+    import check_plugin
+
+    def boom(*args, **kwargs):
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr(check_plugin, "check_plugin_vegas_elements", boom)
+    results = check_plugin.check_one(
+        "vegas-live-stub", [str(STUB.parent)], [(W, H)], {}, {}, False, None,
+        False, None, None)
+    vegas = [r for r in results if r.mode == "vegas elements"]
+    assert len(vegas) == 1 and "boom" in vegas[0].error

test/test_plugin_update_listener.py

@@ -0,0 +1,143 @@
+"""PluginManager's update listeners: told the moment a plugin's data may have changed.
+
+Vegas live elements redraw a plugin when its update() completes. Before these
+listeners the only signal was a set drained by the Vegas tick every ~4s; the
+listener hears it at once. It is called on the update worker with the
+plugin's lock still held, which is why a listener may only hand off.
+"""
+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.base_plugin import BasePlugin  # noqa: E402
+from src.plugin_system.plugin_manager import PluginManager  # noqa: E402
+from src.plugin_system.plugin_state import PluginState  # noqa: E402
+
+
+@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()
+
+
+class _Plugin:
+    def __init__(self, fail=False):
+        self.enabled = True
+        self.fail = fail
+        self.updates = 0
+
+    def update(self):
+        self.updates += 1
+        if self.fail:
+            raise RuntimeError("no data")
+
+    def display(self, force_clear=False):
+        return True
+
+
+def _install(pm, plugin, plugin_id="p"):
+    pm.plugins[plugin_id] = plugin
+    pm._update_interval_cache[plugin_id] = 0.01
+    pm.state_manager.set_state(plugin_id, PluginState.ENABLED)
+    return plugin_id
+
+
+def _wait(predicate, timeout=5.0):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if predicate():
+            return True
+        time.sleep(0.01)
+    return False
+
+
+def test_a_completed_update_calls_the_listener_with_the_lock_held(pm):
+    plugin_id = _install(pm, _Plugin())
+    heard = []
+
+    def listener(pid):
+        heard.append((pid, pm.get_plugin_lock(pid).locked()))
+
+    pm.add_update_listener(listener)
+    pm.run_scheduled_updates()
+    assert _wait(lambda: heard)
+    assert heard[0] == (plugin_id, True)
+    # The poll still sees it too: the set is filled before listeners run.
+    assert plugin_id in pm.drain_completed_updates()
+
+
+def test_a_failed_update_is_not_reported(pm):
+    _install(pm, _Plugin(fail=True))
+    heard = []
+    pm.add_update_listener(heard.append)
+    pm.run_scheduled_updates()
+    assert _wait(lambda: pm.plugins["p"].updates == 1)
+    time.sleep(0.1)
+    assert heard == []
+
+
+def test_a_listener_that_raises_does_not_stop_the_others(pm):
+    heard = []
+
+    def broken(_pid):
+        raise ValueError("listener bug")
+
+    pm.add_update_listener(broken)
+    pm.add_update_listener(heard.append)
+    pm._note_update_completed("p")     # must not raise
+    assert heard == ["p"]
+    assert "p" in pm.drain_completed_updates()
+
+
+def test_adding_twice_calls_once_and_removing_stops_it(pm):
+    heard = []
+    pm.add_update_listener(heard.append)
+    pm.add_update_listener(heard.append)
+    pm.notify_data_changed("x")
+    assert heard == ["x"]
+    pm.remove_update_listener(heard.append)
+    pm.notify_data_changed("y")
+    assert heard == ["x"]
+
+
+def test_notify_data_changed_reaches_listeners_but_not_the_poll(pm):
+    heard = []
+    pm.add_update_listener(heard.append)
+    pm.notify_data_changed("q")
+    assert heard == ["q"]
+    assert pm.drain_completed_updates() == []
+
+
+def test_a_plugin_can_report_data_that_arrived_on_its_own_thread(pm):
+    class Pushed(BasePlugin):
+        def update(self):
+            pass
+
+        def display(self, force_clear=False):
+            pass
+
+    plugin = Pushed.__new__(Pushed)
+    plugin.plugin_id = "pushed"
+    plugin.plugin_manager = pm
+    heard = []
+    pm.add_update_listener(heard.append)
+    thread = threading.Thread(target=plugin.notify_vegas_data_changed)
+    thread.start()
+    thread.join()
+    assert heard == ["pushed"]
+
+
+def test_a_bare_manager_has_no_listeners_to_call():
+    manager = PluginManager.__new__(PluginManager)
+    manager._completed_updates = set()
+    manager._completed_updates_lock = threading.Lock()
+    manager._note_update_completed("p")
+    manager.add_update_listener(lambda pid: None)
+    manager.remove_update_listener(lambda pid: None)

test/test_vegas_elements_api.py

@@ -0,0 +1,331 @@
+"""Live Vegas elements: the plugin API and how the adapter fetches them.
+
+A plugin opts in by implementing get_vegas_elements(); the adapter then asks
+for elements instead of pictures, but only where that is safe (the background
+fetch, under the plugin's lock, with live elements switched on) and falls back
+to get_vegas_content() everywhere else. What makes an element live rides in
+its image's ``info`` so it survives the adapter's cache and the pipeline's
+join unchanged. These tests pin all of that.
+"""
+import dataclasses
+import sys
+import threading
+from contextlib import contextmanager
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import numpy as np
+import pytest
+from PIL import Image, ImageDraw
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.plugin_system.base_plugin import BasePlugin  # noqa: E402
+from src.plugin_system.vegas_elements import VegasElement  # noqa: E402
+from src.vegas_mode import elements  # noqa: E402
+from src.vegas_mode.config import VegasModeConfig  # noqa: E402
+from src.vegas_mode.elements import ElementMeta, LiveEpochs  # noqa: E402
+from src.vegas_mode.plugin_adapter import PluginAdapter  # noqa: E402
+
+W, H = 128, 32
+
+
+def _card(width=40, colour=(255, 0, 0), height=H):
+    image = Image.new("RGB", (width, height), (0, 0, 0))
+    ImageDraw.Draw(image).rectangle([0, 0, width - 1, height - 1], outline=colour)
+    return image
+
+
+class _DM:
+    """A display manager with the per-thread canvas Vegas renders on."""
+
+    def __init__(self):
+        self.width, self.height = W, H
+        self.image = Image.new("RGB", (W, H))
+        self.offscreen_widths = []
+
+    @contextmanager
+    def offscreen(self, width=None, height=None):
+        self.offscreen_widths.append(width)
+        yield SimpleNamespace(image=Image.new("RGB", (width or W, H)))
+
+
+class _Plugin(BasePlugin):
+    """A BasePlugin with both the legacy and the element hooks."""
+
+    def __init__(self, elements_result=None, config=None):
+        self.plugin_id = "p"
+        self.config = config or {}
+        self.elements_result = elements_result
+        self.element_calls = 0
+        self.content_calls = 0
+        self.render_widths = []
+        self.plugin_manager = None
+
+    def update(self):
+        pass
+
+    def display(self, force_clear=False):
+        pass
+
+    def get_vegas_content(self):
+        self.content_calls += 1
+        return [_card(40, (0, 0, 255))]
+
+    def get_vegas_elements(self):
+        self.element_calls += 1
+        self.render_widths.append(self.get_vegas_render_width())
+        result = self.elements_result
+        if isinstance(result, Exception):
+            raise result
+        return result() if callable(result) else result
+
+
+class _Legacy(_Plugin):
+    get_vegas_elements = BasePlugin.get_vegas_elements
+
+
+def _adapter(**cfg):
+    lock = threading.Lock()
+    pm = SimpleNamespace(get_plugin_lock=lambda pid: lock)
+    adapter = PluginAdapter(_DM(), VegasModeConfig(**cfg), plugin_manager=pm)
+    adapter.live_elements_enabled = True
+    adapter.live_epochs = LiveEpochs()
+    return adapter
+
+
+def _elements():
+    return [VegasElement("card:a", _card(40, (255, 0, 0)), version=1),
+            VegasElement("sep", _card(10, (90, 90, 90)), live=False),
+            VegasElement("card:b", _card(40, (0, 255, 0)), version=2)]
+
+
+# -- the type and the tag -----------------------------------------------------
+
+
+def test_element_defaults():
+    element = VegasElement("k", _card())
+    assert element.live and element.refresh_hz == 0 and element.version is None
+    with pytest.raises(dataclasses.FrozenInstanceError):
+        element.key = "other"
+
+
+def test_a_tag_survives_what_the_plumbing_does_to_an_image():
+    meta = ElementMeta("p", "k", 3, ((H, 10, 3), 1), 0.0, 0.0)
+    image = elements.tag(_card(), meta)
+    for derived in (image.copy(), image.crop((0, 0, 10, H)), image.convert("RGB"),
+                    image.resize((20, H))):
+        assert elements.meta_of(derived) == meta
+    assert elements.meta_of(elements.untag(image.copy())) is None
+    assert elements.meta_of(_card()) is None
+    assert elements.meta_of(object()) is None
+
+
+def test_pinning_pads_with_black_and_freezes_the_pixels():
+    image, array = elements.pin_element(_card(20), 8)
+    assert image.size == (36, H)
+    assert array.shape == (H, 36, 3)
+    assert not array.flags.writeable
+    assert not array[:, :8].any() and not array[:, -8:].any()
+    assert np.array_equal(array[:, 8:28], np.asarray(_card(20)))
+
+
+def test_the_digest_sees_a_one_pixel_change():
+    _, a = elements.pin_element(_card(20), 0)
+    b = np.array(a)
+    b[5, 5, 0] ^= 1
+    assert elements.pixel_digest(a) != elements.pixel_digest(b)
+    assert elements.pixel_digest(a) == elements.pixel_digest(np.array(a))
+
+
+def test_epochs_move_on_per_plugin_and_never_repeat():
+    epochs = LiveEpochs()
+    assert epochs.get("a") == 0
+    first = epochs.bump("a")
+    second = epochs.bump("b")
+    assert second > first and epochs.get("a") == first
+    assert epochs.bump("a") > second
+
+
+# -- BasePlugin ---------------------------------------------------------------
+
+
+def test_the_base_hooks_do_nothing():
+    plugin = _Legacy()
+    assert plugin.get_vegas_elements() is None
+    assert plugin.redraw_vegas_element("k", 10, H, 0.0) is None
+
+
+def test_notify_vegas_data_changed_reaches_the_plugin_manager():
+    plugin = _Legacy()
+    plugin.plugin_manager = MagicMock()
+    plugin.notify_vegas_data_changed()
+    plugin.plugin_manager.notify_data_changed.assert_called_once_with("p")
+    plugin.plugin_manager = None
+    plugin.notify_vegas_data_changed()          # no manager: nothing to do
+
+
+# -- the adapter's keyed path --------------------------------------------------
+
+
+def test_the_background_fetch_asks_for_elements():
+    adapter = _adapter()
+    adapter.live_epochs.bump("p")
+    plugin = _Plugin(_elements)
+    images = adapter.get_content(plugin, "p", offscreen_only=True)
+    assert plugin.element_calls == 1 and plugin.content_calls == 0
+    metas = [elements.meta_of(img) for img in images]
+    assert [m.key if m else None for m in metas] == ["card:a", None, "card:b"]
+    assert metas[0].epoch == adapter.live_epochs.get("p")
+    assert metas[0].version == 1
+
+
+def test_live_elements_are_pinned_not_trimmed():
+    adapter = _adapter(content_padding=8)
+    plugin = _Plugin(_elements)
+    images = adapter.get_content(plugin, "p", offscreen_only=True)
+    live = [img for img in images if elements.meta_of(img)]
+    assert all(img.width == 40 + 16 for img in live)
+    # The plain separator is trimmed as always: drawn to its edges, it keeps
+    # its width (trimming never widens an image).
+    separator = images[1]
+    assert elements.meta_of(separator) is None
+    assert separator.width == 10
+
+
+def test_an_elements_digest_matches_its_pixels():
+    adapter = _adapter()
+    images = adapter.get_content(_Plugin(_elements), "p", offscreen_only=True)
+    meta = elements.meta_of(images[0])
+    assert meta.digest == elements.pixel_digest(np.asarray(images[0]))
+
+
+def test_elements_render_on_a_canvas_of_their_own_at_the_render_width():
+    adapter = _adapter(render_width_pct=50)
+    plugin = _Plugin(_elements)
+    adapter.get_content(plugin, "p", offscreen_only=True)
+    assert plugin.render_widths == [W // 2]
+    assert adapter.display_manager.offscreen_widths == [W // 2]
+    assert plugin.get_vegas_render_width() == W   # restored afterwards
+
+
+@pytest.mark.parametrize("why", ["disabled", "render thread", "restricted",
+                                 "plugin opted out", "legacy plugin"])
+def test_everywhere_else_the_legacy_content_is_used(why):
+    cfg = {"offscreen_prefetch": False} if why == "restricted" else {}
+    adapter = _adapter(**cfg)
+    plugin_cls = _Legacy if why == "legacy plugin" else _Plugin
+    plugin = plugin_cls(_elements, config={"vegas_live": False}
+                        if why == "plugin opted out" else None)
+    if why == "disabled":
+        adapter.live_elements_enabled = False
+    images = adapter.get_content(plugin, "p", offscreen_only=(why != "render thread"))
+    assert plugin.element_calls == 0
+    if why != "restricted":
+        assert plugin.content_calls == 1
+        assert all(elements.meta_of(img) is None for img in images)
+
+
+def test_a_mock_plugin_is_never_asked_for_elements():
+    adapter = _adapter()
+    plugin = MagicMock()
+    plugin.config = {}
+    plugin.get_vegas_content.return_value = [_card()]
+    adapter.get_content(plugin, "p", offscreen_only=True)
+    plugin.get_vegas_elements.assert_not_called()
+
+
+@pytest.mark.parametrize("result", [None, RuntimeError("boom"), "nonsense",
+                                    [], [object()]])
+def test_a_broken_or_empty_answer_falls_back_to_legacy_content(result):
+    adapter = _adapter()
+    plugin = _Plugin(result)
+    images = adapter.get_content(plugin, "p", offscreen_only=True)
+    assert plugin.content_calls == 1
+    assert images and all(elements.meta_of(img) is None for img in images)
+
+
+def test_duplicate_keys_keep_the_first():
+    adapter = _adapter()
+    plugin = _Plugin(lambda: [VegasElement("k", _card(40)),
+                              VegasElement("k", _card(30))])
+    images = adapter.get_content(plugin, "p", offscreen_only=True)
+    assert len(images) == 1 and images[0].width == 40 + 16
+
+
+def test_elements_are_brought_to_the_display_height_and_rgb():
+    adapter = _adapter()
+    plugin = _Plugin(lambda: [VegasElement("k", _card(40, height=H * 2).convert("RGBA"))])
+    image = adapter.get_content(plugin, "p", offscreen_only=True)[0]
+    assert image.mode == "RGB" and image.height == H
+    assert elements.meta_of(image) is not None
+
+
+def test_a_keyed_fetch_ignores_legacy_content_in_the_cache():
+    # The first compose runs on the render thread and caches legacy content;
+    # the first background fetch after it must still ask for elements.
+    adapter = _adapter()
+    plugin = _Plugin(_elements)
+    adapter.get_content(plugin, "p", offscreen_only=False)
+    assert plugin.content_calls == 1
+    images = adapter.get_content(plugin, "p", offscreen_only=True)
+    assert plugin.element_calls == 1
+    assert elements.meta_of(images[0]) is not None
+
+
+def test_keyed_content_is_cached_with_its_tags():
+    adapter = _adapter()
+    plugin = _Plugin(_elements)
+    adapter.get_content(plugin, "p", offscreen_only=True)
+    again = adapter.get_content(plugin, "p", offscreen_only=True)
+    assert plugin.element_calls == 1
+    assert elements.meta_of(again[0]).key == "card:a"
+
+
+def test_an_element_cropped_to_a_width_budget_is_no_longer_live():
+    adapter = _adapter(max_plugin_width_ratio=0.5)
+    wide = Image.new("RGB", (400, H), (255, 255, 255))
+    plugin = _Plugin(lambda: [VegasElement("map", wide)])
+    image = adapter.get_content(plugin, "p", offscreen_only=True)[0]
+    budget = adapter._width_budget(plugin, "p")
+    assert budget // 2 <= image.width < 400
+    # A real window of the element, not a cut in the middle of its padding.
+    assert np.asarray(image).any()
+    assert elements.meta_of(image) is None
+
+
+def test_an_element_whose_drawing_fits_the_budget_stays_live():
+    adapter = _adapter(max_plugin_width_ratio=0.5)
+    budget = adapter._width_budget(None, "p")
+    # Over the budget only by the black padding pinned on either side.
+    plugin = _Plugin(lambda: [VegasElement("card", _card(budget - 4))])
+    image = adapter.get_content(plugin, "p", offscreen_only=True)[0]
+    assert image.width > budget
+    assert elements.meta_of(image).key == "card"
+
+
+def test_a_padded_solid_image_over_the_budget_is_cropped_not_slivered():
+    # Legacy content has the same margins after trimming, and the same cut
+    # used to land mid-margin.
+    adapter = _adapter(max_plugin_width_ratio=0.5)
+    budget = adapter._width_budget(None, "p")
+    solid = Image.new("RGB", (budget * 3, H), (255, 255, 255))
+    cropped = adapter._apply_width_budget(
+        [_padded(solid, adapter.config.content_padding)], "p", None)
+    assert budget // 2 <= cropped[0].width
+    assert np.asarray(cropped[0]).any()
+
+
+def _padded(image, pad):
+    out = Image.new("RGB", (image.width + 2 * pad, image.height))
+    out.paste(image, (pad, 0))
+    return out
+
+
+def test_rows_under_a_width_budget_keep_their_tags():
+    adapter = _adapter(max_plugin_width_ratio=1.0)
+    plugin = _Plugin(lambda: [VegasElement(f"c{i}", _card(40)) for i in range(6)])
+    images = adapter.get_content(plugin, "p", offscreen_only=True)
+    assert 0 < len(images) < 6
+    assert all(elements.meta_of(img) is not None for img in images)

test/test_vegas_elements_layout.py

@@ -0,0 +1,199 @@
+"""Where live elements land in the Vegas strip, and that the records stay true.
+
+The pipeline records each live element's place (ElementRecord) as it builds
+the strip, in absolute columns that a trim never moves: the strip column is
+``abs_x - _strip_origin``. Everything a live update will do depends on these
+being exact, so the check here is the strongest one available -- after any
+sequence of compose, extend and trim, the strip's pixels at every record are
+the element's own pixels.
+"""
+import sys
+from pathlib import Path
+
+import numpy as np
+from PIL import Image, ImageDraw
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.vegas_mode import elements  # noqa: E402
+from src.vegas_mode.config import VegasModeConfig  # noqa: E402
+from src.vegas_mode.elements import ElementMeta  # noqa: E402
+from src.vegas_mode.render_pipeline import RenderPipeline  # noqa: E402
+
+W, H = 128, 32
+
+
+def _live(pid, key, width, seed):
+    """A tagged, pinned element as the adapter hands it over."""
+    rng = np.random.default_rng(seed)
+    pixels = rng.integers(20, 255, (H, width, 3), dtype=np.uint8)
+    image = Image.frombytes("RGB", (width, H), pixels.tobytes())
+    pinned, array = elements.pin_element(image, 8)
+    return elements.tag(pinned, ElementMeta(
+        pid, key, 1, elements.pixel_digest(array), 0.0, 0.0))
+
+
+def _plain(width, colour=(200, 200, 200)):
+    image = Image.new("RGB", (width, H), (0, 0, 0))
+    ImageDraw.Draw(image).rectangle([0, 0, width - 1, H - 1], outline=colour)
+    return image
+
+
+class _Stream:
+    def __init__(self, groups):
+        self.groups = groups
+        self.plugin_manager = type("PM", (), {"plugins": {}})()
+        self.plugin_adapter = None
+        self.i = 0
+
+    def get_grouped_content_for_composition(self):
+        return self.groups[0]
+
+    def get_active_plugin_ids(self):
+        return [pid for pid, _ in self.groups[0]]
+
+    def take_next_group(self, count=None, offscreen_only=False):
+        self.i += 1
+        return self.groups[self.i] if self.i < len(self.groups) else []
+
+
+class _DM:
+    width, height = W, H
+
+    def __init__(self):
+        self.image = Image.new("RGB", (W, H))
+
+    def set_scrolling_state(self, *a):
+        pass
+
+    def update_display(self):
+        pass
+
+
+def _pipeline(groups, **cfg):
+    cfg.setdefault("continuous_scroll", True)
+    return RenderPipeline(VegasModeConfig(**cfg), _DM(), _Stream(groups))
+
+
+def _assert_records_match(p, images_by_key):
+    strip = p.scroll_helper.cached_array
+    for record in p.live_records():
+        x = record.abs_x - p._strip_origin
+        expected = np.asarray(images_by_key[record.key])
+        assert expected.shape[1] == record.width
+        lo = max(0, x)
+        got = strip[:, lo:x + record.width]
+        assert np.array_equal(got, expected[:, lo - x:]), record.key
+
+
+def _group_images(groups):
+    found = {}
+    for group in groups:
+        for _pid, images in group:
+            for image in images:
+                meta = elements.meta_of(image)
+                if meta:
+                    found[meta.key] = image
+    return found
+
+
+def test_a_compose_records_every_live_element_exactly():
+    groups = [[("a", [_live("a", "a1", 40, 1), _plain(20), _live("a", "a2", 50, 2)]),
+               ("b", [_plain(60)]),
+               ("c", [_live("c", "c1", 70, 3)])]]
+    p = _pipeline(groups, lead_in_width=10)
+    assert p.compose_scroll_content()
+    keys = [r.key for r in p.live_records()]
+    assert keys == ["a1", "a2", "c1"]
+    assert all(r.width == 16 + {"a1": 40, "a2": 50, "c1": 70}[r.key]
+               for r in p.live_records())
+    _assert_records_match(p, _group_images(groups))
+
+
+def test_extensions_and_trims_keep_every_record_true():
+    # A strip long enough that the viewport never runs off its end (trimming
+    # is refused while it wraps), advanced by less than each extension adds.
+    groups = [[("a", [_live("a", "a1", 40, 1), _plain(500)])]]
+    for n in range(12):
+        groups.append([(f"p{n}", [_live(f"p{n}", f"k{n}", 30 + n, 10 + n),
+                                  _plain(25)]),
+                       (f"q{n}", [_plain(45)])])
+    p = _pipeline(groups, lead_in_width=0)
+    assert p.compose_scroll_content()
+    images = _group_images(groups)
+    cuts = 0
+    for _ in range(11):
+        width_before = p.scroll_helper.cached_array.shape[1]
+        p.scroll_helper.scroll_position += 150
+        assert p.scroll_helper.scroll_position + W <= width_before
+        position = p.scroll_helper.scroll_position
+        assert p.extend_scroll_content()
+        cut = int(position - p.scroll_helper.scroll_position)
+        assert 0 <= cut <= width_before
+        cuts += cut
+        assert p._strip_origin == cuts
+        _assert_records_match(p, images)
+    # Records wholly trimmed away are forgotten, and only those.
+    for record in p.live_records():
+        assert record.abs_x + record.width > p._strip_origin
+    assert set(p._record_by_seq) == {r.seq for r in p.live_records()}
+    assert cuts > 0
+
+
+def test_the_first_extension_of_an_empty_strip_starts_at_zero():
+    groups = [[], [("a", [_live("a", "a1", 40, 1)]), ("b", [_live("b", "b1", 20, 2)])]]
+    p = _pipeline(groups)
+    assert p.extend_scroll_content()
+    first = p.live_records()[0]
+    assert first.abs_x == 0 and p._strip_origin == 0
+    _assert_records_match(p, _group_images(groups))
+
+
+def test_plain_content_is_not_recorded():
+    p = _pipeline([[("a", [_plain(80)])], [("b", [_plain(90)])]])
+    assert p.compose_scroll_content()
+    assert p.extend_scroll_content()
+    assert p.live_records() == ()
+
+
+def test_a_new_strip_starts_a_new_generation_with_no_records():
+    groups = [[("a", [_live("a", "a1", 40, 1)])]]
+    p = _pipeline(groups)
+    assert p.compose_scroll_content()
+    gen = p._strip_gen
+    assert p.live_records()
+    assert p.compose_scroll_content()
+    assert p._strip_gen == gen + 1
+    assert [r.key for r in p.live_records()] == ["a1"]
+    p.reset()
+    assert p._strip_gen == gen + 2
+    assert p.live_records() == () and p._strip_origin == 0
+
+
+def test_record_numbers_are_never_reused():
+    groups = [[("a", [_live("a", "a1", 40, 1)])]]
+    p = _pipeline(groups)
+    seen = set()
+    for _ in range(3):
+        assert p.compose_scroll_content()
+        for record in p.live_records():
+            assert record.seq not in seen
+            seen.add(record.seq)
+
+
+def test_static_markers_still_land_where_they_did():
+    # The marker arithmetic now shares _block_starts with the records.
+    class Stream(_Stream):
+        def is_static_plugin(self, pid):
+            return pid == "pause"
+
+    groups = [[("a", [_plain(100)])],
+              [("b", [_plain(60)]), ("pause", []), ("c", [_plain(70)])]]
+    p = RenderPipeline(VegasModeConfig(continuous_scroll=True, lead_in_width=0,
+                                       separator_width=32),
+                       _DM(), Stream(groups))
+    assert p.compose_scroll_content()
+    strip_end = p.scroll_helper.total_scroll_width
+    assert p.extend_scroll_content()
+    # b starts after one separator and ends 60 later; the pause follows b.
+    assert p._static_markers == ((strip_end + 32 + 60, "pause"),)

test/test_vegas_live_gating.py

@@ -0,0 +1,137 @@
+"""When Vegas live elements are on, and that everywhere else nothing changes.
+
+Live elements run only when the config allows them and nothing rules them
+out: multi-display sync (the follower mirrors whole strips only), swap mode,
+the offscreen kill switch, or a display manager with no off-screen canvas.
+While they are on, the coordinator listens for plugin updates and moves each
+plugin's data epoch on; while off, the adapter never asks for elements.
+"""
+import sys
+from contextlib import contextmanager
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+from PIL import Image
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.vegas_mode.coordinator import VegasModeCoordinator  # noqa: E402
+
+
+class _DM:
+    width, height = 128, 32
+
+    def __init__(self, offscreen=True):
+        self.image = Image.new("RGB", (128, 32))
+        if offscreen:
+            self.offscreen = self._offscreen
+
+    @contextmanager
+    def _offscreen(self, width=None, height=None):
+        yield SimpleNamespace(image=Image.new("RGB", (width or 128, 32)))
+
+    def set_scrolling_state(self, *a, **k):
+        pass
+
+    def update_display(self):
+        pass
+
+
+class _PM:
+    def __init__(self):
+        self.plugins = {}
+        self.listeners = []
+
+    def add_update_listener(self, fn):
+        if fn not in self.listeners:
+            self.listeners.append(fn)
+
+    def remove_update_listener(self, fn):
+        self.listeners = [f for f in self.listeners if f != fn]
+
+
+def _coordinator(dm=None, **vegas):
+    vegas.setdefault("enabled", True)
+    config = {"display": {"vegas_scroll": vegas}}
+    return VegasModeCoordinator(config, dm or _DM(), _PM())
+
+
+def test_on_by_default():
+    c = _coordinator()
+    c._apply_live_state()
+    assert c.live_active
+    assert c.plugin_adapter.live_elements_enabled
+    assert c._on_plugin_data_changed in c.plugin_manager.listeners
+
+
+@pytest.mark.parametrize("why,vegas,dm", [
+    ("switched off", {"live_refresh": False}, None),
+    ("swap mode", {"continuous_scroll": False}, None),
+    ("offscreen kill switch", {"offscreen_prefetch": False}, None),
+    ("no offscreen canvas", {}, _DM(offscreen=False)),
+])
+def test_off_when_ruled_out(why, vegas, dm):
+    c = _coordinator(dm, **vegas)
+    c._apply_live_state()
+    assert not c.live_active
+    assert not c.plugin_adapter.live_elements_enabled
+    assert c.plugin_manager.listeners == []
+
+
+@pytest.mark.parametrize("role", ["leader", "follower"])
+def test_off_whenever_sync_is_configured(role):
+    c = _coordinator()
+    c.set_sync_manager(SimpleNamespace(role=role))
+    c._apply_live_state()
+    assert not c.live_active
+    assert not c.plugin_adapter.live_elements_enabled
+
+
+def test_a_standalone_sync_manager_does_not_count():
+    from src.common.sync_manager import SyncRole
+    c = _coordinator()
+    c.set_sync_manager(SimpleNamespace(role=SyncRole.STANDALONE))
+    c._apply_live_state()
+    assert c.live_active
+
+
+def test_stopping_switches_it_off_and_stops_listening():
+    c = _coordinator()
+    c._apply_live_state()
+    c._is_active = True
+    c.stop()
+    assert not c.live_active and not c.plugin_adapter.live_elements_enabled
+    assert c.plugin_manager.listeners == []
+
+
+def test_a_config_change_can_switch_it_off_mid_run():
+    c = _coordinator()
+    c._apply_live_state()
+    c._is_active = True
+    c.stream_manager.refresh = MagicMock()
+    c.update_config({"display": {"vegas_scroll": {"enabled": True,
+                                                   "live_refresh": False}}})
+    c._apply_pending_config()
+    assert not c.live_active
+    assert c.plugin_manager.listeners == []
+
+
+def test_an_update_moves_the_plugins_epoch_on():
+    c = _coordinator()
+    c._apply_live_state()
+    before = c.live_epochs.get("p")
+    for listener in c.plugin_manager.listeners:
+        listener("p")
+    assert c.live_epochs.get("p") > before
+    assert c.plugin_adapter.live_epochs is c.live_epochs
+
+
+def test_the_state_change_is_logged_once(caplog):
+    c = _coordinator(live_refresh=False)
+    with caplog.at_level("INFO"):
+        c._apply_live_state()
+        c._apply_live_state()
+    lines = [r.message for r in caplog.records if "live elements" in r.message]
+    assert lines == ["Vegas live elements off: switched off (vegas_scroll.live_refresh)"]

test/test_vegas_live_integration.py

@@ -0,0 +1,137 @@
+"""Live Vegas elements end to end: a real plugin through the real ticker.
+
+The stub fixture plugin (test/fixtures/plugins/vegas-live-stub) is loaded by
+the real plugin loader onto a real DisplayManager (RGBMatrixEmulator) and a
+real PluginManager, and the real coordinator runs it: the first compose uses
+its ordinary Vegas content, the background prefetch asks it for elements, and
+the strip records where each one landed.
+"""
+import os
+import sys
+import time
+from pathlib import Path
+
+os.environ["EMULATOR"] = "true"
+
+import numpy as np  # noqa: E402
+import pytest  # noqa: E402
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.plugin_system.plugin_state import PluginState  # noqa: E402
+from src.plugin_system.testing.harness import _instantiate  # noqa: E402
+from src.plugin_system.testing.loading import build_full_config, load_harness_spec, load_manifest  # noqa: E402
+from src.vegas_mode import elements  # noqa: E402
+
+STUB = Path(__file__).resolve().parent / "fixtures" / "plugins" / "vegas-live-stub"
+PID = "vegas-live-stub"
+
+
+@pytest.fixture(scope="module")
+def dm(tmp_path_factory):
+    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)
+    manager._snapshot_path = str(
+        tmp_path_factory.mktemp("live") / "led_matrix_preview.png")
+    if manager.matrix is None:
+        pytest.fail("DisplayManager fell back to matrix=None")
+    yield manager
+    DisplayManager._instance = None
+    DisplayManager._initialized = False
+
+
+@pytest.fixture
+def ticker(dm, tmp_path):
+    from src.plugin_system.plugin_manager import PluginManager
+    from src.vegas_mode.coordinator import VegasModeCoordinator
+
+    pm = PluginManager(plugins_dir=str(tmp_path), config_manager=None,
+                       display_manager=dm, cache_manager=None)
+    config = {**build_full_config(STUB, load_harness_spec(STUB), {}),
+              "enabled": True, "map_hz": 4}
+    plugin = _instantiate(PID, load_manifest(STUB), STUB, config, {}, dm)
+    plugin.plugin_manager = pm
+    pm.plugins[PID] = plugin
+    pm.state_manager.set_state(PID, PluginState.ENABLED)
+
+    coordinator = VegasModeCoordinator({"display": {"vegas_scroll": {
+        "enabled": True, "continuous_scroll": True, "plugins_per_cycle": 1,
+        "scroll_speed": 100, "lead_in_width": 0,
+    }}}, dm, pm)
+    yield coordinator, plugin, pm
+    coordinator.stop()
+    pm.stop_update_worker()
+
+
+def _run_until(coordinator, predicate, seconds=10.0):
+    deadline = time.monotonic() + seconds
+    while time.monotonic() < deadline:
+        coordinator.run_frame()
+        if predicate():
+            return True
+        time.sleep(0.002)
+    return False
+
+
+def test_the_prefetched_stub_is_placed_as_live_elements(ticker):
+    coordinator, plugin, _pm = ticker
+    assert coordinator.start()
+    assert coordinator.live_active
+    pipeline = coordinator.render_pipeline
+    # The first compose ran on this thread without the plugin's lock, so it
+    # is plain content.
+    assert pipeline.live_records() == ()
+
+    assert _run_until(coordinator, lambda: pipeline.live_records())
+    keys = [r.key for r in pipeline.live_records()]
+    assert {"card:0", "card:5", "map"} <= set(keys)
+    assert "sep" not in keys
+    assert all(r.plugin_id == PID for r in pipeline.live_records())
+
+    # Each record points at the element's pixels: a card is bordered in its
+    # colour, with content_padding black columns either side.
+    strip = pipeline.scroll_helper.cached_array
+    pad = coordinator.vegas_config.content_padding
+    for record in pipeline.live_records():
+        x = record.abs_x - pipeline._strip_origin
+        if x < 0:
+            continue
+        columns = strip[:, x:x + record.width]
+        assert not columns[:, :pad].any() and not columns[:, -pad:].any()
+        assert columns[:, pad:record.width - pad].any()
+
+
+def test_an_update_moves_the_plugins_epoch_and_new_elements_carry_it(ticker):
+    coordinator, plugin, pm = ticker
+    assert coordinator.start()
+    before = coordinator.live_epochs.get(PID)
+    pm._note_update_completed(PID)
+    epoch = coordinator.live_epochs.get(PID)
+    assert epoch > before
+    coordinator.plugin_adapter.invalidate_cache(PID)
+    images = coordinator.plugin_adapter.get_content(plugin, PID, offscreen_only=True)
+    metas = [elements.meta_of(img) for img in images if elements.meta_of(img)]
+    assert metas and all(m.epoch == epoch for m in metas)
+
+
+def test_with_live_refresh_off_the_same_run_is_plain_content(dm, tmp_path, ticker):
+    coordinator, _plugin, _pm = ticker
+    coordinator.vegas_config.live_refresh = False
+    assert coordinator.start()
+    assert not coordinator.live_active
+    pipeline = coordinator.render_pipeline
+    start_width = pipeline.scroll_helper.total_scroll_width
+    assert _run_until(
+        coordinator,
+        lambda: pipeline.stats.get('extensions', 0) >= 1, seconds=10.0)
+    assert pipeline.live_records() == ()
+    assert np.asarray(pipeline.scroll_helper.cached_array).shape[1] > 0
+    assert start_width > 0