fix(backup): restore must not depend on owning the file it replaces

Follow-up from testing the previous commit on real hardware, where the
secrets fix turned out to be both too narrow and slightly wrong.

Too narrow: config.json, wifi_config.json and ytm_auth.json are installed
root-owned and group-readable exactly like the secrets file, so all four
were unrestorable by the web service, not just one. `shutil.copy2` opens
the destination for writing, which needs permission on the *existing
file*; the web user could create files in that directory all day and
still not replace them.

Slightly wrong: the previous commit loosened the secrets file to
group-writable. That was treating the symptom. The real error was
deciding ownership from `ledmatrix.service` -- the display service, which
runs as root and only ever *reads* secrets -- when the account that
*writes* them is the web interface, which deliberately does not run as
root. Ownership now follows the web service's user and the mode stays
640.

`_copy_file` writes a temporary file alongside the target and renames
over it. That needs only directory permission, so a restore no longer
cares who owns the destination, and it is atomic: a crash mid-restore can
no longer leave a half-written config. The destination's mode is carried
across so restoring secrets does not widen them to the umask, and its
owner is carried across too when the OS allows it -- only root can hand a
file to another user, so a restore run by the web service keeps its own
ownership rather than pretending to preserve root's.

Verified on a device with all four config files set root-owned 640 and
unwritable by the web user: before, every one failed with EACCES; after,
the restore reports success with no errors and all four sections
restored, mode still 640, root still able to read them and the web
service still able to write them.

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

.github/workflows/release-version-check.yml

@@ -0,0 +1,40 @@
+name: Release version check
+
+# A release tag, the CHANGELOG, and src.__version__ must agree. They have not
+# always: v3.1.0 was tagged while src/__init__.py still said "1.0.0", which
+# silently exempted every device installed from that release from plugin
+# compatibility warnings. See docs/SPORTS_UNIFICATION.md (phase B4).
+on:
+  push:
+    tags: ["v*"]
+  release:
+    types: [published]
+  # Pre-flight: run this against the tag you are about to create.
+  workflow_dispatch:
+    inputs:
+      tag:
+        description: "Tag to check (e.g. v3.2.0)"
+        required: true
+        type: string
+
+permissions:
+  contents: read
+
+jobs:
+  version-matches-tag:
+    name: Tag matches src.__version__
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+        with:
+          persist-credentials: false
+
+      - uses: actions/setup-python@0b93645e9fea7318ecaed2b359559ac225c90a2b # v5.3.0
+        with:
+          python-version: "3.12"
+
+      # No dependencies: the script reads src/__init__.py and CHANGELOG.md only.
+      - name: Assert the tag, CHANGELOG and src.__version__ agree
+        run: python scripts/check_release_version.py "${TAG}"
+        env:
+          TAG: ${{ inputs.tag || github.ref_name }}

.github/workflows/test.yml

@@ -76,4 +76,12 @@ jobs:
             test/test_sports_core_promotions.py \
             test/test_sports_modes_promotions.py \
             test/test_sports_capabilities.py \
-            test/test_sports_scroll.py
+            test/test_sports_scroll.py \
+            test/test_version_consistency.py \
+            test/test_plugin_compatibility_gate.py \
+            test/test_install_preserves_existing.py \
+            test/test_core_owned_config_keys.py \
+            test/test_async_plugin_updates.py \
+            test/test_registry_id_resolution.py \
+            test/test_backup_manager.py \
+            test/test_plugin_update_reservation.py

CHANGELOG.md

@@ -29,6 +29,21 @@ Adoption is deliberately staged: the modules below ship here, plugins adopt them
 behind guarded imports, and only then do the bundled copies go away. Nothing in
 this release changes what an existing plugin loads.
 
