fix(plugins): report updates that completed, not ones that were queued

run_scheduled_updates_with_changes() snapshotted plugin_last_update,
called run_scheduled_updates(), and diffed the two to answer "whose data
just changed".

But run_scheduled_updates() only enqueues. The work runs on the update
worker and stamps plugin_last_update there, after this method has already
returned, so the two snapshots were always identical and the result was
always an empty list. The only path that ever worked was the synchronous
kill-switch, where update() runs inline.

Vegas is the caller. That empty list is what feeds mark_plugin_updated(),
which drops the cached content for a plugin whose data moved -- so a
segment kept scrolling whatever it was first built from. It is the
failure the coordinator's own comments describe: last night's live game
still drawn as live the next morning. On a live rig: zero update ticks in
twenty minutes, with weather, stocks and news all updating on schedule.

The worker now records each completed update in a ledger and the call
drains it, reporting what has finished since the previous poll rather
than what this call enqueued. That costs one tick of latency -- Vegas
polls every ~4s -- and is correct whichever side of the queue the work
lands on. Failure paths are excluded: they stamp the timestamp too, to
space out retries, but no fresh data exists.

Verified on the rig it was found on: 0 update ticks before, 208 in
twenty-five minutes after, naming real plugins.

The behavioural tests here would pass with both production call sites
deleted, which mutation testing caught -- they drive the ledger directly.
So there is also a structural test asserting the invariant at the source:
wherever a successful update stamps plugin_last_update, it must record
the completion. Writing it immediately caught that _record_update_failure
stamps the same field and must not be included.

Mutation-checked: removing either call site, removing both, and dropping
the drain's clear are all caught.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Udr6MfaFLUPhX5Fgo67Jf5

.github/workflows/test.yml

@@ -79,4 +79,5 @@ jobs:
             test/test_sports_scroll.py \
             test/test_version_consistency.py \
             test/test_plugin_compatibility_gate.py \
-            test/test_install_preserves_existing.py
+            test/test_install_preserves_existing.py \
+            test/test_core_owned_config_keys.py

src/plugin_system/compatibility.py

@@ -27,6 +27,7 @@ fixes their version string. See `docs/SPORTS_UNIFICATION.md`, phase B4.
 
 from __future__ import annotations
 
+import re
 from typing import Any, Dict, Optional, Tuple
 
 # Below this, the core's self-reported version is not evidence of anything.
@@ -39,7 +40,20 @@ def parse_semver(value: Any) -> Optional[Tuple[int, int, int]]:
     3-tuple, or ``None`` when unparseable. A leading ``v`` is tolerated."""
     if not isinstance(value, str):
         return None
-    parts = value.strip().lstrip('v').split('.')
+    text = value.strip().lstrip('v')
+    # Drop the prerelease/build suffix before scraping digits. Without this the
+    # scrape pulls them into the numbers: "3.2.0+build42" parsed as (3, 2, 42)
+    # and "3.2.0-rc1" as (3, 2, 1) -- a release candidate ranking *above* its
+    # own release, and a build of 3.2.0 failing an exact "3.2.0" match.
+    #
+    # Prereleases compare equal to their release here rather than below it.
+    # Full prerelease ordering is more than any caller needs, and equal is far
+    # closer to right than the old behaviour.
+    for sep in ('+', '-'):
+        head, found, _tail = text.partition(sep)
+        if found:
+            text = head
+    parts = text.split('.')
     try:
         nums = [int(''.join(ch for ch in p if ch.isdigit()) or 0) for p in parts[:3]]
     except ValueError:
@@ -49,6 +63,92 @@ def parse_semver(value: Any) -> Optional[Tuple[int, int, int]]:
     return tuple(nums)  # type: ignore[return-value]
 
 
+# `parse_semver` is deliberately lenient — it strips non-digits and yields
+# (0, 0, 0) for a string with no numbers at all, which is fine for a floor
+# (a floor of 0.0.0 never blocks anything) but wrong for a range, where the
+# same leniency would turn an unreadable spec into a *refusal*. Range specs
+# are therefore validated against this first, so garbage reads as "no
+# evidence" rather than "incompatible".
+_VERSION_TOKEN = re.compile(r"^v?\d+(\.\d+){0,2}(-[\w.-]+)?(\+[\w.-]+)?$")
+
+
+def _parse_strict(value: str) -> Optional[Tuple[int, int, int]]:
+    """`parse_semver`, but ``None`` unless the string really looks like one."""
+    if not isinstance(value, str) or not _VERSION_TOKEN.match(value.strip()):
+        return None
+    return parse_semver(value)
+
+
+def _satisfies_range(core: Tuple[int, int, int], spec: str) -> Optional[bool]:
+    """Does ``core`` satisfy one `compatible_versions` entry?
+
+    Returns ``None`` when the spec cannot be parsed — the caller treats that as
+    "no evidence" rather than as a refusal, so an unrecognised spelling never
+    costs a user a working install.
+
+    Supports the forms `schema/manifest_schema.json` permits: `>=`, `<=`, `>`,
+    `<`, `~`, `^`, a bare exact version, and an inclusive `A - B` range.
+    Prerelease/build suffixes are tolerated and ignored, matching `parse_semver`.
+    """
+    spec = spec.strip()
+    if not spec:
+        return None
+
+    if " - " in spec:  # inclusive range, e.g. "2.0.0 - 3.1.0"
+        low_raw, _, high_raw = spec.partition(" - ")
+        low, high = _parse_strict(low_raw), _parse_strict(high_raw)
+        if low is None or high is None:
+            return None
+        return low <= core <= high
+
+    for op in (">=", "<=", ">", "<", "~", "^"):
+        if spec.startswith(op):
+            target = _parse_strict(spec[len(op):])
+            if target is None:
+                return None
+            if op == ">=":
+                return core >= target
+            if op == "<=":
+                return core <= target
+            if op == ">":
+                return core > target
+            if op == "<":
+                return core < target
+            if op == "~":
+                # Patch-level changes only: >=X.Y.Z, <X.(Y+1).0
+                return target <= core < (target[0], target[1] + 1, 0)
+            # "^": minor and patch changes: >=X.Y.Z, <(X+1).0.0
+            return target <= core < (target[0] + 1, 0, 0)
+
+    exact = _parse_strict(spec)
+    return None if exact is None else core == exact
+
+
+def satisfies_compatible_versions(
+    manifest: Dict[str, Any], core: Tuple[int, int, int]
+) -> Optional[bool]:
+    """Evaluate the manifest's `compatible_versions` array against ``core``.
+
+    The array is a set of *alternatives*: satisfying any one entry means the
+    plugin declares itself compatible. Returns ``None`` when the field is
+    absent or no entry could be parsed, so callers can distinguish "declared
+    incompatible" from "did not say".
+
+    This is the field `schema/manifest_schema.json` marks **required**, and it
+    is the only one that can express an upper bound — `ledmatrix_min_version`
+    is a floor and cannot say "not compatible with 4.x".
+    """
+    specs = manifest.get('compatible_versions')
+    if not isinstance(specs, list) or not specs:
+        return None
+
+    verdicts = [_satisfies_range(core, s) for s in specs if isinstance(s, str)]
+    parsed = [v for v in verdicts if v is not None]
+    if not parsed:
+        return None
+    return any(parsed)
+
+
 def declared_min_version(manifest: Dict[str, Any]) -> Optional[str]:
     """The core version this plugin says it needs, or ``None`` if it doesn't say.
 
@@ -56,16 +156,25 @@ def declared_min_version(manifest: Dict[str, Any]) -> Optional[str]:
     of `ledmatrix_min_version` (`store_manager._validate_manifest_fields` flags
     it); both are read because a large share of published manifests still carry
     the old one.
+
+    Container types are validated rather than assumed. A hand-edited or
+    third-party manifest can carry `requires` as a list or `versions` as a
+    mapping, and both used to raise out of here (`AttributeError` and
+    `KeyError` respectively). That now matters far more than it did: the
+    untrustworthy-core branch of :func:`check` calls this for *every* manifest,
+    so one malformed file would take down the install path rather than just
+    itself. A shape we do not recognise means "no declared floor".
     """
-    declared = (
-        manifest.get('min_ledmatrix_version')
-        or (manifest.get('requires') or {}).get('min_ledmatrix_version')
-    )
+    declared = manifest.get('min_ledmatrix_version')
+    if not declared:
+        requires = manifest.get('requires')
+        if isinstance(requires, dict):
+            declared = requires.get('min_ledmatrix_version')
     if declared:
         return declared
 
-    versions = manifest.get('versions') or []
-    if versions and isinstance(versions[0], dict):
+    versions = manifest.get('versions')
+    if isinstance(versions, list) and versions and isinstance(versions[0], dict):
         return (versions[0].get('ledmatrix_min_version')
                 or versions[0].get('ledmatrix_min'))
     return None
@@ -74,26 +183,68 @@ def declared_min_version(manifest: Dict[str, Any]) -> Optional[str]:
 def check(manifest: Dict[str, Any], core_version: str) -> Tuple[bool, Optional[str]]:
     """Return ``(compatible, reason)``.
 