+**This is also the first release that *enforces* `ledmatrix_min_version`.**
+Before it, the floor was advisory — the loader logged a warning and continued,
+and the plugin store never compared the core version at all, so an update could
+deliver a plugin that could not run. From 3.2.0 the store refuses such an
+install. That matters for the sunset rule: a plugin may only delete its bundled
+fallback once the cores in the field actually enforce the floor, which means
+waiting for 3.2.0 to be widely installed rather than merely released. See
+`docs/SPORTS_UNIFICATION.md`, phase B6.
+
+One deliberate exception: a core reporting a version below `2.0.0` is treated as
+*unknown* rather than old and is never blocked. The v3.1.0 release ships
+`__version__ = "1.0.0"` (the tag was cut before the string was bumped), and
+nearly every published manifest floors at `2.0.0` — so blocking on that number
+would lock those users out of the plugin store entirely.
+
 ### Added
 - `src/element_style.py` — per-element style resolver backing the
   `x-style-elements` config-schema extension. Already consumed (behind guarded
@@ -71,8 +86,37 @@ this release changes what an existing plugin loads.
   override point — see `docs/SPORTS_UNIFICATION.md` for where the line falls
   and why.
 
+- `src/plugin_system/compatibility.py` — the single place that answers "can this
+  plugin run on this core?", shared by the loader (advisory, at load time) and
+  the store (blocking, at install/update time) so the two cannot drift. Reads
+  every spelling published manifests use, including the deprecated
+  `versions[].ledmatrix_min`. It does **not** yet evaluate `compatible_versions`,
+  which is the schema-required field and can express upper bounds; closing that
+  is tracked in `docs/SPORTS_UNIFICATION.md` before B6.
+- `scripts/check_release_version.py` and a `Release version check` workflow —
+  assert that a tag, the newest CHANGELOG heading and `src.__version__` agree,
+  on pushed `v*` tags and published releases. Runnable via `workflow_dispatch`
+  to check a tag *before* creating it. Added because `v3.1.0` was tagged six
+  weeks before `src/__init__.py` was bumped to match, which is why devices
+  installed from that release report `1.0.0`.
+
 ### Changed
 - `src/__init__.py` bumped to **3.2.0** — the number the sunset rule keys on.
+- **The plugin store refuses an incompatible install.**
+  `StoreManager.install_plugin` now checks the downloaded manifest's declared
+  floor against `src.__version__` and refuses when the plugin needs a newer
+  core. The check sits in `install_plugin` because `_reinstall_with_rollback`
+  calls it, so a refused *update* restores the version the user already had.
+  Refusal requires evidence: an undeclared floor, an unparseable version on
+  either side, or an untrustworthy core version all allow the install.
+- **A failed install no longer destroys the plugin it replaced.**
+  `install_plugin` previously deleted the existing plugin directory before
+  downloading, so any later failure — a dropped connection, a malformed
+  manifest, or the new compatibility refusal — left the user with nothing. The
+  existing copy is now set aside and restored if the install fails, matching
+  the protection `_reinstall_with_rollback` already gave the update path.
+- `web_interface.__version__` re-exports `src.__version__` instead of carrying
+  its own hardcoded `"3.0.0"`, which had drifted two majors from the core.
 - **Live games are no longer dropped when the feed omits a game clock.**
   `SportsLive._is_game_really_over` previously (in the baseball and UFC
   plugin lineages) coerced a missing or non-string clock to the literal
@@ -86,6 +130,12 @@ this release changes what an existing plugin loads.
   there means kickoff rather than expiry.
 
 ### Fixed
+- **Plugin updates could hang the web request thread.** The per-plugin reinstall
+  locks were non-reentrant, and `_reinstall_with_rollback` holds one across its
+  call to `install_plugin` — which now takes the same lock to protect the
+  set-aside/restore above. That nesting deadlocked
+  `update_plugin → _reinstall_with_rollback → install_plugin`, the standard
+  path for every monorepo plugin update. The locks are now `RLock`s.
 - `FontManager` resolves `assets/fonts` against the core install root instead
   of the process working directory, so font loading works when the process
   starts elsewhere (e.g. the plugin safety harness on CI).

README.md

@@ -357,6 +357,12 @@ sudo bash ./first_time_install.sh
 
 This single script installs services, dependencies, configures permissions and sudoers, and validates the setup.
 
+It finishes by asking whether to reboot. If you run it non-interactively — piped, over a script, or with `-y` — there is no one to ask, so **it reboots immediately without prompting**. Pass `--no-reboot-prompt` to install without rebooting:
+
+```bash
+sudo bash ./first_time_install.sh -y --no-reboot-prompt
+```
+
 </details>
 
 </details>

first_time_install.sh

@@ -1483,26 +1483,37 @@ if [ -f "$PROJECT_ROOT_DIR/config/config.json" ]; then
 fi
 
 # Set proper permissions for secrets file (restrictive: owner rw, group r)
-# If service runs as root, set ownership to root so it can read as owner
-# Otherwise, use ACTUAL_USER and rely on group membership
+# Owned by whoever WRITES the file, which is the web interface.
+#
+# This used to read the User= of ledmatrix.service — the display service —
+# and, finding root, hand the file to root:ledmatrix 640. But the display
+# service only ever reads secrets, and root can read any file regardless of
+# mode. The account that *writes* them is the web interface: it saves config
+# edits and performs backup restores, and it deliberately does not run as root
+# (a web server should not). So a root-owned, group-read-only file left the web
+# UI unable to write its own secrets, and restoring a backup failed with
+# "Permission denied: config_secrets.json" while every other file in the same
+# backup restored fine.
+#
+# Owning by the writer keeps the tighter 640 rather than loosening to
+# group-writable, and root still reads it as superuser.
 if [ -f "$PROJECT_ROOT_DIR/config/config_secrets.json" ]; then
-    # Check if service runs as root (from service file or template)
-    SERVICE_USER="root"
-    if [ -f "/etc/systemd/system/ledmatrix.service" ]; then
-        SERVICE_USER=$(grep "^User=" /etc/systemd/system/ledmatrix.service | cut -d'=' -f2 || echo "root")
-    elif [ -f "$PROJECT_ROOT_DIR/systemd/ledmatrix.service" ]; then
-        SERVICE_USER=$(grep "^User=" "$PROJECT_ROOT_DIR/systemd/ledmatrix.service" | cut -d'=' -f2 || echo "root")
-    fi
-    
-    if [ "$SERVICE_USER" = "root" ]; then
-        # Service runs as root - set ownership to root so it can read as owner
-        chown "root:$LEDMATRIX_GROUP" "$PROJECT_ROOT_DIR/config/config_secrets.json" || true
-        echo "✓ Secrets file permissions set (root:ledmatrix for root service)"
-    else
-        # Service runs as regular user - use ACTUAL_USER and rely on group membership
-        chown "$ACTUAL_USER:$LEDMATRIX_GROUP" "$PROJECT_ROOT_DIR/config/config_secrets.json" || true
-        echo "✓ Secrets file permissions set ($ACTUAL_USER:ledmatrix)"
+    # The web service is the writer; fall back to the display service, then to
+    # the installing user, so an unusual layout still lands somewhere sensible.
+    SECRETS_OWNER=""
+    for unit in "/etc/systemd/system/ledmatrix-web.service" \
+                "$PROJECT_ROOT_DIR/systemd/ledmatrix-web.service"; do
+        if [ -f "$unit" ]; then
+            SECRETS_OWNER=$(grep -m1 "^User=" "$unit" | cut -d'=' -f2)
+            [ -n "$SECRETS_OWNER" ] && break
+        fi
+    done
+    if [ -z "$SECRETS_OWNER" ]; then
+        SECRETS_OWNER="$ACTUAL_USER"
     fi
+    # A root-owned file is only correct when the writer really is root.
+    chown "$SECRETS_OWNER:$LEDMATRIX_GROUP" "$PROJECT_ROOT_DIR/config/config_secrets.json" || true
+    echo "✓ Secrets file owned by the web service user ($SECRETS_OWNER:$LEDMATRIX_GROUP, mode 640)"
     chmod 640 "$PROJECT_ROOT_DIR/config/config_secrets.json"
 fi
 

src/backup_manager.py

@@ -16,6 +16,7 @@ import json
 import logging
 import os
 import shutil
+import stat
 import socket
 import tempfile
 import zipfile
@@ -82,6 +83,10 @@ BUNDLED_FONTS: frozenset[str] = frozenset({
 _CONFIG_REL = Path("config/config.json")
 _SECRETS_REL = Path("config/config_secrets.json")
 _WIFI_REL = Path("config/wifi_config.json")
+# Sits in config/ next to the three above and is pure user state — a
+# YouTube Music session that has to be re-authenticated by hand if lost.
+# It was omitted from backups, so a restore silently signed the user out.
+_YTM_REL = Path("config/ytm_auth.json")
 _FONTS_REL = Path("assets/fonts")
 _PLUGIN_UPLOADS_REL = Path("assets/plugins")
 _STATE_REL = Path("data/plugin_state.json")
@@ -303,6 +308,9 @@ def create_backup(
             if (project_root / _WIFI_REL).exists():
                 zf.write(project_root / _WIFI_REL, _WIFI_REL.as_posix())
                 contents.append("wifi")
+            if (project_root / _YTM_REL).exists():
+                zf.write(project_root / _YTM_REL, _YTM_REL.as_posix())
+                contents.append("ytm_auth")
 
             # User-uploaded fonts.
             user_fonts = iter_user_fonts(project_root)
@@ -348,6 +356,7 @@ def preview_backup_contents(project_root: Path) -> Dict[str, Any]:
         "has_config": (project_root / _CONFIG_REL).exists(),
         "has_secrets": (project_root / _SECRETS_REL).exists(),
         "has_wifi": (project_root / _WIFI_REL).exists(),
+        "has_ytm_auth": (project_root / _YTM_REL).exists(),
         "user_fonts": [p.name for p in iter_user_fonts(project_root)],
         "plugin_uploads": len(iter_plugin_uploads(project_root)),
         "plugins": list_installed_plugins(project_root),
@@ -429,6 +438,8 @@ def validate_backup(zip_path: Path) -> Tuple[bool, str, Dict[str, Any]]:
                 detected.append("secrets")
             if _WIFI_REL.as_posix() in names:
                 detected.append("wifi")
+            if _YTM_REL.as_posix() in names:
+                detected.append("ytm_auth")
             if any(n.startswith(_FONTS_REL.as_posix() + "/") for n in names):
                 detected.append("fonts")
             if any(
@@ -481,8 +492,61 @@ def _extract_zip_safe(zip_path: Path, dest_dir: Path) -> None:
 
 
 def _copy_file(src: Path, dst: Path) -> None:
+    """Replace ``dst`` with ``src``, atomically, without needing to own ``dst``.
+
+    ``shutil.copy2`` opens the destination for writing, so it needs write
+    permission on the *existing file*. Several config files are installed
+    root-owned and group-readable while the web interface — which is what runs
+    a restore — deliberately runs as a non-root user. Restoring those failed
+    with EACCES even though the account could create files in the same
+    directory perfectly well.
+
+    Writing a temporary file alongside and renaming over the target needs only
+    directory permission, which the web user has. It is also atomic: a crash
+    mid-restore can no longer leave a half-written config behind.
+
+    The destination's existing mode is preserved when there is one, so
+    restoring secrets does not silently widen them to the umask default.
+    """
     dst.parent.mkdir(parents=True, exist_ok=True)
-    shutil.copy2(src, dst)
+
+    existing_mode: Optional[int] = None
+    existing_owner: Optional[Tuple[int, int]] = None
+    if dst.exists():
+        try:
+            info = dst.stat()
+            existing_mode = stat.S_IMODE(info.st_mode)
+            existing_owner = (info.st_uid, info.st_gid)
+        except OSError:
+            existing_mode = None
+            existing_owner = None
+
+    fd, tmp_name = tempfile.mkstemp(dir=str(dst.parent), prefix=f".{dst.name}.", suffix=".tmp")
+    os.close(fd)
+    tmp_path = Path(tmp_name)
+    try:
+        shutil.copyfile(src, tmp_path)
+        if existing_mode is not None:
+            os.chmod(tmp_path, existing_mode)
+        else:
+            shutil.copymode(src, tmp_path)
+        if existing_owner is not None:
+            # Replacing a file creates a new inode owned by whoever is running,
+            # which would silently move a root-owned config to the web user.
+            # Carry the previous owner across when the OS permits it — only
+            # root can hand a file to another user, so this is best-effort and
+            # a plain restore as the web user simply keeps its own ownership.
+            try:
+                os.chown(tmp_path, existing_owner[0], existing_owner[1])
+            except (OSError, PermissionError):
+                pass
+        os.replace(tmp_path, dst)
+    except BaseException:
+        try:
+            tmp_path.unlink()
+        except OSError:
+            pass
+        raise
 
 
 def restore_backup(
@@ -546,6 +610,18 @@ def restore_backup(
         elif (tmp_dir / _WIFI_REL).exists():
             result.skipped.append("wifi")
 
+        # YouTube Music session. Follows restore_wifi rather than getting its
+        # own flag: it is device-local auth in the same sense, and a separate
+        # toggle for one file would be noise in the restore dialog.
+        if options.restore_wifi and (tmp_dir / _YTM_REL).exists():
+            try:
+                _copy_file(tmp_dir / _YTM_REL, project_root / _YTM_REL)
+                result.restored.append("ytm_auth")
+            except OSError as e:
+                result.errors.append(f"Failed to restore ytm_auth.json: {e}")
+        elif (tmp_dir / _YTM_REL).exists():
+            result.skipped.append("ytm_auth")
+
         # User fonts — skip anything that collides with a bundled font.
         tmp_fonts = tmp_dir / _FONTS_REL
         if options.restore_fonts and tmp_fonts.exists():

src/base_classes/data_sources.py

@@ -44,9 +44,16 @@ class DataSource(ABC):
         """Fetch standings for a sport/league."""
     
     def get_headers(self) -> Dict[str, str]:
-        """Get headers for API requests."""
+        """Get headers for API requests.
+
+        The agent carries the project URL deliberately. Around 2026-08-04 ESPN
+        began returning 403 for bare custom tokens like 'LEDMatrix/1.0' — and
+        for browser-style strings — while accepting an agent that identifies
+        the client and links to it. An Accept header alone does not rescue the
+        bare form when the request goes out through requests.
+        """
         return {
-            'User-Agent': 'LEDMatrix/1.0',
+            'User-Agent': 'LEDMatrix/1.0 (+https://github.com/ChuckBuilds/LEDMatrix)',
             'Accept': 'application/json'
         }
 

src/common/api_helper.py

@@ -56,7 +56,9 @@ class APIHelper:
         
         # Default headers
         self.session.headers.update({
-            'User-Agent': 'LEDMatrix-Common/1.0',
+            # Identifies the client and links to it: ESPN began 403ing bare
+            # custom tokens (and browser strings) around 2026-08-04.
+            'User-Agent': 'LEDMatrix/1.0 (+https://github.com/ChuckBuilds/LEDMatrix)',
             'Accept': 'application/json',
             'Accept-Language': 'en-US,en;q=0.9',
             'Accept-Encoding': 'gzip, deflate, br',

src/plugin_system/compatibility.py

@@ -0,0 +1,253 @@
+"""One place that answers "can this plugin run on this core?".
+
+Two callers ask that question and they must not drift apart:
+
+- `PluginLoader._warn_if_incompatible` — at load time, **advisory**. A plugin
+  already on disk keeps loading regardless, because the guarded-import pattern
+  means most incompatibilities degrade rather than break.
+- `PluginStoreManager.install_plugin` — at install/update time, **blocking**.
+  This is the point where refusing costs the user nothing (they keep the
+  version they already had) and allowing can cost them a plugin that fails to
+  load with only a log line to explain it.
+
+## The trustworthiness problem
+
+The core's own `__version__` has not always been right. `v3.1.0` was tagged
+2026-05-31 while `src/__init__.py` still said `"1.0.0"`; the bump landed
+2026-07-12. Devices installed from that release report `1.0.0` — below the
+floor that essentially every published plugin declares.
+
+So a core reporting a version below `TRUSTWORTHY_FLOOR` is treated as
+**unknown, not old**: it neither warns nor blocks. Blocking on it would be far
+worse than the problem being solved — nearly every manifest in the ecosystem
+floors at `2.0.0`, so a strict gate would stop those users installing *any*
+plugin. They are unprotected until they update the core, which is also what
+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.
+# See the module docstring.
+TRUSTWORTHY_FLOOR: Tuple[int, int, int] = (2, 0, 0)
+
+
+def parse_semver(value: Any) -> Optional[Tuple[int, int, int]]:
+    """Parse ``X.Y.Z`` (extra parts and suffixes ignored) into a comparable
+    3-tuple, or ``None`` when unparseable. A leading ``v`` is tolerated."""
+    if not isinstance(value, str):
+        return None
+    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:
+        return None
+    while len(nums) < 3:
+        nums.append(0)
+    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.
+
+    Checked in order of specificity. `ledmatrix_min` is the deprecated spelling
+    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')
+    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')
+    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
+
+
+def check(manifest: Dict[str, Any], core_version: str) -> Tuple[bool, Optional[str]]:
+    """Return ``(compatible, reason)``.
+
+    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 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."
+        )
+
+    return True, None

src/plugin_system/plugin_loader.py

@@ -702,34 +702,25 @@ class PluginLoader:
         newer than the running core. Advisory only — never raises — so a
         plugin that guards optional features with try/except keeps working.
         """
-        declared = (
-            manifest.get('min_ledmatrix_version')
-            or manifest.get('requires', {}).get('min_ledmatrix_version')
-        )
-        if not declared:
-            versions = manifest.get('versions') or []
-            if versions and isinstance(versions[0], dict):
-                declared = (versions[0].get('ledmatrix_min_version')
-                            or versions[0].get('ledmatrix_min'))
-        needed = self._parse_semver(declared)
-        if needed is None:
+        from src import __version__ as core_version
+        from src.plugin_system import compatibility
+
+        compatible, _reason = compatibility.check(manifest, core_version)
+        if compatible:
+            # Distinguish "fine" from "couldn't tell" for anyone reading logs:
+            # a core below the trustworthy floor is skipped, not cleared.
+            current = compatibility.parse_semver(core_version)
+            if current is None or current < compatibility.TRUSTWORTHY_FLOOR:
+                self.logger.debug(
+                    "Skipping version compatibility check for %s: core __version__ "
+                    "(%s) is below the ecosystem floor", plugin_id, core_version)
             return
 
-        from src import __version__ as core_version
-        current = self._parse_semver(core_version)
-        # Anti-spam guard: if the core's own version number is stale (below
-        # the ecosystem floor every shipped plugin declares), comparing would
-        # warn on nearly everything — skip with a debug note instead.
-        if current is None or current < (2, 0, 0):
-            self.logger.debug(
-                "Skipping version compatibility check for %s: core __version__ "
-                "(%s) is below the ecosystem floor", plugin_id, core_version)
-            return
-        if needed > current:
-            self.logger.warning(
-                "Plugin %s declares min LEDMatrix version %s but this core is %s — "
-                "features it relies on may be missing; update the core or expect "
-                "degraded fallbacks", plugin_id, declared, core_version)
+        declared = compatibility.declared_min_version(manifest)
+        self.logger.warning(
+            "Plugin %s declares min LEDMatrix version %s but this core is %s — "
+            "features it relies on may be missing; update the core or expect "
+            "degraded fallbacks", plugin_id, declared, core_version)
 
     def load_plugin(
         self,

src/plugin_system/plugin_manager.py

@@ -113,6 +113,15 @@ class PluginManager:
         self._update_queue: "queue.Queue[Optional[Tuple[str, float]]]" = queue.Queue()
         self._pending_updates: set = set()
         self._pending_lock = threading.Lock()
+        # Serializes the "is this plugin eligible?" -> "claim it (RUNNING)"
+        # transition. Two schedulers run concurrently in practice — the render
+        # loop's _tick_plugin_updates() and Vegas mode's vegas-plugin-tick
+        # daemon thread, which is never joined — so without this both can
+        # observe ENABLED and both call update() on the same plugin. Held only
+        # across the check and the state transition, never across update()
+        # itself: that would serialize slow plugins behind each other and
+        # reintroduce the stall the async worker exists to avoid.
+        self._reservation_lock = threading.Lock()
         self._plugin_locks: Dict[str, threading.Lock] = {}
         self._plugin_locks_guard = threading.Lock()
         self._update_worker: Optional[threading.Thread] = None
@@ -395,6 +404,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 +459,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.
@@ -777,25 +817,69 @@ class PluginManager:
             if self.health_tracker and self.health_tracker.should_skip_plugin(plugin_id):
                 continue
 
-            # Check if plugin can execute
-            if not self.state_manager.can_execute(plugin_id):
-                continue
-
             interval = self._get_plugin_update_interval(plugin_id, plugin_instance)
             if interval is None:
                 continue
 
-            with self._plugin_last_update_lock:
-                last_update = self.plugin_last_update.get(plugin_id, 0.0)
+            # Eligibility check, due check and the RUNNING transition happen
+            # together, so a concurrent scheduler cannot claim the same plugin.
+            if not self._reserve_for_update(plugin_id, current_time, interval):
+                continue
 
-            if last_update == 0.0 or (current_time - last_update) >= interval:
-                if self._synchronous_updates:
-                    # Kill-switch path: the original inline execution
-                    # (blocks the caller until update() completes/times out)
-                    self.state_manager.set_state(plugin_id, PluginState.RUNNING)
-                    self._execute_update_now(plugin_id, plugin_instance, current_time)
-                else:
-                    self._enqueue_update(plugin_id, current_time)
+            if self._synchronous_updates:
+                # Kill-switch path: the original inline execution
+                # (blocks the caller until update() completes/times out)
+                self._execute_update_now(plugin_id, plugin_instance, current_time)
+            else:
+                self._enqueue_update(plugin_id, current_time)
+
+    def _reserve_for_update(
+        self,
+        plugin_id: str,
+        current_time: Optional[float] = None,
+        interval: Optional[float] = None,
+    ) -> bool:
+        """Atomically claim a plugin for update, returning True if we won it.
+
+        can_execute() and the RUNNING transition have to happen under one lock.
+        As two separate calls, two scheduler threads can both see ENABLED and
+        both go on to run the same plugin's update() concurrently — unsafe for
+        any plugin that isn't reentrant (shared mutable state, a non-thread-safe
+        HTTP session or cache).
+
+        The due-time check is inside the lock too. Leaving it outside would let
+        a second thread that had already decided "due" claim the plugin the
+        instant the first finished, running update() twice in one interval.
+
+        Args:
+            plugin_id: Plugin to claim.
+            current_time: Now, for the due check. Omit to skip that check.
+            interval: Seconds between updates. Omit to skip the due check.
+
+        Returns:
+            True if this caller reserved the plugin and must dispatch it,
+            False if it is ineligible, not yet due, or already claimed.
+        """
+        with self._reservation_lock:
+            if not self.state_manager.can_execute(plugin_id):
+                return False
+
+            if current_time is not None and interval is not None:
+                with self._plugin_last_update_lock:
+                    last_update = self.plugin_last_update.get(plugin_id, 0.0)
+                if last_update != 0.0 and (current_time - last_update) < interval:
+                    return False
+
+            self.state_manager.set_state(plugin_id, PluginState.RUNNING)
+            return True
+
+    def _release_reservation(self, plugin_id: str) -> None:
+        """Hand a claimed plugin back when it never got dispatched.
+
+        Without this a plugin reserved but not queued would sit in RUNNING
+        forever, and can_execute() would refuse it on every later tick.
+        """
+        self.state_manager.set_state(plugin_id, PluginState.ENABLED)
 
     def get_plugin_lock(self, plugin_id: str) -> threading.Lock:
         """Per-plugin lock keeping update() and display() mutually exclusive.
@@ -812,16 +896,40 @@ class PluginManager:
             return lock
 
     def _enqueue_update(self, plugin_id: str, scheduled_time: float) -> None:
-        """Queue a due update for the background worker (dedup while pending)."""
+        """Queue an already-reserved update for the background worker.
+
+        The caller has reserved the plugin (RUNNING), which is what blocks
+        re-entry and shows the truthful state in the web UI while the item
+        waits its turn. The pending set stays as a second line of defence; if
+        it ever fires the reservation has to be handed back, or the plugin
+        would sit in RUNNING with nothing queued to release it.
+        """
         with self._pending_lock:
             if plugin_id in self._pending_updates:
+                self.logger.warning(
+                    "Plugin %s reserved for update but already queued; "
+                    "releasing the reservation", plugin_id)
+                self._release_reservation(plugin_id)
                 return
             self._pending_updates.add(plugin_id)
-        # RUNNING is set at enqueue time so can_execute() blocks re-entry and
-        # the web UI shows the truthful state while the item waits its turn.
-        self.state_manager.set_state(plugin_id, PluginState.RUNNING)
-        self._ensure_update_worker()
-        self._update_queue.put((plugin_id, scheduled_time))
+        try:
+            self._ensure_update_worker()
+            self._update_queue.put((plugin_id, scheduled_time))
+        except Exception as exc:  # pylint: disable=broad-except
+            # Thread.start() raises RuntimeError when the OS refuses a new
+            # thread — a real condition on a Pi under memory pressure. Nothing
+            # is queued to release the plugin at that point, so the claim has to
+            # be undone here, or it sits in RUNNING with nothing to clear it and
+            # can_execute() refuses it for the rest of the process. Swallowed
+            # rather than raised so the remaining plugins in this tick still get
+            # their turn.
+            self.logger.error(
+                "Could not queue update for plugin %s (%s: %s); releasing the "
+                "reservation so the next tick can retry",
+                plugin_id, type(exc).__name__, exc, exc_info=True)
+            with self._pending_lock:
+                self._pending_updates.discard(plugin_id)
+            self._release_reservation(plugin_id)
 
     def _ensure_update_worker(self) -> None:
         if self._update_worker is not None and self._update_worker.is_alive():
@@ -906,6 +1014,14 @@ class PluginManager:
                     return
                 finished['done'] = True
             try:
+                # Drop the queue reservation *before* the state goes back to
+                # ENABLED. The other order leaves a window where a scheduler
+                # sees ENABLED, reserves the plugin, then finds it still in
+                # _pending_updates -- the enqueue is dropped and the plugin
+                # would sit in RUNNING with nothing left to release it.
+                if lock is not None:
+                    with self._pending_lock:
+                        self._pending_updates.discard(plugin_id)
                 if success:
                     with self._plugin_last_update_lock:
                         self.plugin_last_update[plugin_id] = scheduled_time
@@ -918,8 +1034,6 @@ class PluginManager:
             finally:
                 if lock is not None:
                     lock.release()
-                    with self._pending_lock:
-                        self._pending_updates.discard(plugin_id)
 
         if lock is None:
             # Synchronous / no-lock path: unchanged behavior.
@@ -1010,13 +1124,12 @@ class PluginManager:
             if not hasattr(plugin_instance, "update"):
                 continue
             
-            # Check if plugin can execute
-            if not self.state_manager.can_execute(plugin_id):
+            # Eligibility check and the RUNNING transition together, so a
+            # concurrent scheduler cannot claim the same plugin (see
+            # _reserve_for_update).
+            if not self._reserve_for_update(plugin_id):
                 continue
-            
-            # Update state to RUNNING
-            self.state_manager.set_state(plugin_id, PluginState.RUNNING)
-            
+
             try:
                 success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
                 if success:

src/plugin_system/store_manager.py

@@ -149,18 +149,27 @@ class PluginStoreManager:
         # loser can end up renaming the winner's in-progress install aside
         # mid-download, stealing its own rollback safety net. Keyed by
         # plugin_id so unrelated plugins still update concurrently.
-        self._reinstall_locks: Dict[str, threading.Lock] = {}
+        # Reentrant: install_plugin takes this lock, and _reinstall_with_rollback
+        # holds it across its call to install_plugin. A plain Lock would
+        # self-deadlock on that nesting.
+        self._reinstall_locks: Dict[str, "threading.RLock"] = {}
         self._reinstall_locks_guard = threading.Lock()
 
         # Ensure plugins directory exists
         self.plugins_dir.mkdir(exist_ok=True)
 
-    def _get_reinstall_lock(self, plugin_id: str) -> threading.Lock:
-        """Lazily create (or fetch) the per-plugin reinstall lock."""
+    def _get_reinstall_lock(self, plugin_id: str):
+        """Lazily create (or fetch) the per-plugin reinstall lock.
+
+        Reentrant by necessity: `install_plugin` acquires it to protect its
+        set-aside/restore, and `_reinstall_with_rollback` holds it across its
+        own call to `install_plugin`. With a plain `Lock` that nesting
+        deadlocks the request thread.
+        """
         with self._reinstall_locks_guard:
             lock = self._reinstall_locks.get(plugin_id)
             if lock is None:
-                lock = threading.Lock()
+                lock = threading.RLock()
                 self._reinstall_locks[plugin_id] = lock
             return lock
 
@@ -1134,7 +1143,7 @@ class PluginStoreManager:
         """
         registry = self.fetch_registry()
         plugins = registry.get('plugins', []) or []
-        plugin_info = next((p for p in plugins if p['id'] == plugin_id), None)
+        plugin_info = self._match_registry_entry(plugins, plugin_id)
 
         if not plugin_info:
             return None
@@ -1174,6 +1183,37 @@ class PluginStoreManager:
 
         return plugin_info
 
+    @staticmethod
+    def _match_registry_entry(plugins: List[Dict], plugin_id: str) -> Optional[Dict]:
+        """Find a registry entry by its id, or by the directory it installs to.
+
+        Four shipped plugins have a registry ``id`` that differs from the ``id``
+        in their own manifest: ``weather`` installs to ``plugins/ledmatrix-weather``,
+        and likewise stocks, music and leaderboard. Installation already prefers
+        the manifest id for the directory name, so on disk, in ``config.json``
+        and in a backup manifest those plugins are called ``ledmatrix-weather``.
+
+        Only the registry calls them ``weather``, and nothing resolved that in
+        reverse: restoring a backup asked the store for ``ledmatrix-weather``
+        and got "Plugin not found in registry", silently dropping four enabled
+        plugins from a restored device.
+
+        Matching ``plugin_path`` fixes it without renaming any published id,
+        which would orphan ``plugin_state.json`` entries keyed on the old ones.
+        Exact id always wins, so an entry whose *path* happens to collide with
+        another entry's id cannot shadow it.
+        """
+        if not plugin_id:
+            return None
+        exact = next((p for p in plugins if p.get('id') == plugin_id), None)
+        if exact is not None:
+            return exact
+        for entry in plugins:
+            path = (entry.get('plugin_path') or '').rstrip('/')
+            if path and path.rsplit('/', 1)[-1] == plugin_id:
+                return entry
+        return None
+
     def get_registry_info(self, plugin_id: str) -> Optional[Dict]:
         """
         Get plugin information from the registry cache only (no GitHub API calls).
@@ -1189,9 +1229,93 @@ class PluginStoreManager:
         """
         registry = self.fetch_registry()
         plugins = registry.get('plugins', []) or []
-        return next((p for p in plugins if p.get('id') == plugin_id), None)
+        return self._match_registry_entry(plugins, plugin_id)
     
     def install_plugin(self, plugin_id: str, branch: Optional[str] = None) -> bool:
+        """Install a plugin, keeping any existing install until the new one is
+        known good.
+
+        `_install_plugin_impl` deletes the existing directory *before*
+        downloading, so every failure after that point — a dropped connection, a
+        malformed manifest, or the compatibility gate refusing the new version —
+        left the user with no plugin at all. `_reinstall_with_rollback` gives the
+        *update* path exactly this protection; a direct install had none, and the
+        compatibility gate added a new way to reach it.
+
+        Pass-through when nothing is installed, and when called from
+        `_reinstall_with_rollback`, which has already moved the old copy aside.
+
+        The aside name embeds '.standalone-backup-' so plugin discovery
+        (`plugin_manager._scan_directory_for_plugins`) skips it even though it
+        still holds a manifest.json.
+
+        Held under the per-plugin reinstall lock for the same reason
+        `_reinstall_with_rollback` is: the web UI runs Flask with
+        threaded=True, so a double-clicked Install button gives two threads the
+        same plugin_id. Interleaved, one thread's restore would delete the
+        other's freshly installed copy. The lock is reentrant because the
+        rollback path already holds it when it calls in here.
+        """
+        with self._get_reinstall_lock(plugin_id):
+            plugin_path = self.plugins_dir / plugin_id
+            if not plugin_path.exists():
+                return self._install_plugin_impl(plugin_id, branch)
+
+            backup_path = plugin_path.with_name(
+                f"{plugin_path.name}.standalone-backup-preinstall")
+            if backup_path.exists() and not self._safe_remove_directory(backup_path):
+                # Can't stage a safety net. Better to attempt the install than
+                # to refuse outright, which is what callers got before this
+                # existed.
+                self.logger.warning(
+                    "Could not clear stale pre-install backup for %s at %s; "
+                    "installing without a rollback net", plugin_id, backup_path)
+                return self._install_plugin_impl(plugin_id, branch)
+
+            try:
+                plugin_path.rename(backup_path)
+            except OSError as e:
+                self.logger.warning(
+                    "Could not set aside existing install of %s (%s); "
+                    "installing without a rollback net", plugin_id, e)
+                return self._install_plugin_impl(plugin_id, branch)
+
+            try:
+                installed = self._install_plugin_impl(plugin_id, branch)
+            except Exception:
+                self._restore_preinstall_backup(plugin_id, plugin_path, backup_path)
+                raise
+
+            if installed:
+                if not self._safe_remove_directory(backup_path):
+                    self.logger.warning(
+                        "Install of %s succeeded but the previous copy at %s "
+                        "could not be removed; it will be cleared on the next "
+                        "install", plugin_id, backup_path)
+                return True
+
+            self._restore_preinstall_backup(plugin_id, plugin_path, backup_path)
+            return False
+
+    def _restore_preinstall_backup(
+        self, plugin_id: str, plugin_path: Path, backup_path: Path
+    ) -> None:
+        """Put the previous install back after a failed (re)install."""
+        self.logger.error(
+            "Install of %s failed; restoring the previous version", plugin_id)
+        try:
+            if plugin_path.exists():
+                # Partial download debris from the failed install.
+                self._safe_remove_directory(plugin_path)
+            backup_path.rename(plugin_path)
+            self.logger.info("Restored previous install of %s", plugin_id)
+        except OSError as e:
+            self.logger.error(
+                "CRITICAL: could not restore %s from %s: %s. The previous "
+                "install is preserved there — rename it back manually.",
+                plugin_id, backup_path, e)
+
+    def _install_plugin_impl(self, plugin_id: str, branch: Optional[str] = None) -> bool:
         """
         Install a plugin from the official registry. Always installs the latest commit
         from the repository's default branch (or specified branch).
@@ -1333,6 +1457,26 @@ class PluginStoreManager:
                     self._safe_remove_directory(plugin_path)
                     return False
 
+                # Refuse a plugin that needs a newer core than this one. The
+                # registry carries no compatibility field, so the floor is only
+                # knowable once the files are down — checking here, before
+                # dependency installation, is the earliest possible point.
+                #
+                # Refusing costs the user nothing: on an update this returns
+                # False and _reinstall_with_rollback restores the version they
+                # already had. Allowing it costs them a plugin that raises
+                # ModuleNotFoundError at load and is reported only as one line
+                # in the journal. See docs/SPORTS_UNIFICATION.md (phase B4/B6).
+                from src import __version__ as core_version
+                from src.plugin_system import compatibility
+
+                compatible, reason = compatibility.check(manifest, core_version)
+                if not compatible:
+                    self.logger.error(
+                        "Refusing to install %s: %s", plugin_id, reason)
+                    self._safe_remove_directory(plugin_path)
+                    return False
+
                 if 'entry_point' not in manifest:
                     manifest['entry_point'] = 'manager.py'
                     manifest_modified = True

test/test_backup_manager.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import stat
 import zipfile
 from pathlib import Path
 
@@ -41,6 +42,13 @@ def _make_project(root: Path) -> Path:
         json.dumps({"ap_mode": {"ssid": "LEDMatrix"}}),
         encoding="utf-8",
     )
+    # Device-local auth that lives in config/ like the three above. It was
+    # omitted from backups, so a restore silently signed the user out of
+    # YouTube Music and they had to re-authenticate by hand.
+    (root / "config" / "ytm_auth.json").write_text(
+        json.dumps({"token": "YTM-TOKEN"}),
+        encoding="utf-8",
+    )
 
     fonts = root / "assets" / "fonts"
     fonts.mkdir(parents=True)
@@ -240,6 +248,10 @@ def test_restore_roundtrip(project: Path, empty_project: Path, tmp_path: Path) -
     restored_secrets = json.loads((empty_project / "config" / "config_secrets.json").read_text())
     assert restored_secrets["ledmatrix-weather"]["api_key"] == "SECRET"
 
+    assert "ytm_auth" in result.restored
+    restored_ytm = json.loads((empty_project / "config" / "ytm_auth.json").read_text())
+    assert restored_ytm["token"] == "YTM-TOKEN"
+
     # User font restored, bundled font untouched.
     assert (empty_project / "assets" / "fonts" / "my-custom-font.ttf").read_bytes() == b"\x00\x01USER"
     assert (empty_project / "assets" / "fonts" / "5x7.bdf").read_text() == "BUNDLED"
@@ -282,3 +294,39 @@ def test_restore_rejects_malicious_zip(empty_project: Path, tmp_path: Path) -> N
     # validate_backup catches it before extraction.
     assert not result.success
     assert any("unsafe" in e.lower() for e in result.errors)
+
+
+def test_restore_over_a_file_the_user_cannot_write(
+    project: Path, empty_project: Path, tmp_path: Path
+) -> None:
+    """Restore must not need write permission on the destination *file*.
+
+    Reproduces what a fresh install leaves behind: config files owned by root
+    and only group-readable, while the web interface that performs the restore
+    runs as a non-root user. shutil.copy2 opens the destination for writing and
+    failed with EACCES; writing alongside and renaming needs only directory
+    permission, which that account has.
+
+    Simulated here by making the destination read-only — the owner cannot
+    open it for writing either, but can still replace it within its directory.
+    """
+    zip_path = create_backup(project, output_dir=tmp_path / "exports")
+
+    # Pre-existing, read-only destinations.
+    (empty_project / "config").mkdir(parents=True, exist_ok=True)
+    for name in ("config.json", "config_secrets.json", "wifi_config.json", "ytm_auth.json"):
+        target = empty_project / "config" / name
+        target.write_text("{}", encoding="utf-8")
+        target.chmod(0o444)
+
+    result = restore_backup(zip_path, empty_project, RestoreOptions())
+
+    assert result.success, result.errors
+    for section in ("config", "secrets", "wifi", "ytm_auth"):
+        assert section in result.restored, f"{section} not restored: {result.errors}"
+
+    restored = json.loads((empty_project / "config" / "config.json").read_text())
+    assert restored["my-plugin"]["favorites"] == ["A", "B"]
+
+    # The destination's mode is preserved rather than widened to the umask.
+    assert stat.S_IMODE((empty_project / "config" / "config_secrets.json").stat().st_mode) == 0o444

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_install_preserves_existing.py

@@ -0,0 +1,222 @@
+"""A failed (re)install must not destroy the working plugin it replaced.
+
+`_install_plugin_impl` deletes the existing plugin directory *before* it
+downloads anything, so any failure after that point used to leave the user with
+nothing. The update path was protected — `_reinstall_with_rollback` renames the
+old copy aside first — but a direct `install_plugin` was not, and the
+compatibility gate added a new way to fail late: a plugin whose declared floor
+exceeds the running core is now refused *after* the old copy is already gone.
+
+Concretely, without the wrapper: a user on core 3.1.0 with a working
+hockey-scoreboard clicks Install; the new manifest floors at 3.2.0; the gate
+refuses; the plugin they had is deleted. Floors are hand-written and can be
+over-declared, so this could remove a plugin that was working fine.
+"""
+
+import json
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from src.plugin_system.store_manager import PluginStoreManager
+
+
+@pytest.fixture
+def store(tmp_path):
+    plugins_dir = tmp_path / "plugin-repos"
+    plugins_dir.mkdir()
+    mgr = PluginStoreManager(plugins_dir=str(plugins_dir))
+    mgr.logger = MagicMock()
+    return mgr, plugins_dir
+
+
+def _existing_install(plugins_dir: Path, plugin_id: str, marker: str) -> Path:
+    path = plugins_dir / plugin_id
+    path.mkdir(parents=True)
+    (path / "manifest.json").write_text(
+        json.dumps({"id": plugin_id, "name": plugin_id, "class_name": "P",
+                    "display_modes": ["a"], "version": "1.0.0"}),
+        encoding="utf-8")
+    (path / "marker.txt").write_text(marker, encoding="utf-8")
+    return path
+
+
+class TestFailedInstallPreservesPrevious:
+    def test_failed_install_restores_the_old_copy(self, store, monkeypatch):
+        mgr, plugins_dir = store
+        path = _existing_install(plugins_dir, "hockey-scoreboard", "the-original")
+
+        monkeypatch.setattr(mgr, "_install_plugin_impl", lambda *a, **k: False)
+
+        assert mgr.install_plugin("hockey-scoreboard") is False
+        assert path.exists(), "the previous install must be restored"
+        assert (path / "marker.txt").read_text() == "the-original"
+
+    def test_raising_install_restores_and_reraises(self, store, monkeypatch):
+        mgr, plugins_dir = store
+        path = _existing_install(plugins_dir, "hockey-scoreboard", "the-original")
+
+        def boom(*a, **k):
+            raise RuntimeError("network died mid-install")
+
+        monkeypatch.setattr(mgr, "_install_plugin_impl", boom)
+
+        with pytest.raises(RuntimeError):
+            mgr.install_plugin("hockey-scoreboard")
+        assert path.exists()
+        assert (path / "marker.txt").read_text() == "the-original"
+
+    def test_successful_install_clears_the_backup(self, store, monkeypatch):
+        mgr, plugins_dir = store
+        _existing_install(plugins_dir, "hockey-scoreboard", "the-original")
+
+        def succeed(plugin_id, branch=None):
+            _existing_install(plugins_dir, plugin_id, "the-new-one")
+            return True
+
+        monkeypatch.setattr(mgr, "_install_plugin_impl", succeed)
+
+        assert mgr.install_plugin("hockey-scoreboard") is True
+        assert (plugins_dir / "hockey-scoreboard" / "marker.txt").read_text() == "the-new-one"
+        leftovers = [p.name for p in plugins_dir.iterdir() if "backup" in p.name]
+        assert not leftovers, f"backup left behind: {leftovers}"
+
+    def test_backup_name_is_invisible_to_plugin_discovery(self, store, monkeypatch):
+        """A backup that discovery can see becomes a duplicate plugin entry;
+        the marker '.standalone-backup-' is what makes it skip."""
+        mgr, plugins_dir = store
+        _existing_install(plugins_dir, "hockey-scoreboard", "the-original")
+
+        seen = {}
+
+        def capture(plugin_id, branch=None):
+            seen["dirs"] = sorted(p.name for p in plugins_dir.iterdir())
+            return False
+
+        monkeypatch.setattr(mgr, "_install_plugin_impl", capture)
+        mgr.install_plugin("hockey-scoreboard")
+
+        backups = [d for d in seen["dirs"] if d != "hockey-scoreboard"]
+        assert backups, "expected the old copy to be set aside during install"
+        for name in backups:
+            assert ".standalone-backup-" in name, (
+                f"{name} would be picked up by "
+                "plugin_manager._scan_directory_for_plugins as a real plugin")
+
+    def test_fresh_install_is_a_pass_through(self, store, monkeypatch):
+        """Nothing installed means nothing to protect; don't create stray dirs."""
+        mgr, plugins_dir = store
+        calls = []
+        monkeypatch.setattr(
+            mgr, "_install_plugin_impl",
+            lambda *a, **k: calls.append(a) or True)
+
+        assert mgr.install_plugin("brand-new") is True
+        assert calls, "the implementation must still be called"
+        assert list(plugins_dir.iterdir()) == []
+
+    def test_stale_backup_from_a_crash_does_not_block(self, store, monkeypatch):
+        mgr, plugins_dir = store
+        _existing_install(plugins_dir, "hockey-scoreboard", "the-original")
+        stale = plugins_dir / "hockey-scoreboard.standalone-backup-preinstall"
+        stale.mkdir()
+        (stale / "junk.txt").write_text("from a previous crash", encoding="utf-8")
+
+        monkeypatch.setattr(mgr, "_install_plugin_impl", lambda *a, **k: False)
+
+        assert mgr.install_plugin("hockey-scoreboard") is False
+        assert (plugins_dir / "hockey-scoreboard" / "marker.txt").read_text() == "the-original"
+
+
+class TestUpdatePathStillWorks:
+    def test_reinstall_with_rollback_is_not_double_wrapped(self, store, monkeypatch):
+        """_reinstall_with_rollback moves the plugin aside itself, so by the
+        time install_plugin runs there is nothing at the original path and the
+        wrapper must be a pass-through rather than staging a second backup."""
+        mgr, plugins_dir = store
+        path = _existing_install(plugins_dir, "hockey-scoreboard", "the-original")
+
+        observed = {}
+
+        def impl(plugin_id, branch=None):
+            observed["dirs"] = sorted(p.name for p in plugins_dir.iterdir())
+            return False
+
+        monkeypatch.setattr(mgr, "_install_plugin_impl", impl)
+
+        assert mgr._reinstall_with_rollback("hockey-scoreboard", path) is False
+        # Exactly one aside directory existed during the attempt — rollback's.
+        assert observed["dirs"] == ["hockey-scoreboard.standalone-backup-migrating"]
+        # And the user still has their plugin.
+        assert (plugins_dir / "hockey-scoreboard" / "marker.txt").read_text() == "the-original"
+
+
+class TestConcurrency:
+    """The web UI runs Flask threaded, so a double-clicked Install button puts
+    two threads on the same plugin_id. `_reinstall_with_rollback` already
+    guarded against this; the install wrapper has to as well, or one thread's
+    restore deletes the other's freshly installed copy."""
+
+    def test_rollback_calling_install_does_not_deadlock(self, store, monkeypatch):
+        """The rollback path holds the per-plugin lock across its call to
+        install_plugin. A non-reentrant lock would hang the request thread
+        forever — this test would time out rather than fail."""
+        import threading
+
+        mgr, plugins_dir = store
+        path = _existing_install(plugins_dir, "hockey-scoreboard", "the-original")
+        monkeypatch.setattr(
+            mgr, "_install_plugin_impl",
+            lambda pid, branch=None: bool(_existing_install(plugins_dir, pid, "new")))
+
+        done = threading.Event()
+        result = {}
+
+        def run():
+            result["ok"] = mgr._reinstall_with_rollback("hockey-scoreboard", path)
+            done.set()
+
+        t = threading.Thread(target=run, daemon=True)
+        t.start()
+        assert done.wait(timeout=10), (
+            "install_plugin deadlocked when called from _reinstall_with_rollback "
+            "— the per-plugin lock must be reentrant"
+        )
+        assert result["ok"] is True
+
+    def test_concurrent_installs_serialize(self, store, monkeypatch):
+        """Two threads installing the same plugin must not interleave their
+        set-aside/restore, and the survivor must be a complete install."""
+        import threading
+
+        mgr, plugins_dir = store
+        _existing_install(plugins_dir, "hockey-scoreboard", "the-original")
+
+        in_flight = []
+        overlap = []
+
+        def slow_impl(plugin_id, branch=None):
+            in_flight.append(1)
+            if len(in_flight) > 1:
+                overlap.append(1)
+            threading.Event().wait(0.05)
+            _existing_install(plugins_dir, plugin_id, "installed")
+            in_flight.pop()
+            return True
+
+        monkeypatch.setattr(mgr, "_install_plugin_impl", slow_impl)
+
+        threads = [threading.Thread(target=mgr.install_plugin,
+                                    args=("hockey-scoreboard",), daemon=True)
+                   for _ in range(2)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(timeout=10)
+            assert not t.is_alive(), "concurrent install hung"
+
+        assert not overlap, "two installs of the same plugin ran concurrently"
+        assert (plugins_dir / "hockey-scoreboard" / "marker.txt").exists()
+        leftovers = [p.name for p in plugins_dir.iterdir() if "backup" in p.name]
+        assert not leftovers, f"backup left behind: {leftovers}"

test/test_plugin_compatibility_gate.py

@@ -0,0 +1,430 @@
+"""The install/update gate, and the shared compatibility rules behind it.
+
+Before this existed, `ledmatrix_min_version` was decoration: the loader logged
+an advisory warning and the store never looked at the core version at all, so a
+routine store update happily delivered a plugin that could not run. Deleting a
+plugin's bundled fallback under those conditions would have handed un-updated
+users a scoreboard that fails to load with one line in the journal.
+
+The rules being pinned here, in priority order:
+
+1. Refuse only on **evidence**. Undeclared floor, unparseable version on either
+   side, or a core whose self-reported version is untrustworthy → allow. A
+   wrong refusal breaks a working install; a wrong allowance degrades to the
+   behavior we already had.
+2. A core below `TRUSTWORTHY_FLOOR` is *unknown*, not old. The v3.1.0 release
+   reports `1.0.0` while nearly every manifest floors at `2.0.0`; blocking on
+   that number would stop those users installing anything at all.
+3. The loader and the store must agree, because they read the same manifests.
+"""
+
+import json
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from src.plugin_system import compatibility
+
+
+# --------------------------------------------------------------------------
+# Floor resolution — every spelling published plugins actually use
+# --------------------------------------------------------------------------
+
+class TestDeclaredMinVersion:
+    def test_top_level_min_ledmatrix_version(self):
+        assert compatibility.declared_min_version(
+            {"min_ledmatrix_version": "3.2.0"}) == "3.2.0"
+
+    def test_requires_block(self):
+        assert compatibility.declared_min_version(
+            {"requires": {"min_ledmatrix_version": "3.1.0"}}) == "3.1.0"
+
+    def test_versions_array_new_spelling(self):
+        assert compatibility.declared_min_version(
+            {"versions": [{"ledmatrix_min_version": "3.2.0"}]}) == "3.2.0"
+
+    def test_versions_array_deprecated_spelling(self):
+        """Most published manifests still say `ledmatrix_min`; ignoring it
+        would silently exempt them from the gate."""
+        assert compatibility.declared_min_version(
+            {"versions": [{"ledmatrix_min": "2.0.0"}]}) == "2.0.0"
+
+    def test_absent(self):
+        assert compatibility.declared_min_version({"id": "x"}) is None
+
+    def test_requires_present_but_null(self):
+        assert compatibility.declared_min_version({"requires": None}) is None
+
+
+# --------------------------------------------------------------------------
+# The decision itself
+# --------------------------------------------------------------------------
+
+class TestCheck:
+    def test_blocks_when_plugin_needs_a_newer_core(self):
+        ok, reason = compatibility.check(
+            {"name": "Hockey Scoreboard", "min_ledmatrix_version": "3.2.0"}, "3.1.0")
+        assert ok is False
+        assert "3.2.0" in reason and "3.1.0" in reason
+        assert "Hockey Scoreboard" in reason
+
+    def test_allows_equal_version(self):
+        ok, _ = compatibility.check({"min_ledmatrix_version": "3.2.0"}, "3.2.0")
+        assert ok is True
+
+    def test_allows_newer_core(self):
+        ok, _ = compatibility.check({"min_ledmatrix_version": "3.2.0"}, "4.0.0")
+        assert ok is True
+
+    def test_allows_when_no_floor_declared(self):
+        ok, reason = compatibility.check({"id": "x"}, "3.2.0")
+        assert ok is True and reason is None
+
+    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 *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": "2.0.0"}, "1.0.0")
+        assert ok is True and reason is None
+
+    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")
+        assert ok is True
+
+    def test_v_prefix_tolerated_on_both_sides(self):
+        ok, _ = compatibility.check({"min_ledmatrix_version": "v3.3.0"}, "v3.2.0")
+        assert ok is False
+
+    @pytest.mark.parametrize("floor,core,expected_ok", [
+        ("3.2.0", "3.2.1", True),
+        ("3.2.1", "3.2.0", False),
+        ("3.10.0", "3.9.0", False),   # numeric compare, not lexical
+        ("3.9.0", "3.10.0", True),
+    ])
+    def test_ordering(self, floor, core, expected_ok):
+        ok, _ = compatibility.check({"min_ledmatrix_version": floor}, core)
+        assert ok is expected_ok
+
+
+# --------------------------------------------------------------------------
+# The gate in install_plugin
+# --------------------------------------------------------------------------
+
+def _write_plugin(plugins_dir: Path, plugin_id: str, manifest: dict) -> Path:
+    path = plugins_dir / plugin_id
+    path.mkdir(parents=True)
+    (path / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
+    (path / "manager.py").write_text("class P: pass\n", encoding="utf-8")
+    return path
+
+
+@pytest.fixture
+def store(tmp_path, monkeypatch):
+    """A PluginStoreManager whose download step is stubbed to drop a plugin
+    directory in place, so the test exercises the post-download validation
+    path without touching the network."""
+    from src.plugin_system.store_manager import PluginStoreManager
+
+    plugins_dir = tmp_path / "plugin-repos"
+    plugins_dir.mkdir()
+    mgr = PluginStoreManager(plugins_dir=str(plugins_dir))
+    mgr.logger = MagicMock()
+    return mgr, plugins_dir
+
+
+class TestInstallGate:
+    """`install_plugin` is the chokepoint: `_reinstall_with_rollback` calls it,
+    so gating there covers updates too, and a refused update restores the
+    version the user already had."""
+
+    def _install_with_manifest(self, store, manifest, core_version, monkeypatch):
+        mgr, plugins_dir = store
+        plugin_id = manifest["id"]
+
+        monkeypatch.setattr(
+            mgr, "get_plugin_info",
+            lambda *a, **k: {"repo": "https://example.invalid/r",
+                             "plugin_path": f"plugins/{plugin_id}",
+                             "branch": "main"})
+        # Stand in for the download: put the files where install_plugin expects.
+        monkeypatch.setattr(
+            mgr, "_install_from_monorepo",
+            lambda *a, **k: bool(_write_plugin(plugins_dir, plugin_id, manifest)))
+        monkeypatch.setattr(mgr, "_install_from_monorepo_api", lambda *a, **k: False)
+        monkeypatch.setattr(mgr, "_install_dependencies", lambda *a, **k: True)
+
+        import src
+        monkeypatch.setattr(src, "__version__", core_version)
+        return mgr.install_plugin(plugin_id), plugins_dir / plugin_id
+
+    def test_refuses_and_leaves_nothing_behind(self, store, monkeypatch):
+        manifest = {
+            "id": "needs-newer", "name": "Needs Newer", "class_name": "P",
+            "display_modes": ["a"], "min_ledmatrix_version": "9.9.9",
+        }
+        ok, path = self._install_with_manifest(store, manifest, "3.2.0", monkeypatch)
+
+        assert ok is False, "install must refuse a plugin that needs a newer core"
+        assert not path.exists(), (
+            "a refused install must not leave a half-installed directory — "
+            "plugin discovery would pick it up and fail to load it")
+
+    def test_allows_a_compatible_plugin(self, store, monkeypatch):
+        manifest = {
+            "id": "fine", "name": "Fine", "class_name": "P",
+            "display_modes": ["a"], "min_ledmatrix_version": "3.0.0",
+        }
+        ok, path = self._install_with_manifest(store, manifest, "3.2.0", monkeypatch)
+
+        assert ok is True
+        assert (path / "manifest.json").exists()
+
+    def test_untrustworthy_core_does_not_block_installs(self, store, monkeypatch):
+        """Regression guard for the worst possible outcome of this feature:
+        users on the v3.1.0 release (which reports 1.0.0) must not be locked
+        out of the plugin store entirely."""
+        manifest = {
+            "id": "floored", "name": "Floored", "class_name": "P",
+            "display_modes": ["a"], "versions": [{"ledmatrix_min": "2.0.0"}],
+        }
+        ok, path = self._install_with_manifest(store, manifest, "1.0.0", monkeypatch)
+
+        assert ok is True, (
+            "a core below the trustworthy floor must not block installs — "
+            "nearly every published manifest floors at 2.0.0")
+        assert (path / "manifest.json").exists()
+
+
+class TestLoaderAndStoreAgree:
+    """Both read the same manifests; a disagreement means one of them is
+    lying to the user."""
+
+    @pytest.mark.parametrize("manifest,core,expected", [
+        ({"min_ledmatrix_version": "3.2.0"}, "3.1.0", False),
+        ({"versions": [{"ledmatrix_min": "2.0.0"}]}, "3.2.0", True),
+        ({"versions": [{"ledmatrix_min_version": "9.0.0"}]}, "3.2.0", False),
+        ({}, "3.2.0", True),
+    ])
+    def test_same_verdict(self, manifest, core, expected):
+        from src.plugin_system.plugin_loader import PluginLoader
+
+        store_ok, _ = compatibility.check(manifest, core)
+        assert store_ok is expected
+
+        # The loader resolves the floor through the same helper, so a
+        # divergence in spelling handling would show up here.
+        loader_needed = compatibility.parse_semver(
+            compatibility.declared_min_version(manifest))
+        current = compatibility.parse_semver(core)
+        loader_would_warn = (
+            loader_needed is not None
+            and current is not None
+            and current >= compatibility.TRUSTWORTHY_FLOOR
+            and loader_needed > current
+        )
+        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_plugin_update_reservation.py

@@ -0,0 +1,335 @@
+"""Plugin update scheduling must be atomic (issue #401).
+
+`run_scheduled_updates()` decided whether to update a plugin with a
+check-then-act sequence: `can_execute()` and `set_state(RUNNING)` were separate
+calls with nothing between them, so two scheduler threads could both observe
+ENABLED and both go on to call the same plugin's `update()`.
+
+Two schedulers really do run at once. The render loop calls
+`_tick_plugin_updates()`, and Vegas mode fires its own `vegas-plugin-tick`
+daemon thread every few seconds; that thread is never joined when
+`VegasModeCoordinator.play()` returns, so a slow `update()` still in flight can
+overlap the next tick from the main loop.
+
+A plugin whose `update()` runs twice at once is unsafe unless it happens to be
+reentrant — shared mutable state, a non-thread-safe HTTP session or cache all
+break. These tests pin the reservation that closes it, and the state
+bookkeeping that has to survive it: a plugin reserved but never dispatched must
+not be stranded in RUNNING, because `can_execute()` would then refuse it
+forever.
+"""
+
+import os
+import sys
+import threading
+import time
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.plugin_system.plugin_manager import PluginManager  # noqa: E402
+from src.plugin_system.plugin_state import PluginState  # noqa: E402
+
+
+class OverlapDetectingPlugin:
+    """Records the high-water mark of concurrent update() calls."""
+
+    def __init__(self, update_seconds=0.25):
+        self.enabled = True
+        self.update_seconds = update_seconds
+        self.update_calls = 0
+        self.max_concurrent = 0
+        self._active = 0
+        self._guard = threading.Lock()
+
+    def update(self):
+        with self._guard:
+            self._active += 1
+            self.update_calls += 1
+            self.max_concurrent = max(self.max_concurrent, self._active)
+        try:
+            time.sleep(self.update_seconds)
+        finally:
+            with self._guard:
+                self._active -= 1
+        return True
+
+    def display(self, force_clear=False):
+        return True
+
+
+@pytest.fixture
+def pm(tmp_path):
+    manager = PluginManager(plugins_dir=str(tmp_path), config_manager=None,
+                            display_manager=None, cache_manager=None)
+    yield manager
+    manager.stop_update_worker()
+
+
+def _install(pm, plugin, plugin_id="racy-plugin", interval=0.01):
+    pm.plugins[plugin_id] = plugin
+    pm._update_interval_cache[plugin_id] = interval
+    pm.state_manager.set_state(plugin_id, PluginState.ENABLED)
+    return plugin_id
+
+
+def _widen_check_then_act_window(pm, delay=0.02):
+    """Hold every scheduler thread inside the eligibility check at once.
+
+    The real gap between can_execute() and the RUNNING transition is a couple
+    of bytecodes wide, so a plain thread race under the GIL almost never lands
+    in it — an unfixed scheduler looks correct in a test that just hammers it.
+    Delaying the check reproduces the interleaving that Vegas's tick thread and
+    the render loop actually produce when a slow update() overlaps the next
+    tick, and it is what makes these tests fail without the reservation.
+
+    Once the check and the transition are under one lock, the delay only
+    serializes the schedulers: the losers observe RUNNING and back off.
+    """
+    real_can_execute = pm.state_manager.can_execute
+
+    def slow_can_execute(plugin_id):
+        result = real_can_execute(plugin_id)
+        time.sleep(delay)
+        return result
+
+    pm.state_manager.can_execute = slow_can_execute
+
+
+def _hammer(target, threads=8, rounds=1):
+    """Run `target` on N threads released simultaneously by a barrier."""
+    errors = []
+    barrier = threading.Barrier(threads)
+
+    def runner():
+        try:
+            barrier.wait(timeout=5)
+            for _ in range(rounds):
+                target()
+        except Exception as exc:  # noqa: BLE001 - surfaced by the assertion
+            errors.append(exc)
+
+    workers = [threading.Thread(target=runner) for _ in range(threads)]
+    for worker in workers:
+        worker.start()
+    for worker in workers:
+        worker.join(timeout=30)
+    assert not errors, f"worker raised: {errors[0]!r}"
+
+
+class TestReservationAtomicity:
+    """The check-and-claim itself, independent of any dispatch path."""
+
+    def test_only_one_caller_wins_the_reservation(self, pm):
+        plugin_id = _install(pm, OverlapDetectingPlugin())
+        _widen_check_then_act_window(pm)
+        wins = []
+        lock = threading.Lock()
+
+        def claim():
+            if pm._reserve_for_update(plugin_id):
+                with lock:
+                    wins.append(threading.current_thread().name)
+
+        _hammer(claim, threads=16)
+        assert len(wins) == 1, (
+            f"{len(wins)} threads reserved the same plugin concurrently; "
+            "the eligibility check and the RUNNING transition are not atomic")
+        assert pm.state_manager.get_state(plugin_id) == PluginState.RUNNING
+
+    def test_reservation_refused_while_running(self, pm):
+        plugin_id = _install(pm, OverlapDetectingPlugin())
+        assert pm._reserve_for_update(plugin_id) is True
+        assert pm._reserve_for_update(plugin_id) is False
+
+    def test_reservation_can_be_handed_back(self, pm):
+        plugin_id = _install(pm, OverlapDetectingPlugin())
+        assert pm._reserve_for_update(plugin_id) is True
+        pm._release_reservation(plugin_id)
+        assert pm.state_manager.get_state(plugin_id) == PluginState.ENABLED
+        assert pm._reserve_for_update(plugin_id) is True, \
+            "a released reservation must be claimable again"
+
+    def test_due_check_is_inside_the_reservation(self, pm):
+        """Two threads that both decided 'due' must not both get a turn.
+
+        With the due check outside the lock, the loser of the race could claim
+        the plugin the moment the winner finished, running update() twice
+        inside one interval.
+        """
+        plugin_id = _install(pm, OverlapDetectingPlugin(), interval=60.0)
+        now = time.time()
+        assert pm._reserve_for_update(plugin_id, now, 60.0) is True
+        pm.plugin_last_update[plugin_id] = now
+        pm._release_reservation(plugin_id)
+        assert pm._reserve_for_update(plugin_id, now, 60.0) is False, \
+            "plugin updated just now must not be due again"
+
+
+class TestNoConcurrentUpdate:
+    """The end-to-end invariant the issue is actually about."""
+
+    def test_synchronous_path_never_overlaps(self, pm):
+        """The kill-switch path ran update() inline with no dedup at all."""
+        pm._synchronous_updates = True
+        plugin = OverlapDetectingPlugin(update_seconds=0.25)
+        _install(pm, plugin)
+        _widen_check_then_act_window(pm)
+
+        _hammer(pm.run_scheduled_updates, threads=8)
+
+        assert plugin.max_concurrent == 1, (
+            f"update() ran {plugin.max_concurrent}x concurrently on the "
+            "synchronous path")
+
+    def test_update_all_plugins_never_overlaps(self, pm):
+        plugin = OverlapDetectingPlugin(update_seconds=0.25)
+        _install(pm, plugin)
+        _widen_check_then_act_window(pm)
+
+        _hammer(pm.update_all_plugins, threads=8)
+
+        assert plugin.max_concurrent == 1, (
+            f"update() ran {plugin.max_concurrent}x concurrently via "
+            "update_all_plugins()")
+
+    def test_async_path_never_overlaps(self, pm):
+        plugin = OverlapDetectingPlugin(update_seconds=0.2)
+        plugin_id = _install(pm, plugin)
+
+        _hammer(pm.run_scheduled_updates, threads=8, rounds=3)
+
+        # Wait for an update to have both started and finished. Polling only
+        # `_active` races the worker: before it picks the item up nothing is
+        # active yet, so the loop would fall straight through and assert on a
+        # plugin that never ran.
+        deadline = time.monotonic() + 15
+        while time.monotonic() < deadline:
+            if plugin.update_calls >= 1 and plugin._active == 0:
+                break
+            time.sleep(0.05)
+
+        assert plugin.update_calls >= 1, "no update ran on the async path"
+        assert plugin.max_concurrent == 1, (
+            f"update() ran {plugin.max_concurrent}x concurrently on the "
+            "async path")
+        assert plugin_id in pm.plugins
+
+
+class TestNoStrandedState:
+    """A reservation that is never dispatched must not wedge the plugin."""
+
+    def test_plugin_returns_to_enabled_after_async_updates(self, pm):
+        plugin = OverlapDetectingPlugin(update_seconds=0.1)
+        plugin_id = _install(pm, plugin)
+
+        _hammer(pm.run_scheduled_updates, threads=6, rounds=2)
+
+        deadline = time.monotonic() + 15
+        while time.monotonic() < deadline:
+            if (plugin.update_calls >= 1
+                    and pm.state_manager.get_state(plugin_id) == PluginState.ENABLED
+                    and not pm._pending_updates):
+                break
+            time.sleep(0.05)
+
+        # ENABLED is also the starting state, so without this the assertion
+        # below would pass on a plugin that never got scheduled at all.
+        assert plugin.update_calls >= 1, "no update ran; the state assertion would be vacuous"
+        assert pm.state_manager.get_state(plugin_id) == PluginState.ENABLED, \
+            "plugin stranded in RUNNING; can_execute() would refuse it forever"
+        assert not pm._pending_updates, "pending set not drained"
+
+    def test_pending_cleared_before_state_reset(self, pm):
+        """Invariant: never-RUNNING implies never-pending.
+
+        _finish() used to clear the pending entry *after* flipping the state
+        back to ENABLED. In that window a scheduler could reserve the plugin
+        and then have its enqueue dropped by the pending-dedup.
+        """
+        plugin = OverlapDetectingPlugin(update_seconds=0.05)
+        plugin_id = _install(pm, plugin)
+
+        violations = []
+        stop = threading.Event()
+
+        def watcher():
+            while not stop.is_set():
+                state = pm.state_manager.get_state(plugin_id)
+                if state != PluginState.RUNNING and plugin_id in pm._pending_updates:
+                    violations.append(state)
+                time.sleep(0.001)
+
+        thread = threading.Thread(target=watcher, daemon=True)
+        thread.start()
+        try:
+            for _ in range(15):
+                pm.run_scheduled_updates()
+                time.sleep(0.05)
+        finally:
+            stop.set()
+            thread.join(timeout=5)
+
+        assert not violations, (
+            f"{len(violations)} sample(s) saw a non-RUNNING plugin still in "
+            "_pending_updates")
+
+
+class TestDispatchFailure:
+    """A reservation must survive the dispatch itself failing.
+
+    _enqueue_update() claims the plugin, adds it to the pending set, and only
+    then starts the worker and queues the item. threading.Thread.start() raises
+    RuntimeError when the OS refuses a new thread — not hypothetical on a Pi
+    under memory or thread pressure. Nothing is queued to release the plugin at
+    that point, so without an explicit rollback it stays RUNNING with a stale
+    pending entry and can_execute() refuses it for the rest of the process.
+    """
+
+    def test_reservation_released_when_the_worker_cannot_start(self, pm):
+        plugin_id = _install(pm, OverlapDetectingPlugin())
+
+        def refuse_to_start():
+            raise RuntimeError("can't start new thread")
+
+        pm._ensure_update_worker = refuse_to_start
+
+        assert pm._reserve_for_update(plugin_id) is True
+        pm._enqueue_update(plugin_id, time.time())
+
+        assert pm.state_manager.get_state(plugin_id) == PluginState.ENABLED, \
+            "plugin left in RUNNING after a failed dispatch; can_execute() " \
+            "would refuse it forever"
+        assert plugin_id not in pm._pending_updates, \
+            "stale pending entry would make the next enqueue hit the dedup"
+        assert pm._reserve_for_update(plugin_id) is True, \
+            "plugin must be claimable again on the next tick"
+
+    def test_dispatch_failure_does_not_abort_the_rest_of_the_tick(self, pm):
+        """One plugin failing to queue must not skip the others in that tick."""
+        first = OverlapDetectingPlugin()
+        second = OverlapDetectingPlugin()
+        _install(pm, first, plugin_id="plugin-a")
+        _install(pm, second, plugin_id="plugin-b")
+
+        calls = []
+        real_ensure = pm._ensure_update_worker
+
+        def fail_first_only():
+            calls.append(1)
+            if len(calls) == 1:
+                raise RuntimeError("can't start new thread")
+            return real_ensure()
+
+        pm._ensure_update_worker = fail_first_only
+
+        pm.run_scheduled_updates()  # must not propagate the RuntimeError
+
+        assert len(calls) == 2, (
+            "run_scheduled_updates() stopped after the failing plugin; the "
+            "exception escaped the enqueue")
+        for plugin_id in ("plugin-a", "plugin-b"):
+            assert pm.state_manager.get_state(plugin_id) in (
+                PluginState.ENABLED, PluginState.RUNNING), \
+                f"{plugin_id} left in an unexpected state"

test/test_registry_id_resolution.py

@@ -0,0 +1,107 @@
+"""A plugin must be findable in the registry by the id it calls itself.
+
+Four shipped plugins have a registry ``id`` that differs from the ``id`` in
+their own ``manifest.json``:
+
+    directory / manifest.json id      registry id
+    ledmatrix-weather                 weather
+    ledmatrix-stocks                  stocks
+    ledmatrix-music                   music
+    ledmatrix-leaderboard             leaderboard
+
+The installer already knows about this: it deliberately names the install
+directory after the *manifest* id (store_manager, "Use manifest ID for
+directory name"), and warns when the two disagree. So on disk, in
+``config.json`` and in a backup manifest, these plugins are called
+``ledmatrix-weather``. Only the registry calls them ``weather``.
+
+Nothing resolved that in reverse. Asking the store to install
+``ledmatrix-weather`` -- which is exactly what restoring a backup does --
+failed with "Plugin not found in registry", and four enabled plugins went
+missing from a restored device with no error surfaced to the user.
+
+Renaming the registry ids would orphan existing ``plugin_state.json`` entries
+keyed on the old ones, so the lookup resolves ``plugin_path`` instead: the
+registry already records ``plugins/ledmatrix-weather``, which is unambiguous
+and needs no published identity to change.
+"""
+
+import os
+import sys
+from typing import Any, Dict, List, Optional
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.plugin_system.store_manager import PluginStoreManager  # noqa: E402
+
+
+# Shaped like the real registry: id and plugin_path basename disagree for the
+# first entry, agree for the second.
+REGISTRY: Dict[str, List[Dict[str, Any]]] = {
+    "plugins": [
+        {
+            "id": "weather",
+            "name": "Weather",
+            "plugin_path": "plugins/ledmatrix-weather",
+            "repo": "https://github.com/ChuckBuilds/ledmatrix-plugins",
+        },
+        {
+            "id": "ledmatrix-flights",
+            "name": "Flights",
+            "plugin_path": "plugins/ledmatrix-flights",
+            "repo": "https://github.com/ChuckBuilds/ledmatrix-plugins",
+        },
+        {
+            "id": "third-party",
+            "name": "Third Party",
+            "plugin_path": "",
+            "repo": "https://github.com/someone/thing",
+        },
+    ]
+}
+
+
+@pytest.fixture
+def store(monkeypatch: pytest.MonkeyPatch) -> PluginStoreManager:
+    manager = PluginStoreManager.__new__(PluginStoreManager)
+    monkeypatch.setattr(manager, "fetch_registry", lambda *a, **k: REGISTRY, raising=False)
+    return manager
+
+
+def _ids(entry: Optional[Dict[str, Any]]) -> Optional[str]:
+    return entry.get("id") if entry else None
+
+
+class TestRegistryLookupByManifestId:
+    def test_exact_registry_id_still_resolves(self, store: PluginStoreManager) -> None:
+        assert _ids(store.get_registry_info("weather")) == "weather"
+
+    def test_manifest_id_resolves_via_plugin_path(self, store: PluginStoreManager) -> None:
+        """The case that broke restore: asked by the name on disk."""
+        assert _ids(store.get_registry_info("ledmatrix-weather")) == "weather", (
+            "a plugin installed as 'ledmatrix-weather' could not be found in a "
+            "registry that lists it under plugin_path plugins/ledmatrix-weather")
+
+    def test_matching_id_and_path_unaffected(self, store: PluginStoreManager) -> None:
+        assert _ids(store.get_registry_info("ledmatrix-flights")) == "ledmatrix-flights"
+
+    def test_unknown_plugin_still_returns_none(self, store: PluginStoreManager) -> None:
+        assert store.get_registry_info("no-such-plugin") is None
+
+    def test_empty_plugin_path_is_not_a_wildcard(self, store: PluginStoreManager) -> None:
+        """Third-party entries carry plugin_path "" — that must not match ""."""
+        assert store.get_registry_info("") is None
+
+    def test_exact_id_wins_over_a_path_match(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        """If some other entry's path collides with a real id, id wins."""
+        registry = {
+            "plugins": [
+                {"id": "decoy", "plugin_path": "plugins/weather"},
+                {"id": "weather", "plugin_path": "plugins/ledmatrix-weather"},
+            ]
+        }
+        manager = PluginStoreManager.__new__(PluginStoreManager)
+        monkeypatch.setattr(manager, "fetch_registry", lambda *a, **k: registry, raising=False)
+        assert _ids(manager.get_registry_info("weather")) == "weather"

web_interface/blueprints/api_v3.py

@@ -7843,7 +7843,31 @@ def clear_old_errors():
 # Backup / Restore
 # ---------------------------------------------------------------------------
 
-_BACKUP_EXPORT_DIR = PROJECT_ROOT / "config" / "backups" / "exports"
+def _resolve_backup_export_dir() -> Path:
+    """Where exported backups live: beside the install, not inside it.
+
+    They used to be written to ``<project>/config/backups/exports``. That is
+    inside the directory a reinstall deletes, so the documented recovery path
+    -- export a backup, then reinstall -- destroyed the backup it had just
+    told the user to make. Anyone who downloaded the ZIP was fine; anyone
+    relying on the on-device copy was not.
+
+    Falls back to the old location when the parent directory is not writable,
+    so an unusual layout degrades to previous behaviour instead of failing to
+    export at all.
+    """
+    preferred = PROJECT_ROOT.parent / "ledmatrix-backups"
+    try:
+        preferred.mkdir(parents=True, exist_ok=True)
+        probe = preferred / ".writetest"
+        probe.write_text("", encoding="utf-8")
+        probe.unlink()
+        return preferred
+    except OSError:
+        return PROJECT_ROOT / "config" / "backups" / "exports"
+
+
+_BACKUP_EXPORT_DIR = _resolve_backup_export_dir()
 
 
 def _safe_backup_path(filename: str) -> Path:
@@ -7997,7 +8021,22 @@ def backup_restore():
 
         data = result.to_dict()
         if not result.success:
-            return jsonify({'status': 'error', 'message': 'Restore had errors', 'data': data}), 500
+            # Name what failed, and what nonetheless landed. A restore is
+            # partial far more often than it is total -- a fresh install can
+            # leave config_secrets.json unwritable by the web service, so
+            # config restores and secrets do not. "Restore had errors" alone
+            # left the user unable to tell a wholly failed restore from one
+            # that quietly dropped their API keys.
+            failed_plugins = [p.get('plugin_id') for p in (result.plugins_failed or [])]
+            parts = []
+            if result.restored:
+                parts.append(f"restored: {', '.join(result.restored)}")
+            if result.errors:
+                parts.append(f"failed: {'; '.join(result.errors)}")
+            if failed_plugins:
+                parts.append(f"plugins not reinstalled: {', '.join(failed_plugins)}")
+            message = 'Restore incomplete — ' + ('. '.join(parts) if parts else 'see logs')
+            return jsonify({'status': 'error', 'message': message, 'data': data}), 500
         return jsonify({'status': 'success', 'data': data})
     except Exception as e:
         logger.error("backup_restore failed: %s", e, exc_info=True)