-    ``compatible`` is False **only** when the plugin declares a parseable floor,
-    the core reports a parseable and trustworthy version, and the floor is
-    genuinely above it. Every uncertain case resolves to compatible: an
-    undeclared floor, an unparseable version on either side, or a core whose
-    version is below `TRUSTWORTHY_FLOOR`. Refusing on a guess would break
-    working installs, which is the more expensive mistake here.
+    Two fields can say a plugin is incompatible and **the more restrictive
+    wins**:
+
+    - `compatible_versions` — the schema-required array of semver ranges, and
+      the only one that can express an upper bound.
+    - `ledmatrix_min_version` (or the deprecated `ledmatrix_min`) — the
+      per-release floor inside `versions[]`.
+
+    They agree across every published manifest today except `7-segment-clock`,
+    but they *can* disagree, and a plugin that says `["2.0.0 - 2.9.9"]` means
+    "not compatible with 3.x" no matter what its floor says.
+
+    ``compatible`` is False **only** on evidence: the core reports a parseable,
+    trustworthy version and a field genuinely excludes it. Every uncertain case
+    resolves to compatible — nothing declared, an unparseable version on either
+    side, or a core below `TRUSTWORTHY_FLOOR`. Refusing on a guess breaks a
+    working install, which is the more expensive mistake here.
 
     ``reason`` is user-facing text, present only when incompatible.
     """
+    current = parse_semver(core_version)
+    name = manifest.get('name') or manifest.get('id') or 'This plugin'
+
+    if current is None or current < TRUSTWORTHY_FLOOR:
+        # The version is not evidence of what this core HAS. But a floor above
+        # the ecosystem baseline says the plugin needs modules that arrived
+        # *after* 2.0.0 — and a core reporting below that either is the v3.1.0
+        # release (which ships __version__ = "1.0.0" and has none of the 3.2.0
+        # modules) or is genuinely ancient. Either way it will not have them.
+        #
+        # This is the only protection available to that population: they cannot
+        # be told apart from a real 1.0.0 install, so the gate cannot reason
+        # about them, and the *plugin's* guarded-import fallback disappears at
+        # the B6 sunset. Refusing the install leaves them on the version they
+        # already run instead of handing them one that fails to load.
+        #
+        # Floors at or below 2.0.0 are still allowed, which is every manifest
+        # published today — so this does not lock anyone out of the store.
+        declared = declared_min_version(manifest)
+        needed = parse_semver(declared)
+        if needed is not None and needed > TRUSTWORTHY_FLOOR:
+            return False, (
+                f"{name} requires LEDMatrix {declared} or newer. This system "
+                f"reports {core_version}, which is too old to identify "
+                f"reliably — update LEDMatrix, then install it."
+            )
+        return True, None
+
+    # Ranges first: they are the canonical field and can rule out a core that
+    # clears the floor.
+    if satisfies_compatible_versions(manifest, current) is False:
+        specs = ", ".join(
+            s for s in manifest.get('compatible_versions', []) if isinstance(s, str))
+        return False, (
+            f"{name} supports LEDMatrix {specs}, but this system is running "
+            f"{core_version}. Install a build in that range, or a plugin "
+            f"version that supports {core_version}."
+        )
+
     declared = declared_min_version(manifest)
     needed = parse_semver(declared)
-    if needed is None:
-        return True, None
-
-    current = parse_semver(core_version)
-    if current is None or current < TRUSTWORTHY_FLOOR:
-        return True, None
-
-    if needed > current:
-        name = manifest.get('name') or manifest.get('id') or 'This plugin'
+    if needed is not None and needed > current:
         return False, (
             f"{name} requires LEDMatrix {declared} or newer, but this system is "
             f"running {core_version}. Update LEDMatrix first, then install it."

src/plugin_system/plugin_manager.py

@@ -116,6 +116,14 @@ class PluginManager:
         self._plugin_locks: Dict[str, threading.Lock] = {}
         self._plugin_locks_guard = threading.Lock()
         self._update_worker: Optional[threading.Thread] = None
+        # Plugin ids whose update() has finished since the last time anyone
+        # asked. Updates are dispatched to a worker thread, so a caller that
+        # wants to know "whose data just changed" cannot learn it by diffing
+        # plugin_last_update around run_scheduled_updates() -- that call only
+        # enqueues, and the timestamp is stamped later, on the worker. See
+        # run_scheduled_updates_with_changes().
+        self._completed_updates: set = set()
+        self._completed_updates_lock = threading.Lock()
         self._synchronous_updates = False
         if self.config_manager is not None:
             try:
@@ -395,6 +403,37 @@ class PluginManager:
             self.state_manager.set_state(plugin_id, PluginState.ERROR, error=e)
             return False
     
+    #: Config keys the **core** reads out of a plugin's own config block. The
+    #: plugin never declares them, so a schema with
+    #: ``"additionalProperties": false`` — 37 of the 42 published ones — reports
+    #: them as violations and the plugin gets flagged degraded in the web UI for
+    #: using a documented core feature.
+    #:
+    #: Listed explicitly rather than matched on a ``vegas_`` prefix, because
+    #: ``vegas_mode`` is the opposite case: plugins *do* declare that one, and a
+    #: 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``).
+    CORE_OWNED_CONFIG_KEYS = frozenset({
+        'vegas_width_pct',
+        'vegas_overflow',
+        'vegas_max_width_screens',
+    })
+
+    def _strip_core_owned_keys(self, config: Dict[str, Any]) -> Dict[str, Any]:
+        """A shallow copy of ``config`` without the core's own tuning keys.
+
+        Only the top level is touched, and only when such a key is present, so
+        the common case allocates nothing extra.
+        """
+        if not isinstance(config, dict):
+            return config
+        if not self.CORE_OWNED_CONFIG_KEYS.intersection(config):
+            return config
+        return {k: v for k, v in config.items()
+                if k not in self.CORE_OWNED_CONFIG_KEYS}
+
     def _validate_config_schema_soft(self, plugin_id: str, config: Dict[str, Any]) -> None:
         """Validate a plugin's config against its JSON schema — warn/degrade only.
 
@@ -419,7 +458,7 @@ class PluginManager:
 
         try:
             is_valid, errors = self.schema_manager.validate_config_against_schema(
-                config, schema, plugin_id
+                self._strip_core_owned_keys(config), schema, plugin_id
             )
         except Exception as e:  # pragma: no cover - defensive
             # Validation machinery itself failed — do not penalise the plugin.
@@ -909,6 +948,7 @@ class PluginManager:
                 if success:
                     with self._plugin_last_update_lock:
                         self.plugin_last_update[plugin_id] = scheduled_time
+                    self._note_update_completed(plugin_id)
                     self.state_manager.record_update(plugin_id)
                     self.state_manager.set_state(plugin_id, PluginState.ENABLED)
                     if self.health_tracker:
@@ -975,28 +1015,41 @@ class PluginManager:
 
     def run_scheduled_updates_with_changes(self, current_time: Optional[float] = None) -> List[str]:
         """
-        Like run_scheduled_updates(), but also returns the plugin_ids whose
-        plugin_last_update timestamp actually advanced during this call.
+        Like run_scheduled_updates(), but also reports which plugins have
+        fresh data -- the ids whose update() has finished since the last
+        call, not necessarily the ones enqueued by this one.
 
-        The before/after snapshots and the update pass itself are each
-        individually lock-protected against concurrent plugin_last_update
-        mutation (Vegas mode calls this from its own background
-        update-tick thread, racing the main render loop's plugin updates),
-        so callers get an atomic "who got fresh data" answer without
-        reaching into plugin_last_update themselves. The lock is not held
-        across the update pass so slow/blocking plugin update() calls don't
-        serialize against other plugin_last_update readers.
+        That distinction is the whole point. This used to snapshot
+        plugin_last_update, call run_scheduled_updates(), and diff. But
+        run_scheduled_updates() only *enqueues*: the work runs on the
+        update worker and the timestamp is stamped there, after this method
+        has already returned. The two snapshots were therefore always
+        identical and the result was always empty, so Vegas never learned
+        that any plugin's data had changed and kept scrolling whatever a
+        segment was first built from -- last night's live game still drawn
+        as live the next morning. The only path that ever worked was the
+        synchronous kill-switch, where update() runs inline.
+
+        Reporting completions instead of enqueues costs a poll's worth of
+        latency (the Vegas tick runs every ~4s) and is correct regardless of
+        which side of the queue the work lands on.
         """
-        with self._plugin_last_update_lock:
-            old_times = dict(self.plugin_last_update)
-
         self.run_scheduled_updates(current_time)
+        return self.drain_completed_updates()
 
-        with self._plugin_last_update_lock:
-            return [
-                plugin_id for plugin_id, new_time in self.plugin_last_update.items()
-                if new_time > old_times.get(plugin_id, 0.0)
-            ]
+    def _note_update_completed(self, plugin_id: str) -> None:
+        """Record that a plugin's update() finished, for the next poll."""
+        with self._completed_updates_lock:
+            self._completed_updates.add(plugin_id)
+
+    def drain_completed_updates(self) -> List[str]:
+        """Return and clear the plugin ids whose update() has since finished."""
+        with self._completed_updates_lock:
+            if not self._completed_updates:
+                return []
+            done = sorted(self._completed_updates)
+            self._completed_updates.clear()
+            return done
 
     def update_all_plugins(self) -> None:
         """
@@ -1022,6 +1075,7 @@ class PluginManager:
                 if success:
                     with self._plugin_last_update_lock:
                         self.plugin_last_update[plugin_id] = time.time()
+                    self._note_update_completed(plugin_id)
                     self.state_manager.record_update(plugin_id)
                     self.state_manager.set_state(plugin_id, PluginState.ENABLED)
                 else:

test/test_core_owned_config_keys.py

@@ -0,0 +1,98 @@
+"""The core's own tuning keys must not make a plugin look broken.
+
+`vegas_width_pct`, `vegas_overflow` and `vegas_max_width_screens` are read by
+the *core* out of each plugin's config block — `vegas_mode/plugin_adapter.py`
+and `base_plugin.py`. No plugin declares them, and 37 of the 42 published
+config schemas set `"additionalProperties": false`, so schema validation
+reported them as violations.
+
+That is not just log noise: `_validate_config_schema_soft` sets `degraded` in
+the health tracker, which the web UI surfaces. Measured on a real device, **9
+of 27 installed plugins** were flagged degraded purely for using a documented
+core feature — including `baseball-scoreboard` and `f1-scoreboard`.
+
+The fix strips those keys before validating. It deliberately does *not* match
+on a `vegas_` prefix: `vegas_mode` is plugin-owned and declared in schemas, and
+a prefix rule would silently stop validating it.
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from src.plugin_system.plugin_manager import PluginManager
+
+
+STRICT_SCHEMA = {
+    "type": "object",
+    "additionalProperties": False,
+    "properties": {
+        "enabled": {"type": "boolean"},
+        "vegas_mode": {"type": "string"},  # plugin-owned, must stay validated
+    },
+}
+
+
+@pytest.fixture
+def manager():
+    mgr = PluginManager.__new__(PluginManager)  # skip the heavy constructor
+    mgr.logger = MagicMock()
+    mgr.schema_manager = MagicMock()
+    mgr._set_degraded_safe = MagicMock()
+    return mgr
+
+
+class TestStripCoreOwnedKeys:
+    def test_removes_every_core_owned_key(self, manager):
+        cfg = {"enabled": True, "vegas_width_pct": 50,
+               "vegas_overflow": "wrap", "vegas_max_width_screens": 2}
+        assert manager._strip_core_owned_keys(cfg) == {"enabled": True}
+
+    def test_leaves_plugin_owned_vegas_mode_alone(self, manager):
+        """A prefix rule would have eaten this one."""
+        cfg = {"enabled": True, "vegas_mode": "scroll"}
+        assert manager._strip_core_owned_keys(cfg) == cfg
+
+    def test_returns_the_same_object_when_nothing_to_strip(self, manager):
+        cfg = {"enabled": True}
+        assert manager._strip_core_owned_keys(cfg) is cfg
+
+    def test_does_not_mutate_the_caller_config(self, manager):
+        cfg = {"enabled": True, "vegas_width_pct": 50}
+        manager._strip_core_owned_keys(cfg)
+        assert "vegas_width_pct" in cfg, "the live plugin config was mutated"
+
+    def test_tolerates_a_non_dict(self, manager):
+        assert manager._strip_core_owned_keys(None) is None
+
+
+class TestSoftValidation:
+    def _validate_with(self, manager, config, valid=True, errors=()):
+        manager.schema_manager.load_schema.return_value = STRICT_SCHEMA
+        manager.schema_manager.validate_config_against_schema.return_value = (
+            valid, list(errors))
+        manager._validate_config_schema_soft("baseball-scoreboard", config)
+        return manager.schema_manager.validate_config_against_schema.call_args
+
+    def test_core_keys_never_reach_the_validator(self, manager):
+        """The regression: these keys reaching a strict schema is what flagged
+        9 of 27 plugins degraded."""
+        args = self._validate_with(
+            manager, {"enabled": True, "vegas_width_pct": 50})
+        validated = args[0][0]
+        assert "vegas_width_pct" not in validated
+        assert validated == {"enabled": True}
+
+    def test_plugin_owned_keys_still_reach_the_validator(self, manager):
+        args = self._validate_with(
+            manager, {"enabled": True, "vegas_mode": "scroll"})
+        assert args[0][0]["vegas_mode"] == "scroll"
+
+    def test_a_genuine_violation_is_still_reported(self, manager):
+        """Stripping core keys must not turn the check into a no-op."""
+        self._validate_with(
+            manager, {"enabled": True, "typo_key": 1},
+            valid=False, errors=["Field root: 'typo_key' was unexpected"])
+        manager._set_degraded_safe.assert_called()
+        reason = manager._set_degraded_safe.call_args[0][1]
+        assert reason and "typo_key" in reason

test/test_plugin_compatibility_gate.py

@@ -81,17 +81,24 @@ class TestCheck:
         ok, reason = compatibility.check({"id": "x"}, "3.2.0")
         assert ok is True and reason is None
 
-    def test_untrustworthy_core_version_allows_everything(self):
+    def test_untrustworthy_core_allows_todays_ecosystem_floor(self):
         """The v3.1.0 release reports 1.0.0. Nearly every manifest floors at
-        2.0.0, so blocking here would stop those users installing any plugin
-        at all — strictly worse than the problem being solved."""
+        2.0.0, so blocking *that* would stop those users installing any plugin
+        at all — strictly worse than the problem being solved.
+
+        A floor ABOVE 2.0.0 is refused instead; see
+        TestUntrustworthyCoreAndTheSunset for why that case is different."""
         ok, reason = compatibility.check(
-            {"min_ledmatrix_version": "3.2.0"}, "1.0.0")
+            {"min_ledmatrix_version": "2.0.0"}, "1.0.0")
         assert ok is True and reason is None
 
-    def test_unparseable_core_version_allows(self):
-        ok, _ = compatibility.check({"min_ledmatrix_version": "3.2.0"}, "not-a-version")
+    def test_unparseable_core_version_allows_the_ecosystem_floor(self):
+        """An unidentifiable core is treated exactly like an untrustworthy one:
+        today's 2.0.0 floor is allowed, a post-sunset floor is not."""
+        ok, _ = compatibility.check({"min_ledmatrix_version": "2.0.0"}, "not-a-version")
         assert ok is True
+        ok, _ = compatibility.check({"min_ledmatrix_version": "3.2.0"}, "not-a-version")
+        assert ok is False
 
     def test_unparseable_floor_allows(self):
         ok, _ = compatibility.check({"min_ledmatrix_version": {"nope": 1}}, "3.2.0")
@@ -230,3 +237,194 @@ class TestLoaderAndStoreAgree:
         )
         assert loader_would_warn is (not expected)
         assert hasattr(PluginLoader, "_warn_if_incompatible")
+
+
+# --------------------------------------------------------------------------
+# compatible_versions — the schema-required field, and the only one that can
+# express an upper bound
+# --------------------------------------------------------------------------
+
+class TestCompatibleVersions:
+    @pytest.mark.parametrize("spec,core,expected", [
+        (">=2.0.0", "3.2.0", True),
+        (">=2.0.0", "1.9.9", False),
+        ("<=3.0.0", "3.2.0", False),
+        ("<=3.0.0", "2.9.0", True),
+        (">3.2.0", "3.2.0", False),
+        ("<4.0.0", "3.2.0", True),
+        ("3.2.0", "3.2.0", True),      # bare == exact match
+        ("3.2.0", "3.2.1", False),
+        ("~3.2.0", "3.2.9", True),     # patch-level only
+        ("~3.2.0", "3.3.0", False),
+        ("^3.2.0", "3.9.9", True),     # minor + patch
+        ("^3.2.0", "4.0.0", False),
+        ("2.0.0 - 3.2.0", "3.2.0", True),   # inclusive both ends
+        ("2.0.0 - 3.2.0", "2.0.0", True),
+        ("2.0.0 - 3.2.0", "3.2.1", False),
+        ("v3.2.0", "3.2.0", True),     # leading v tolerated
+        ("3.2.0-beta.1", "3.2.0", True),  # prerelease suffix ignored
+    ])
+    def test_range_forms(self, spec, core, expected):
+        got = compatibility.satisfies_compatible_versions(
+            {"compatible_versions": [spec]}, compatibility.parse_semver(core))
+        assert got is expected, f"{spec!r} vs {core}"
+
+    def test_array_is_alternatives_not_conjunction(self):
+        """Satisfying any one entry is enough — otherwise ['<2.0.0','>=3.0.0']
+        could never be satisfied by anything."""
+        m = {"compatible_versions": ["<2.0.0", ">=3.0.0"]}
+        assert compatibility.satisfies_compatible_versions(
+            m, compatibility.parse_semver("3.2.0")) is True
+
+    def test_absent_or_unparseable_is_no_evidence(self):
+        core = compatibility.parse_semver("3.2.0")
+        assert compatibility.satisfies_compatible_versions({}, core) is None
+        assert compatibility.satisfies_compatible_versions(
+            {"compatible_versions": []}, core) is None
+        assert compatibility.satisfies_compatible_versions(
+            {"compatible_versions": ["not a version"]}, core) is None
+        # One unparseable entry alongside a good one must not poison the result.
+        assert compatibility.satisfies_compatible_versions(
+            {"compatible_versions": ["garbage", ">=2.0.0"]}, core) is True
+
+
+class TestMoreRestrictiveWins:
+    def test_upper_bound_blocks_a_core_that_clears_the_floor(self):
+        """The gap this closes: the floor says 2.0.0 and the core is 3.2.0, so
+        the floor alone would allow it — but the plugin said it stops at 2.x."""
+        m = {"name": "Legacy Plugin",
+             "compatible_versions": ["2.0.0 - 2.9.9"],
+             "versions": [{"ledmatrix_min_version": "2.0.0"}]}
+        ok, reason = compatibility.check(m, "3.2.0")
+        assert ok is False
+        assert "2.0.0 - 2.9.9" in reason and "3.2.0" in reason
+
+    def test_floor_blocks_when_ranges_would_allow(self):
+        m = {"name": "Needs Newer",
+             "compatible_versions": [">=1.0.0"],
+             "versions": [{"ledmatrix_min_version": "9.9.9"}]}
+        ok, reason = compatibility.check(m, "3.2.0")
+        assert ok is False
+        assert "9.9.9" in reason
+
+    def test_both_satisfied_allows(self):
+        m = {"compatible_versions": [">=2.0.0"],
+             "versions": [{"ledmatrix_min_version": "2.0.0"}]}
+        assert compatibility.check(m, "3.2.0") == (True, None)
+
+    def test_untrustworthy_core_still_bypasses_both_checks(self):
+        """A core reporting 1.0.0 fails `>=2.0.0`, which 41 of 42 published
+        manifests declare. Blocking there would empty the plugin store for
+        exactly the users who cannot be helped by it."""
+        m = {"compatible_versions": [">=2.0.0"],
+             "versions": [{"ledmatrix_min_version": "2.0.0"}]}
+        assert compatibility.check(m, "1.0.0") == (True, None)
+
+
+class TestUntrustworthyCoreAndTheSunset:
+    """The population B6 would otherwise break.
+
+    A device installed from the v3.1.0 release reports `__version__ = "1.0.0"`.
+    The gate cannot tell it apart from a genuine 1.0.0 install, so it cannot
+    reason about what that core actually has — and at the B6 sunset the
+    plugin's guarded-import fallback is gone. Without this rule the store hands
+    those users a 3.2.0-floored plugin that fails to load, and nothing else in
+    the system protects them.
+
+    The rule: on an untrustworthy core, refuse a floor *above* the ecosystem
+    baseline, allow anything at or below it. Every manifest published today
+    floors at exactly 2.0.0, so nobody is locked out of the store.
+    """
+
+    UNTRUSTWORTHY = ["1.0.0", "0.9.0", "1.9.9"]
+
+    @pytest.mark.parametrize("core", UNTRUSTWORTHY)
+    def test_refuses_a_post_sunset_floor(self, core):
+        m = {"name": "Hockey Scoreboard",
+             "versions": [{"ledmatrix_min_version": "3.2.0"}]}
+        ok, reason = compatibility.check(m, core)
+        assert ok is False, (
+            f"core {core} must not receive a 3.2.0-floored plugin: after the "
+            "sunset there is no fallback and it will fail to load"
+        )
+        assert "3.2.0" in reason and core in reason
+
+    @pytest.mark.parametrize("core", UNTRUSTWORTHY)
+    def test_still_allows_todays_ecosystem_floor(self, core):
+        """Regression guard: every published manifest floors at 2.0.0. If this
+        starts refusing, those users lose the plugin store entirely."""
+        m = {"versions": [{"ledmatrix_min": "2.0.0"}]}
+        assert compatibility.check(m, core) == (True, None)
+
+    @pytest.mark.parametrize("core", UNTRUSTWORTHY)
+    def test_still_allows_an_undeclared_floor(self, core):
+        assert compatibility.check({"id": "x"}, core) == (True, None)
+
+    def test_trustworthy_core_below_the_floor_is_unaffected(self):
+        """A core that reports 3.1.0 is believable and already handled by the
+        ordinary comparison — not by this rule."""
+        m = {"name": "P", "versions": [{"ledmatrix_min_version": "3.2.0"}]}
+        ok, reason = compatibility.check(m, "3.1.0")
+        assert ok is False
+        assert "too old to identify" not in reason, (
+            "a believable version should get the ordinary message"
+        )
+
+
+class TestMalformedManifests:
+    """A manifest we cannot parse must read as "no declared floor", not raise.
+
+    This matters more since the untrustworthy-core branch of check() began
+    resolving the floor for *every* manifest: one hand-edited or third-party
+    file with the wrong shape would take down the whole install path rather
+    than just itself.
+    """
+
+    @pytest.mark.parametrize("manifest", [
+        {"requires": ["python>=3.9"]},          # a list, not a mapping
+        {"requires": "python>=3.9"},            # a bare string
+        {"versions": {"a": 1}},                 # a mapping, not a list
+        {"versions": "1.0.0"},                  # a bare string
+        {"versions": [None]},                   # a list of the wrong thing
+        {"versions": []},
+    ])
+    def test_shape_errors_read_as_no_floor(self, manifest):
+        assert compatibility.declared_min_version(manifest) is None
+        assert compatibility.check(manifest, "1.0.0") == (True, None)
+        assert compatibility.check(manifest, "3.2.0") == (True, None)
+
+    def test_a_valid_requires_block_still_works(self):
+        assert compatibility.declared_min_version(
+            {"requires": {"min_ledmatrix_version": "3.2.0"}}) == "3.2.0"
+
+
+class TestSuffixedVersions:
+    """Prerelease and build metadata must not leak into the numbers.
+
+    The digit scrape used to pull them in: "3.2.0+build42" became (3, 2, 42)
+    and "3.2.0-rc1" became (3, 2, 1) — a release candidate ranking above its
+    own release. Both fed reject decisions.
+    """
+
+    @pytest.mark.parametrize("text,expected", [
+        ("3.2.0", (3, 2, 0)),
+        ("3.2.0+build42", (3, 2, 0)),
+        ("3.2.0-rc1", (3, 2, 0)),
+        ("3.2.0-rc.1+build.9", (3, 2, 0)),
+        ("v3.2.0+build42", (3, 2, 0)),
+    ])
+    def test_suffixes_are_dropped(self, text, expected):
+        assert compatibility.parse_semver(text) == expected
+
+    def test_a_build_of_the_pinned_version_is_not_refused(self):
+        """The regression: an exact "3.2.0" pin refused a core running
+        3.2.0+build42, which is that same version."""
+        ok, reason = compatibility.check(
+            {"compatible_versions": ["3.2.0"]}, "3.2.0+build42")
+        assert ok is True, f"refused a build of the pinned version: {reason}"
+
+    def test_a_release_candidate_does_not_outrank_its_release(self):
+        m = {"versions": [{"ledmatrix_min_version": "3.2.0"}]}
+        assert compatibility.check(m, "3.2.0-rc1")[0] is True
+        # ...and still refuses something genuinely older.
+        assert compatibility.check(m, "3.1.0-rc1")[0] is False

test/test_update_change_reporting.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""
+Tests that "which plugins have fresh data" survives the async update worker.
+
+Regression under test: run_scheduled_updates_with_changes() snapshotted
+plugin_last_update, called run_scheduled_updates(), and diffed the two. But
+run_scheduled_updates() only *enqueues* -- the work runs on the update worker
+and stamps the timestamp there, after the method has already returned. The
+snapshots were therefore always identical and the result always empty.
+
+Vegas depends on that result: it is what calls mark_plugin_updated(), which
+drops the cached content for a plugin whose data changed. With it always
+empty, a segment kept scrolling whatever it was first built from -- the
+"last night's live game still drawn as live the next morning" failure the
+coordinator comments describe. Observed on a live rig: zero update ticks in
+twenty minutes, with weather, stocks and news all updating.
+
+Run: python -m pytest test/test_update_change_reporting.py -v
+"""
+
+import ast
+import inspect
+import sys
+import threading
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.plugin_system.plugin_manager import PluginManager  # noqa: E402
+
+
+def _manager():
+    """A PluginManager with only the update-reporting state initialised."""
+    manager = PluginManager.__new__(PluginManager)
+    manager._completed_updates = set()
+    manager._completed_updates_lock = threading.Lock()
+    return manager
+
+
+class DrainCompletedUpdates(unittest.TestCase):
+    def setUp(self):
+        self.manager = _manager()
+
+    def test_nothing_completed_reports_nothing(self):
+        self.assertEqual(self.manager.drain_completed_updates(), [])
+
+    def test_a_completed_update_is_reported(self):
+        self.manager._note_update_completed("news")
+        self.assertEqual(self.manager.drain_completed_updates(), ["news"])
+
+    def test_draining_clears_so_the_next_poll_is_empty(self):
+        self.manager._note_update_completed("news")
+        self.manager.drain_completed_updates()
+        self.assertEqual(
+            self.manager.drain_completed_updates(), [],
+            "a plugin must be reported once per update, not on every poll, "
+            "or Vegas would drop its cached content every few seconds")
+
+    def test_repeated_completions_between_polls_collapse(self):
+        for _ in range(5):
+            self.manager._note_update_completed("weather")
+        self.assertEqual(self.manager.drain_completed_updates(), ["weather"])
+
+    def test_multiple_plugins_are_all_reported(self):
+        for plugin_id in ("news", "weather", "ledmatrix-stocks"):
+            self.manager._note_update_completed(plugin_id)
+        self.assertEqual(self.manager.drain_completed_updates(),
+                         ["ledmatrix-stocks", "news", "weather"])
+
+
+class CompletionReportingIsAsyncSafe(unittest.TestCase):
+    """The point of the change: completion may land after the call returns."""
+
+    def setUp(self):
+        self.manager = _manager()
+
+    def test_an_update_completing_after_the_call_is_still_reported(self):
+        """The exact shape of the bug.
+
+        The enqueueing call sees nothing, because the worker has not run yet.
+        The next poll must report it -- under the old diff it was lost, since
+        the second snapshot was taken before the worker ever stamped.
+        """
+        first = self.manager.drain_completed_updates()
+        self.assertEqual(first, [], "nothing has finished yet")
+
+        # The worker finishes some time later, on its own thread.
+        worker = threading.Thread(
+            target=self.manager._note_update_completed, args=("news",))
+        worker.start()
+        worker.join()
+
+        self.assertEqual(
+            self.manager.drain_completed_updates(), ["news"],
+            "an update that finishes between polls must still be reported")
+
+    def test_concurrent_completions_are_not_lost(self):
+        ids = ["plugin-%02d" % i for i in range(40)]
+        threads = [threading.Thread(target=self.manager._note_update_completed,
+                                    args=(pid,)) for pid in ids]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+        self.assertEqual(self.manager.drain_completed_updates(), sorted(ids))
+
+    def test_a_completion_during_a_drain_is_not_swallowed(self):
+        """A drain must not clear an entry it did not report."""
+        self.manager._note_update_completed("news")
+        reported = self.manager.drain_completed_updates()
+        # ...worker finishes another one immediately afterwards
+        self.manager._note_update_completed("weather")
+        self.assertEqual(reported, ["news"])
+        self.assertEqual(self.manager.drain_completed_updates(), ["weather"])
+
+
+class EveryStampRecordsACompletion(unittest.TestCase):
+    """The ledger is only correct if the production paths actually fill it.
+
+    Asserting on the mechanics alone passes even when nothing calls
+    _note_update_completed -- verified by deleting the call sites, which the
+    behavioural tests above did not notice. This checks the invariant at the
+    source: wherever a successful update stamps plugin_last_update, it must
+    also record the completion, or Vegas silently stops being told.
+    """
+
+    def test_success_paths_record_the_completion(self):
+        import src.plugin_system.plugin_manager as pm
+
+        tree = ast.parse(inspect.getsource(pm))
+        stamps = []
+        for node in ast.walk(tree):
+            if not isinstance(node, ast.With):
+                continue
+            # `with self._plugin_last_update_lock:` blocks that stamp a real
+            # time on success. Two stamps are deliberately excluded: the 0.0
+            # written at registration, and the failure path, which backs the
+            # timestamp off to space out retries -- neither means fresh data.
+            assigns_time = any(
+                isinstance(stmt, ast.Assign)
+                and any(isinstance(t, ast.Subscript)
+                        and getattr(t.value, "attr", None) == "plugin_last_update"
+                        for t in stmt.targets)
+                and not (isinstance(stmt.value, ast.Constant)
+                         and stmt.value.value == 0.0)
+                and "failure" not in ast.dump(stmt.value)
+                for stmt in node.body
+            )
+            if assigns_time:
+                stamps.append(node)
+
+        self.assertGreaterEqual(
+            len(stamps), 2,
+            "expected the worker and inline success paths to stamp the time; "
+            "if this drops, the search below is looking at the wrong thing")
+
+        for stamp in stamps:
+            enclosing = self._enclosing_function(tree, stamp)
+            calls = [n for n in ast.walk(enclosing)
+                     if isinstance(n, ast.Call)
+                     and getattr(n.func, "attr", None) == "_note_update_completed"]
+            self.assertTrue(
+                calls,
+                "%s stamps plugin_last_update on success but never calls "
+                "_note_update_completed, so a plugin's fresh data would never "
+                "be reported and Vegas would keep its stale cached content"
+                % enclosing.name)
+
+    @staticmethod
+    def _enclosing_function(tree, target):
+        best = None
+        for node in ast.walk(tree):
+            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                if node.lineno <= target.lineno <= (node.end_lineno or node.lineno):
+                    if best is None or node.lineno > best.lineno:
+                        best = node
+        return best
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)