fix: harden the health-state repair and confirm journald took effect

Second review round; all three findings were valid and two were bugs in the
repair added last commit.

The repair could raise out of itself. An unhashable circuit_state (a list or
dict on disk) hit `value in {...}` and raised TypeError -- from the code
whose whole job is to stop a malformed record crashing the caller. It now
requires a str before the membership test.

bool is a subclass of int, so True passed the timestamp check and then
compared as 1.0: enough to expire a cooldown the instant the breaker opened,
while False would stop the elapsed check firing at all. Timestamps now
exclude bool explicitly.

The regression test for the original crash was seeded with a record that
*contained* circuit_state, so it passed against the old raw-return behaviour
too -- the counters are read with .get(), so circuit_state is the only field
whose absence used to raise. Reseeded to omit it, and it now fails against
raw-return as intended.

journald: drop-ins apply in lexical order, so a local file sorting after
ledmatrix-persistent.conf still wins and writing ours proves nothing. The
effective Storage is re-read afterwards and a warning naming the diagnostic
command is printed if persistence is still not active, rather than reporting
a success that was not verified.

Full suite 2934 passed, same single pre-existing tmpfs failure.

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

.github/workflows/test.yml

@@ -72,4 +72,4 @@ jobs:
             --ignore=test/plugins \
             --cov=src --cov=web_interface \
             --cov-report=term \
-            --cov-fail-under=52
+            --cov-fail-under=48

config/config.template.json

@@ -130,6 +130,9 @@
         "plugin_rotation_order": [],
         "use_short_date_format": true,
         "vegas_scroll": {
+      "live_in_ticker": false,
+      "live_weight": 3,
+      "favorite_live_weight": 5,
             "enabled": false,
             "scroll_speed": 50,
             "separator_width": 32,

docs/ADVANCED_FEATURES.md

@@ -64,10 +64,98 @@ JSON is optional.
 | `target_fps` | `125` | Target frame rate |
 | `buffer_ahead` | `2` | Number of plugins buffered ahead |
 
-This table is a subset — `display.vegas_scroll` supports 26 keys in
+This table is a subset — `display.vegas_scroll` supports 30 keys in
 total. See the full list in
 [CONFIG_REFERENCE.md](CONFIG_REFERENCE.md#displayvegas_scroll--continuous-scroll-mode).
 
+### Live Content in the Ticker
+
+By default, live content **preempts** Vegas mode: while any plugin reports
+live priority, the display controller refuses to run the ticker and shows
+that plugin's full-screen display instead. You get a big readable scoreboard,
+but the marquee stops entirely for the duration of the game.
+
+Set `live_in_ticker` to keep the ticker running and let live content take
+**extra turns inside it** instead:
+
+```json
+"vegas_scroll": {
+  "live_in_ticker": true,
+  "live_weight": 3,
+  "favorite_live_weight": 5
+}
+```
+
+#### Why weights exist
+
+The rotation is otherwise a strict round robin — every plugin appears exactly
+once per cycle. With a dozen plugins enabled, a live score comes round once a
+lap and can be minutes old by the time you see it. A weight of *N* gives a
+plugin *N* slots per cycle.
+
+The slots are placed by **Smooth Weighted Round-Robin**, the same scheduler
+the sports plugins use internally to rotate their own games. The important
+property is that repeats are *spread through the cycle* rather than clumped:
+three appearances in a row followed by a long silence would be worse than not
+boosting at all.
+
+Twelve plugins, with a favorite's baseball game and an ordinary live hockey
+game (`live_weight: 3`, `favorite_live_weight: 5`):
+
+```
+baseball > hockey > weather  > clock  > baseball
+stocks   > news   > flights  > baseball > hockey
+calendar > f1     > music    > baseball > tides
+birds    > hockey > baseball
+```
+
+18 slots for 12 plugins. Baseball appears 5 times, hockey 3, everything else
+once, and no plugin ever appears twice in a row — **including across the seam**
+where the cycle loops back on itself. Smooth Weighted Round-Robin schedules the
+heaviest item first and usually last as well, so the strip would otherwise show
+it twice running at exactly the one join a within-cycle check cannot see. The
+trailing repeat is moved into the widest remaining gap. Where a double is
+unavoidable — a plugin holding most of the slots has to neighbour itself — the
+schedule is left as it is.
+
+#### Where the weight comes from
+
+For each plugin in the rotation, in order:
+
+1. **The plugin's own answer.** If it implements
+   `get_vegas_priority_weight()` and returns a number, that wins. This is the
+   only route for favorite-team awareness — the core can see *that* a game is
+   live, but not *whose*, so a scoreboard has to say so itself.
+2. **The core's default.** When the plugin returns `None` (the base-class
+   default), a plugin where both `has_live_priority()` and `has_live_content()`
+   are true gets `live_weight`.
+3. **Everything else** gets 1.
+
+Because of step 2, **existing plugins need no changes** — any scoreboard with
+`live_priority` enabled already gets extra turns. Step 1 is opt-in, for
+plugins that want to distinguish a favorite's game from any other live game.
+
+Weights are clamped to 1–10. A weight of 1 is no boost; a weight below 1 would
+drop the plugin from the rotation entirely, which is never what is meant.
+
+#### Things worth knowing
+
+- **Weights are per plugin, not per game.** A scoreboard showing four live
+  games still occupies one slot at a time, rotating its own games within that
+  slot using its own `favorite_live_boost`. This controls how often the
+  *plugin* comes round.
+- **The ticker is zero-sum.** Giving baseball 5 slots does not make the cycle
+  faster; it makes the cycle *longer* and everything else proportionally
+  rarer. If you want live scores sooner in wall-clock terms, pair this with a
+  smaller `plugins_per_cycle`.
+- **Frequency is not freshness.** Each appearance redraws from the plugin's
+  current data (`refresh_updated_plugins()` drops cached content when a
+  plugin's data changes), but how current that data is depends on the
+  plugin's own `live_update_interval`. Showing a stale score five times a lap
+  is no better than showing it once.
+- **Everything still appears.** A boost never starves another plugin out of
+  the cycle; low-weight plugins keep their single slot.
+
 ### Per-Plugin Configuration
 
 Override Vegas behavior for specific plugins:

docs/CONFIG_REFERENCE.md

@@ -104,7 +104,8 @@ logical image to multiple chained physical panels.
 ## `display.vegas_scroll` — continuous scroll mode
 
 Read by `src/vegas_mode/config.py` (`VegasScrollConfig.from_config`). See
-[ADVANCED_FEATURES.md](ADVANCED_FEATURES.md) for behavior details.
+[ADVANCED_FEATURES.md](ADVANCED_FEATURES.md) for behavior details, including
+[live content in the ticker](ADVANCED_FEATURES.md#live-content-in-the-ticker).
 
 | Key | Type / default |
 |---|---|
@@ -135,6 +136,9 @@ Read by `src/vegas_mode/config.py` (`VegasScrollConfig.from_config`). See
 | `max_cycle_duration` | int, `240` |
 | `frame_based_scrolling` | bool, `true` — frame-count-based scroll stepping |
 | `scroll_delay` | float, `0.02` — seconds between scroll updates (~50 FPS) |
+| `live_in_ticker` | bool, `false` — keep scrolling during live games instead of handing the display to a full-screen scoreboard |
+| `live_weight` | int, `3` (1–10) — slots per cycle for a plugin with live content |
+| `favorite_live_weight` | int, `5` (1–10) — slots per cycle when a plugin reports a favorite team is live |
 
 ## `sync` — multi-display synchronization
 

docs/LOW_MEMORY_BOARDS.md

@@ -0,0 +1,115 @@
+# Running on Low-Memory Boards
+
+Applies to the Pi Zero 2 W (512 MB), Pi 3 / 3B+ (1 GB), and the 1 GB Pi 4.
+If your board has 2 GB or more you can skip this document.
+
+## The failure this prevents
+
+The display process is the largest thing on the board. On a 1 GB Pi 3B+ with
+around 20 plugins enabled it settles near **600 MB of 905 MB usable**, leaving
+under 200 MB of headroom for everything else.
+
+When that headroom runs out, the board does not crash cleanly. `fork()` starts
+failing, and because a new process is needed to do almost anything, the
+symptoms look nothing like "out of memory":
+
+| What you see | Why |
+|---|---|
+| SSH accepts the connection then closes it instantly, before any banner | `sshd` forks a session per connection; the fork fails |
+| The web UI still responds quickly | Already running, serves from existing threads, forks nothing |
+| Ping is perfect, 0% loss | Handled entirely in the kernel |
+| The panel is dark | The display process was killed and cannot be respawned |
+| The clock is wrong after the next boot | `fake-hwclock`'s periodic save is a scheduled job, and it cannot fork either |
+
+The board looks healthy from the outside and cannot be logged into. Only a
+power cycle clears it. If you are here because SSH stopped working, also see
+[SSH_UNAVAILABLE_AFTER_INSTALL.md](SSH_UNAVAILABLE_AFTER_INSTALL.md), which
+covers the more common cause (AP mode).
+
+## Check your headroom
+
+```bash
+free -m
+ps -eo rss,comm --sort=-rss | head -5
+```
+
+If `MemAvailable` is under ~150 MB while the display is running, you are close
+to the edge. To watch it over time:
+
+```bash
+watch -n 30 'free -m | head -2'
+```
+
+Available memory that falls steadily rather than holding flat means you will
+reach the wall; it is a question of when.
+
+## What to do
+
+**1. Enable the memory cgroup controller.** Without it, the `MemoryMax=85%` in
+`systemd/ledmatrix.service` is accepted by systemd and silently ignored, so the
+service has no ceiling and a runaway takes the whole board down instead of just
+restarting. Raspberry Pi firmware disables this controller by default.
+
+`first_time_install.sh` does this for you. To check it took effect:
+
+```bash
+grep memory /sys/fs/cgroup/cgroup.controllers
+```
+
+If that prints nothing, add `cgroup_enable=memory cgroup_memory=1` to the
+kernel command line and reboot. Edit whichever file your image uses —
+`/boot/firmware/cmdline.txt` on current Raspberry Pi OS, `/boot/cmdline.txt` on
+older layouts (the installer checks the first and falls back to the second).
+Everything must stay on a single line.
+
+This changes the failure mode from "the board becomes unreachable" to "the
+display service restarts". It is a safety net, not a fix.
+
+**2. Run fewer plugins.** This is the actual remedy. Every enabled plugin costs
+memory permanently — its module, its parsed config, and its cached API
+responses. On a 512 MB or 1 GB board, keep the enabled set small and prefer
+plugins that poll infrequently.
+
+**3. Lower the cache ceiling.** The in-memory cache is sized from total RAM
+(150 entries at 1 GB and below, up to 1500 at 8 GB). To go lower still:
+
+```ini
+# /etc/systemd/system/ledmatrix.service.d/override.conf
+[Service]
+Environment=LEDMATRIX_CACHE_MAX_ENTRIES=75
+```
+
+Writing the file does not change the running service. Reload systemd and
+restart it:
+
+```bash
+sudo systemctl daemon-reload
+sudo systemctl restart ledmatrix
+```
+
+Fewer entries means more API calls, so lower this only while you are actually
+short of memory.
+
+**4. Consider `MemoryHigh`.** `MemoryMax` kills and restarts. `MemoryHigh`
+throttles and reclaims instead, which is gentler — but on a board where the
+process genuinely wants more than the limit, sustained reclaim can stall the
+render loop and show as visible stutter on the panel. Add it only if you prefer
+degraded output to a restart:
+
+```ini
+[Service]
+MemoryHigh=70%
+```
+
+## Keep your logs
+
+These images default to volatile journald storage, so every reboot destroys the
+logs — including the ones explaining why the board rebooted. `first_time_install.sh`
+enables persistent storage capped at 64 MB. To confirm:
+
+```bash
+journalctl --list-boots
+```
+
+More than one boot listed means logs are surviving reboots. If only one is
+listed, journald is still writing to `/run` (tmpfs).

docs/PLUGIN_API_REFERENCE.md

@@ -170,6 +170,47 @@ Default returns `False`.
 List of display modes to show during a live takeover. Default returns the
 plugin's `display_modes` from its manifest.
 
+#### `get_vegas_priority_weight() -> Optional[int]`
+
+How many slots per Vegas cycle this plugin should get. Default returns
+`None`, which defers to the core.
+
+The Vegas ticker is otherwise a strict round robin — every plugin appears
+exactly once per cycle — so with a dozen plugins enabled a live score can be
+minutes stale by the time it comes round. A weight of *N* gives the plugin
+*N* slots per cycle, spread evenly through it rather than clumped.
+
+**You usually do not need this.** When the hook returns `None`, the core
+already gives a plugin `vegas_scroll.live_weight` whenever
+`has_live_priority()` and `has_live_content()` are both true. Live sports get
+extra turns with no code at all.
+
+Implement it only when the plugin knows something the core cannot. The
+motivating case is favorite teams — the core can see *that* a game is live,
+but not *whose*:
+
+```python
+def get_vegas_priority_weight(self):
+    if not (self.has_live_priority() and self.has_live_content()):
+        return None                       # let the core decide
+    vegas = self.global_config.get('display', {}).get('vegas_scroll', {})
+    if self._favorite_is_live():
+        return vegas.get('favorite_live_weight', 5)
+    return vegas.get('live_weight', 3)
+```
+
+The weight is per *plugin*, not per game: a scoreboard showing four live games
+still occupies one slot at a time and rotates its own games within it. Values
+are clamped to 1–10 by the caller. An exception here is caught and logged, and
+the core then falls back to its own live-content check — so a plugin whose
+weight calculation is broken still gets `live_weight` for a game that really
+is live, rather than being demoted to 1.
+
+Only consulted when the user has set `vegas_scroll.live_in_ticker`. With the
+default (`false`) live content preempts Vegas entirely and there is no ticker
+to be weighted within. See
+[ADVANCED_FEATURES.md](ADVANCED_FEATURES.md#live-content-in-the-ticker).
+
 ### Vegas scroll hooks
 
 Vegas mode shows multiple plugins as a single continuous scroll instead of

docs/README.md

@@ -14,6 +14,7 @@ the one-shot installer. The pages here go deeper.
 5. [TROUBLESHOOTING.md](TROUBLESHOOTING.md) — common issues and fixes
 6. [SSH_UNAVAILABLE_AFTER_INSTALL.md](SSH_UNAVAILABLE_AFTER_INSTALL.md) — recovering SSH after install
 7. [CONFIG_DEBUGGING.md](CONFIG_DEBUGGING.md) — diagnosing config problems
+8. [LOW_MEMORY_BOARDS.md](LOW_MEMORY_BOARDS.md) — Pi Zero 2 W / 3B+ / 1GB Pi 4 memory limits
 
 ## I want to write a plugin
 

docs/SSH_UNAVAILABLE_AFTER_INSTALL.md

@@ -20,7 +20,22 @@ The installation script:
 - Installs and configures `dnsmasq` (DHCP server for AP mode)
 - These services can interfere with normal WiFi client mode
 
-### 3. Reboot After Installation
+### 3. The Board Ran Out of Memory
+
+On a 512MB or 1GB board, memory exhaustion stops `sshd` being able to fork a
+session process. The connection is accepted and then closed immediately, before
+any banner:
+
+```text
+kex_exchange_identification: Connection closed by remote host
+```
+
+The giveaway is that the board is otherwise healthy — ping is clean and the web
+UI still responds — but nothing that needs to start a new process works, and
+the panel is usually dark. Only a power cycle clears it. See
+[LOW_MEMORY_BOARDS.md](LOW_MEMORY_BOARDS.md).
+
+### 4. Reboot After Installation
 
 If the script reboots the Pi (which it recommends), network services may restart in a different state, potentially triggering AP mode.
 
@@ -190,11 +205,23 @@ The web interface allows you to:
 
 ## Summary
 
-**SSH becomes unavailable because**:
+**SSH becomes unavailable because** — two unrelated causes, and they need
+different responses:
+
+*AP mode (most common):*
 - WiFi monitor service enables AP mode when WiFi disconnects
 - AP mode switches WiFi from client to access point mode
 - Pi loses connection to your original network
 
+*Memory exhaustion (low-memory boards):*
+- The board runs out of memory, so `sshd` cannot fork a session process
+- The connection is accepted and closed before any banner
+- Ping still answers and the web UI still responds, so it looks healthy
+- The panel is usually dark and the service cannot restart
+- **Only a power cycle clears this** — there is no remote recovery, because
+  every remote route needs a new process
+- Prevention and tuning: [LOW_MEMORY_BOARDS.md](LOW_MEMORY_BOARDS.md)
+
 **To regain SSH**:
 1. Connect to **LEDMatrix-Setup** AP network (password: `ledmatrix123`)
 2. SSH to `192.168.4.1`

first_time_install.sh

@@ -1688,6 +1688,94 @@ else
     echo "✗ $CMDLINE_FILE not found; skipping isolcpus optimization"
 fi
 
+# Enable the memory cgroup controller (idempotent).
+# The Pi firmware boots with cgroup_disable=memory, so systemd's MemoryMax= is
+# accepted and silently ignored — the display service then has no ceiling, and
+# a runaway takes the whole board down (sshd can no longer fork, the panel goes
+# dark) rather than just restarting the one service.
+if [ "$SKIP_PERF" != "1" ] && [ -f "$CMDLINE_FILE" ]; then
+    # Both parameters are required for the memory controller, and they can get
+    # separated -- an image, another tool or a half-applied earlier run can
+    # leave one without the other. Checking only cgroup_enable=memory would
+    # report success while MemoryMax= silently does nothing, so each is checked
+    # and appended independently.
+    cgroup_missing=""
+    for cgroup_param in cgroup_enable=memory cgroup_memory=1; do
+        if ! grep -qw "$cgroup_param" "$CMDLINE_FILE"; then
+            cgroup_missing="$cgroup_missing $cgroup_param"
+        fi
+    done
+    if [ -z "$cgroup_missing" ]; then
+        echo "cgroup memory parameters already present in $CMDLINE_FILE"
+    else
+        echo "Adding${cgroup_missing} to $CMDLINE_FILE..."
+        cp "$CMDLINE_FILE" "$CMDLINE_FILE.bak" 2>/dev/null || true
+        # The kernel command line must stay on one line.
+        sed -i "1 s|\$|${cgroup_missing}|" "$CMDLINE_FILE"
+        echo "  Takes effect after reboot. Verify with:"
+        echo "    grep memory /sys/fs/cgroup/cgroup.controllers"
+    fi
+fi
+
+# Persist the journal (idempotent).
+# These images default to volatile storage: journald keeps everything in /run
+# (tmpfs), so every reboot destroys the logs — including the ones that would
+# explain why the board rebooted. Capped so an SD card is not worn out by logs.
+# A non-empty /var/log/journal does not prove journald is configured the way
+# this needs: the directory survives a switch back to volatile storage, and it
+# says nothing about whether a size cap is set. Read the effective
+# configuration instead, and only write the keys the user has not set
+# themselves so an explicit local limit is preserved.
+journald_effective() {
+    # systemd-analyze merges journald.conf with every drop-in; grep is the
+    # fallback for images that ship without it.
+    if command -v systemd-analyze >/dev/null 2>&1 &&
+       systemd-analyze cat-config systemd/journald.conf >/dev/null 2>&1; then
+        systemd-analyze cat-config systemd/journald.conf 2>/dev/null
+    else
+        cat /etc/systemd/journald.conf /etc/systemd/journald.conf.d/*.conf 2>/dev/null
+    fi
+}
+journald_conf="$(journald_effective)"
+journald_storage="$(printf '%s\n' "$journald_conf" | grep -E '^[[:space:]]*Storage=' | tail -n1 | cut -d= -f2 | tr -d '[:space:]')"
+journald_cap="$(printf '%s\n' "$journald_conf" | grep -E '^[[:space:]]*SystemMaxUse=' | tail -n1 | cut -d= -f2 | tr -d '[:space:]')"
+
+if [ "$journald_storage" = "persistent" ] && [ -n "$journald_cap" ]; then
+    echo "Persistent journald storage already configured (SystemMaxUse=$journald_cap)"
+else
+    echo "Enabling persistent journald storage..."
+    mkdir -p /etc/systemd/journald.conf.d
+    {
+        echo "# Installed by LEDMatrix first_time_install.sh"
+        echo "[Journal]"
+        echo "Storage=persistent"
+        if [ -n "$journald_cap" ]; then
+            echo "# SystemMaxUse left to your existing setting ($journald_cap)"
+        else
+            # Capped so logs cannot wear out or fill an SD card.
+            echo "SystemMaxUse=64M"
+        fi
+    } > /etc/systemd/journald.conf.d/ledmatrix-persistent.conf
+    mkdir -p /var/log/journal
+    systemd-tmpfiles --create --prefix /var/log/journal >/dev/null 2>&1 || true
+    systemctl restart systemd-journald >/dev/null 2>&1 || true
+
+    # Drop-ins are applied in lexical order, so a locally added file that sorts
+    # after ledmatrix-persistent.conf (zz-local.conf and friends) still wins.
+    # Writing the file is not evidence it took effect -- re-read and say so
+    # plainly rather than reporting success we cannot confirm.
+    journald_now="$(journald_effective | grep -E '^[[:space:]]*Storage=' | tail -n1 | cut -d= -f2 | tr -d '[:space:]')"
+    if [ "$journald_now" = "persistent" ]; then
+        echo "  Persistent journald storage active"
+    else
+        echo "  WARNING: journald storage is still '${journald_now:-unset}' after"
+        echo "  writing /etc/systemd/journald.conf.d/ledmatrix-persistent.conf."
+        echo "  Another drop-in that sorts later is overriding it. Check:"
+        echo "    systemd-analyze cat-config systemd/journald.conf | grep -n Storage="
+        echo "  Logs will not survive a reboot until that is resolved."
+    fi
+fi
+
 # Ensure dtparam=audio=off in config.txt (idempotent)
 if [ "$SKIP_PERF" = "1" ]; then
     : # skipped

scripts/download_pixlet.sh

@@ -24,9 +24,29 @@ echo "========================================"
 # Auto-detect latest version if needed
 if [ "$PIXLET_VERSION" = "latest" ]; then
     echo "Detecting latest version..."
-    PIXLET_VERSION=$(curl -s "https://api.github.com/repos/${REPO}/releases/latest" | grep '"tag_name"' | sed -E 's/.*"([^"]+)".*/\1/')
-    if [ -z "$PIXLET_VERSION" ]; then
-        echo "Failed to detect latest version, using fallback"
+    # When this response arrives on a single line -- as it did on the device
+    # where Starlark apps were failing -- `grep '"tag_name"'` matches the whole
+    # document and a greedy `sed 's/.*"([^"]+)".*/\1/'` captures the LAST
+    # quoted token in it rather than the tag. That resolved to
+    # "mentions_count", which built a download URL for a release that does not
+    # exist. (The API is pretty-printed by default, which is why the old
+    # command looks correct when you try it by hand -- but the formatting is
+    # not something to depend on.) Match the field itself and take the value
+    # after it, which is right for either shape.
+    PIXLET_VERSION=$(curl -fsSL "https://api.github.com/repos/${REPO}/releases/latest" \
+        | grep -o '"tag_name"[[:space:]]*:[[:space:]]*"[^"]*"' \
+        | head -n1 \
+        | sed -E 's/.*:[[:space:]]*"([^"]*)".*/\1/')
+
+    # A wrong-but-non-empty value is what made the old bug silent, so check the
+    # shape rather than just that something came back. Anchored at both ends: a
+    # partial match would accept "v0.53garbage" or "0.53" and build a URL for a
+    # release that cannot exist, which is the failure this check is here to
+    # stop. Every tronbyt/pixlet release to date is vX.Y.Z; the optional suffix
+    # leaves room for a future -rc.1 or +build tag.
+    if ! printf '%s' "$PIXLET_VERSION" \
+        | grep -qE '^v[0-9]+\.[0-9]+\.[0-9]+([-+][0-9A-Za-z.-]+)?$'; then
+        echo "Could not detect the latest version (got: '${PIXLET_VERSION:-<empty>}'), using fallback"
         PIXLET_VERSION="v0.50.2"
     fi
 fi
@@ -67,8 +87,26 @@ download_binary() {
     temp_dir=$(mktemp -d -p "$PROJECT_ROOT" -t pixlet_download.XXXXXXXXXX)
     local temp_file="$temp_dir/$archive_name"
 
-    if ! curl -L -o "$temp_file" "$url" 2>/dev/null; then
-        echo "✗ Failed to download $arch"
+    # -f so an HTTP error is a failure. Without it curl writes the 404 body
+    # to the file and exits 0, and the first sign of trouble is tar saying
+    # "not in gzip format" about what is actually a page of HTML.
+    if ! curl -fL -o "$temp_file" "$url" 2>/dev/null; then
+        echo "✗ Failed to download $arch from $url"
+        rm -rf "$temp_dir"
+        return 1
+    fi
+
+    # Belt and braces: a mirror or proxy can return 200 with an error page.
+    if ! gzip -t "$temp_file" 2>/dev/null; then
+        echo "✗ Downloaded file is not a gzip archive: $url"
+        # These bytes come from whatever answered the request, so strip
+        # everything non-printable before echoing them: an error page carrying
+        # terminal escapes would otherwise be able to rewrite this output or
+        # bury it in a CI log. Printable characters are kept rather than
+        # hex-encoding the lot, because "<!DOCTYPE html>" is the diagnostic.
+        local first_bytes
+        first_bytes=$(head -c 60 "$temp_file" | tr -cd '[:print:]')
+        printf '  (first bytes: %s)\n' "$first_bytes"
         rm -rf "$temp_dir"
         return 1
     fi

src/cache/memory_cache.py

@@ -4,11 +4,58 @@ Memory Cache
 Handles in-memory caching with TTL support, size limits, and automatic cleanup.
 """
 
+import os
 import time
 import threading
 import logging
 from typing import Dict, Any, Optional
 
+# Historical fixed ceiling, kept as the fallback when RAM cannot be read.
+DEFAULT_MAX_SIZE = 1000
+
+
+def _total_memory_mb() -> Optional[float]:
+    """Physical RAM in MB, or None where /proc/meminfo is unavailable."""
+    try:
+        with open('/proc/meminfo', 'r', encoding='utf-8') as fh:
+            for line in fh:
+                if line.startswith('MemTotal:'):
+                    return int(line.split()[1]) / 1024
+    except (OSError, ValueError, IndexError):
+        return None
+    return None
+
+
+def default_max_size() -> int:
+    """Entry ceiling scaled to this machine's RAM.
+
+    One fixed ceiling cannot serve both a 512 MB Pi Zero 2 W and an 8 GB Pi 5.
+    Entries here are parsed API payloads that routinely run tens of kilobytes
+    each, so a thousand of them is a comfortable cache on a large board and a
+    substantial fraction of total RAM on a small one — where the process
+    competing for that RAM is also driving the panel. Set
+    LEDMATRIX_CACHE_MAX_ENTRIES to override.
+    """
+    override = os.environ.get('LEDMATRIX_CACHE_MAX_ENTRIES')
+    if override:
+        try:
+            value = int(override)
+            if value > 0:
+                return value
+        except ValueError:
+            pass
+
+    total_mb = _total_memory_mb()
+    if total_mb is None:
+        return DEFAULT_MAX_SIZE
+    if total_mb < 1536:      # 512 MB and 1 GB boards
+        return 150
+    if total_mb < 3072:      # 2 GB
+        return 400
+    if total_mb < 6144:      # 4 GB
+        return 800
+    return 1500              # 8 GB and up
+
 
 class MemoryCache:
     """Manages in-memory cache with TTL and size limits."""
@@ -87,6 +134,32 @@ class MemoryCache:
         with self._lock:
             self._cache[key] = value
             self._timestamps[key] = time.time()
+            # Enforce the ceiling here rather than leaving it to the periodic
+            # cleanup, which only runs every cleanup_interval seconds (300 by
+            # default). A burst of inserts between two sweeps could otherwise
+            # take the cache far past _max_size, which is the memory growth this
+            # limit exists to prevent -- and on a 1GB board that is the
+            # difference between a bounded cache and an unreachable Pi.
+            self._evict_over_limit_locked()
+
+    def _evict_over_limit_locked(self) -> int:
+        """Drop oldest entries until the cache is within _max_size.
+
+        Caller must hold self._lock. Returns the number of entries removed.
+        """
+        excess = len(self._cache) - self._max_size
+        if excess <= 0:
+            return 0
+        oldest = sorted(
+            self._timestamps.items(),
+            key=lambda item: float(item[1]) if isinstance(item[1], (int, float)) else 0.0
+        )
+        removed = 0
+        for key, _ in oldest[:excess]:
+            self._cache.pop(key, None)
+            self._timestamps.pop(key, None)
+            removed += 1
+        return removed
     
     def clear(self, key: Optional[str] = None) -> None:
         """
@@ -143,22 +216,8 @@ class MemoryCache:
                 self._timestamps.pop(key, None)
                 removed_count += 1
             
-            # Enforce size limit by removing oldest entries if cache is too large
-            if len(self._cache) > self._max_size:
-                # Sort by timestamp (oldest first)
-                sorted_entries = sorted(
-                    self._timestamps.items(),
-                    key=lambda x: float(x[1]) if isinstance(x[1], (int, float)) else 0
-                )
-                
-                # Remove oldest entries until we're under the limit
-                excess_count = len(self._cache) - self._max_size
-                for i in range(excess_count):
-                    if i < len(sorted_entries):
-                        key = sorted_entries[i][0]
-                        self._cache.pop(key, None)
-                        self._timestamps.pop(key, None)
-                        removed_count += 1
+            # Same ceiling enforcement set() uses, so the two cannot drift.
+            removed_count += self._evict_over_limit_locked()
             
             self._last_cleanup = current_time
             

src/cache_manager.py

@@ -33,7 +33,7 @@ import logging
 import threading
 import tempfile
 from src.exceptions import CacheError
-from src.cache.memory_cache import MemoryCache
+from src.cache.memory_cache import MemoryCache, default_max_size
 from src.cache.disk_cache import DiskCache
 from src.cache.cache_strategy import CacheStrategy
 from src.cache.cache_metrics import CacheMetrics
@@ -84,7 +84,9 @@ class CacheManager:
             self.logger.warning("ConfigManager not available, using default cache intervals")
         
         # Initialize cache components using composition
-        self._memory_cache_component = MemoryCache(max_size=1000, cleanup_interval=300.0)
+        self._memory_cache_component = MemoryCache(
+            max_size=default_max_size(), cleanup_interval=300.0
+        )
         self._disk_cache_component = DiskCache(cache_dir=self.cache_dir, logger=self.logger)
         self._strategy_component = CacheStrategy(config_manager=self.config_manager, logger=self.logger)
         self._metrics_component = CacheMetrics(logger=self.logger)

src/common/logo_helper.py

@@ -6,8 +6,6 @@ Extracted from LEDMatrix core to provide reusable functionality for plugins.
 """
 
 import logging
-import os
-import tempfile
 from pathlib import Path
 from typing import Dict, List, Optional, Union
 
@@ -21,10 +19,6 @@ from src.common.permission_utils import (
 )
 
 
-# Well above any real team logo; bounds what a remote URL can write to disk.
-MAX_LOGO_BYTES = 10 * 1024 * 1024
-
-
 class LogoHelper:
     """
     Helper class for logo loading, caching, and resizing.
@@ -232,10 +226,7 @@ class LogoHelper:
         return {
             'cached_logos': len(self._logo_cache),
             'cache_size_limit': self.cache_size,
-            'cache_usage_percent': (
-                (len(self._logo_cache) / self.cache_size) * 100
-                if self.cache_size else 0
-            ),
+            'cache_usage_percent': (len(self._logo_cache) / self.cache_size) * 100
         }
     
     def _resize_logo(self, logo: Image.Image, max_width: Optional[int] = None, 
@@ -267,60 +258,17 @@ class LogoHelper:
         self._cache_order.append(cache_key)
     
     def _download_logo(self, url: str, file_path: Path) -> None:
-        """Download logo from URL.
-
-        The response size is capped and the saved file is verified as a
-        decodable image before it is left on disk: a logo URL is remote
-        input, and without this an oversized or malformed response would
-        be cached for every later load_logo() call to trip over.
-
-        The body is streamed and counted as it arrives rather than read
-        through response.content, which buffers the whole thing first —
-        a server that omits Content-Length and never stops sending would
-        exhaust memory before any size check could run. Nothing lands at
-        file_path until the download completes and decodes, so a failed
-        download cannot leave a truncated logo behind either.
-        """
+        """Download logo from URL."""
         # Ensure directory exists with proper permissions
         ensure_directory_permissions(file_path.parent, get_assets_dir_mode())
         
-        # A unique temp name, not a fixed "<name>.part": two plugins can
-        # ask for the same logo at once, and a shared name would let them
-        # interleave writes into one file, publish the mixture, or delete
-        # each other's partial. Same directory, so os.replace stays atomic.
-        fd, tmp_name = tempfile.mkstemp(
-            dir=str(file_path.parent), prefix=file_path.name + '.', suffix='.part')
-        tmp_path = Path(tmp_name)
-        try:
-            # fdopen outermost so the descriptor mkstemp handed back is
-            # always adopted and closed, including when the request itself
-            # raises — load_logo_with_download swallows that, so a leak
-            # here would accumulate quietly on a URL that keeps failing.
-            with os.fdopen(fd, 'wb') as f:
-                with self.session.get(url, timeout=30, stream=True) as response:
+        # Download with timeout
+        response = self.session.get(url, timeout=30)
         response.raise_for_status()
-                    downloaded = 0
-                    for chunk in response.iter_content(chunk_size=64 * 1024):
-                        if not chunk:
-                            continue
-                        downloaded += len(chunk)
-                        if downloaded > MAX_LOGO_BYTES:
-                            raise ValueError(
-                                f"Logo at {url} exceeds the "
-                                f"{MAX_LOGO_BYTES}-byte limit; not saved")
-                        f.write(chunk)
         
-            # Verify it decodes before it becomes the cached logo. PIL
-            # raises DecompressionBombError past its own pixel limit; a
-            # partial or non-image response raises UnidentifiedImageError
-            # (an OSError subclass).
-            with Image.open(tmp_path) as probe:
-                probe.load()
-
-            os.replace(tmp_path, file_path)
-        except BaseException:
-            tmp_path.unlink(missing_ok=True)
-            raise
+        # Save to file
+        with open(file_path, 'wb') as f:
+            f.write(response.content)
         
         # Set proper file permissions after saving
         ensure_file_permissions(file_path, get_assets_file_mode())

src/common/sync_manager.py

@@ -19,7 +19,6 @@ Port default: 5765 (UDP). Open this port on both Pis if ufw is active:
 
 import io
 import json
-import math
 import os
 import socket
 import struct
@@ -38,13 +37,6 @@ _RAW_MAGIC = b'SYNC_RAW'
 _RAW_HEADER = struct.Struct('<HH')  # width, height (uint16 LE)
 
 
-# Upper bound on a decoded frame/scroll image. Generous for any real scroll
-# image (a leader's full cycle is long but only panel-height tall), and low
-# enough that a crafted image from any host on the LAN cannot force a large
-# allocation on the render thread. Applied on both receive paths — the TCP
-# image server and the follower's legacy-PNG UDP fallback.
-_MAX_FRAME_W, _MAX_FRAME_H = 100_000, 256
-
 SYNC_PORT = 5765
 HELLO_INTERVAL = 5.0       # follower broadcasts hello every 5 s
 HEARTBEAT_INTERVAL = 2.0   # follower sends heartbeat every 2 s
@@ -109,7 +101,6 @@ class DisplaySyncManager:
         self._peer_chain: int = 0
         self._last_heartbeat_time: float = 0.0
         self._leader_width: int = 0  # set by display_controller after init
-        self._oversized_frame_warned: bool = False
 
         # Follower state
         self._follower_state = FollowerState.STANDALONE
@@ -183,10 +174,6 @@ class DisplaySyncManager:
                 continue
             except Exception as exc:
                 self.logger.debug("Sync leader recv error: %s", exc)
-                # Brief backoff: a socket left in a bad state raises
-                # immediately, which would otherwise spin this thread at
-                # 100% CPU logging the same error.
-                time.sleep(0.1)
 
     def _handle_hello(self, msg: dict, sender_ip: str) -> None:
         hw = self._hw_config
@@ -286,10 +273,11 @@ class DisplaySyncManager:
                             break
                         data.extend(chunk)
                     img = Image.open(io.BytesIO(data))
-                    if img.width > _MAX_FRAME_W or img.height > _MAX_FRAME_H:
+                    _MAX_W, _MAX_H = 100_000, 256  # generous for any real scroll image
+                    if img.width > _MAX_W or img.height > _MAX_H:
                         self.logger.warning(
                             "Sync: rejected oversized scroll image %dx%d (max %dx%d) from %s",
-                            img.width, img.height, _MAX_FRAME_W, _MAX_FRAME_H, addr,
+                            img.width, img.height, _MAX_W, _MAX_H, addr,
                         )
                         continue
                     try:
@@ -408,7 +396,7 @@ class DisplaySyncManager:
             data = header + arr.tobytes()
             if len(data) <= 65000:
                 self._send_sock.sendto(data, (self._peer_ip, self.port))
-            elif not self._oversized_frame_warned:
+            elif not getattr(self, '_oversized_frame_warned', False):
                 self._oversized_frame_warned = True
                 self.logger.warning(
                     "Sync: frame too large for UDP (%d bytes, max 65000) — "
@@ -463,8 +451,25 @@ class DisplaySyncManager:
         )
         self.write_status_file()
 
-    def _handle_received_frame(self, img: Image.Image, sender_ip: str) -> None:
-        """Record a decoded leader frame and enter follower mode if needed."""
+    def _follower_recv_loop(self) -> None:
+        while self._running:
+            try:
+                data, addr = self._recv_sock.recvfrom(65535)
+                sender_ip = addr[0]
+
+                if data[:8] == _RAW_MAGIC or len(data) > 512:
+                    # Frame data: prefer magic-tagged raw RGB; fall back to legacy PNG
+                    try:
+                        if data[:8] == _RAW_MAGIC:
+                            w, h = _RAW_HEADER.unpack(data[8:12])
+                            raw = data[12:]
+                            img = Image.frombuffer(
+                                "RGB", (w, h), raw, "raw", "RGB", 0, 1
+                            )
+                        else:
+                            # Fallback: try legacy PNG
+                            img = Image.open(io.BytesIO(data))
+                            img.load()
                         with self._frame_lock:
                             self._latest_frame = img
                         self._last_leader_frame_time = time.time()
@@ -477,62 +482,12 @@ class DisplaySyncManager:
                                 sender_ip,
                             )
                             self.write_status_file()
-
-    def _follower_recv_loop(self) -> None:
-        while self._running:
-            try:
-                data, addr = self._recv_sock.recvfrom(65535)
-                sender_ip = addr[0]
-
-                if data[:8] == _RAW_MAGIC:
-                    # Magic-tagged raw RGB frame — self-describing, no guessing.
-                    try:
-                        w, h = _RAW_HEADER.unpack(data[8:12])
-                        raw = data[12:]
-                        img = Image.frombuffer(
-                            "RGB", (w, h), raw, "raw", "RGB", 0, 1
-                        )
-                        self._handle_received_frame(img, sender_ip)
                     except Exception as exc:
                         self.logger.debug("Sync: frame decode error: %s", exc)
                 else:
-                    # No magic prefix. Whether the payload parses as JSON
-                    # decides between a control message and a legacy
-                    # (pre-magic) PNG frame — both wire formats are
-                    # self-describing, so no size heuristic is needed. A
-                    # >512-byte control message used to be misrouted into
-                    # image decode and silently dropped.
+                    # Control message
                     try:
                         msg = json.loads(data.decode("utf-8"))
-                    except (json.JSONDecodeError, UnicodeDecodeError):
-                        # Not JSON — try a legacy PNG frame.
-                        try:
-                            img = Image.open(io.BytesIO(data))
-                            if img.width > _MAX_FRAME_W or img.height > _MAX_FRAME_H:
-                                # Same cap the TCP image path applies: decode
-                                # is deferred until load(), so check first.
-                                self.logger.debug(
-                                    "Sync: rejected oversized legacy frame %dx%d from %s",
-                                    img.width, img.height, sender_ip,
-                                )
-                                continue
-                            img.load()
-                            self._handle_received_frame(img, sender_ip)
-                        except Exception as exc:
-                            self.logger.debug("Sync: frame decode error: %s", exc)
-                        continue
-
-                    # It parsed, so it is a control message and never a
-                    # frame. Read and validate its fields under a guard —
-                    # a UDP payload is attacker-shaped, so a non-object
-                    # body makes .get() raise AttributeError and an "sx"
-                    # carrying a non-numeric x raises ValueError/TypeError
-                    # — but dispatch the callback *outside* it. Running
-                    # the callback in here would let a fault in someone
-                    # else's code read as a malformed packet and be
-                    # logged as one.
-                    fire_new_cycle = False
-                    try:
                         t = msg.get("t")
                         if t == "hello_ack":
                             self._leader_ip = sender_ip
@@ -546,17 +501,7 @@ class DisplaySyncManager:
                             self.write_status_file()
                         elif t == "sx":
                             # Vegas scroll-position sync — tiny message, renders locally
-                            scroll_x = float(msg["x"])
-                            if not math.isfinite(scroll_x):
-                                # json.loads accepts the NaN/Infinity literals,
-                                # and float("nan") accepts the strings, so a
-                                # non-finite x reaches here intact. Left alone
-                                # it poisons every offset computed from it —
-                                # NaN comparisons are all false, so the
-                                # follower renders a frame it can never scroll
-                                # back from. Treat it as malformed.
-                                raise ValueError(f"non-finite scroll x: {msg['x']!r}")
-                            self._latest_scroll_x = scroll_x
+                            self._latest_scroll_x = float(msg["x"])
                             self._last_leader_frame_time = time.time()
                             self._leader_ip = sender_ip
                             if self._follower_state == FollowerState.STANDALONE:
@@ -566,22 +511,19 @@ class DisplaySyncManager:
                                     sender_ip,
                                 )
                                 self.write_status_file()
-                                fire_new_cycle = True  # build initial scroll image
+                                if self._on_new_cycle:
+                                    self._on_new_cycle()  # build initial scroll image
                         elif t == "nc":
                             # Leader started a new scroll cycle — rebuild local image
-                            fire_new_cycle = True
-                    except (KeyError, AttributeError, TypeError, ValueError) as exc:
-                        self.logger.debug("Sync: malformed control message: %s", exc)
-                        continue
-
-                    if fire_new_cycle and self._on_new_cycle:
+                            if self._on_new_cycle:
                                 self._on_new_cycle()
+                    except (json.JSONDecodeError, UnicodeDecodeError, KeyError):
+                        pass
 
             except socket.timeout:
                 continue
             except Exception as exc:
                 self.logger.debug("Sync follower recv error: %s", exc)
-                time.sleep(0.1)
 
     def _follower_announce_loop(self) -> None:
         hw = self._hw_config

src/display_controller.py

@@ -1694,6 +1694,12 @@ class DisplayController:
                 logger.warning("Error checking live priority for %s: %s", mode_name, e)
         return live
 
+    def _vegas_keeps_live_in_ticker(self) -> bool:
+        """Whether live content should stay in the ticker instead of preempting it."""
+        coordinator = getattr(self, 'vegas_coordinator', None)
+        config = getattr(coordinator, 'vegas_config', None)
+        return bool(getattr(config, 'live_in_ticker', False))
+
     def _check_live_priority(self, advance=False):
         """Return the live-priority mode to display, or None if nothing is live.
 
@@ -1907,14 +1913,24 @@ class DisplayController:
                 # Check for live priority content and switch to it immediately.
                 # advance=True so multiple simultaneously-live games take turns
                 # (round-robin) instead of pinning to the first plugin.
-                if not self.on_demand_active and not wifi_status_data:
+                # Skipped when the ticker is keeping live content: switching
+                # the rotation underneath Vegas would move current_mode_index
+                # and stash a resume point for a takeover that never happens.
+                if (not self.on_demand_active and not wifi_status_data
+                        and not (self._is_vegas_mode_active()
+                                 and self._vegas_keeps_live_in_ticker())):
                     live_priority_mode = self._check_live_priority(advance=True)
                     self._apply_live_priority(live_priority_mode)
 
                 # Vegas scroll mode - continuous ticker across all plugins
                 # Priority: on-demand > wifi-status > live-priority > vegas > normal rotation
                 if self._is_vegas_mode_active() and not wifi_status_data:
-                    live_mode = self._check_live_priority()
+                    # Live content normally preempts the ticker entirely. With
+                    # vegas_scroll.live_in_ticker the marquee keeps running and
+                    # the live plugin takes extra turns inside it instead --
+                    # see StreamManager._apply_priority_weights.
+                    live_mode = (None if self._vegas_keeps_live_in_ticker()
+                                 else self._check_live_priority())
                     if not live_mode:
                         try:
                             # Run Vegas mode iteration

src/plugin_system/base_plugin.py

@@ -555,6 +555,48 @@ class BasePlugin(ABC):
         """
         return False
 
+    def get_vegas_priority_weight(self) -> Optional[int]:
+        """How many slots per Vegas cycle this plugin should get, or None.
+
+        The Vegas ticker is otherwise a strict round robin: every plugin
+        appears exactly once per cycle. With a dozen plugins enabled that puts
+        minutes between a live score and its next appearance. A weight of N
+        gives the plugin N slots per cycle, spread evenly through it rather
+        than clumped together.
+
+        Return ``None`` (the default) to let the core decide. It gives a
+        plugin ``vegas_scroll.live_weight`` when ``has_live_priority()`` and
+        ``has_live_content()`` are both true, and 1 otherwise -- so live sports
+        already get extra turns without implementing this at all.
+
+        Implement it only when the plugin knows something the core cannot. The
+        motivating case is favorite teams: the core can see *that* a game is
+        live but not *whose*, so a scoreboard that wants its favorite's game
+        shown more often than other live games has to say so::
+
+            def get_vegas_priority_weight(self):
+                if not (self.has_live_priority() and self.has_live_content()):
+                    return None                      # let the core decide
+                cfg = self.global_config.get('display', {}).get('vegas_scroll', {})
+                if self._favorite_is_live():
+                    return cfg.get('favorite_live_weight', 5)
+                return cfg.get('live_weight', 3)
+
+        The weight is per *plugin*, not per game. A scoreboard showing four
+        live games still occupies one slot at a time and rotates its own games
+        within that slot; this controls how often the plugin itself comes
+        round.
+
+        Raising is safe: the core logs it and falls back to its own
+        live-content check, so a broken weight calculation costs the plugin
+        the favorite distinction but not the live boost.
+
+        Returns:
+            Slots per cycle (clamped to 1..10 by the caller), or None to
+            defer to the core's own live-content weighting.
+        """
+        return None
+
     def get_live_modes(self) -> List[str]:
         """
         Get list of display modes that should be used during live priority takeover.

src/plugin_system/plugin_health.py

@@ -7,7 +7,7 @@ and circuit breaker state. Provides automatic recovery mechanisms.
 
 import time
 import logging
-from typing import Dict, Optional, Any
+from typing import Dict, Optional, Any, Tuple
 from enum import Enum
 
 
@@ -64,10 +64,48 @@ class PluginHealthTracker:
             cache_key, max_age=None, memory_ttl=0 if force_reload else None
         )
 
-        if cached:
-            return cached
+        if isinstance(cached, dict) and cached:
+            # Complete it rather than trusting it: a persisted record can be
+            # missing fields the callers index directly (a partial write, a
+            # restored backup, an older schema), and returning it verbatim makes
+            # record_success / record_failure raise KeyError, which takes the
+            # display down in a restart loop that survives reboots because the
+            # bad entry is on disk.
+            state, repaired = self._repair_health_state(cached)
+            if repaired:
+                self.logger.warning(
+                    f"Repaired health state for {plugin_id}: "
+                    f"{sorted(repaired)} missing or invalid, using defaults for those."
+                )
+            return state
 
-        # Default state
+        # Not a dict at all: written by something other than
+        # _save_health_state (a key collision, a corrupted entry). Nothing to
+        # salvage.
+        if cached is not None and not isinstance(cached, dict):
+            self.logger.warning(
+                f"Discarding malformed health state for {plugin_id}: expected "
+                f"dict, got {type(cached).__name__}. Falling back to defaults."
+            )
+
+        return self._default_health_state()
+    
+    def _save_health_state(self, plugin_id: str, state: Dict[str, Any]) -> None:
+        """Save health state to cache."""
+        cache_key = self._get_health_key(plugin_id)
+        self.cache_manager.set(cache_key, state)  # Persist indefinitely
+        self._health_state[plugin_id] = state
+    
+    # The fields callers index directly (state['circuit_state'] and friends).
+    # A cached dict missing any of them raises KeyError deep in record_success /
+    # record_failure, so the value is completed before it is handed out.
+    _COUNTER_FIELDS = ('consecutive_failures', 'total_failures', 'total_successes')
+    _TIMESTAMP_FIELDS = ('last_success_time', 'last_failure_time',
+                         'circuit_opened_time', 'half_open_start_time')
+
+    @staticmethod
+    def _default_health_state() -> Dict[str, Any]:
+        """A fresh state with every field the callers expect."""
         return {
             'consecutive_failures': 0,
             'total_failures': 0,
@@ -77,14 +115,55 @@ class PluginHealthTracker:
             'circuit_state': CircuitState.CLOSED.value,
             'circuit_opened_time': None,
             'half_open_start_time': None,
-            'last_error': None
+            'last_error': None,
         }
 
-    def _save_health_state(self, plugin_id: str, state: Dict[str, Any]) -> None:
-        """Save health state to cache."""
-        cache_key = self._get_health_key(plugin_id)
-        self.cache_manager.set(cache_key, state)  # Persist indefinitely
-        self._health_state[plugin_id] = state
+    @classmethod
+    def _repair_health_state(cls, cached: Dict[str, Any]) -> Tuple[Dict[str, Any], list]:
+        """Return `cached` completed against the defaults, plus what was repaired.
+
+        Per-field rather than all-or-nothing: a record that has real failure
+        counts but is missing `last_error` should keep the counts, not be reset
+        to healthy. Only values that are absent or the wrong type fall back to
+        the default, so a partial or older-schema record survives with whatever
+        it does carry, while every field the callers index is guaranteed present
+        and of a usable type.
+        """
+        state = cls._default_health_state()
+        repaired = []
+        for field, default in state.items():
+            if field not in cached:
+                repaired.append(field)
+                continue
+            value = cached[field]
+            if field in cls._COUNTER_FIELDS:
+                ok = isinstance(value, int) and not isinstance(value, bool) and value >= 0
+            elif field in cls._TIMESTAMP_FIELDS:
+                # bool is a subclass of int, so True would pass as a timestamp
+                # and then compare as 1.0 -- expiring a cooldown the instant it
+                # opens, or (False) making the elapsed check never fire.
+                ok = value is None or (
+                    isinstance(value, (int, float)) and not isinstance(value, bool)
+                )
+            elif field == 'circuit_state':
+                # Membership first requires the value to be hashable: a list or
+                # dict here would raise TypeError out of the repair itself,
+                # which is the crash this whole path exists to prevent.
+                ok = isinstance(value, str) and value in {
+                    member.value for member in CircuitState
+                }
+            else:  # last_error
+                ok = value is None or isinstance(value, str)
+            if ok:
+                state[field] = value
+            else:
+                repaired.append(field)
+        # Anything the schema has since grown (degraded, degraded_reason) is
+        # read with .get() by its callers, so carry it through untouched.
+        for field, value in cached.items():
+            if field not in state:
+                state[field] = value
+        return state, repaired
 
     def get_health_state(self, plugin_id: str, force_reload: bool = False) -> Dict[str, Any]:
         """Get current health state for a plugin.

src/plugin_system/plugin_loader.py

@@ -14,7 +14,7 @@ import sys
 import subprocess
 import threading
 from pathlib import Path
-from typing import Dict, Any, Optional, Tuple, Type
+from typing import Dict, Any, List, Optional, Tuple, Type
 import logging
 
 from packaging.requirements import InvalidRequirement, Requirement
@@ -45,6 +45,76 @@ def requirements_has_real_deps(requirements_file: str) -> bool:
     return False
 
 
+def _extra_dependencies(dist_name: str, extras) -> Optional[List[Requirement]]:
+    """Dependencies a distribution declares *only* behind the given extras.
+
+    Returns None when the installed metadata cannot be read or parsed, so the
+    caller can fall back to running pip rather than assuming anything.
+    """
+    try:
+        meta = importlib.metadata.metadata(dist_name)
+    except importlib.metadata.PackageNotFoundError:
+        return None
+
+    gated: List[Requirement] = []
+    for raw in meta.get_all('Requires-Dist') or []:
+        try:
+            dep = Requirement(raw)
+        except InvalidRequirement:
+            return None
+        if dep.marker is None:
+            continue
+        # Keep only what the distribution gates behind an extra we asked for:
+        # satisfied when `extra` is that name, but not when no extra is
+        # requested. A marker that holds either way (python_version, sys_platform)
+        # belongs to the base install and is already covered by the version check.
+        if dep.marker.evaluate({'extra': ''}):
+            continue
+        if any(dep.marker.evaluate({'extra': extra}) for extra in extras):
+            gated.append(dep)
+    return gated
+
+
+def _extras_are_satisfied(req: Requirement, _visited: Optional[set] = None) -> bool:
+    """Check the dependencies pulled in by req's extras are installed.
+
+    Follows extras through nested extras. A gated dependency can itself request
+    one (`requests[socks]`), and checking only that `requests` is installed at
+    an acceptable version says nothing about whether the socks extra's own
+    dependency is there -- so the caller would skip pip and the plugin would
+    fail at import instead. Plain dependencies are still checked one level
+    deep, which is all that is needed to tell "the extra was installed" from
+    "the extra was never installed".
+
+    `_visited` carries the (distribution, extras) pairs already seen, so a
+    dependency cycle between extras terminates instead of recursing forever.
+    Anything unreadable returns False, so the caller still falls through to pip.
+    """
+    if _visited is None:
+        _visited = set()
+    marker = (req.name.lower(), frozenset(e.lower() for e in req.extras))
+    if marker in _visited:
+        # Already accounted for higher up the chain; treating a cycle as
+        # satisfied here is safe because the outer frame still has to pass.
+        return True
+    _visited.add(marker)
+
+    gated = _extra_dependencies(req.name, req.extras)
+    if gated is None:
+        return False
+
+    for dep in gated:
+        try:
+            dep_version = importlib.metadata.version(dep.name)
+        except importlib.metadata.PackageNotFoundError:
+            return False
+        if dep.specifier and not dep.specifier.contains(dep_version, prereleases=True):
+            return False
+        if dep.extras and not _extras_are_satisfied(dep, _visited):
+            return False
+    return True
+
+
 def requirements_are_satisfied(requirements_file: str) -> bool:
     """
     Check whether every real requirement line in requirements.txt is already
@@ -76,9 +146,6 @@ def requirements_are_satisfied(requirements_file: str) -> bool:
         except InvalidRequirement:
             return False
 
-        if req.extras:
-            return False  # verifying extras' sub-dependencies isn't worth it here
-
         if req.marker is not None and not req.marker.evaluate():
             continue  # not applicable on this platform/interpreter
 
@@ -90,6 +157,9 @@ def requirements_are_satisfied(requirements_file: str) -> bool:
         if req.specifier and not req.specifier.contains(installed_version, prereleases=True):
             return False
 
+        if req.extras and not _extras_are_satisfied(req):
+            return False
+
     return True
 
 

src/plugin_system/plugin_manager.py

@@ -125,6 +125,14 @@ class PluginManager:
         self._plugin_locks: Dict[str, threading.Lock] = {}
         self._plugin_locks_guard = threading.Lock()
         self._update_worker: Optional[threading.Thread] = None
+        # Plugin ids whose update() has finished since the last time anyone
+        # asked. Updates are dispatched to a worker thread, so a caller that
+        # wants to know "whose data just changed" cannot learn it by diffing
+        # plugin_last_update around run_scheduled_updates() -- that call only
+        # enqueues, and the timestamp is stamped later, on the worker. See
+        # run_scheduled_updates_with_changes().
+        self._completed_updates: set = set()
+        self._completed_updates_lock = threading.Lock()
         self._synchronous_updates = False
         if self.config_manager is not None:
             try:
@@ -1025,6 +1033,7 @@ class PluginManager:
                 if success:
                     with self._plugin_last_update_lock:
                         self.plugin_last_update[plugin_id] = scheduled_time
+                    self._note_update_completed(plugin_id)
                     self.state_manager.record_update(plugin_id)
                     self.state_manager.set_state(plugin_id, PluginState.ENABLED)
                     if self.health_tracker:
@@ -1089,28 +1098,41 @@ class PluginManager:
 
     def run_scheduled_updates_with_changes(self, current_time: Optional[float] = None) -> List[str]:
         """
-        Like run_scheduled_updates(), but also returns the plugin_ids whose
-        plugin_last_update timestamp actually advanced during this call.
+        Like run_scheduled_updates(), but also reports which plugins have
+        fresh data -- the ids whose update() has finished since the last
+        call, not necessarily the ones enqueued by this one.
 
-        The before/after snapshots and the update pass itself are each
-        individually lock-protected against concurrent plugin_last_update
-        mutation (Vegas mode calls this from its own background
-        update-tick thread, racing the main render loop's plugin updates),
-        so callers get an atomic "who got fresh data" answer without
-        reaching into plugin_last_update themselves. The lock is not held
-        across the update pass so slow/blocking plugin update() calls don't
-        serialize against other plugin_last_update readers.
+        That distinction is the whole point. This used to snapshot
+        plugin_last_update, call run_scheduled_updates(), and diff. But
+        run_scheduled_updates() only *enqueues*: the work runs on the
+        update worker and the timestamp is stamped there, after this method
+        has already returned. The two snapshots were therefore always
+        identical and the result was always empty, so Vegas never learned
+        that any plugin's data had changed and kept scrolling whatever a
+        segment was first built from -- last night's live game still drawn
+        as live the next morning. The only path that ever worked was the
+        synchronous kill-switch, where update() runs inline.
+
+        Reporting completions instead of enqueues costs a poll's worth of
+        latency (the Vegas tick runs every ~4s) and is correct regardless of
+        which side of the queue the work lands on.
         """
-        with self._plugin_last_update_lock:
-            old_times = dict(self.plugin_last_update)
-
         self.run_scheduled_updates(current_time)
+        return self.drain_completed_updates()
 
-        with self._plugin_last_update_lock:
-            return [
-                plugin_id for plugin_id, new_time in self.plugin_last_update.items()
-                if new_time > old_times.get(plugin_id, 0.0)
-            ]
+    def _note_update_completed(self, plugin_id: str) -> None:
+        """Record that a plugin's update() finished, for the next poll."""
+        with self._completed_updates_lock:
+            self._completed_updates.add(plugin_id)
+
+    def drain_completed_updates(self) -> List[str]:
+        """Return and clear the plugin ids whose update() has since finished."""
+        with self._completed_updates_lock:
+            if not self._completed_updates:
+                return []
+            done = sorted(self._completed_updates)
+            self._completed_updates.clear()
+            return done
 
     def update_all_plugins(self) -> None:
         """
@@ -1135,6 +1157,7 @@ class PluginManager:
                 if success:
                     with self._plugin_last_update_lock:
                         self.plugin_last_update[plugin_id] = time.time()
+                    self._note_update_completed(plugin_id)
                     self.state_manager.record_update(plugin_id)
                     self.state_manager.set_state(plugin_id, PluginState.ENABLED)
                 else:

src/vegas_mode/config.py

@@ -125,6 +125,32 @@ class VegasModeConfig:
     plugin_order: List[str] = field(default_factory=list)
     excluded_plugins: Set[str] = field(default_factory=set)
 
+    # --- Live content in the ticker -------------------------------------
+    #
+    # By default a live game preempts Vegas entirely: the display controller
+    # refuses to run the ticker while any plugin reports live priority, and you
+    # get the full-screen scoreboard instead. Set live_in_ticker to keep the
+    # marquee running and let live content take extra turns within it.
+    #
+    # The rotation is otherwise a strict round robin -- every plugin appears
+    # exactly once per cycle -- so with a dozen plugins enabled a live score
+    # comes round once a lap and can be minutes old on screen. Weighting lets a
+    # plugin claim several slots per cycle instead.
+    #
+    # Weights are per plugin, not per game: a scoreboard showing four live
+    # games still occupies one slot at a time, and rotates its own games within
+    # that slot using its own favorite_live_boost.
+    live_in_ticker: bool = False
+
+    # Slots per cycle for a plugin reporting live content. 1 disables the boost
+    # and restores the plain round robin.
+    live_weight: int = 3
+
+    # Slots per cycle for a plugin whose live content involves a favorite team.
+    # Only plugins implementing get_vegas_priority_weight() can claim this --
+    # the core cannot tell whose game is on, so the plugin reports it.
+    favorite_live_weight: int = 5
+
     # Performance settings
     target_fps: int = 125  # Target frame rate
     buffer_ahead: int = 2  # Number of plugins to buffer ahead
@@ -175,6 +201,12 @@ class VegasModeConfig:
             overflow_mode=str(vegas_config.get('overflow_mode', 'rotate')),
             plugin_order=list(vegas_config.get('plugin_order', [])),
             excluded_plugins=set(vegas_config.get('excluded_plugins', [])),
+            live_in_ticker=bool(vegas_config.get('live_in_ticker', False)),
+            # Clamped: a weight below 1 would drop the plugin from the rotation
+            # entirely, and a very large one starves everything else.
+            live_weight=max(1, min(10, int(vegas_config.get('live_weight', 3)))),
+            favorite_live_weight=max(
+                1, min(10, int(vegas_config.get('favorite_live_weight', 5)))),
             target_fps=int(vegas_config.get('target_fps', 125)),
             buffer_ahead=int(vegas_config.get('buffer_ahead', 2)),
             frame_based_scrolling=vegas_config.get('frame_based_scrolling', True),
@@ -204,6 +236,9 @@ class VegasModeConfig:
             'lead_in_width': self.lead_in_width,
             'plugins_per_cycle': self.plugins_per_cycle,
             'max_plugin_width_ratio': self.max_plugin_width_ratio,
+            'live_in_ticker': self.live_in_ticker,
+            'live_weight': self.live_weight,
+            'favorite_live_weight': self.favorite_live_weight,
             'overflow_mode': self.overflow_mode,
             'plugin_order': self.plugin_order,
             'excluded_plugins': list(self.excluded_plugins),
@@ -371,6 +406,15 @@ class VegasModeConfig:
 
         if 'enabled' in vegas_config:
             self.enabled = vegas_config['enabled']
+        if 'live_in_ticker' in vegas_config:
+            self.live_in_ticker = bool(vegas_config['live_in_ticker'])
+        # Clamped exactly as from_config does: a weight below 1 would drop the
+        # plugin from the rotation, and a huge one starves everything else.
+        if 'live_weight' in vegas_config:
+            self.live_weight = max(1, min(10, int(vegas_config['live_weight'])))
+        if 'favorite_live_weight' in vegas_config:
+            self.favorite_live_weight = max(
+                1, min(10, int(vegas_config['favorite_live_weight'])))
         if 'scroll_speed' in vegas_config:
             self.scroll_speed = float(vegas_config['scroll_speed'])
         if 'separator_width' in vegas_config:

src/vegas_mode/coordinator.py

@@ -528,6 +528,12 @@ class VegasModeCoordinator:
         if not self._live_priority_check:
             return False
 
+        if self.vegas_config.live_in_ticker:
+            # The ticker keeps live content rather than yielding to it; the
+            # extra turns are arranged in the rotation itself, so there is
+            # nothing to pause for.
+            return False
+
         try:
             live_mode = self._live_priority_check()
             if live_mode:

src/vegas_mode/stream_manager.py

@@ -406,6 +406,8 @@ class StreamManager:
         )
         logger.info("Ordered plugins: %s", ordered_plugins)
 
+        ordered_plugins = self._apply_priority_weights(ordered_plugins)
+
         # Atomically update shared state under lock to avoid races with prefetchers
         with self._buffer_lock:
             self._ordered_plugins = ordered_plugins
@@ -417,6 +419,143 @@ class StreamManager:
 
         logger.info("=" * 60)
 
+    def _plugin_weight(self, plugin_id: str) -> int:
+        """Slots per cycle for one plugin.
+
+        A plugin may answer for itself via get_vegas_priority_weight() -- the
+        only way favorite-team awareness can reach here, since the core can see
+        that a game is live but not whose. When it declines (returns None, the
+        default), live content earns ``live_weight`` and everything else 1.
+        """
+        plugin = None
+        try:
+            plugin = self.plugin_manager.plugins.get(plugin_id)
+        except (AttributeError, TypeError):
+            return 1
+        if plugin is None:
+            return 1
+
+        try:
+            if hasattr(plugin, 'get_vegas_priority_weight'):
+                declared = plugin.get_vegas_priority_weight()
+                if declared is not None:
+                    return max(1, min(10, int(declared)))
+        except Exception:
+            # Deliberately falls through to the core's own live check rather
+            # than demoting to 1. The plugin's weight calculation is broken,
+            # but has_live_priority() and has_live_content() are separate
+            # methods guarded separately below -- a plugin that genuinely has
+            # a live game should still get live_weight for it.
+            logger.exception("[%s] get_vegas_priority_weight() failed", plugin_id)
+
+        try:
+            if (hasattr(plugin, 'has_live_priority')
+                    and hasattr(plugin, 'has_live_content')
+                    and plugin.has_live_priority()
+                    and plugin.has_live_content()):
+                return self.config.live_weight
+        except Exception:
+            logger.exception("[%s] live-content check failed", plugin_id)
+        return 1
+
+    def _apply_priority_weights(self, ordered: List[str]) -> List[str]:
+        """Expand the rotation so weighted plugins take several turns per cycle.
+
+        Smooth Weighted Round-Robin, the same scheduler the sports plugins use
+        to rotate their own games: a plugin of weight N appears N times per
+        cycle, and the repeats are spaced through the cycle rather than
+        clumped, so a live score is never three-in-a-row followed by a long
+        silence.
+
+        Returns the input unchanged when nothing is weighted, which is both the
+        common case and the pre-existing behaviour.
+        """
+        if not ordered or not self.config.live_in_ticker:
+            return ordered
+
+        weights = {pid: self._plugin_weight(pid) for pid in ordered}
+        total = sum(weights.values())
+        if total <= len(ordered):
+            return ordered          # nothing boosted; plain round robin
+
+        current = {pid: 0 for pid in ordered}
+        schedule: List[str] = []
+        for _ in range(total):
+            for pid in ordered:
+                current[pid] += weights[pid]
+            picked = max(current, key=lambda p: current[p])
+            current[picked] -= total
+            schedule.append(picked)
+
+        schedule = self._unclump_seam(schedule)
+
+        boosted = {p: w for p, w in weights.items() if w > 1}
+        logger.info(
+            "Vegas rotation weighted: %d slots for %d plugins (boosted: %s)",
+            len(schedule), len(ordered), boosted)
+        return schedule
+
+    @staticmethod
+    def _unclump_seam(schedule: List[str]) -> List[str]:
+        """Stop the heaviest plugin sitting on both ends of the cycle.
+
+        Smooth Weighted Round-Robin spaces repeats well *within* a pass, but
+        it schedules the heaviest item first and often last too. The strip
+        loops, so those two are neighbours: the one place the marquee shows
+        the same plugin twice running is the seam between cycles.
+
+        Rotating the list cannot fix this. Rotation preserves the cyclic order
+        exactly, so it only moves where the seam is drawn, not the adjacency
+        itself. The trailing entry has to be swapped with one from the middle
+        whose neighbours differ from it, which breaks the pair without
+        creating another.
+
+        Left alone when no such position exists -- a rotation short enough or
+        lopsided enough to have none is one where the plugin is unavoidably
+        adjacent to itself anyway.
+        """
+        if len(schedule) < 3 or schedule[0] != schedule[-1]:
+            return schedule
+
+        repeated = schedule[-1]
+        size = len(schedule)
+
+        def cyclic_doubles(seq) -> int:
+            return sum(1 for i in range(size) if seq[i] == seq[(i + 1) % size])
+
+        def clearance(seq, value) -> int:
+            """Smallest cyclic gap between appearances of `value`."""
+            at = [i for i, v in enumerate(seq) if v == value]
+            if len(at) < 2:
+                return size
+            return min(min((b - a) % size, (a - b) % size)
+                       for i, a in enumerate(at) for b in at[i + 1:])
+
+        # Try each swap and judge the result, rather than reasoning about which
+        # neighbours the two moved elements will end up with. That reasoning is
+        # where the first version went wrong: it guarded the slot `repeated`
+        # moves into but not the one the displaced element lands in, so
+        # ['a','b','c','d','x','y','x','a'] came back ending ['x','x'] -- the
+        # seam duplicate traded for a fresh one.
+        best = None
+        best_clearance = -1
+        for j in range(1, size - 1):
+            candidate = list(schedule)
+            candidate[j], candidate[-1] = candidate[-1], candidate[j]
+            if cyclic_doubles(candidate):
+                continue
+            # Among the repairs that work, prefer the one that leaves the
+            # boosted plugin most evenly spread; taking the first that merely
+            # fits moved a repeat from a gap of 7 into a gap of 2.
+            spread = clearance(candidate, repeated)
+            if spread > best_clearance:
+                best, best_clearance = candidate, spread
+
+        # None exists when the value is unavoidably adjacent to itself -- a
+        # plugin holding most of the slots has to be. Schedule it as it is
+        # rather than refuse.
+        return best if best is not None else schedule
+
     def _prefetch_content(self, count: int = 1) -> None:
         """
         Prefetch content for upcoming plugins.

src/web_interface/api_helpers.py

@@ -30,14 +30,16 @@ def success_response(
     """
     response_data = create_success_response(data, message, metadata)
     
-    # Timing is merged into whatever the caller passed, without inventing a
-    # metadata block for responses that have neither.
-    enriched = dict(metadata) if metadata is not None else {}
-    if hasattr(request, 'start_time'):
-        enriched['response_time_ms'] = int((time.time() - request.start_time) * 1000)
+    # Add request metadata if available
+    if metadata is None:
+        metadata = {}
     
-    if metadata is not None or enriched:
-        response_data['metadata'] = enriched
+    # Add timing if request start time is available
+    if hasattr(request, 'start_time'):
+        metadata['response_time_ms'] = int((time.time() - request.start_time) * 1000)
+    
+    if metadata:
+        response_data['metadata'] = metadata
     
     return jsonify(response_data)
 

src/web_interface/error_handler.py

@@ -77,6 +77,25 @@ def describe_exception(exc: BaseException,
     """
     message = str(exc).strip()
     text = f"{type(exc).__name__}: {message}" if message else type(exc).__name__
+    return redact_text(text, max_length)
+
+
+def redact_text(text: str, max_length: int = _MAX_DETAIL_LENGTH) -> str:
+    """Make arbitrary text safe to hand back over HTTP.
+
+    Split out of describe_exception because exceptions are not the only thing
+    worth returning: a subprocess's stderr, or a message a helper script
+    printed, is just as useful to a user and just as capable of carrying a
+    token or a password in it.
+
+    Args:
+        text: The text to redact
+        max_length: Truncate beyond this many characters
+
+    Returns:
+        A single line, credentials replaced, length capped.
+    """
+    text = text or ''
     # Order matters: the URL and header forms are more specific than the
     # generic key=value pattern, which would otherwise chew the scheme.
     text = _REDACT_URL_USERINFO.sub(r'\1<redacted>\3', text)
@@ -142,16 +161,13 @@ def create_success_response(
         "status": "success"
     }
     
-    # All three use `is not None` rather than truthiness: "" and {} are
-    # values a caller chose to send, and dropping them silently would make
-    # the response shape depend on the data.
     if data is not None:
         response["data"] = data
     
-    if message is not None:
+    if message:
         response["message"] = message
     
-    if metadata is not None:
+    if metadata:
         response["metadata"] = metadata
     
     return response

src/web_interface/errors.py

@@ -89,11 +89,7 @@ class WebInterfaceError:
         self.category = category or self._infer_category(error_code)
         self.details = details
         self.context = context or {}
-        # `is None`, not truthiness: an explicit [] means "this caller has
-        # no suggestions to offer", which the default list would override.
-        self.suggested_fixes = (
-            suggested_fixes if suggested_fixes is not None
-            else self._get_default_suggestions(error_code))
+        self.suggested_fixes = suggested_fixes or self._get_default_suggestions(error_code)
         self.original_error = original_error
     
     def _infer_category(self, error_code: ErrorCode) -> ErrorCategory:

src/web_interface/validators.py

@@ -43,15 +43,10 @@ def validate_image_url(url: str) -> Tuple[bool, Optional[str]]:
     if any(handler in url_lower for handler in ['onerror=', 'onload=', 'onclick=']):
         return False, "Event handlers not allowed in URLs"
     
-    # Reject directory traversal anywhere, not only in relative paths:
-    # http://host/../secret is as much a traversal attempt as /../secret.
-    if '..' in url:
-        return False, "Invalid path: directory traversal not allowed"
-
     # Allow relative paths starting with /
     if url.startswith('/'):
-        # // would be a protocol-relative URL, not a local path
-        if url.startswith('//'):
+        # Validate it's a safe relative path (no directory traversal)
+        if '..' in url or url.startswith('//'):
             return False, "Invalid relative path"
         return True, None
     
@@ -109,11 +104,10 @@ def validate_file_upload(filename: str, max_size_mb: int = 10,
     if '..' in filename or '/' in filename or '\\' in filename:
         return False, "Filename contains invalid characters"
     
-    # Check extension if specified. Both sides are lowercased: the caller's
-    # list is as likely to hold '.TTF' as the filename is.
+    # Check extension if specified
     if allowed_extensions:
         file_ext = Path(filename).suffix.lower()
-        if file_ext not in [ext.lower() for ext in allowed_extensions]:
+        if file_ext not in allowed_extensions:
             return False, f"File extension must be one of: {', '.join(allowed_extensions)}"
     
     return True, None
@@ -153,8 +147,7 @@ def validate_numeric_range(value: float, min_val: Optional[float] = None,
     Returns:
         Tuple of (is_valid, error_message)
     """
-    # bool is an int subclass, so True would otherwise validate as 1.
-    if not isinstance(value, (int, float)) or isinstance(value, bool):
+    if not isinstance(value, (int, float)):
         return False, "Value must be a number"
     
     if min_val is not None and value < min_val:
@@ -190,15 +183,7 @@ def validate_string_length(text: str, min_length: Optional[int] = None,
 
 def sanitize_plugin_config(config: dict) -> dict:
     """
-    Restrict a plugin config to safe key names and value types.
-
-    Drops keys that are not plain identifiers and values that are not
-    JSON-ish scalars, lists, or dicts, recursing into the latter two.
-
-    String values are returned **unescaped**: output escaping is the
-    template layer's job, and escaping here would store the escaped form
-    in config.json. Do not read this function as XSS protection for
-    rendered output.
+    Sanitize plugin configuration input to prevent injection.
     
     Args:
         config: Configuration dictionary

systemd/ledmatrix-wifi-monitor.service

@@ -10,8 +10,8 @@ WorkingDirectory=__PROJECT_ROOT_DIR__
 ExecStart=/usr/bin/python3 __PROJECT_ROOT_DIR__/scripts/utils/wifi_monitor_daemon.py --interval 30
 Restart=on-failure
 RestartSec=10
-StandardOutput=syslog
-StandardError=syslog
+StandardOutput=journal
+StandardError=journal
 SyslogIdentifier=ledmatrix-wifi-monitor
 
 [Install]

systemd/ledmatrix.service

@@ -9,8 +9,25 @@ User=root
 WorkingDirectory=__PROJECT_ROOT_DIR__
 Environment=PYTHONDONTWRITEBYTECODE=1
 ExecStart=/usr/bin/python3 __PROJECT_ROOT_DIR__/run.py
-Restart=on-failure
+# Restart=always, not on-failure: run.py exiting 0 (a clean shutdown path taken
+# for a reason that no longer applies, e.g. a config reload) would otherwise leave
+# the service stopped and the panel dark indefinitely, with systemd considering
+# that a successful outcome and never bringing it back.
+Restart=always
 RestartSec=10
+# Memory ceiling as a share of physical RAM, so one unit file suits a 512 MB
+# Pi Zero 2 W and an 8 GB Pi 5 alike. This is a backstop, not a tuning knob: it
+# turns "the board runs out of memory, stops being able to fork, and takes sshd
+# and the panel down together until someone pulls the plug" into "this one
+# service restarts".
+#
+# NOTE: Raspberry Pi firmware boots the kernel with cgroup_disable=memory, and
+# systemd accepts this setting and then silently ignores it. Verify with:
+#     grep memory /sys/fs/cgroup/cgroup.controllers
+# If that prints nothing, add "cgroup_enable=memory cgroup_memory=1" to
+# /boot/firmware/cmdline.txt (all on line 1) and reboot. first_time_install.sh
+# does this for you.
+MemoryMax=85%
 StandardOutput=journal
 StandardError=journal
 SyslogIdentifier=ledmatrix

test/_api_v3_test_helpers.py

@@ -1,75 +0,0 @@
-"""
-Shared scaffolding for api_v3 blueprint tests.
-
-Not a test module (the leading underscore keeps pytest from collecting
-it). It is the pytest-fixture equivalent of ``_make_client()`` in
-test_uninstall_and_reconcile_endpoint.py, which is unittest-style and
-requires ``self.addCleanup``.
-
-The api_v3 blueprint keeps its managers as attributes on a module-level
-singleton, not in Flask app state, so replacing them with mocks leaks
-into every later test that imports api_v3 unless the originals are put
-back. ``api_v3_client`` snapshots and restores them around each test.
-"""
-
-from unittest.mock import MagicMock
-
-import pytest
-from flask import Flask
-
-
-# Every manager attribute the blueprint reads. Anything missing here keeps
-# whatever a previously-run test left on the singleton.
-API_V3_MANAGER_ATTRS = (
-    'config_manager', 'plugin_manager', 'plugin_store_manager',
-    'plugin_state_manager', 'saved_repositories_manager', 'schema_manager',
-    'operation_queue', 'operation_history', 'cache_manager',
-)
-
-_SENTINEL = object()
-
-
-def build_app(blueprint):
-    app = Flask(__name__)
-    app.config['TESTING'] = True
-    app.config['SECRET_KEY'] = 'test'
-    app.register_blueprint(blueprint, url_prefix='/api/v3')
-    return app
-
-
-@pytest.fixture
-def api_v3_module():
-    """The api_v3 module with every manager replaced by a MagicMock.
-
-    Restores the original attributes afterwards. Tests point individual
-    managers at real objects (a ConfigManager over tmp_path, say) or set
-    them to None to exercise the not-initialized branches.
-    """
-    from web_interface.blueprints import api_v3 as module
-
-    originals = {
-        name: getattr(module.api_v3, name, _SENTINEL)
-        for name in API_V3_MANAGER_ATTRS
-    }
-    for name in API_V3_MANAGER_ATTRS:
-        setattr(module.api_v3, name, MagicMock())
-    # Default to the direct path; queue tests opt in explicitly.
-    module.api_v3.operation_queue = None
-
-    yield module
-
-    for name, original in originals.items():
-        if original is _SENTINEL:
-            if hasattr(module.api_v3, name):
-                try:
-                    delattr(module.api_v3, name)
-                except AttributeError:
-                    pass
-        else:
-            setattr(module.api_v3, name, original)
-
-
-@pytest.fixture
-def api_v3_client(api_v3_module):
-    """Flask test client wired to the mocked blueprint."""
-    return build_app(api_v3_module.api_v3).test_client()

test/test_api_v3_calendar_credentials.py

@@ -1,226 +0,0 @@
-"""
-Endpoint tests for POST /plugins/calendar/upload-credentials.
-
-The endpoint takes an uploaded Google OAuth credentials file, writes it
-into the calendar plugin's directory as credentials.json at mode 0600, and
-copies any previous file aside first. It had no tests.
-
-Regression coverage for two fixed bugs:
-- The OAuth-shape check sat inside `except Exception: pass`, so a valid
-  JSON document that is not an object — a bare `42`, a list, a string —
-  raised TypeError on the membership test, was swallowed, and got saved
-  as credentials.json anyway.
-- Each overwrite created a timestamped backup and nothing ever removed
-  them, so every re-upload left another complete copy of the user's OAuth
-  client credentials in the plugin directory, indefinitely.
-"""
-
-import io
-import json
-import os
-import stat
-import sys
-import time
-from pathlib import Path
-from types import SimpleNamespace
-
-import pytest
-
-sys.path.insert(0, str(Path(__file__).parent.parent))
-
-from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
-
-URL = "/api/v3/plugins/calendar/upload-credentials"
-
-VALID_CREDENTIALS = {
-    "installed": {
-        "client_id": "abc.apps.googleusercontent.com",
-        "client_secret": "shh",
-        "redirect_uris": ["http://localhost"],
-    }
-}
-
-
-@pytest.fixture
-def plugin_dir(tmp_path, api_v3_module):
-    directory = tmp_path / "plugins" / "calendar"
-    directory.mkdir(parents=True)
-    api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(directory)
-    return directory
-
-
-def upload(client, content, filename="credentials.json"):
-    # bytes are sent verbatim (to exercise malformed input); anything else
-    # is serialized, so None becomes the JSON literal null rather than an
-    # empty body.
-    payload = content if isinstance(content, bytes) else json.dumps(content).encode()
-    return client.post(
-        URL,
-        data={"file": (io.BytesIO(payload), filename)},
-        content_type="multipart/form-data",
-    )
-
-
-def backups(plugin_dir):
-    return sorted(plugin_dir.glob("credentials.json.backup.*"))
-
-
-class TestRequestValidation:
-    def test_no_file_part_is_a_400(self, api_v3_client, plugin_dir):
-        response = api_v3_client.post(URL, data={}, content_type="multipart/form-data")
-        assert response.status_code == 400
-        assert "No file provided" in response.get_json()["message"]
-
-    def test_empty_filename_is_a_400(self, api_v3_client, plugin_dir):
-        response = upload(api_v3_client, VALID_CREDENTIALS, filename="")
-        assert response.status_code == 400
-
-    @pytest.mark.parametrize("filename", ["creds.txt", "creds.pem", "creds"])
-    def test_non_json_extension_is_a_400(self, api_v3_client, plugin_dir, filename):
-        response = upload(api_v3_client, VALID_CREDENTIALS, filename=filename)
-        assert response.status_code == 400
-        assert "JSON file" in response.get_json()["message"]
-
-    def test_uppercase_json_extension_accepted(self, api_v3_client, plugin_dir):
-        assert upload(api_v3_client, VALID_CREDENTIALS,
-                      filename="CREDENTIALS.JSON").status_code == 200
-
-    def test_oversized_file_is_a_400(self, api_v3_client, plugin_dir):
-        response = upload(api_v3_client, b"x" * (1024 * 1024 + 1))
-        assert response.status_code == 400
-        assert "1MB" in response.get_json()["message"]
-        assert not (plugin_dir / "credentials.json").exists()
-
-    def test_invalid_json_is_a_400(self, api_v3_client, plugin_dir):
-        response = upload(api_v3_client, b"{not json")
-        assert response.status_code == 400
-        assert "not valid JSON" in response.get_json()["message"]
-        assert not (plugin_dir / "credentials.json").exists()
-
-    def test_missing_plugin_directory_is_a_404(self, api_v3_client, api_v3_module, tmp_path):
-        api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(
-            tmp_path / "not-installed")
-        assert upload(api_v3_client, VALID_CREDENTIALS).status_code == 404
-
-
-class TestOAuthShapeValidation:
-    def test_installed_key_accepted(self, api_v3_client, plugin_dir):
-        assert upload(api_v3_client, VALID_CREDENTIALS).status_code == 200
-
-    def test_web_key_accepted(self, api_v3_client, plugin_dir):
-        assert upload(api_v3_client, {"web": {"client_id": "x"}}).status_code == 200
-
-    def test_object_without_oauth_keys_is_a_400(self, api_v3_client, plugin_dir):
-        response = upload(api_v3_client, {"something": "else"})
-        assert response.status_code == 400
-        assert "valid Google OAuth" in response.get_json()["message"]
-        assert not (plugin_dir / "credentials.json").exists()
-
-    @pytest.mark.parametrize("content", [42, "a string", [1, 2, 3], True, None])
-    def test_valid_json_that_is_not_an_object_is_rejected(
-            self, api_v3_client, plugin_dir, content):
-        # Regression: `'installed' not in 42` raises TypeError, which the
-        # bare `except Exception: pass` swallowed — the file was then saved
-        # as credentials.json despite being unusable as credentials.
-        response = upload(api_v3_client, content)
-        assert response.status_code == 400
-        assert "valid Google OAuth" in response.get_json()["message"]
-        assert not (plugin_dir / "credentials.json").exists()
-
-
-class TestSaving:
-    def test_file_written_with_contents_intact(self, api_v3_client, plugin_dir):
-        response = upload(api_v3_client, VALID_CREDENTIALS)
-        assert response.status_code == 200
-        saved = json.loads((plugin_dir / "credentials.json").read_text())
-        assert saved == VALID_CREDENTIALS
-
-    def test_response_reports_the_path(self, api_v3_client, plugin_dir):
-        body = upload(api_v3_client, VALID_CREDENTIALS).get_json()
-        assert body["path"].endswith("credentials.json")
-
-    def test_permissions_are_owner_only(self, api_v3_client, plugin_dir):
-        upload(api_v3_client, VALID_CREDENTIALS)
-        mode = stat.S_IMODE((plugin_dir / "credentials.json").stat().st_mode)
-        assert mode == 0o600
-
-    def test_first_upload_creates_no_backup(self, api_v3_client, plugin_dir):
-        upload(api_v3_client, VALID_CREDENTIALS)
-        assert backups(plugin_dir) == []
-
-    def test_overwrite_backs_up_the_previous_file(self, api_v3_client, plugin_dir):
-        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"old": 1}}))
-        upload(api_v3_client, VALID_CREDENTIALS)
-        assert len(backups(plugin_dir)) == 1
-        assert json.loads(backups(plugin_dir)[0].read_text()) == {"installed": {"old": 1}}
-        assert json.loads((plugin_dir / "credentials.json").read_text()) == VALID_CREDENTIALS
-
-
-class TestBackupPruning:
-    def _seed(self, plugin_dir, count):
-        """Create `count` backups with distinct, increasing mtimes."""
-        now = int(time.time())
-        for i in range(count):
-            path = plugin_dir / f"credentials.json.backup.{now - (count - i) * 10}"
-            path.write_text(json.dumps({"installed": {"gen": i}}))
-            os.utime(path, (now - (count - i) * 10, now - (count - i) * 10))
-
-    def test_old_backups_are_pruned(self, api_v3_client, plugin_dir):
-        # Regression: nothing ever removed these, so a plugin directory
-        # accumulated one full copy of the user's OAuth credentials per
-        # re-upload, forever.
-        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
-        self._seed(plugin_dir, 7)
-        assert len(backups(plugin_dir)) == 7
-
-        upload(api_v3_client, VALID_CREDENTIALS)
-        assert len(backups(plugin_dir)) == 5
-
-    def test_the_newest_backups_are_the_ones_kept(self, api_v3_client, plugin_dir):
-        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
-        self._seed(plugin_dir, 7)
-
-        upload(api_v3_client, VALID_CREDENTIALS)
-        remaining = backups(plugin_dir)
-        # The just-created backup (of "cur") plus the four newest seeds.
-        contents = [json.loads(p.read_text()) for p in remaining]
-        assert {"installed": {"cur": 1}} in contents
-        assert {"installed": {"gen": 0}} not in contents  # oldest seed gone
-
-    def test_under_the_limit_nothing_is_removed(self, api_v3_client, plugin_dir):
-        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
-        self._seed(plugin_dir, 2)
-        upload(api_v3_client, VALID_CREDENTIALS)
-        assert len(backups(plugin_dir)) == 3  # 2 seeded + 1 new
-
-    def test_repeated_uploads_stay_bounded(
-            self, api_v3_client, plugin_dir, api_v3_module, monkeypatch):
-        # The backup filename carries int(time.time()), so uploads inside
-        # the same second all write the same name and overwrite each other.
-        # Advance a fake clock a second per round — otherwise this never
-        # reaches six backups and the bound holds for the wrong reason.
-        clock = {"now": int(time.time())}
-        monkeypatch.setattr(
-            api_v3_module, "time", SimpleNamespace(time=lambda: clock["now"]))
-        for i in range(10):
-            clock["now"] += 1
-            upload(api_v3_client, {"installed": {"round": i}})
-            os.utime(plugin_dir / "credentials.json",
-                     (clock["now"], clock["now"]))
-        remaining = backups(plugin_dir)
-        assert len(remaining) == 5
-        # And they are the five most recent rounds, not an arbitrary five.
-        kept = sorted(int(p.name.rsplit(".", 1)[1]) for p in remaining)
-        assert kept == [clock["now"] - 4 + i for i in range(5)]
-
-    def test_unremovable_backup_does_not_fail_the_upload(
-            self, api_v3_client, plugin_dir, monkeypatch):
-        (plugin_dir / "credentials.json").write_text(json.dumps({"installed": {"cur": 1}}))
-        self._seed(plugin_dir, 7)
-
-        def refuse(self):
-            raise OSError("read-only filesystem")
-        monkeypatch.setattr(Path, "unlink", refuse)
-
-        # Pruning is housekeeping; failing it must not lose the upload.
-        assert upload(api_v3_client, VALID_CREDENTIALS).status_code == 200

test/test_api_v3_music_auth_endpoints.py

@@ -1,302 +0,0 @@
-"""
-Endpoint tests for /plugins/authenticate/spotify and .../ytm.
-
-The Spotify step-2 handler writes a Python wrapper script to a temp file
-with the user's redirect URL embedded in it, then runs that file through
-subprocess. That is the most dangerous shape in the blueprint and had no
-tests: the URL is user input reaching generated source code.
-
-The two endpoints are NOT symmetrical, despite the matching names. Only
-Spotify has a two-step flow, a wrapper script, and a redirect_url; YTM
-just runs its script directly.
-
-Regression coverage for one fixed bug: the wrapper file was unlinked in
-the success/failure branch and again in the TimeoutExpired handler, so
-any other failure from subprocess.run — the interpreter missing, a fork
-failure, an interrupted call — left a temp file containing the user's
-redirect URL behind.
-"""
-
-import ast
-import json
-import os
-import subprocess
-import sys
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-sys.path.insert(0, str(Path(__file__).parent.parent))
-
-from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
-
-
-@pytest.fixture
-def plugin_dir(tmp_path, api_v3_module):
-    """A plugin directory containing both auth scripts."""
-    directory = tmp_path / "plugins" / "ledmatrix-music"
-    directory.mkdir(parents=True)
-    (directory / "authenticate_spotify.py").write_text("print('spotify')\n")
-    (directory / "authenticate_ytm.py").write_text("print('ytm')\n")
-    api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(directory)
-    return directory
-
-
-def completed(returncode=0, stdout="ok", stderr=""):
-    return subprocess.CompletedProcess(
-        args=["python3"], returncode=returncode, stdout=stdout, stderr=stderr)
-
-
-class TestSpotifyPreconditions:
-    URL = "/api/v3/plugins/authenticate/spotify"
-
-    def test_missing_plugin_directory_is_404(self, api_v3_client, api_v3_module, tmp_path):
-        api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(
-            tmp_path / "not-installed")
-        response = api_v3_client.post(self.URL, json={})
-        assert response.status_code == 404
-        assert response.get_json()["message"] == "Plugin not found"
-
-    def test_none_plugin_directory_is_404(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = None
-        assert api_v3_client.post(self.URL, json={}).status_code == 404
-
-    def test_missing_auth_script_is_404(self, api_v3_client, plugin_dir):
-        (plugin_dir / "authenticate_spotify.py").unlink()
-        response = api_v3_client.post(self.URL, json={})
-        assert response.status_code == 404
-        assert "script not found" in response.get_json()["message"]
-
-
-class TestSpotifyStepTwo:
-    """redirect_url present — the wrapper-script path."""
-
-    URL = "/api/v3/plugins/authenticate/spotify"
-
-    def test_success(self, api_v3_client, plugin_dir):
-        with patch.object(subprocess, "run", return_value=completed(0, "done")):
-            response = api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
-        assert response.status_code == 200
-        body = response.get_json()
-        assert body["status"] == "success"
-        assert body["output"] == "done"
-
-    def test_script_failure_is_a_400_with_combined_output(self, api_v3_client, plugin_dir):
-        with patch.object(subprocess, "run", return_value=completed(1, "out", "err")):
-            response = api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
-        assert response.status_code == 400
-        assert response.get_json()["output"] == "outerr"
-
-    def test_timeout_is_a_408(self, api_v3_client, plugin_dir):
-        with patch.object(subprocess, "run",
-                          side_effect=subprocess.TimeoutExpired("python3", 120)):
-            response = api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
-        assert response.status_code == 408
-        assert "timed out" in response.get_json()["message"]
-
-    def test_runs_a_list_argv_never_a_shell(self, api_v3_client, plugin_dir):
-        with patch.object(subprocess, "run", return_value=completed()) as run:
-            api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
-        args, kwargs = run.call_args
-        assert isinstance(args[0], list)
-        assert args[0][0] == "python3"
-        assert kwargs.get("shell") in (None, False)
-
-    def test_timeout_is_bounded(self, api_v3_client, plugin_dir):
-        with patch.object(subprocess, "run", return_value=completed()) as run:
-            api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
-        assert run.call_args.kwargs["timeout"] == 120
-
-
-class TestSpotifyWrapperCleanup:
-    URL = "/api/v3/plugins/authenticate/spotify"
-
-    def _wrapper_paths_after(self, api_v3_client, run_mock):
-        """Run the endpoint and return the wrapper path subprocess saw."""
-        seen = {}
-
-        def capture(args, **kwargs):
-            seen["path"] = args[1]
-            return run_mock(args, **kwargs)
-
-        with patch.object(subprocess, "run", side_effect=capture):
-            api_v3_client.post(self.URL, json={"redirect_url": "http://cb/?code=x"})
-        return seen["path"]
-
-    def test_removed_after_success(self, api_v3_client, plugin_dir):
-        path = self._wrapper_paths_after(api_v3_client, lambda *a, **kw: completed())
-        assert not os.path.exists(path)
-
-    def test_removed_after_script_failure(self, api_v3_client, plugin_dir):
-        path = self._wrapper_paths_after(
-            api_v3_client, lambda *a, **kw: completed(1, "out", "err"))
-        assert not os.path.exists(path)
-
-    def test_removed_after_timeout(self, api_v3_client, plugin_dir):
-        def raise_timeout(*a, **kw):
-            raise subprocess.TimeoutExpired("python3", 120)
-        path = self._wrapper_paths_after(api_v3_client, raise_timeout)
-        assert not os.path.exists(path)
-
-    def test_removed_when_subprocess_cannot_start(self, api_v3_client, plugin_dir):
-        # Regression: cleanup lived in the success/failure branch and in the
-        # TimeoutExpired handler only. An OSError from subprocess.run itself
-        # — no interpreter, fork failure — skipped both and left the wrapper,
-        # which contains the user's redirect URL, on disk.
-        def raise_oserror(*a, **kw):
-            raise OSError("[Errno 12] Cannot allocate memory")
-        path = self._wrapper_paths_after(api_v3_client, raise_oserror)
-        assert not os.path.exists(path)
-
-
-class TestSpotifyRedirectUrlIsNotInjectable:
-    """The wrapper embeds redirect_url into generated Python source."""
-
-    URL = "/api/v3/plugins/authenticate/spotify"
-
-    ADVERSARIAL = [
-        '''http://cb/?code=x"''',
-        """http://cb/?code=x'""",
-        'http://cb/?code=x\\',
-        'http://cb/?code=x\nimport os; os.system("id")',
-        'http://cb/?code=x"""\nimport os\n"""',
-        "http://cb/?code=x'''",
-        'http://cb/?code=x\\"\\n',
-        '"; import os; os.system("id"); "',
-    ]
-
-    def _wrapper_source(self, api_v3_client, redirect_url):
-        captured = {}
-
-        def capture(args, **kwargs):
-            captured["source"] = Path(args[1]).read_text()
-            return completed()
-
-        with patch.object(subprocess, "run", side_effect=capture):
-            api_v3_client.post(self.URL, json={"redirect_url": redirect_url})
-        return captured["source"]
-
-    @pytest.mark.parametrize("redirect_url", ADVERSARIAL)
-    def test_wrapper_is_still_valid_python(self, api_v3_client, plugin_dir, redirect_url):
-        # If escaping failed, the generated file would not parse at all.
-        source = self._wrapper_source(api_v3_client, redirect_url)
-        ast.parse(source)
-
-    @pytest.mark.parametrize("redirect_url", ADVERSARIAL)
-    def test_url_survives_as_one_string_literal(
-            self, api_v3_client, plugin_dir, redirect_url):
-        # Stronger than "it parses": the URL must still be a single string
-        # assigned to redirect_url, not code that escaped into statements.
-        source = self._wrapper_source(api_v3_client, redirect_url)
-        tree = ast.parse(source)
-        assigned = [
-            node.value.value for node in ast.walk(tree)
-            if isinstance(node, ast.Assign)
-            and isinstance(node.value, ast.Constant)
-            and any(getattr(t, "id", None) == "redirect_url" for t in node.targets)
-        ]
-        assert assigned == [redirect_url.strip()]
-
-    def test_injected_call_does_not_become_a_statement(self, api_v3_client, plugin_dir):
-        source = self._wrapper_source(
-            api_v3_client, 'http://cb/\nimport os; os.system("id")')
-        tree = ast.parse(source)
-        imported = {
-            alias.name for node in ast.walk(tree)
-            if isinstance(node, ast.Import) for alias in node.names
-        }
-        # The wrapper legitimately imports sys, subprocess and os; what it
-        # must not gain is a *call* smuggled in through the URL.
-        calls = [
-            node for node in ast.walk(tree)
-            if isinstance(node, ast.Call)
-            and isinstance(node.func, ast.Attribute)
-            and node.func.attr == "system"
-        ]
-        assert calls == []
-
-
-class TestSpotifyStepOne:
-    """No redirect_url — the OAuth-URL path, which imports the script."""
-
-    URL = "/api/v3/plugins/authenticate/spotify"
-
-    def test_script_without_credentials_helper_is_an_error(
-            self, api_v3_client, plugin_dir):
-        # The stub script defines neither get_auth_url nor
-        # load_spotify_credentials, so no URL can be produced.
-        response = api_v3_client.post(self.URL, json={})
-        assert response.status_code in (400, 500)
-        assert response.get_json()["status"] == "error"
-
-    def test_unusable_credentials_do_not_leak_into_the_response(
-            self, api_v3_client, plugin_dir):
-        (plugin_dir / "authenticate_spotify.py").write_text(
-            "def load_spotify_credentials():\n"
-            "    return ('id-abc', 'super-secret-value', None)\n"
-        )
-        response = api_v3_client.post(self.URL, json={})
-        assert "super-secret-value" not in response.get_data(as_text=True)
-
-    def test_script_raising_on_import_is_handled(self, api_v3_client, plugin_dir):
-        (plugin_dir / "authenticate_spotify.py").write_text("raise RuntimeError('boom')\n")
-        response = api_v3_client.post(self.URL, json={})
-        assert response.status_code == 500
-        assert response.get_json()["status"] == "error"
-
-    def test_bodyless_post_reaches_step_one(self, api_v3_client, plugin_dir):
-        # Covered by the silent=True fix: previously a 500 from body parsing.
-        response = api_v3_client.post(self.URL)
-        assert response.status_code in (400, 500)
-        assert response.get_json()["status"] == "error"
-
-    def test_whitespace_redirect_url_is_treated_as_absent(
-            self, api_v3_client, plugin_dir):
-        with patch.object(subprocess, "run", return_value=completed()) as run:
-            api_v3_client.post(self.URL, json={"redirect_url": "   "})
-        # Step 2 never runs, so no wrapper is executed.
-        run.assert_not_called()
-
-
-class TestYouTubeMusic:
-    """No wrapper script and no redirect_url — deliberately not symmetric."""
-
-    URL = "/api/v3/plugins/authenticate/ytm"
-
-    def test_missing_plugin_directory_is_404(self, api_v3_client, api_v3_module, tmp_path):
-        api_v3_module.api_v3.plugin_manager.get_plugin_directory.return_value = str(
-            tmp_path / "not-installed")
-        assert api_v3_client.post(self.URL).status_code == 404
-
-    def test_missing_script_is_404(self, api_v3_client, plugin_dir):
-        (plugin_dir / "authenticate_ytm.py").unlink()
-        response = api_v3_client.post(self.URL)
-        assert response.status_code == 404
-        assert "script not found" in response.get_json()["message"]
-
-    def test_success(self, api_v3_client, plugin_dir):
-        with patch.object(subprocess, "run", return_value=completed(0, "authorized")):
-            response = api_v3_client.post(self.URL)
-        assert response.status_code == 200
-        assert response.get_json()["output"] == "authorized"
-
-    def test_failure_is_a_400_with_combined_output(self, api_v3_client, plugin_dir):
-        with patch.object(subprocess, "run", return_value=completed(1, "out", "err")):
-            response = api_v3_client.post(self.URL)
-        assert response.status_code == 400
-        assert response.get_json()["output"] == "outerr"
-
-    def test_timeout_is_a_408(self, api_v3_client, plugin_dir):
-        with patch.object(subprocess, "run",
-                          side_effect=subprocess.TimeoutExpired("python3", 60)):
-            assert api_v3_client.post(self.URL).status_code == 408
-
-    def test_runs_the_script_directly_without_a_shell(self, api_v3_client, plugin_dir):
-        with patch.object(subprocess, "run", return_value=completed()) as run:
-            api_v3_client.post(self.URL)
-        args, kwargs = run.call_args
-        assert args[0][0] == "python3"
-        assert args[0][1].endswith("authenticate_ytm.py")
-        assert kwargs.get("shell") in (None, False)
-        assert kwargs["timeout"] == 60

test/test_api_v3_optional_body.py

@@ -1,136 +0,0 @@
-"""
-Regression tests: POST endpoints whose body is optional must accept a
-request that has no body at all.
-
-Six handlers in api_v3 read their body as ``request.get_json() or {}``.
-The ``or {}`` states the intent plainly — every field is optional, so a
-bodyless POST should fall back to defaults. But ``get_json()`` without
-``silent=True`` raises ``UnsupportedMediaType`` when the request carries
-no JSON Content-Type, and it raises *before* ``or {}`` is evaluated. Each
-handler's catch-all then turned that into a 500.
-
-So the natural way to call these endpoints — a POST with no body, which
-is what curl, a fetch() without options, and most HTTP clients send by
-default — failed on every one of them. The shipped UI always sends a JSON
-object, which is why this went unnoticed.
-
-This file covers the endpoints whose bodyless behaviour is not already
-tested in their own suite.
-"""
-
-import re
-import sys
-from pathlib import Path
-from unittest.mock import MagicMock
-
-import pytest
-
-sys.path.insert(0, str(Path(__file__).parent.parent))
-
-from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
-
-
-class TestOnDemandStart:
-    URL = "/api/v3/display/on-demand/start"
-
-    def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module):
-        response = api_v3_client.post(self.URL)
-        # The endpoint may still reject the request on its own terms (no
-        # plugin_id, nothing to display); what it must not do is fail with
-        # a 500 raised out of body parsing.
-        assert response.status_code != 500
-
-    def test_json_body_still_works(self, api_v3_client, api_v3_module):
-        assert api_v3_client.post(self.URL, json={}).status_code != 500
-
-
-class TestResetPluginConfig:
-    URL = "/api/v3/plugins/config/reset"
-
-    def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module):
-        assert api_v3_client.post(self.URL).status_code != 500
-
-    def test_json_body_still_works(self, api_v3_client, api_v3_module):
-        assert api_v3_client.post(self.URL, json={}).status_code != 500
-
-
-class TestDeleteOfTheDayJson:
-    URL = "/api/v3/plugins/of-the-day/json/delete"
-
-    def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module):
-        assert api_v3_client.post(self.URL).status_code != 500
-
-    def test_json_body_still_works(self, api_v3_client, api_v3_module):
-        assert api_v3_client.post(self.URL, json={}).status_code != 500
-
-
-class TestPluginLimits:
-    URL = "/api/v3/plugins/clock/limits"
-
-    def test_bodyless_post_is_not_a_server_error(self, api_v3_client, api_v3_module):
-        assert api_v3_client.post(self.URL).status_code != 500
-
-
-class TestMissingBodyGivesTheDeclaredError:
-    """Handlers that answer "No data provided" must actually be able to.
-
-    A second group of handlers reads `data = request.get_json()` and then
-    guards with `if not data: return 400`. That guard is unreachable for a
-    request with no JSON body, because get_json() raises first — so the
-    caller got a 500 "an error occurred; see logs for details" instead of
-    the 400 the handler plainly intends to send.
-    """
-
-    @pytest.mark.parametrize("url", [
-        "/api/v3/plugins/install",
-        "/api/v3/plugins/install-from-url",
-        "/api/v3/plugins/registry-from-url",
-        "/api/v3/config/raw/main",
-        "/api/v3/config/raw/secrets",
-        "/api/v3/cache/delete",
-    ])
-    def test_bodyless_post_gets_a_400_not_a_500(self, api_v3_client, api_v3_module, url):
-        response = api_v3_client.post(url)
-        assert response.status_code == 400, (
-            f"{url} answered {response.status_code}: "
-            f"{response.get_data(as_text=True)[:200]}")
-
-    @pytest.mark.parametrize("url", [
-        "/api/v3/plugins/install",
-        "/api/v3/config/raw/main",
-    ])
-    def test_malformed_json_gets_a_400_not_a_500(self, api_v3_client, api_v3_module, url):
-        response = api_v3_client.post(
-            url, data="{not json", content_type="application/json")
-        assert response.status_code == 400
-
-
-class TestNoBodyReadContradictsItsOwnGuard:
-    SOURCE = Path(__file__).parent.parent / "web_interface/blueprints/api_v3.py"
-
-    def test_no_or_default_read_is_unguarded(self):
-        """`get_json() or <default>` is a contradiction without silent=True.
-
-        Writing `or {}` declares the body optional; omitting silent=True
-        means the call raises before the default can apply.
-        """
-        offenders = [
-            line.strip() for line in self.SOURCE.read_text().splitlines()
-            if "request.get_json()" in line and " or " in line
-        ]
-        assert offenders == [], (
-            "these reads declare a default but raise before reaching it; "
-            f"use get_json(silent=True): {offenders}")
-
-    def test_no_not_data_guard_is_unreachable(self):
-        """A `if not data:` guard needs a read that can actually return None."""
-        lines = self.SOURCE.read_text().splitlines()
-        offenders = []
-        for i, line in enumerate(lines):
-            if re.search(r"=\s*request\.get_json\(\)\s*$", line):
-                window = "\n".join(lines[i + 1:i + 3])
-                if re.search(r"if\s+(not\s+data\b|data\s+is\s+None)", window):
-                    offenders.append(f"line {i + 1}: {line.strip()}")
-        assert offenders == [], (
-            "these handlers guard on a missing body but raise before the "
-            f"guard runs; use get_json(silent=True): {offenders}")

test/test_api_v3_plugin_install_endpoints.py

@@ -1,302 +0,0 @@
-"""
-Endpoint tests for POST /plugins/install and POST /plugins/install-from-url.
-
-Both were only ever tested at the PluginStoreManager layer, so the route
-logic — the queue-vs-direct branch, schema invalidation, plugin discovery,
-state and history recording — was unexercised.
-
-/plugins/install carries the same install logic twice: once inside the
-operation-queue callback and once in the direct fallback. The paired
-tests below assert both branches produce the same side effects, so the
-duplication cannot quietly drift.
-"""
-
-import sys
-from pathlib import Path
-from unittest.mock import MagicMock
-
-import pytest
-
-sys.path.insert(0, str(Path(__file__).parent.parent))
-
-from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
-
-INSTALL = "/api/v3/plugins/install"
-FROM_URL = "/api/v3/plugins/install-from-url"
-
-
-@pytest.fixture
-def queued(api_v3_module):
-    """Enable the operation queue and run its callback synchronously."""
-    queue = MagicMock()
-
-    def enqueue(operation_type, plugin_id, operation_callback=None):
-        queue.callback_result = operation_callback(MagicMock())
-        return "op-123"
-
-    queue.enqueue_operation.side_effect = enqueue
-    api_v3_module.api_v3.operation_queue = queue
-    return queue
-
-
-def side_effects(module):
-    """The manager calls a successful install is expected to make."""
-    api = module.api_v3
-    return {
-        "schema_invalidated": api.schema_manager.invalidate_cache.call_args_list,
-        "discovered": api.plugin_manager.discover_plugins.call_count,
-        "loaded": api.plugin_manager.load_plugin.call_args_list,
-        "state_set": api.plugin_state_manager.set_plugin_installed.call_args_list,
-        "history": api.operation_history.record_operation.call_args_list,
-    }
-
-
-class TestInstallValidation:
-    def test_uninitialized_store_manager_is_a_500(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager = None
-        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
-        assert response.status_code == 500
-        assert "not initialized" in response.get_json()["message"]
-
-    def test_missing_plugin_id_is_a_400(self, api_v3_client, api_v3_module):
-        response = api_v3_client.post(INSTALL, json={})
-        assert response.status_code == 400
-        assert "plugin_id required" in response.get_json()["message"]
-        api_v3_module.api_v3.plugin_store_manager.install_plugin.assert_not_called()
-
-    def test_empty_body_is_a_400(self, api_v3_client, api_v3_module):
-        assert api_v3_client.post(INSTALL, json=None).status_code == 400
-
-
-class TestInstallDirectPath:
-    """operation_queue is None — the fallback branch."""
-
-    def test_success(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
-        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
-        assert response.status_code == 200
-        assert response.get_json()["status"] == "success"
-
-    def test_success_side_effects(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
-        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
-        effects = side_effects(api_v3_module)
-        assert effects["schema_invalidated"] == [(("clock",), {})]
-        assert effects["discovered"] == 1
-        assert effects["loaded"] == [(("clock",), {})]
-        assert effects["state_set"] == [(("clock",), {})]
-        assert effects["history"][0].kwargs["status"] == "success"
-
-    def test_branch_forwarded_to_the_manager(self, api_v3_client, api_v3_module):
-        manager = api_v3_module.api_v3.plugin_store_manager
-        manager.install_plugin.return_value = True
-        api_v3_client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
-        manager.install_plugin.assert_called_once_with("clock", branch="dev")
-
-    def test_branch_named_in_the_message(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
-        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
-        assert "(branch: dev)" in response.get_json()["message"]
-
-    def test_failure_is_a_500(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
-        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
-        assert response.status_code == 500
-        assert "Failed to install" in response.get_json()["message"]
-
-    def test_failure_mentions_missing_registry_entry(self, api_v3_client, api_v3_module):
-        manager = api_v3_module.api_v3.plugin_store_manager
-        manager.install_plugin.return_value = False
-        manager.get_plugin_info.return_value = None
-        response = api_v3_client.post(INSTALL, json={"plugin_id": "ghost"})
-        assert "not found in registry" in response.get_json()["message"]
-
-    def test_failure_omits_registry_note_when_plugin_is_known(
-            self, api_v3_client, api_v3_module):
-        manager = api_v3_module.api_v3.plugin_store_manager
-        manager.install_plugin.return_value = False
-        manager.get_plugin_info.return_value = {"id": "clock"}
-        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
-        assert "not found in registry" not in response.get_json()["message"]
-
-    def test_failure_recorded_in_history(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
-        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
-        record = api_v3_module.api_v3.operation_history.record_operation.call_args
-        assert record.kwargs["status"] == "failed"
-
-    def test_no_side_effects_on_failure(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
-        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
-        effects = side_effects(api_v3_module)
-        assert effects["schema_invalidated"] == []
-        assert effects["loaded"] == []
-        assert effects["state_set"] == []
-
-
-class TestInstallQueuedPath:
-    """operation_queue present — the callback branch."""
-
-    def test_returns_an_operation_id(self, api_v3_client, api_v3_module, queued):
-        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
-        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
-        assert response.status_code == 200
-        assert response.get_json()["data"]["operation_id"] == "op-123"
-
-    def test_message_says_queued(self, api_v3_client, api_v3_module, queued):
-        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
-        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
-        assert "queued" in response.get_json()["message"]
-
-    def test_callback_success_side_effects(self, api_v3_client, api_v3_module, queued):
-        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
-        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
-        effects = side_effects(api_v3_module)
-        assert effects["schema_invalidated"] == [(("clock",), {})]
-        assert effects["discovered"] == 1
-        assert effects["loaded"] == [(("clock",), {})]
-        assert effects["state_set"] == [(("clock",), {})]
-        assert effects["history"][0].kwargs["status"] == "success"
-
-    def test_callback_reports_success(self, api_v3_client, api_v3_module, queued):
-        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
-        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
-        assert queued.callback_result["success"] is True
-
-    def test_callback_failure_raises_for_the_queue(self, api_v3_client, api_v3_module, queued):
-        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
-        # The callback signals failure by raising, so the queue can mark the
-        # operation failed; the route's catch-all turns it into a 500.
-        response = api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
-        assert response.status_code == 500
-
-    def test_callback_failure_recorded_in_history(self, api_v3_client, api_v3_module, queued):
-        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = False
-        api_v3_client.post(INSTALL, json={"plugin_id": "clock"})
-        record = api_v3_module.api_v3.operation_history.record_operation.call_args
-        assert record.kwargs["status"] == "failed"
-
-    def test_branch_forwarded_from_the_callback(self, api_v3_client, api_v3_module, queued):
-        manager = api_v3_module.api_v3.plugin_store_manager
-        manager.install_plugin.return_value = True
-        api_v3_client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
-        manager.install_plugin.assert_called_once_with("clock", branch="dev")
-
-
-class TestInstallPathsAgree:
-    """The queue callback and the direct fallback duplicate the same logic."""
-
-    def _run(self, client, module, install_ok, queue):
-        module.api_v3.plugin_store_manager.install_plugin.return_value = install_ok
-        client.post(INSTALL, json={"plugin_id": "clock", "branch": "dev"})
-        return side_effects(module)
-
-    def test_success_side_effects_match(self, api_v3_client, api_v3_module):
-        direct = self._run(api_v3_client, api_v3_module, True, None)
-
-        # Reset and re-run through the queue.
-        for mock in (api_v3_module.api_v3.schema_manager,
-                     api_v3_module.api_v3.plugin_manager,
-                     api_v3_module.api_v3.plugin_state_manager,
-                     api_v3_module.api_v3.operation_history):
-            mock.reset_mock()
-        queue = MagicMock()
-        queue.enqueue_operation.side_effect = (
-            lambda t, p, operation_callback=None: operation_callback(MagicMock()) and "op")
-        api_v3_module.api_v3.operation_queue = queue
-        queued = self._run(api_v3_client, api_v3_module, True, queue)
-
-        assert direct["schema_invalidated"] == queued["schema_invalidated"]
-        assert direct["discovered"] == queued["discovered"]
-        assert direct["loaded"] == queued["loaded"]
-        assert direct["state_set"] == queued["state_set"]
-        assert (direct["history"][0].kwargs["status"]
-                == queued["history"][0].kwargs["status"])
-        assert (direct["history"][0].kwargs["details"]
-                == queued["history"][0].kwargs["details"])
-
-    def test_only_the_message_wording_differs(self, api_v3_client, api_v3_module):
-        # Characterized: the direct path says "Plugin installed
-        # successfully" while the queue callback says "Plugin clock
-        # installed successfully". Cosmetic, and the queue's text is
-        # internal to the operation record rather than the HTTP response.
-        api_v3_module.api_v3.plugin_store_manager.install_plugin.return_value = True
-        direct = api_v3_client.post(INSTALL, json={"plugin_id": "clock"}).get_json()
-        assert direct["message"] == "Plugin installed successfully"
-
-
-class TestInstallFromUrl:
-    def test_uninitialized_store_manager_is_a_500(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager = None
-        assert api_v3_client.post(FROM_URL, json={"repo_url": "http://x"}).status_code == 500
-
-    def test_missing_repo_url_is_a_400(self, api_v3_client, api_v3_module):
-        response = api_v3_client.post(FROM_URL, json={})
-        assert response.status_code == 400
-        assert "repo_url required" in response.get_json()["message"]
-
-    def test_success(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
-            "success": True, "plugin_id": "clock", "name": "Clock"}
-        response = api_v3_client.post(FROM_URL, json={"repo_url": "https://github.com/o/r"})
-        assert response.status_code == 200
-        body = response.get_json()
-        assert body["plugin_id"] == "clock"
-        assert body["name"] == "Clock"
-
-    def test_all_optional_arguments_forwarded(self, api_v3_client, api_v3_module):
-        manager = api_v3_module.api_v3.plugin_store_manager
-        manager.install_from_url.return_value = {"success": True, "plugin_id": "clock"}
-        api_v3_client.post(FROM_URL, json={
-            "repo_url": "  https://github.com/o/r  ",
-            "plugin_id": "clock",
-            "plugin_path": "plugins/clock",
-            "branch": "dev",
-        })
-        manager.install_from_url.assert_called_once_with(
-            repo_url="https://github.com/o/r",
-            plugin_id="clock",
-            plugin_path="plugins/clock",
-            branch="dev",
-        )
-
-    def test_success_invalidates_schema_and_loads_plugin(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
-            "success": True, "plugin_id": "clock"}
-        api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
-        api_v3_module.api_v3.schema_manager.invalidate_cache.assert_called_once_with("clock")
-        api_v3_module.api_v3.plugin_manager.load_plugin.assert_called_once_with("clock")
-
-    def test_success_without_plugin_id_skips_discovery(self, api_v3_client, api_v3_module):
-        # install_from_url can succeed without naming the plugin; there is
-        # then nothing to invalidate or load.
-        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
-            "success": True, "plugin_id": None}
-        api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
-        api_v3_module.api_v3.schema_manager.invalidate_cache.assert_not_called()
-        api_v3_module.api_v3.plugin_manager.load_plugin.assert_not_called()
-
-    def test_branch_from_result_included(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
-            "success": True, "plugin_id": "clock", "branch": "dev"}
-        body = api_v3_client.post(FROM_URL, json={"repo_url": "http://x"}).get_json()
-        assert body["branch"] == "dev"
-        assert "(branch: dev)" in body["message"]
-
-    def test_failure_reports_the_managers_error(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
-            "success": False, "error": "repo not found"}
-        response = api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
-        assert response.status_code == 500
-        assert response.get_json()["message"] == "repo not found"
-
-    def test_failure_without_error_uses_fallback_text(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.install_from_url.return_value = {
-            "success": False}
-        response = api_v3_client.post(FROM_URL, json={"repo_url": "http://x"})
-        assert "Failed to install plugin from URL" in response.get_json()["message"]
-
-    def test_manager_exception_is_a_500(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.install_from_url.side_effect = (
-            RuntimeError("boom"))
-        assert api_v3_client.post(FROM_URL, json={"repo_url": "http://x"}).status_code == 500

test/test_api_v3_registry_endpoints.py

@@ -1,179 +0,0 @@
-"""
-Endpoint tests for the plugin-registry routes in api_v3:
-POST /plugins/store/refresh and POST /plugins/registry-from-url.
-
-Both reach out to the network through PluginStoreManager (mocked here) and
-had no endpoint-level coverage; registry-from-url in particular takes a
-user-supplied URL and hands it straight to the manager.
-"""
-
-import sys
-from pathlib import Path
-
-import pytest
-
-sys.path.insert(0, str(Path(__file__).parent.parent))
-
-from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
-
-
-class TestRefreshPluginStore:
-    URL = "/api/v3/plugins/store/refresh"
-
-    def test_uninitialized_manager_is_a_500(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager = None
-        response = api_v3_client.post(self.URL, json={})
-        assert response.status_code == 500
-        assert "not initialized" in response.get_json()["message"]
-
-    def test_success_reports_plugin_count(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {
-            "plugins": [{"id": "a"}, {"id": "b"}, {"id": "c"}]}
-        response = api_v3_client.post(self.URL, json={})
-        assert response.status_code == 200
-        assert response.get_json()["plugin_count"] == 3
-
-    def test_forces_a_refresh_rather_than_using_cache(self, api_v3_client, api_v3_module):
-        manager = api_v3_module.api_v3.plugin_store_manager
-        manager.fetch_registry.return_value = {"plugins": []}
-        api_v3_client.post(self.URL, json={})
-        manager.fetch_registry.assert_called_once_with(force_refresh=True)
-
-    def test_empty_registry_reports_zero(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {}
-        response = api_v3_client.post(self.URL, json={})
-        assert response.get_json()["plugin_count"] == 0
-
-    def test_no_body_is_accepted(self, api_v3_client, api_v3_module):
-        # Regression: `request.get_json() or {}` says a missing body is
-        # fine, but get_json() raises UnsupportedMediaType before `or {}`
-        # is reached, so a bodyless POST — the natural way to call a
-        # refresh endpoint — came back 500.
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
-        assert api_v3_client.post(self.URL).status_code == 200
-
-    def test_body_without_json_content_type_is_accepted(
-            self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
-        response = api_v3_client.post(self.URL, data="", content_type="text/plain")
-        assert response.status_code == 200
-
-    def test_malformed_json_body_falls_back_to_defaults(
-            self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
-        response = api_v3_client.post(
-            self.URL, data="{not json", content_type="application/json")
-        assert response.status_code == 200
-
-    @pytest.mark.parametrize("key", ["fetch_commit_info", "fetch_latest_versions"])
-    def test_either_commit_info_key_extends_the_message(
-            self, api_v3_client, api_v3_module, key):
-        # fetch_latest_versions is the older spelling; both must work.
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
-        response = api_v3_client.post(self.URL, json={key: True})
-        assert "commit metadata" in response.get_json()["message"]
-
-    def test_message_stays_plain_without_the_flag(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry.return_value = {"plugins": []}
-        response = api_v3_client.post(self.URL, json={})
-        assert response.get_json()["message"] == "Plugin store refreshed"
-
-    def test_network_failure_is_a_500(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry.side_effect = (
-            ConnectionError("github unreachable"))
-        response = api_v3_client.post(self.URL, json={})
-        assert response.status_code == 500
-        assert response.get_json()["message"] == "An error occurred; see logs for details"
-
-    def test_failure_body_carries_no_traceback_or_paths(
-            self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry.side_effect = (
-            RuntimeError("failed at /home/user/LEDMatrix/src/secret.py line 42"))
-        body = api_v3_client.post(self.URL, json={}).get_json()
-        assert "Traceback" not in str(body)
-        # `details` is describe_exception output: one line, type-named,
-        # credential-redacted. It may quote the message, but never a stack.
-        assert body["details"].startswith("RuntimeError:")
-        assert "\n" not in body["details"]
-
-
-class TestRegistryFromUrl:
-    URL = "/api/v3/plugins/registry-from-url"
-
-    def test_uninitialized_manager_is_a_500(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager = None
-        response = api_v3_client.post(self.URL, json={"repo_url": "http://x"})
-        assert response.status_code == 500
-
-    def test_missing_repo_url_is_a_400(self, api_v3_client, api_v3_module):
-        response = api_v3_client.post(self.URL, json={})
-        assert response.status_code == 400
-        assert "repo_url required" in response.get_json()["message"]
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.assert_not_called()
-
-    def test_success_returns_the_plugin_list(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = {
-            "plugins": [{"id": "clock"}]}
-        response = api_v3_client.post(
-            self.URL, json={"repo_url": "https://github.com/o/r"})
-        assert response.status_code == 200
-        body = response.get_json()
-        assert body["plugins"] == [{"id": "clock"}]
-        assert body["registry_url"] == "https://github.com/o/r"
-
-    def test_url_is_trimmed_before_use(self, api_v3_client, api_v3_module):
-        manager = api_v3_module.api_v3.plugin_store_manager
-        manager.fetch_registry_from_url.return_value = {"plugins": []}
-        api_v3_client.post(self.URL, json={"repo_url": "  https://github.com/o/r  "})
-        manager.fetch_registry_from_url.assert_called_once_with("https://github.com/o/r")
-
-    def test_registry_without_plugins_key_returns_empty_list(
-            self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = {
-            "other": 1}
-        response = api_v3_client.post(self.URL, json={"repo_url": "http://x"})
-        assert response.get_json()["plugins"] == []
-
-    def test_no_registry_found_is_a_400(self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = None
-        response = api_v3_client.post(self.URL, json={"repo_url": "http://x/not-a-registry"})
-        assert response.status_code == 400
-        assert "Failed to fetch registry" in response.get_json()["message"]
-
-    @pytest.mark.parametrize("url", [
-        "not a url",
-        "javascript:alert(1)",
-        "file:///etc/passwd",
-        "http://localhost:8080/admin",
-    ])
-    def test_unusable_urls_fail_cleanly(self, api_v3_client, api_v3_module, url):
-        # Characterization: the handler performs no URL validation of its
-        # own — whatever the manager makes of the URL decides the outcome.
-        # What is pinned here is that a rejected URL produces a clean 400
-        # rather than a traceback or a 500.
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.return_value = None
-        response = api_v3_client.post(self.URL, json={"repo_url": url})
-        assert response.status_code == 400
-        assert "Traceback" not in str(response.get_json())
-
-    def test_fetch_exception_is_a_500_without_internals(
-            self, api_v3_client, api_v3_module):
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.side_effect = (
-            ValueError("parse failed in /srv/app/internal.py"))
-        response = api_v3_client.post(self.URL, json={"repo_url": "http://x"})
-        assert response.status_code == 500
-        body = response.get_json()
-        assert body["message"] == "An error occurred; see logs for details"
-        assert "Traceback" not in str(body)
-
-    def test_non_string_repo_url_is_rejected(self, api_v3_client, api_v3_module):
-        # Regression: .strip() on a non-string raised, and the catch-all
-        # reported the caller's own mistake as a server fault.
-        response = api_v3_client.post(self.URL, json={"repo_url": 12345})
-        assert response.status_code == 400
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.assert_not_called()
-
-    def test_blank_repo_url_is_rejected(self, api_v3_client, api_v3_module):
-        response = api_v3_client.post(self.URL, json={"repo_url": "   "})
-        assert response.status_code == 400
-        api_v3_module.api_v3.plugin_store_manager.fetch_registry_from_url.assert_not_called()

test/test_api_v3_wifi_endpoints.py

@@ -1,240 +0,0 @@
-"""
-Endpoint tests for the /wifi/* routes in api_v3.
-
-These routes drive the host's actual networking — connecting, dropping a
-connection, switching the radio off — and had no endpoint-level tests at
-all. WiFiManager is mocked throughout; nothing here may touch real
-networking.
-
-Each handler does `from src.wifi_manager import WiFiManager` inside the
-function body, so the patch target is the class at its definition site.
-"""
-
-import sys
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-sys.path.insert(0, str(Path(__file__).parent.parent))
-
-from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
-
-
-@pytest.fixture
-def wifi_manager():
-    """Patch WiFiManager where it is defined; yield the instance mock."""
-    with patch("src.wifi_manager.WiFiManager") as cls:
-        instance = MagicMock()
-        cls.return_value = instance
-        yield instance
-
-
-class TestConnect:
-    URL = "/api/v3/wifi/connect"
-
-    def test_success(self, api_v3_client, wifi_manager):
-        wifi_manager.connect_to_network.return_value = (True, "Connected to HomeNet")
-        response = api_v3_client.post(self.URL, json={"ssid": "HomeNet", "password": "pw"})
-        assert response.status_code == 200
-        assert response.get_json()["message"] == "Connected to HomeNet"
-        wifi_manager.connect_to_network.assert_called_once_with("HomeNet", "pw")
-
-    def test_missing_body_rejected(self, api_v3_client, wifi_manager):
-        response = api_v3_client.post(self.URL, json={})
-        assert response.status_code == 400
-        wifi_manager.connect_to_network.assert_not_called()
-
-    def test_missing_ssid_rejected(self, api_v3_client, wifi_manager):
-        response = api_v3_client.post(self.URL, json={"password": "pw"})
-        assert response.status_code == 400
-        assert "SSID is required" in response.get_json()["message"]
-        wifi_manager.connect_to_network.assert_not_called()
-
-    @pytest.mark.parametrize("ssid", ["", "   ", "\t"])
-    def test_blank_ssid_rejected(self, api_v3_client, wifi_manager, ssid):
-        response = api_v3_client.post(self.URL, json={"ssid": ssid})
-        assert response.status_code == 400
-        wifi_manager.connect_to_network.assert_not_called()
-
-    def test_ssid_is_trimmed(self, api_v3_client, wifi_manager):
-        wifi_manager.connect_to_network.return_value = (True, "ok")
-        api_v3_client.post(self.URL, json={"ssid": "  HomeNet  "})
-        wifi_manager.connect_to_network.assert_called_once_with("HomeNet", "")
-
-    def test_missing_password_becomes_empty_string(self, api_v3_client, wifi_manager):
-        wifi_manager.connect_to_network.return_value = (True, "ok")
-        api_v3_client.post(self.URL, json={"ssid": "OpenNet"})
-        wifi_manager.connect_to_network.assert_called_once_with("OpenNet", "")
-
-    def test_null_password_becomes_empty_string(self, api_v3_client, wifi_manager):
-        wifi_manager.connect_to_network.return_value = (True, "ok")
-        api_v3_client.post(self.URL, json={"ssid": "OpenNet", "password": None})
-        wifi_manager.connect_to_network.assert_called_once_with("OpenNet", "")
-
-    def test_failure_reports_the_managers_reason(self, api_v3_client, wifi_manager):
-        wifi_manager.connect_to_network.return_value = (False, "Bad password")
-        response = api_v3_client.post(self.URL, json={"ssid": "HomeNet"})
-        assert response.status_code == 400
-        assert response.get_json()["message"] == "Bad password"
-
-    def test_failure_without_reason_uses_fallback_text(self, api_v3_client, wifi_manager):
-        wifi_manager.connect_to_network.return_value = (False, None)
-        response = api_v3_client.post(self.URL, json={"ssid": "HomeNet"})
-        assert response.status_code == 400
-        assert response.get_json()["message"] == "Failed to connect to network"
-
-    def test_manager_exception_is_a_500_without_leaking_internals(
-            self, api_v3_client, wifi_manager):
-        wifi_manager.connect_to_network.side_effect = RuntimeError(
-            "/usr/lib/secret/path blew up")
-        response = api_v3_client.post(self.URL, json={"ssid": "HomeNet"})
-        assert response.status_code == 500
-        body = response.get_json()
-        assert body["message"] == "An error occurred; see logs for details"
-        # `details` comes from describe_exception, which is deliberately
-        # safe to return (redacted, capped) — it names the type.
-        assert "RuntimeError" in body["details"]
-
-
-class TestDisconnect:
-    URL = "/api/v3/wifi/disconnect"
-
-    def test_success(self, api_v3_client, wifi_manager):
-        wifi_manager.disconnect_from_network.return_value = (True, "Disconnected")
-        response = api_v3_client.post(self.URL)
-        assert response.status_code == 200
-        assert response.get_json()["message"] == "Disconnected"
-
-    def test_failure(self, api_v3_client, wifi_manager):
-        wifi_manager.disconnect_from_network.return_value = (False, "Not connected")
-        response = api_v3_client.post(self.URL)
-        assert response.status_code == 400
-        assert response.get_json()["message"] == "Not connected"
-
-    def test_failure_without_reason_uses_fallback(self, api_v3_client, wifi_manager):
-        wifi_manager.disconnect_from_network.return_value = (False, "")
-        response = api_v3_client.post(self.URL)
-        assert response.get_json()["message"] == "Failed to disconnect from network"
-
-    def test_exception_is_a_500(self, api_v3_client, wifi_manager):
-        wifi_manager.disconnect_from_network.side_effect = OSError("nmcli missing")
-        assert api_v3_client.post(self.URL).status_code == 500
-
-
-class TestApMode:
-    ENABLE = "/api/v3/wifi/ap/enable"
-    DISABLE = "/api/v3/wifi/ap/disable"
-
-    def test_enable_success(self, api_v3_client, wifi_manager):
-        wifi_manager.enable_ap_mode.return_value = (True, "AP enabled")
-        response = api_v3_client.post(self.ENABLE, json={})
-        assert response.status_code == 200
-        wifi_manager.enable_ap_mode.assert_called_once_with(force=False)
-
-    @pytest.mark.parametrize("raw,expected", [
-        (True, True), (False, False),
-        ("true", True), ("TRUE", True), ("1", True),
-        ("false", False), ("no", False), ("yes", False),
-        (1, False),  # only real True or the listed strings count
-    ])
-    def test_force_coercion(self, api_v3_client, wifi_manager, raw, expected):
-        wifi_manager.enable_ap_mode.return_value = (True, "ok")
-        api_v3_client.post(self.ENABLE, json={"force": raw})
-        wifi_manager.enable_ap_mode.assert_called_once_with(force=expected)
-
-    def test_enable_without_body(self, api_v3_client, wifi_manager):
-        wifi_manager.enable_ap_mode.return_value = (True, "ok")
-        assert api_v3_client.post(self.ENABLE).status_code == 200
-
-    def test_enable_failure(self, api_v3_client, wifi_manager):
-        wifi_manager.enable_ap_mode.return_value = (False, "hostapd missing")
-        response = api_v3_client.post(self.ENABLE, json={})
-        assert response.status_code == 400
-        assert response.get_json()["message"] == "hostapd missing"
-
-    def test_disable_success(self, api_v3_client, wifi_manager):
-        wifi_manager.disable_ap_mode.return_value = (True, "AP disabled")
-        assert api_v3_client.post(self.DISABLE).status_code == 200
-
-    def test_disable_failure(self, api_v3_client, wifi_manager):
-        wifi_manager.disable_ap_mode.return_value = (False, "not running")
-        assert api_v3_client.post(self.DISABLE).status_code == 400
-
-    def test_enable_exception_is_a_500(self, api_v3_client, wifi_manager):
-        wifi_manager.enable_ap_mode.side_effect = RuntimeError("boom")
-        assert api_v3_client.post(self.ENABLE, json={}).status_code == 500
-
-
-class TestRadio:
-    URL = "/api/v3/wifi/radio"
-
-    def test_get_state(self, api_v3_client, wifi_manager):
-        wifi_manager.get_wifi_radio_state.return_value = {
-            "enabled": True, "ethernet_connected": False}
-        response = api_v3_client.get(self.URL)
-        assert response.status_code == 200
-        assert response.get_json()["data"]["enabled"] is True
-
-    def test_get_state_exception_is_a_500(self, api_v3_client, wifi_manager):
-        wifi_manager.get_wifi_radio_state.side_effect = OSError("rfkill missing")
-        assert api_v3_client.get(self.URL).status_code == 500
-
-    def test_enabled_is_required(self, api_v3_client, wifi_manager):
-        response = api_v3_client.post(self.URL, json={})
-        assert response.status_code == 400
-        assert "enabled is required" in response.get_json()["message"]
-        wifi_manager.set_wifi_radio.assert_not_called()
-
-    def test_enable_success(self, api_v3_client, wifi_manager):
-        wifi_manager.set_wifi_radio.return_value = (True, "Radio on", None)
-        wifi_manager.get_wifi_radio_state.return_value = {"enabled": True}
-        response = api_v3_client.post(self.URL, json={"enabled": True})
-        assert response.status_code == 200
-        wifi_manager.set_wifi_radio.assert_called_once_with(True, force=False)
-
-    @pytest.mark.parametrize("raw,expected", [
-        (True, True), ("true", True), ("1", True), ("yes", True),
-        (False, False), ("false", False), ("off", False), (0, False),
-    ])
-    def test_enabled_coercion_is_string_aware(
-            self, api_v3_client, wifi_manager, raw, expected):
-        # bool("false") is True, so the endpoint parses strings explicitly
-        # rather than trusting truthiness — it is a public contract, not
-        # only the shipped UI which always sends real JSON booleans.
-        wifi_manager.set_wifi_radio.return_value = (True, "ok", None)
-        wifi_manager.get_wifi_radio_state.return_value = {}
-        api_v3_client.post(self.URL, json={"enabled": raw})
-        wifi_manager.set_wifi_radio.assert_called_once_with(expected, force=False)
-
-    def test_force_passed_through(self, api_v3_client, wifi_manager):
-        wifi_manager.set_wifi_radio.return_value = (True, "ok", None)
-        wifi_manager.get_wifi_radio_state.return_value = {}
-        api_v3_client.post(self.URL, json={"enabled": False, "force": "true"})
-        wifi_manager.set_wifi_radio.assert_called_once_with(False, force=True)
-
-    def test_refusal_reports_reason(self, api_v3_client, wifi_manager):
-        # Disabling the radio without Ethernet would lock the user out of
-        # this very interface, so the manager can refuse with a reason.
-        wifi_manager.set_wifi_radio.return_value = (
-            False, "Refusing: no wired fallback", "no_ethernet")
-        response = api_v3_client.post(self.URL, json={"enabled": False})
-        assert response.status_code == 400
-        body = response.get_json()
-        assert body["reason"] == "no_ethernet"
-        assert "Refusing" in body["message"]
-
-    def test_exception_is_a_500(self, api_v3_client, wifi_manager):
-        wifi_manager.set_wifi_radio.side_effect = RuntimeError("boom")
-        assert api_v3_client.post(self.URL, json={"enabled": True}).status_code == 500
-
-
-class TestNoRealNetworking:
-    def test_wifi_manager_is_never_constructed_for_real(self, api_v3_client):
-        # Guard against a future refactor moving the import to module level,
-        # where the fixture's patch of the definition site would stop
-        # applying and the tests would start driving real networking.
-        with patch("src.wifi_manager.WiFiManager") as cls:
-            cls.return_value.disconnect_from_network.return_value = (True, "ok")
-            api_v3_client.post("/api/v3/wifi/disconnect")
-            assert cls.called

test/test_cache_manager.py

@@ -458,3 +458,26 @@ class TestDiskCacheWriteEconomy:
         cache = DiskCache(cache_dir=str(tmp_path))
         cache.set("k", {"when": datetime(2026, 7, 12, 10, 30)})
         assert cache.get("k") == {"when": "2026-07-12T10:30:00"}
+
+
+# --- the ceiling has to hold between cleanup sweeps ---------------------------
+
+def test_memory_cache_enforces_ceiling_on_every_write():
+    """_cleanup_memory_cache only runs every cleanup_interval seconds (300 by
+    default). If set() accepted entries without bound in between, a burst could
+    take the cache far past max_size -- which is the unbounded growth the limit
+    exists to prevent, and on a 1GB board the difference between a bounded cache
+    and a Pi that cannot fork.
+    """
+    from src.cache.memory_cache import MemoryCache
+
+    cache = MemoryCache(max_size=150, cleanup_interval=300.0)
+    for i in range(1000):
+        cache.set(f"k{i}", {"v": i})
+
+    assert len(cache._cache) <= 150
+    # The timestamp map has to be evicted alongside the values, or it becomes
+    # the leak instead.
+    assert len(cache._timestamps) <= 150
+    assert cache.get("k999") is not None, "the newest write must survive"
+    assert cache.get("k0") is None, "the oldest must be the one evicted"

test/test_logo_helper.py

@@ -1,423 +0,0 @@
-"""
-Tests for src/common/logo_helper.py — logo loading, LRU caching, resizing,
-and download-with-fallback. Previously untested: nothing in test/ referenced
-this module at all.
-
-Real PIL images under tmp_path are used rather than mocked ones, since
-load_logo() does real Path.exists() and Image.open() calls; only the HTTP
-session and the permission helpers are patched.
-
-Regression coverage for two fixed bugs:
-- _download_logo wrote response.content to disk with no size cap and no
-  check that the bytes decoded as an image, so a hostile or broken URL
-  could leave arbitrary/oversized content cached in the assets directory.
-- get_cache_stats() divided by self.cache_size unguarded, raising
-  ZeroDivisionError for a helper constructed with cache_size=0.
-"""
-
-import logging
-import tempfile
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-import pytest
-import requests
-from PIL import Image, UnidentifiedImageError
-
-from src.common.logo_helper import MAX_LOGO_BYTES, LogoHelper
-
-
-@pytest.fixture(autouse=True)
-def _no_real_chmod(monkeypatch):
-    # Keep the permission helpers out of the way: their own env detection
-    # is not what these tests are about.
-    monkeypatch.setattr("src.common.logo_helper.ensure_directory_permissions", MagicMock())
-    monkeypatch.setattr("src.common.logo_helper.ensure_file_permissions", MagicMock())
-
-
-@pytest.fixture
-def helper():
-    return LogoHelper(display_width=64, display_height=32,
-                      logger=logging.getLogger("test.logo_helper"))
-
-
-def write_logo(path: Path, size=(20, 20), color=(255, 0, 0), fmt="PNG") -> Path:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    Image.new("RGB", size, color).save(path, format=fmt)
-    return path
-
-
-def fake_response(content: bytes, chunk_size: int = 64 * 1024):
-    """Stand-in for a streamed requests.Response.
-
-    _download_logo opens `with session.get(..., stream=True)` and reads
-    through iter_content(), so the fake has to be a context manager that
-    yields the body in pieces rather than exposing it as .content.
-    Chunking is the fake's own, not the caller's, so a test can dribble a
-    body out in small pieces.
-    """
-    response = MagicMock()
-    response.__enter__.return_value = response
-    response.__exit__.return_value = False
-    response.raise_for_status = MagicMock()
-
-    def _iter_content(*_args, **_kwargs):
-        for i in range(0, len(content), chunk_size):
-            yield content[i:i + chunk_size]
-
-    response.iter_content = _iter_content
-    return response
-
-
-def endless_response(chunk: bytes = b"\x00" * 65536):
-    """A server that declares no length and never stops sending.
-
-    This is the case response.content could not survive: it buffers to
-    completion, so the size check never got a chance to run.
-    """
-    response = MagicMock()
-    response.__enter__.return_value = response
-    response.__exit__.return_value = False
-    response.raise_for_status = MagicMock()
-
-    def _iter_content(*_args, **_kwargs):
-        while True:
-            yield chunk
-
-    response.iter_content = _iter_content
-    return response
-
-
-def png_bytes(size=(20, 20), color=(0, 128, 0)) -> bytes:
-    import io
-    buf = io.BytesIO()
-    Image.new("RGB", size, color).save(buf, format="PNG")
-    return buf.getvalue()
-
-
-class TestLoadLogo:
-    def test_loads_and_converts_to_rgba(self, helper, tmp_path):
-        path = write_logo(tmp_path / "PHI.png")
-        logo = helper.load_logo("PHI", path)
-        assert logo is not None
-        assert logo.mode == "RGBA"
-
-    def test_missing_file_returns_none(self, helper, tmp_path, caplog):
-        with caplog.at_level(logging.WARNING):
-            assert helper.load_logo("NOPE", tmp_path / "missing.png") is None
-        assert "Logo not found" in caplog.text
-
-    def test_second_load_is_served_from_cache(self, helper, tmp_path):
-        path = write_logo(tmp_path / "PHI.png")
-        first = helper.load_logo("PHI", path)
-        path.unlink()  # cache hit must not touch the filesystem
-        assert helper.load_logo("PHI", path) is first
-
-    def test_cache_key_includes_requested_size(self, helper, tmp_path):
-        # A panel-size change must not hand back a logo sized for the old
-        # dimensions, so the two sizes get separate cache entries.
-        path = write_logo(tmp_path / "PHI.png", size=(100, 100))
-        small = helper.load_logo("PHI", path, max_width=10, max_height=10)
-        large = helper.load_logo("PHI", path, max_width=50, max_height=50)
-        assert small is not large
-        assert small.size != large.size
-        assert len(helper._logo_cache) == 2
-
-    def test_default_size_is_one_and_a_half_display(self, helper, tmp_path):
-        path = write_logo(tmp_path / "PHI.png", size=(500, 500))
-        logo = helper.load_logo("PHI", path)
-        assert logo.width <= int(64 * 1.5)
-        assert logo.height <= int(32 * 1.5)
-
-    def test_smaller_image_is_not_upscaled(self, helper, tmp_path):
-        path = write_logo(tmp_path / "PHI.png", size=(8, 8))
-        assert helper.load_logo("PHI", path, max_width=64, max_height=64).size == (8, 8)
-
-    def test_larger_image_is_downscaled_preserving_aspect(self, helper, tmp_path):
-        path = write_logo(tmp_path / "PHI.png", size=(200, 100))
-        logo = helper.load_logo("PHI", path, max_width=50, max_height=50)
-        assert logo.width <= 50 and logo.height <= 50
-        assert logo.width == 50 and logo.height == 25  # 2:1 preserved
-
-    def test_string_path_accepted(self, helper, tmp_path):
-        path = write_logo(tmp_path / "PHI.png")
-        assert helper.load_logo("PHI", str(path)) is not None
-
-    def test_corrupt_file_returns_none(self, helper, tmp_path, caplog):
-        bad = tmp_path / "bad.png"
-        bad.write_bytes(b"not an image")
-        with caplog.at_level(logging.ERROR):
-            assert helper.load_logo("BAD", bad) is None
-        assert "Error loading logo" in caplog.text
-
-
-class TestCacheManagement:
-    def test_lru_evicts_oldest(self, tmp_path):
-        helper = LogoHelper(64, 32, cache_size=2, logger=MagicMock())
-        paths = [write_logo(tmp_path / f"T{i}.png") for i in range(3)]
-        for i, path in enumerate(paths):
-            helper.load_logo(f"T{i}", path)
-        assert len(helper._logo_cache) == 2
-        assert not any(k.startswith("T0_") for k in helper._logo_cache)
-
-    def test_cache_hit_refreshes_lru_position(self, tmp_path):
-        helper = LogoHelper(64, 32, cache_size=2, logger=MagicMock())
-        a, b, c = [write_logo(tmp_path / f"{n}.png") for n in ("A", "B", "C")]
-        helper.load_logo("A", a)
-        helper.load_logo("B", b)
-        helper.load_logo("A", a)   # A is now most-recently used
-        helper.load_logo("C", c)   # evicts B, not A
-        assert any(k.startswith("A_") for k in helper._logo_cache)
-        assert not any(k.startswith("B_") for k in helper._logo_cache)
-
-    def test_clear_cache_empties_both_structures(self, helper, tmp_path):
-        helper.load_logo("PHI", write_logo(tmp_path / "PHI.png"))
-        helper.clear_cache()
-        assert helper._logo_cache == {}
-        assert helper._cache_order == []
-
-    def test_cache_stats(self, tmp_path):
-        helper = LogoHelper(64, 32, cache_size=4, logger=MagicMock())
-        helper.load_logo("PHI", write_logo(tmp_path / "PHI.png"))
-        stats = helper.get_cache_stats()
-        assert stats["cached_logos"] == 1
-        assert stats["cache_size_limit"] == 4
-        assert stats["cache_usage_percent"] == 25
-
-    def test_zero_cache_size_does_not_divide_by_zero(self):
-        # Regression: this raised ZeroDivisionError.
-        stats = LogoHelper(64, 32, cache_size=0, logger=MagicMock()).get_cache_stats()
-        assert stats["cache_usage_percent"] == 0
-        assert stats["cache_size_limit"] == 0
-
-
-class TestLoadLogoWithDownload:
-    def test_existing_file_skips_download(self, helper, tmp_path):
-        path = write_logo(tmp_path / "PHI.png")
-        helper.session.get = MagicMock()
-        assert helper.load_logo_with_download("PHI", path, "http://x/logo.png") is not None
-        helper.session.get.assert_not_called()
-
-    def test_downloads_then_loads(self, helper, tmp_path):
-        path = tmp_path / "PHI.png"
-        helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
-        logo = helper.load_logo_with_download("PHI", path, "http://x/logo.png")
-        assert logo is not None
-        assert path.exists()
-        # stream=True is load-bearing: it is what lets the size cap apply
-        # before the body is buffered.
-        helper.session.get.assert_called_once_with(
-            "http://x/logo.png", timeout=30, stream=True)
-
-    def test_download_failure_falls_back_to_placeholder(self, helper, tmp_path):
-        helper.session.get = MagicMock(
-            side_effect=requests.RequestException("connection reset"))
-        logo = helper.load_logo_with_download(
-            "PHI", tmp_path / "PHI.png", "http://x/logo.png",
-            max_width=20, max_height=20)
-        assert logo is not None and logo.size == (20, 20)  # placeholder
-
-    def test_http_error_falls_back_to_placeholder(self, helper, tmp_path):
-        response = fake_response(b"")
-        response.raise_for_status.side_effect = requests.HTTPError("404")
-        helper.session.get = MagicMock(return_value=response)
-        logo = helper.load_logo_with_download(
-            "PHI", tmp_path / "PHI.png", "http://x/logo.png",
-            max_width=20, max_height=20)
-        assert logo is not None and logo.size == (20, 20)
-
-    def test_no_url_and_no_file_gives_placeholder(self, helper, tmp_path):
-        logo = helper.load_logo_with_download(
-            "PHI", tmp_path / "missing.png", None, max_width=16, max_height=16)
-        assert logo is not None and logo.size == (16, 16)
-
-
-class TestDownloadLogo:
-    def test_writes_file_and_sets_permissions(self, helper, tmp_path):
-        path = tmp_path / "assets" / "PHI.png"
-        # Directory creation is ensure_directory_permissions' job, and the
-        # autouse fixture stubs it out — so make the directory here.
-        path.parent.mkdir()
-        helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
-        with patch("src.common.logo_helper.ensure_directory_permissions") as dirs, \
-             patch("src.common.logo_helper.ensure_file_permissions") as files:
-            helper._download_logo("http://x/logo.png", path)
-        assert path.exists()
-        dirs.assert_called_once()
-        files.assert_called_once()
-        assert dirs.call_args[0][0] == path.parent
-
-    def test_oversized_response_is_rejected_without_writing(self, helper, tmp_path):
-        # Regression: an unbounded response.content was written straight to
-        # disk, so a hostile URL chose how many bytes landed in assets/.
-        path = tmp_path / "huge.png"
-        helper.session.get = MagicMock(
-            return_value=fake_response(b"\x00" * (MAX_LOGO_BYTES + 1)))
-        with pytest.raises(ValueError, match="exceeds the"):
-            helper._download_logo("http://x/huge.png", path)
-        assert not path.exists()
-
-    def test_unbounded_response_is_aborted_at_the_cap(self, helper, tmp_path):
-        # Regression: the cap used to be checked against response.content,
-        # which buffers the whole body first — so a server that omits
-        # Content-Length and never stops sending exhausted memory before
-        # the check could run. Streaming counts bytes as they arrive, so
-        # this terminates instead of hanging.
-        path = tmp_path / "endless.png"
-        helper.session.get = MagicMock(return_value=endless_response())
-        with pytest.raises(ValueError, match="exceeds the"):
-            helper._download_logo("http://x/endless.png", path)
-        assert not path.exists()
-
-    def test_no_partial_file_is_left_when_the_stream_dies(self, helper, tmp_path):
-        # A transfer that fails midway must not leave a truncated logo
-        # where the real one belongs — load_logo() would cache it.
-        path = tmp_path / "cut.png"
-        real = png_bytes()
-
-        def _dies_midway(*_args, **_kwargs):
-            yield real[:20]
-            raise OSError("connection reset")
-
-        response = MagicMock()
-        response.__enter__.return_value = response
-        response.__exit__.return_value = False
-        response.raise_for_status = MagicMock()
-        response.iter_content = _dies_midway
-        helper.session.get = MagicMock(return_value=response)
-
-        with pytest.raises(OSError):
-            helper._download_logo("http://x/cut.png", path)
-        assert not path.exists()
-        assert list(tmp_path.glob("*.part")) == []
-
-    def test_concurrent_downloads_do_not_share_a_temp_file(self, helper, tmp_path):
-        # Two plugins can ask for the same logo at once. A fixed
-        # "<name>.part" would let them interleave writes into one file and
-        # publish the mixture; each download gets its own temp name.
-        path = tmp_path / "PHI.png"
-        seen = []
-        real_mkstemp = tempfile.mkstemp
-
-        def record(*args, **kwargs):
-            fd, name = real_mkstemp(*args, **kwargs)
-            seen.append(name)
-            return fd, name
-
-        with patch("src.common.logo_helper.tempfile.mkstemp", side_effect=record):
-            helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
-            helper._download_logo("http://x/logo.png", path)
-            helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
-            helper._download_logo("http://x/logo.png", path)
-
-        assert len(seen) == 2 and seen[0] != seen[1]
-        assert path.exists()
-        assert list(tmp_path.glob("*.part")) == []  # both cleaned up
-
-    def test_request_failure_leaves_no_temp_file(self, helper, tmp_path):
-        # mkstemp creates the file up front, so an error before any bytes
-        # arrive still has something to clean up.
-        helper.session.get = MagicMock(
-            side_effect=requests.RequestException("connection reset"))
-        with pytest.raises(requests.RequestException):
-            helper._download_logo("http://x/logo.png", tmp_path / "PHI.png")
-        assert list(tmp_path.glob("*")) == []
-
-    def test_non_image_response_is_deleted_and_raises(self, helper, tmp_path):
-        # Regression: undecodable bytes stayed on disk, so every later
-        # load_logo() call hit the corrupt file instead of re-downloading.
-        path = tmp_path / "bad.png"
-        helper.session.get = MagicMock(return_value=fake_response(b"<html>404</html>"))
-        # Specifically Pillow's identify failure, not any OSError: the
-        # point is that the bytes did not decode, and OSError alone would
-        # also admit unrelated filesystem faults.
-        with pytest.raises(UnidentifiedImageError):
-            helper._download_logo("http://x/bad.png", path)
-        assert not path.exists()
-        assert list(tmp_path.glob("*.part")) == []
-
-    def test_decompression_bomb_is_deleted_and_raises(self, helper, tmp_path, monkeypatch):
-        path = tmp_path / "bomb.png"
-        helper.session.get = MagicMock(return_value=fake_response(png_bytes()))
-
-        class Bomb:
-            def __enter__(self):
-                return self
-
-            def __exit__(self, *a):
-                return False
-
-            def load(self):
-                raise Image.DecompressionBombError("too many pixels")
-
-        monkeypatch.setattr("src.common.logo_helper.Image.open", lambda *a, **kw: Bomb())
-        with pytest.raises(Image.DecompressionBombError):
-            helper._download_logo("http://x/bomb.png", path)
-        assert not path.exists()
-
-    def test_bad_download_surfaces_as_placeholder_not_crash(self, helper, tmp_path):
-        # The new guards raise, and load_logo_with_download's existing
-        # broad except turns that into the placeholder path.
-        helper.session.get = MagicMock(return_value=fake_response(b"garbage"))
-        logo = helper.load_logo_with_download(
-            "PHI", tmp_path / "PHI.png", "http://x/bad.png",
-            max_width=12, max_height=12)
-        assert logo is not None and logo.size == (12, 12)
-
-
-class TestLogoVariations:
-    def test_plain_abbreviation_returns_itself(self, helper):
-        assert helper.get_logo_variations("PHI") == ["PHI"]
-
-    def test_ampersand_expanded(self, helper):
-        assert "TAAND M" in helper.get_logo_variations("TA& M")
-
-    def test_and_contracted(self, helper):
-        assert "T&M" in helper.get_logo_variations("TANDM")
-
-    def test_special_case_appends_known_aliases(self, helper):
-        variations = helper.get_logo_variations("TA&M")
-        assert "TAMU" in variations and "TEXASAM" in variations
-        assert "TAANDM" in variations  # the generic & rule still applies
-
-
-class TestNormalizeAbbreviation:
-    def test_uppercases_and_strips(self, helper):
-        assert helper.normalize_abbreviation("  phi  ") == "PHI"
-
-    def test_ampersand_becomes_and(self, helper):
-        assert helper.normalize_abbreviation("TA&M") == "TAANDM"
-
-    def test_internal_spaces_removed(self, helper):
-        assert helper.normalize_abbreviation("New York") == "NEWYORK"
-
-    def test_deliberately_differs_from_logo_downloader(self, helper):
-        # Pinned, not a bug: LogoDownloader.normalize_abbreviation replaces
-        # filesystem-unsafe characters but keeps spaces, and plugins call
-        # that one. Changing either changes which logo filenames resolve on
-        # existing installs. Both docstrings say so explicitly.
-        from src.logo_downloader import LogoDownloader
-        assert helper.normalize_abbreviation("New York") == "NEWYORK"
-        assert LogoDownloader.normalize_abbreviation("New York") == "NEW YORK"
-
-
-class TestPlaceholderLogo:
-    def test_uses_requested_dimensions(self, helper):
-        assert helper._create_placeholder_logo("PHI", 30, 20).size == (30, 20)
-
-    def test_defaults_to_one_and_a_half_display(self, helper):
-        assert helper._create_placeholder_logo("PHI").size == (96, 48)
-
-    def test_is_rgba(self, helper):
-        assert helper._create_placeholder_logo("PHI", 10, 10).mode == "RGBA"
-
-    def test_invalid_dimensions_return_none(self, helper, caplog):
-        with caplog.at_level(logging.ERROR):
-            assert helper._create_placeholder_logo("PHI", -5, -5) is None
-        assert "Error creating placeholder" in caplog.text
-
-
-class TestSessionConfiguration:
-    def test_user_agent_and_accept_headers(self, helper):
-        assert helper.session.headers["User-Agent"] == "LEDMatrix-Common/1.0"
-        assert helper.session.headers["Accept"] == "image/*"

test/test_pixlet_download.py

@@ -0,0 +1,180 @@
+"""
+Tests for scripts/download_pixlet.sh -- release-tag resolution and download guards.
+
+Background: Starlark apps render through the pixlet binary, and the installer
+that fetches it failed silently. It resolved the release tag by grepping the
+GitHub API response for '"tag_name"' and taking the last quoted token on the
+match with a greedy sed. When the response arrives on one line that token is
+"mentions_count", not the tag, so the script built a URL for a release that
+cannot exist -- and `curl -L -o` without -f wrote the 404 body to the file and
+exited 0, so the first sign of trouble was tar reporting "not in gzip format"
+about a page of HTML.
+
+The API is pretty-printed by default, which is exactly why this needs a test:
+by hand the old command looks correct, and the failure only appears when the
+formatting changes. These drive the real script with a stubbed curl on PATH, so
+both response shapes are covered without touching the network.
+"""
+
+import re
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+SCRIPT = Path(__file__).resolve().parent.parent / "scripts" / "download_pixlet.sh"
+
+PRETTY = """{
+  "url": "https://api.github.com/repos/tronbyt/pixlet/releases/12345",
+  "id": 12345,
+  "tag_name": "v0.53.1",
+  "name": "v0.53.1",
+  "draft": false,
+  "prerelease": false,
+  "mentions_count": 3
+}
+"""
+
+# The shape that broke it: one line, and the last quoted token is not the tag.
+MINIFIED = (
+    '{"url":"https://api.github.com/repos/tronbyt/pixlet/releases/12345",'
+    '"id":12345,"tag_name":"v0.53.1","name":"v0.53.1","draft":false,'
+    '"prerelease":false,"mentions_count":3}'
+)
+
+
+def run_script(tmp_path, api_body, download=None):
+    """Run the real script against a stubbed curl.
+
+    Args:
+        api_body: what the stub returns for the api.github.com request.
+        download: bytes to write for a release-asset request, or None to make
+            that request fail the way `curl -f` does on an HTTP error.
+    """
+    root = tmp_path / "project"
+    (root / "scripts").mkdir(parents=True)
+    shutil.copy(SCRIPT, root / "scripts" / "download_pixlet.sh")
+
+    api_file = tmp_path / "api.json"
+    api_file.write_text(api_body)
+
+    stub_dir = tmp_path / "stub"
+    stub_dir.mkdir()
+    asset_file = tmp_path / "asset.bin"
+    if download is not None:
+        asset_file.write_bytes(download)
+
+    # Stands in for curl, including the -f semantics the fix turns on: without
+    # -f, real curl writes the error body to the output file and exits 0, which
+    # is what let a 404 masquerade as a successful download. The stub has to
+    # honour that or a test of the fix would pass against the old script too.
+    (stub_dir / "curl").write_text(f"""#!/bin/bash
+out=""
+url=""
+fail_on_error=0
+while [ $# -gt 0 ]; do
+    case "$1" in
+        -o) out="$2"; shift 2 ;;
+        -*f*) fail_on_error=1; shift ;;
+        -*) shift ;;
+        *) url="$1"; shift ;;
+    esac
+done
+if [[ "$url" == *api.github.com* ]]; then
+    cat {api_file}
+    exit 0
+fi
+if [ -f "{asset_file}" ]; then
+    cp "{asset_file}" "$out"
+    exit 0
+fi
+# No asset: stand in for an HTTP 404.
+if [ "$fail_on_error" = "1" ]; then
+    exit 22
+fi
+printf '<!DOCTYPE html><html>404 Not Found</html>' > "$out"
+exit 0
+""")
+    (stub_dir / "curl").chmod(0o755)
+
+    return subprocess.run(
+        ["bash", str(root / "scripts" / "download_pixlet.sh")],
+        capture_output=True, text=True,
+        env={"PATH": f"{stub_dir}:/usr/bin:/bin:/usr/sbin:/sbin",
+             "PIXLET_VERSION": "latest"},
+    )
+
+
+def resolved_version(result):
+    match = re.search(r"^Version: (.+)$", result.stdout, re.M)
+    assert match, f"no version line in output:\n{result.stdout}"
+    return match.group(1).strip()
+
+
+def test_script_is_syntactically_valid():
+    result = subprocess.run(["bash", "-n", str(SCRIPT)], capture_output=True, text=True)
+    assert result.returncode == 0, result.stderr
+
+
+@pytest.mark.parametrize("body,label", [(PRETTY, "pretty"), (MINIFIED, "minified")])
+def test_tag_is_resolved_from_either_response_shape(tmp_path, body, label):
+    """The minified case is the regression: the last quoted token there is
+    "mentions_count", which is what the old greedy sed captured."""
+    result = run_script(tmp_path, body)
+    assert resolved_version(result) == "v0.53.1", f"{label}: {result.stdout}"
+    assert "mentions_count" not in result.stdout
+
+
+@pytest.mark.parametrize(
+    "tag",
+    ["mentions_count", "v0.53garbage", "0.53", "v0.5", "", "v0.53.1 ; echo pwned"],
+)
+def test_a_tag_that_is_not_a_release_falls_back(tmp_path, tag):
+    """A wrong-but-non-empty value is what made the original bug silent, so the
+    check is on the shape. Partial matches must not pass: "v0.53garbage" and
+    "0.53" would build a URL for a release that cannot exist."""
+    result = run_script(tmp_path, '{"tag_name": "%s"}' % tag)
+    assert resolved_version(result) == "v0.50.2", result.stdout
+    assert "using fallback" in result.stdout
+
+
+@pytest.mark.parametrize("tag", ["v0.53.1", "v1.0.0", "v0.54.0-rc.1", "v1.2.3+build.4"])
+def test_real_release_tag_shapes_are_accepted(tmp_path, tag):
+    assert resolved_version(run_script(tmp_path, '{"tag_name": "%s"}' % tag)) == tag
+
+
+def test_an_http_error_is_reported_as_a_failed_download(tmp_path):
+    """Without curl -f the 404 body lands in the file and curl exits 0, so the
+    failure surfaced two steps later as tar complaining about gzip -- about
+    what was really a page of HTML. It has to be reported where it happened.
+
+    Both versions end at 0/1, so asserting only on the count would pass against
+    the old script; the discriminating part is which layer reports it.
+    """
+    result = run_script(tmp_path, PRETTY, download=None)
+    assert "Download complete: 0/1 succeeded" in result.stdout
+    assert "✓ Downloaded" not in result.stdout
+    assert "Failed to download" in result.stdout
+    assert "Failed to extract" not in result.stdout, (
+        "an HTTP error should not surface as an extraction failure")
+
+
+def test_a_non_archive_response_is_rejected_before_extraction(tmp_path):
+    result = run_script(tmp_path, PRETTY, download=b"<!DOCTYPE html><html>502 Bad Gateway")
+    assert "not a gzip archive" in result.stdout
+    assert "Download complete: 0/1 succeeded" in result.stdout
+
+
+def test_the_diagnostic_cannot_smuggle_terminal_escapes(tmp_path):
+    """Those bytes come from whatever answered the request. An error page
+    carrying escapes must not be able to rewrite the output or bury it."""
+    hostile = b"<!DOCTYPE html>\x1b[2J\x1b[31mgone\x1b[0m\rHTTP 200 OK\x08\x08"
+    result = run_script(tmp_path, PRETTY, download=hostile)
+    assert "not a gzip archive" in result.stdout
+    printed = re.search(r"^\s*\(first bytes: (.*)\)$", result.stdout, re.M)
+    assert printed, f"no diagnostic line:\n{result.stdout}"
+    assert "DOCTYPE" in printed.group(1), "the useful part of the page was dropped"
+    for forbidden in ("\x1b", "\r", "\x08", "\x00"):
+        assert forbidden not in printed.group(1), (
+            f"control byte {forbidden!r} reached the terminal")

test/test_plugin_health.py

@@ -91,3 +91,101 @@ def test_force_reload_refreshes_stale_in_memory_snapshot():
 
     # and it asked the cache to bypass the in-memory tier (memory_ttl=0).
     assert any(c.kwargs.get("memory_ttl") == 0 for c in cache.get.call_args_list)
+
+
+# --- persisted state that does not match the current schema -------------------
+#
+# A record on disk can be missing fields the callers index directly: a partial
+# write, a restored backup, or a state written by an older schema. Returning it
+# verbatim raises KeyError inside record_success / record_failure, which takes
+# the display down in a restart loop that survives reboots, because the bad
+# entry is on disk and gets read again on the way back up. Observed in the wild
+# as `plugin clock-simple operation failed: 'circuit_state'`, repeating ~50x a
+# minute with the panel frozen.
+
+_INDEXED_FIELDS = (
+    "consecutive_failures", "total_failures", "total_successes",
+    "last_success_time", "last_failure_time", "circuit_state",
+    "circuit_opened_time", "half_open_start_time", "last_error",
+)
+
+
+def _tracker_reading(persisted):
+    cache = _cache()
+    cache.get.return_value = persisted
+    return PluginHealthTracker(cache)
+
+
+def test_partial_state_is_completed_not_returned_raw():
+    """The shape seen in the wild: one field, everything else absent."""
+    state = _tracker_reading({"circuit_state": "closed"}).get_health_state("p")
+    for field in _INDEXED_FIELDS:
+        assert field in state, f"{field} missing; callers index it directly"
+
+
+def test_repair_keeps_real_failure_history():
+    """A record with genuine counts must not be reset to healthy just because
+    an optional field is absent -- that would clear a tripped breaker."""
+    state = _tracker_reading({
+        "consecutive_failures": 5,
+        "total_failures": 5,
+        "circuit_state": "open",
+    }).get_health_state("p")
+    assert state["consecutive_failures"] == 5
+    assert state["total_failures"] == 5
+    assert state["circuit_state"] == "open"
+
+
+def test_wrong_types_fall_back_per_field():
+    """A counter persisted as a string would pass a membership check and then
+    fail on the first += 1; an unknown circuit_state would take a branch the
+    breaker has no handling for."""
+    state = _tracker_reading({
+        "consecutive_failures": "3",
+        "circuit_state": "melted",
+        "total_failures": 7,
+    }).get_health_state("p")
+    assert state["consecutive_failures"] == 0
+    assert state["circuit_state"] == CircuitState.CLOSED.value
+    assert state["total_failures"] == 7, "valid neighbours must survive"
+
+
+def test_newer_fields_are_carried_through():
+    """degraded/degraded_reason are read with .get() and are not part of the
+    indexed set; repairing must not drop them."""
+    state = _tracker_reading({
+        "circuit_state": "closed", "degraded": True, "degraded_reason": "x",
+    }).get_health_state("p")
+    assert state["degraded"] is True
+    assert state["degraded_reason"] == "x"
+
+
+def test_recording_against_a_repaired_state_does_not_raise():
+    """The actual failure: record_failure indexing a field that was not there.
+
+    The seed deliberately omits circuit_state. Seeding a record that *has* it
+    would pass against the old raw-return behaviour too -- the counters are
+    read with .get(), so circuit_state is the only field whose absence used to
+    raise.
+    """
+    tracker = _tracker_reading({"total_failures": 2, "total_successes": 1})
+    tracker.record_failure("p", Exception("boom"))
+    tracker.record_success("p")
+
+
+def test_unhashable_or_boolean_values_are_repaired():
+    """Values that break the repair itself rather than a later caller.
+
+    An unhashable circuit_state raises TypeError inside a set membership test,
+    and bool is a subclass of int, so True would pass as a timestamp and then
+    compare as 1.0 -- expiring a cooldown the moment it opens.
+    """
+    for bad_state in ({"circuit_state": []}, {"circuit_state": {}}):
+        state = _tracker_reading(bad_state).get_health_state("p")
+        assert state["circuit_state"] == CircuitState.CLOSED.value
+
+    state = _tracker_reading({
+        "circuit_opened_time": True, "last_success_time": False,
+    }).get_health_state("p")
+    assert state["circuit_opened_time"] is None
+    assert state["last_success_time"] is None

test/test_update_change_reporting.py

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

test/test_vegas_live_weighting.py

@@ -0,0 +1,300 @@
+"""Tests that live content can take extra turns inside the Vegas ticker.
+
+Vegas was a strict round robin -- every plugin exactly once per cycle -- and
+live content did not appear in it at all, because the display controller
+refused to run the ticker while anything was live. With a dozen plugins
+enabled that left a live score either absent or minutes stale.
+
+Two things change, both off by default. `live_in_ticker` keeps the marquee
+running instead of yielding to a full-screen takeover, and the rotation is
+expanded by Smooth Weighted Round-Robin so a weighted plugin gets several
+slots per cycle, spaced through it rather than clumped.
+
+Weights are per plugin, not per game: a scoreboard showing four live games
+still occupies one slot at a time and rotates its own games within it.
+"""
+
+from unittest.mock import Mock
+
+import pytest
+
+from src.vegas_mode.config import VegasModeConfig
+from src.vegas_mode.stream_manager import StreamManager
+
+
+class FakePlugin:
+    """A plugin that can fail in each place independently.
+
+    hook_raises and live_raises are separate because they mean different
+    things: a broken weight calculation should still leave the core's own
+    live-content check usable, while a plugin that cannot answer whether it is
+    live at all has nothing left to fall back on.
+    """
+
+    def __init__(self, live=False, declared=None, raises=False,
+                 hook_raises=False, live_raises=False):
+        self._live = live
+        self._declared = declared
+        self._hook_raises = hook_raises or raises
+        self._live_raises = live_raises or raises
+        self.enabled = True
+
+    def has_live_priority(self):
+        if self._live_raises:
+            raise RuntimeError("cannot say whether I am live")
+        return self._live
+
+    def has_live_content(self):
+        return self._live
+
+    def get_vegas_priority_weight(self):
+        if self._hook_raises:
+            raise RuntimeError("weight calculation blew up")
+        return self._declared
+
+
+def _manager(plugins, **cfg):
+    config = VegasModeConfig(live_in_ticker=cfg.pop('live_in_ticker', True), **cfg)
+    pm = Mock()
+    pm.plugins = plugins
+    sm = StreamManager.__new__(StreamManager)
+    sm.config = config
+    sm.plugin_manager = pm
+    return sm
+
+
+def _counts(schedule):
+    return {p: schedule.count(p) for p in set(schedule)}
+
+
+def _max_gap(schedule, plugin_id):
+    """Largest gap between consecutive appearances, wrapping around."""
+    at = [i for i, p in enumerate(schedule) if p == plugin_id]
+    if len(at) < 2:
+        return len(schedule)
+    gaps = [b - a for a, b in zip(at, at[1:])]
+    gaps.append(len(schedule) - at[-1] + at[0])
+    return max(gaps)
+
+
+class TestWeightsComeFromTheRightPlace:
+    def test_a_quiet_plugin_gets_one_slot(self):
+        sm = _manager({'clock': FakePlugin()})
+        assert sm._plugin_weight('clock') == 1
+
+    def test_live_content_earns_the_configured_weight(self):
+        sm = _manager({'mlb': FakePlugin(live=True)}, live_weight=4)
+        assert sm._plugin_weight('mlb') == 4
+
+    def test_a_plugin_may_answer_for_itself(self):
+        # The only route for favorite-team awareness: the core can see that a
+        # game is live, not whose.
+        sm = _manager({'mlb': FakePlugin(live=True, declared=7)}, live_weight=3)
+        assert sm._plugin_weight('mlb') == 7
+
+    def test_declaring_none_defers_to_the_core(self):
+        sm = _manager({'mlb': FakePlugin(live=True, declared=None)}, live_weight=3)
+        assert sm._plugin_weight('mlb') == 3
+
+    def test_a_declared_weight_is_clamped(self):
+        sm = _manager({'a': FakePlugin(declared=99), 'b': FakePlugin(declared=0)})
+        assert sm._plugin_weight('a') == 10
+        assert sm._plugin_weight('b') == 1
+
+    def test_a_plugin_that_raises_everywhere_weighs_one(self):
+        sm = _manager({'bad': FakePlugin(raises=True)})
+        assert sm._plugin_weight('bad') == 1
+
+    def test_a_broken_hook_still_earns_the_live_boost(self):
+        # The hook is only how a plugin asks for *more* than live_weight.
+        # Losing it should cost the favorite distinction, not the live boost:
+        # has_live_priority/has_live_content are separate and still work.
+        sm = _manager({'mlb': FakePlugin(live=True, hook_raises=True)},
+                      live_weight=4)
+        assert sm._plugin_weight('mlb') == 4
+
+    def test_a_broken_hook_on_a_quiet_plugin_weighs_one(self):
+        sm = _manager({'clock': FakePlugin(live=False, hook_raises=True)},
+                      live_weight=4)
+        assert sm._plugin_weight('clock') == 1
+
+    def test_a_plugin_that_cannot_say_whether_it_is_live_weighs_one(self):
+        # Nothing left to fall back on, so no boost.
+        sm = _manager({'mlb': FakePlugin(live=True, live_raises=True)},
+                      live_weight=4)
+        assert sm._plugin_weight('mlb') == 1
+
+    def test_an_unknown_plugin_weighs_one(self):
+        assert _manager({})._plugin_weight('ghost') == 1
+
+
+class TestTheSchedule:
+    def test_nothing_weighted_leaves_the_order_untouched(self):
+        order = ['weather', 'clock', 'news']
+        sm = _manager({p: FakePlugin() for p in order})
+        assert sm._apply_priority_weights(order) == order
+
+    def test_off_by_default_the_order_is_untouched(self):
+        order = ['weather', 'mlb', 'news']
+        sm = _manager({'weather': FakePlugin(), 'mlb': FakePlugin(live=True),
+                       'news': FakePlugin()}, live_in_ticker=False, live_weight=3)
+        assert sm._apply_priority_weights(order) == order
+
+    def test_a_live_plugin_takes_its_share_of_slots(self):
+        order = ['weather', 'mlb', 'news', 'clock']
+        sm = _manager({'weather': FakePlugin(), 'mlb': FakePlugin(live=True),
+                       'news': FakePlugin(), 'clock': FakePlugin()},
+                      live_weight=3)
+        schedule = sm._apply_priority_weights(order)
+        counts = _counts(schedule)
+        assert counts['mlb'] == 3, counts
+        assert counts['weather'] == counts['news'] == counts['clock'] == 1, counts
+        assert len(schedule) == 6
+
+    def test_every_plugin_still_appears(self):
+        # A boost must not starve anything out of the cycle.
+        order = ['a', 'b', 'c', 'd', 'e', 'f']
+        plugins = {p: FakePlugin() for p in order}
+        plugins['a'] = FakePlugin(live=True, declared=10)
+        sm = _manager(plugins)
+        schedule = sm._apply_priority_weights(order)
+        assert set(schedule) == set(order), set(order) - set(schedule)
+
+    def test_nothing_doubles_across_the_cycle_seam(self):
+        # The strip loops, so the last slot neighbours the first. Smooth
+        # Weighted Round-Robin schedules the heaviest item first and often
+        # last too, which put the one clump the algorithm exists to avoid at
+        # the one place a within-cycle check cannot see.
+        order = ['baseball', 'weather', 'geochron', 'flights', 'stocks',
+                 'oftheday', 'youtube', 'stocknews', 'leaderboard',
+                 'countdown', 'odds', 'f1', 'football', 'music']
+        plugins = {p: FakePlugin() for p in order}
+        plugins['baseball'] = FakePlugin(live=True, declared=5)
+        plugins['football'] = FakePlugin(live=True, declared=3)
+        schedule = _manager(plugins)._apply_priority_weights(order)
+
+        n = len(schedule)
+        doubles = [schedule[i] for i in range(n)
+                   if schedule[i] == schedule[(i + 1) % n]]
+        assert not doubles, "%r repeats across the seam in %r" % (doubles, schedule)
+
+    def test_the_seam_repair_keeps_every_slot(self):
+        order = ['a', 'b', 'c', 'd', 'e', 'f']
+        plugins = {p: FakePlugin() for p in order}
+        plugins['a'] = FakePlugin(live=True, declared=4)
+        schedule = _manager(plugins)._apply_priority_weights(order)
+        assert _counts(schedule)['a'] == 4, _counts(schedule)
+        assert sorted(schedule) == sorted(
+            ['a'] * 4 + ['b', 'c', 'd', 'e', 'f']), schedule
+
+    def test_the_repair_uses_the_widest_gap(self):
+        # Moving the trailing repeat into the first slot that merely fits
+        # undoes the spacing: on a 28-slot rotation that turned a gap of 7
+        # into a gap of 2, which is more clumped than the seam ever was.
+        order = ['a'] + ['p%d' % i for i in range(13)]
+        plugins = {p: FakePlugin() for p in order}
+        plugins['a'] = FakePlugin(live=True, declared=4)
+        schedule = _manager(plugins)._apply_priority_weights(order)
+        at = [i for i, p in enumerate(schedule) if p == 'a']
+        gaps = [b - a for a, b in zip(at, at[1:])]
+        gaps.append(len(schedule) - at[-1] + at[0])
+        ideal = len(schedule) / len(at)
+        assert min(gaps) >= ideal / 2, "gaps %r for ideal %.1f" % (gaps, ideal)
+
+    def test_an_unavoidable_double_is_left_alone(self):
+        # Five of seven slots are the same plugin, so it must neighbour
+        # itself. Better to schedule it than to refuse or loop forever.
+        order = ['a', 'b', 'c']
+        plugins = {p: FakePlugin() for p in order}
+        plugins['a'] = FakePlugin(live=True, declared=5)
+        schedule = _manager(plugins)._apply_priority_weights(order)
+        assert _counts(schedule) == {'a': 5, 'b': 1, 'c': 1}, _counts(schedule)
+        assert set(schedule) == {'a', 'b', 'c'}
+
+    def test_the_repair_never_creates_a_new_double(self):
+        # The first version guarded the slot the repeated value moves *into*
+        # but not the one the displaced element lands in, so this traded the
+        # seam duplicate for a fresh one and came back ending ['x', 'x'].
+        sm = _manager({})
+        out = sm._unclump_seam(['a', 'b', 'c', 'd', 'x', 'y', 'x', 'a'])
+        n = len(out)
+        doubles = [out[i] for i in range(n) if out[i] == out[(i + 1) % n]]
+        assert not doubles, "%r in %r" % (doubles, out)
+        assert sorted(out) == sorted(['a', 'b', 'c', 'd', 'x', 'y', 'x', 'a'])
+
+    def test_the_last_two_slots_are_a_usable_swap(self):
+        # Reasoning about indices said this candidate was unsafe because
+        # schedule[j] is schedule[-2]; after the swap its neighbour is the
+        # repeated value, not itself. Refusing it left the only repair this
+        # schedule has on the table.
+        assert _manager({})._unclump_seam(['a', 'b', 'c', 'a']) == ['a', 'b', 'a', 'c']
+
+    def test_no_seam_schedule_is_ever_made_worse(self):
+        import random
+        sm = _manager({})
+        random.seed(11)
+        checked = 0
+        for size in range(3, 10):
+            for _ in range(400):
+                original = [random.choice('abcd') for _ in range(size)]
+                if original[0] != original[-1]:
+                    continue
+                checked += 1
+                out = sm._unclump_seam(list(original))
+                n = len(out)
+                before = sum(1 for i in range(n)
+                             if original[i] == original[(i + 1) % n])
+                after = sum(1 for i in range(n) if out[i] == out[(i + 1) % n])
+                assert after <= before, (original, out)
+                assert sorted(out) == sorted(original), (original, out)
+        assert checked > 100, "the generator stopped producing seam cases"
+
+    def test_a_schedule_too_short_to_repair_is_returned_as_is(self):
+        sm = _manager({})
+        assert sm._unclump_seam(['a', 'a']) == ['a', 'a']
+        assert sm._unclump_seam(['a']) == ['a']
+        assert sm._unclump_seam([]) == []
+
+    def test_a_schedule_with_no_seam_clash_is_untouched(self):
+        sm = _manager({})
+        plain = ['a', 'b', 'c', 'a', 'd']
+        assert sm._unclump_seam(plain) == plain
+
+    def test_repeats_are_spread_not_clumped(self):
+        # The point of Smooth Weighted Round-Robin. Three-in-a-row followed by
+        # a long silence would be worse than not boosting at all.
+        order = ['weather', 'mlb', 'news', 'clock', 'stocks', 'f1']
+        plugins = {p: FakePlugin() for p in order}
+        plugins['mlb'] = FakePlugin(live=True)
+        sm = _manager(plugins, live_weight=3)
+        schedule = sm._apply_priority_weights(order)
+
+        assert _counts(schedule)['mlb'] == 3
+        # Evenly spread over 8 slots means a gap of about 3, never 6.
+        assert _max_gap(schedule, 'mlb') <= 4, schedule
+        # And never twice running.
+        assert not any(a == b == 'mlb' for a, b in zip(schedule, schedule[1:])), schedule
+
+    def test_a_favorite_outranks_another_live_game(self):
+        order = ['weather', 'mlb', 'nhl']
+        sm = _manager({'weather': FakePlugin(),
+                       'mlb': FakePlugin(live=True, declared=5),
+                       'nhl': FakePlugin(live=True)}, live_weight=2)
+        counts = _counts(sm._apply_priority_weights(order))
+        assert counts['mlb'] == 5 and counts['nhl'] == 2 and counts['weather'] == 1, counts
+
+    def test_an_empty_rotation_is_harmless(self):
+        assert _manager({})._apply_priority_weights([]) == []
+
+
+class TestConfigParsing:
+    def test_defaults_preserve_todays_behaviour(self):
+        cfg = VegasModeConfig.from_config({})
+        assert cfg.live_in_ticker is False
+        assert cfg.live_weight == 3 and cfg.favorite_live_weight == 5
+
+    @pytest.mark.parametrize("given,expected", [(0, 1), (-4, 1), (99, 10), (4, 4)])
+    def test_weights_are_clamped(self, given, expected):
+        cfg = VegasModeConfig.from_config(
+            {'display': {'vegas_scroll': {'live_weight': given}}})
+        assert cfg.live_weight == expected

test/web_interface/test_api_v3_backup_paths.py

@@ -1,220 +0,0 @@
-"""
-Path-containment tests for the backup file routes:
-GET /backup/download/<filename>, DELETE /backup/<filename>, and the
-listing/validation routes alongside them.
-
-Both filename routes take user input straight from the URL and turn it
-into a filesystem path, one to read and one to unlink. `_safe_backup_path`
-is what stops that from reaching outside the export directory, and it had
-no tests.
-
-This is verification of existing containment, not a fix: no bypass was
-found. The tests exist so that a later "just let dots through" change has
-to argue with something.
-"""
-
-import io
-import json
-import sys
-from pathlib import Path
-from unittest.mock import MagicMock
-
-import pytest
-from flask import Flask
-
-project_root = Path(__file__).parent.parent.parent
-sys.path.insert(0, str(project_root))
-
-from web_interface.blueprints import api_v3 as api_v3_module  # noqa: E402
-from web_interface.blueprints.api_v3 import api_v3  # noqa: E402
-
-_MANAGER_ATTRS = (
-    'config_manager', 'plugin_manager', 'plugin_store_manager',
-    'plugin_state_manager', 'saved_repositories_manager', 'schema_manager',
-    'operation_queue', 'operation_history', 'cache_manager',
-)
-_SENTINEL = object()
-
-# Anything that tries to name a file outside the export directory, or that
-# is not a plain <name>.zip.
-TRAVERSAL_ATTEMPTS = [
-    "../../etc/passwd",
-    "../config.json",
-    "..%2f..%2fetc%2fpasswd",
-    "....//....//etc/passwd",
-    "/etc/passwd",
-    "..\\..\\config.json",
-    "backup.zip/../../../etc/passwd",
-    ".hidden.zip",
-    "backup.txt",
-    "backup.zip.exe",
-    "",
-    ".",
-    "..",
-]
-
-
-@pytest.fixture
-def env(tmp_path, monkeypatch):
-    export_dir = tmp_path / "backups"
-    export_dir.mkdir()
-    monkeypatch.setattr(api_v3_module, "_BACKUP_EXPORT_DIR", export_dir)
-
-    # A file outside the export dir that a traversal would be reaching for.
-    secret = tmp_path / "config.json"
-    secret.write_text(json.dumps({"secret": "do not touch"}))
-
-    originals = {name: getattr(api_v3, name, _SENTINEL) for name in _MANAGER_ATTRS}
-    for name in _MANAGER_ATTRS:
-        setattr(api_v3, name, MagicMock())
-
-    app = Flask(__name__)
-    app.config["TESTING"] = True
-    app.register_blueprint(api_v3, url_prefix="/api/v3")
-
-    class Env:
-        pass
-
-    e = Env()
-    e.client = app.test_client()
-    e.export_dir = export_dir
-    e.secret = secret
-    yield e
-
-    for name, original in originals.items():
-        if original is _SENTINEL:
-            if hasattr(api_v3, name):
-                delattr(api_v3, name)
-        else:
-            setattr(api_v3, name, original)
-
-
-def make_backup(export_dir, name="backup-2026-01-01.zip"):
-    path = export_dir / name
-    path.write_bytes(b"PK\x03\x04fake zip")
-    return path
-
-
-class TestSafeBackupPath:
-    """The containment helper itself."""
-
-    @pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS)
-    def test_rejects_unsafe_names(self, env, filename):
-        assert api_v3_module._safe_backup_path(filename) is None
-
-    def test_rejects_none(self, env):
-        assert api_v3_module._safe_backup_path(None) is None
-
-    @pytest.mark.parametrize("filename", [
-        "backup.zip",
-        "backup-2026-01-01.zip",
-        "backup_2026.01.01-v2.zip",
-        "a.zip",
-    ])
-    def test_accepts_plain_zip_names(self, env, filename):
-        resolved = api_v3_module._safe_backup_path(filename)
-        assert resolved is not None
-        assert resolved.parent == env.export_dir.resolve()
-
-    def test_result_is_always_inside_the_export_dir(self, env):
-        resolved = api_v3_module._safe_backup_path("backup.zip")
-        resolved.relative_to(env.export_dir.resolve())  # raises if outside
-
-    def test_overlong_name_rejected(self, env):
-        assert api_v3_module._safe_backup_path("a" * 250 + ".zip") is None
-
-
-class TestDownload:
-    def test_downloads_an_existing_backup(self, env):
-        make_backup(env.export_dir)
-        response = env.client.get("/api/v3/backup/download/backup-2026-01-01.zip")
-        assert response.status_code == 200
-        assert response.data == b"PK\x03\x04fake zip"
-
-    def test_missing_file_is_a_404(self, env):
-        response = env.client.get("/api/v3/backup/download/never-made.zip")
-        assert response.status_code == 404
-
-    @pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS)
-    def test_traversal_attempts_are_refused(self, env, filename):
-        response = env.client.get(f"/api/v3/backup/download/{filename}")
-        # However the request is turned away — 404 from the containment
-        # check, or 308/405 from routing never matching at all — what
-        # matters is that no file outside the export directory is served.
-        assert response.status_code != 200
-        assert b"do not touch" not in response.data
-
-
-class TestDelete:
-    def test_deletes_an_existing_backup(self, env):
-        path = make_backup(env.export_dir)
-        response = env.client.delete("/api/v3/backup/backup-2026-01-01.zip")
-        assert response.status_code == 200
-        assert not path.exists()
-
-    def test_missing_file_is_a_404(self, env):
-        response = env.client.delete("/api/v3/backup/never-made.zip")
-        assert response.status_code == 404
-
-    @pytest.mark.parametrize("filename", TRAVERSAL_ATTEMPTS)
-    def test_traversal_attempts_delete_nothing(self, env, filename):
-        response = env.client.delete(f"/api/v3/backup/{filename}")
-        assert response.status_code != 200
-        assert env.secret.exists()  # the file a traversal was aiming at
-
-    def test_only_the_named_backup_is_removed(self, env):
-        keep = make_backup(env.export_dir, "keep.zip")
-        drop = make_backup(env.export_dir, "drop.zip")
-        env.client.delete("/api/v3/backup/drop.zip")
-        assert keep.exists()
-        assert not drop.exists()
-
-    def test_directory_with_a_matching_name_is_not_removed(self, env):
-        # The delete loop matches by name but requires a regular file.
-        (env.export_dir / "sneaky.zip").mkdir()
-        response = env.client.delete("/api/v3/backup/sneaky.zip")
-        assert response.status_code == 404
-        assert (env.export_dir / "sneaky.zip").is_dir()
-
-
-class TestList:
-    def test_lists_only_zip_files(self, env):
-        make_backup(env.export_dir, "one.zip")
-        (env.export_dir / "notes.txt").write_text("ignore me")
-        response = env.client.get("/api/v3/backup/list")
-        assert response.status_code == 200
-        names = [entry["filename"] for entry in response.get_json()["data"]]
-        assert names == ["one.zip"]
-
-    def test_empty_directory_lists_nothing(self, env):
-        response = env.client.get("/api/v3/backup/list")
-        assert response.get_json()["data"] == []
-
-    def test_entries_carry_size_and_timestamp(self, env):
-        make_backup(env.export_dir, "one.zip")
-        entry = env.client.get("/api/v3/backup/list").get_json()["data"][0]
-        assert entry["size"] == len(b"PK\x03\x04fake zip")
-        assert entry["created_at"]
-
-
-class TestValidate:
-    def test_missing_file_is_a_400(self, env):
-        response = env.client.post("/api/v3/backup/validate", data={},
-                                   content_type="multipart/form-data")
-        assert response.status_code == 400
-        assert "No backup_file" in response.get_json()["message"]
-
-    def test_invalid_archive_is_a_400(self, env):
-        response = env.client.post(
-            "/api/v3/backup/validate",
-            data={"backup_file": (io.BytesIO(b"not a zip"), "bad.zip")},
-            content_type="multipart/form-data")
-        assert response.status_code == 400
-        assert "Invalid or corrupted" in response.get_json()["message"]
-
-    def test_validation_does_not_leave_temp_files_in_the_export_dir(self, env):
-        env.client.post(
-            "/api/v3/backup/validate",
-            data={"backup_file": (io.BytesIO(b"not a zip"), "bad.zip")},
-            content_type="multipart/form-data")
-        assert list(env.export_dir.iterdir()) == []

test/web_interface/test_api_v3_backup_restore.py

@@ -1,262 +0,0 @@
-"""
-Endpoint tests for POST /backup/restore.
-
-Restore is the most destructive operation the web interface exposes: it
-overwrites config, secrets, WiFi settings and fonts, and reinstalls
-plugins. It had no tests.
-
-restore_backup itself is mocked — this file is about what the route does
-with the request and with the result, not about ZIP handling, which
-belongs to backup_manager's own tests.
-
-Regression coverage for one fixed bug: a malformed `options` field fell
-back to {}, and since every RestoreOptions flag defaults to True, that
-turned a mis-serialized narrow restore into a full one — secrets
-included — with no indication anything had been ignored.
-"""
-
-import io
-import json
-import sys
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-import pytest
-from flask import Flask
-
-project_root = Path(__file__).parent.parent.parent
-sys.path.insert(0, str(project_root))
-
-from web_interface.blueprints.api_v3 import api_v3  # noqa: E402
-
-URL = "/api/v3/backup/restore"
-
-_MANAGER_ATTRS = (
-    'config_manager', 'plugin_manager', 'plugin_store_manager',
-    'plugin_state_manager', 'saved_repositories_manager', 'schema_manager',
-    'operation_queue', 'operation_history', 'cache_manager',
-)
-_SENTINEL = object()
-
-
-class FakeResult:
-    """Stand-in for backup_manager.RestoreResult."""
-
-    def __init__(self, success=True, restored=None, errors=None,
-                 plugins_to_install=None):
-        self.success = success
-        self.restored = restored if restored is not None else ["config"]
-        self.errors = errors or []
-        self.plugins_to_install = plugins_to_install or []
-        self.plugins_installed = []
-        self.plugins_failed = []
-
-    def to_dict(self):
-        return {
-            "success": self.success,
-            "restored": self.restored,
-            "errors": self.errors,
-            "plugins_installed": self.plugins_installed,
-            "plugins_failed": self.plugins_failed,
-        }
-
-
-@pytest.fixture
-def client():
-    originals = {name: getattr(api_v3, name, _SENTINEL) for name in _MANAGER_ATTRS}
-    for name in _MANAGER_ATTRS:
-        setattr(api_v3, name, MagicMock())
-
-    app = Flask(__name__)
-    app.config["TESTING"] = True
-    app.register_blueprint(api_v3, url_prefix="/api/v3")
-    yield app.test_client()
-
-    for name, original in originals.items():
-        if original is _SENTINEL:
-            if hasattr(api_v3, name):
-                delattr(api_v3, name)
-        else:
-            setattr(api_v3, name, original)
-
-
-@pytest.fixture
-def restore():
-    """Patch backup_manager.restore_backup (imported inside the handler)."""
-    with patch("src.backup_manager.restore_backup") as mock:
-        mock.return_value = FakeResult()
-        yield mock
-
-
-def post(client, options=None, filename="backup.zip", content=b"PK\x03\x04fake"):
-    data = {"backup_file": (io.BytesIO(content), filename)}
-    if options is not None:
-        data["options"] = options
-    return client.post(URL, data=data, content_type="multipart/form-data")
-
-
-class TestRequestValidation:
-    def test_missing_file_is_a_400(self, client, restore):
-        response = client.post(URL, data={}, content_type="multipart/form-data")
-        assert response.status_code == 400
-        assert "No backup_file" in response.get_json()["message"]
-        restore.assert_not_called()
-
-    def test_absent_options_defaults_to_a_full_restore(self, client, restore):
-        # Documented default, not the bug: omitting options entirely means
-        # "restore everything".
-        post(client)
-        options = restore.call_args[0][2]
-        assert options.restore_config is True
-        assert options.restore_secrets is True
-        assert options.reinstall_plugins is True
-
-    def test_partial_options_are_honoured(self, client, restore):
-        post(client, options=json.dumps({
-            "restore_secrets": False, "reinstall_plugins": False}))
-        options = restore.call_args[0][2]
-        assert options.restore_secrets is False
-        assert options.reinstall_plugins is False
-        assert options.restore_config is True  # unspecified stays default
-
-    @pytest.mark.parametrize("raw", ["{not json", "", "{'single': 'quotes'}"])
-    def test_malformed_options_are_refused(self, client, restore, raw):
-        # Regression: this fell back to {}, and every flag defaults to
-        # True, so a caller asking for a narrow restore and mis-serializing
-        # it got a full one — secrets overwritten — and no warning.
-        response = post(client, options=raw)
-        assert response.status_code == 400
-        assert "Invalid options" in response.get_json()["message"]
-        restore.assert_not_called()
-
-    @pytest.mark.parametrize("raw", ["[1,2,3]", '"a string"', "42", "true", "null"])
-    def test_options_that_are_not_an_object_are_refused(self, client, restore, raw):
-        response = post(client, options=raw)
-        assert response.status_code == 400
-        restore.assert_not_called()
-
-    def test_empty_object_is_accepted_as_all_defaults(self, client, restore):
-        assert post(client, options="{}").status_code == 200
-        assert restore.call_args[0][2].restore_config is True
-
-
-class TestSuccess:
-    def test_success_returns_the_result(self, client, restore):
-        restore.return_value = FakeResult(success=True, restored=["config", "secrets"])
-        response = post(client)
-        assert response.status_code == 200
-        body = response.get_json()
-        assert body["status"] == "success"
-        assert body["data"]["restored"] == ["config", "secrets"]
-
-    def test_temp_file_is_cleaned_up(self, client, restore):
-        seen = {}
-
-        def capture(path, project_root, options):
-            seen["path"] = Path(path)
-            assert seen["path"].exists()  # present while restoring
-            return FakeResult()
-
-        restore.side_effect = capture
-        post(client)
-        assert not seen["path"].exists()
-
-    def test_temp_file_cleaned_up_even_when_restore_raises(self, client, restore):
-        seen = {}
-
-        def blow_up(path, project_root, options):
-            seen["path"] = Path(path)
-            raise RuntimeError("corrupt archive")
-
-        restore.side_effect = blow_up
-        response = post(client)
-        assert response.status_code == 500
-        assert not seen["path"].exists()
-
-
-class TestPluginReinstall:
-    def test_plugins_are_reinstalled_when_requested(self, client, restore):
-        restore.return_value = FakeResult(
-            plugins_to_install=[{"plugin_id": "clock"}, {"plugin_id": "weather"}])
-        api_v3.plugin_store_manager.install_plugin.return_value = True
-        response = post(client)
-        assert response.status_code == 200
-        assert response.get_json()["data"]["plugins_installed"] == ["clock", "weather"]
-
-    def test_reinstall_skipped_when_not_requested(self, client, restore):
-        restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}])
-        post(client, options=json.dumps({"reinstall_plugins": False}))
-        api_v3.plugin_store_manager.install_plugin.assert_not_called()
-
-    def test_entries_without_a_plugin_id_are_skipped(self, client, restore):
-        restore.return_value = FakeResult(plugins_to_install=[{}, {"plugin_id": "clock"}])
-        api_v3.plugin_store_manager.install_plugin.return_value = True
-        post(client)
-        assert api_v3.plugin_store_manager.install_plugin.call_count == 1
-
-    def test_failed_reinstall_turns_the_whole_restore_into_an_error(
-            self, client, restore):
-        # Pinned as intentional: file restoration succeeded and does not
-        # touch result.errors, but a user whose plugins did not come back
-        # should not be told the restore was a success.
-        restore.return_value = FakeResult(
-            success=True, plugins_to_install=[{"plugin_id": "clock"}])
-        api_v3.plugin_store_manager.install_plugin.return_value = False
-        response = post(client)
-        assert response.status_code == 500
-        body = response.get_json()
-        assert body["status"] == "error"
-        assert "clock" in body["message"]
-
-    def test_message_names_what_landed_and_what_did_not(self, client, restore):
-        restore.return_value = FakeResult(
-            success=True, restored=["config", "fonts"],
-            plugins_to_install=[{"plugin_id": "clock"}])
-        api_v3.plugin_store_manager.install_plugin.return_value = False
-        message = post(client).get_json()["message"]
-        assert "restored: config, fonts" in message
-        assert "plugins not reinstalled: clock" in message
-
-    def test_install_exception_is_recorded_without_leaking_details(
-            self, client, restore):
-        restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}])
-        api_v3.plugin_store_manager.install_plugin.side_effect = RuntimeError(
-            "/srv/internal/path exploded")
-        body = post(client).get_json()
-        failures = body["data"]["plugins_failed"]
-        assert failures[0]["plugin_id"] == "clock"
-        assert "/srv/internal/path" not in json.dumps(body)
-
-    def test_missing_store_manager_is_reported_per_plugin(self, client, restore):
-        restore.return_value = FakeResult(plugins_to_install=[{"plugin_id": "clock"}])
-        api_v3.plugin_store_manager = None
-        with patch("web_interface.blueprints.api_v3.plugin_store_manager", None):
-            body = post(client).get_json()
-        assert body["data"]["plugins_failed"][0]["error"] == "Store manager unavailable"
-
-
-class TestFailureReporting:
-    def test_restore_errors_produce_a_500(self, client, restore):
-        restore.return_value = FakeResult(
-            success=False, restored=[], errors=["config: permission denied"])
-        response = post(client)
-        assert response.status_code == 500
-        assert "permission denied" in response.get_json()["message"]
-
-    def test_partial_restore_names_both_sides(self, client, restore):
-        restore.return_value = FakeResult(
-            success=False, restored=["config"], errors=["secrets: unwritable"])
-        message = post(client).get_json()["message"]
-        assert "restored: config" in message
-        assert "failed: secrets: unwritable" in message
-
-    def test_failure_without_detail_still_says_something(self, client, restore):
-        restore.return_value = FakeResult(success=False, restored=[], errors=[])
-        message = post(client).get_json()["message"]
-        assert "Restore incomplete" in message
-
-    def test_unexpected_exception_is_a_500(self, client, restore):
-        restore.side_effect = RuntimeError("boom")
-        response = post(client)
-        assert response.status_code == 500
-        assert response.get_json()["status"] == "error"

test/web_interface/test_api_v3_config_raw.py

@@ -1,204 +0,0 @@
-"""
-Endpoint tests for POST /config/raw/main and POST /config/raw/secrets.
-
-These write whatever JSON they are given straight to config.json and
-config_secrets.json, bypassing the secret-separation path that
-/config/main and the plugin-config endpoints go through. Given how much
-care the rest of the config surface takes to keep secrets out of
-config.json, an untested pair of endpoints that writes it verbatim is
-worth pinning precisely.
-
-Like test_api_v3_secret_roundtrip.py, these run a REAL ConfigManager over
-tmp_path so the assertions are against files on disk rather than mock
-calls.
-"""
-
-import json
-import sys
-from pathlib import Path
-from unittest.mock import MagicMock
-
-import pytest
-from flask import Flask
-
-project_root = Path(__file__).parent.parent.parent
-sys.path.insert(0, str(project_root))
-
-from src.config_manager import ConfigManager  # noqa: E402
-from src.exceptions import ConfigError  # noqa: E402
-from web_interface.blueprints.api_v3 import api_v3  # noqa: E402
-
-MAIN = "/api/v3/config/raw/main"
-SECRETS = "/api/v3/config/raw/secrets"
-
-
-@pytest.fixture
-def env(tmp_path):
-    config_file = tmp_path / "config.json"
-    config_file.write_text(json.dumps({"timezone": "UTC"}))
-    secrets_file = tmp_path / "config_secrets.json"
-
-    config_manager = ConfigManager(
-        config_path=str(config_file), secrets_path=str(secrets_file))
-    config_manager.template_path = str(tmp_path / "no-template.json")
-
-    _SENTINEL = object()
-    attrs = ('config_manager', 'plugin_manager', 'plugin_store_manager',
-             'plugin_state_manager', 'saved_repositories_manager',
-             'schema_manager', 'operation_queue', 'operation_history',
-             'cache_manager')
-    originals = {name: getattr(api_v3, name, _SENTINEL) for name in attrs}
-
-    for name in attrs:
-        setattr(api_v3, name, MagicMock())
-    api_v3.config_manager = config_manager
-
-    app = Flask(__name__)
-    app.config["TESTING"] = True
-    app.register_blueprint(api_v3, url_prefix="/api/v3")
-
-    class Env:
-        pass
-
-    e = Env()
-    e.client = app.test_client()
-    e.config_manager = config_manager
-    e.config_file = config_file
-    e.secrets_file = secrets_file
-    yield e
-
-    for name, original in originals.items():
-        if original is _SENTINEL:
-            if hasattr(api_v3, name):
-                delattr(api_v3, name)
-        else:
-            setattr(api_v3, name, original)
-
-
-class TestSaveRawMain:
-    def test_writes_the_body_to_config_json(self, env):
-        response = env.client.post(MAIN, json={"timezone": "America/Chicago"})
-        assert response.status_code == 200
-        assert json.loads(env.config_file.read_text()) == {"timezone": "America/Chicago"}
-
-    def test_replaces_rather_than_merges(self, env):
-        env.client.post(MAIN, json={"only": "this"})
-        assert json.loads(env.config_file.read_text()) == {"only": "this"}
-
-    def test_does_not_touch_the_secrets_file(self, env):
-        env.secrets_file.write_text(json.dumps({"weather": {"api_key": "k"}}))
-        env.client.post(MAIN, json={"timezone": "UTC"})
-        assert json.loads(env.secrets_file.read_text()) == {"weather": {"api_key": "k"}}
-
-    def test_uninitialized_manager_is_a_500(self, env):
-        api_v3.config_manager = None
-        response = env.client.post(MAIN, json={"timezone": "UTC"})
-        assert response.status_code == 500
-        assert "not initialized" in response.get_json()["message"]
-
-    def test_empty_object_is_a_400(self, env):
-        response = env.client.post(MAIN, json={})
-        assert response.status_code == 400
-        assert "No data provided" in response.get_json()["message"]
-
-    def test_bodyless_post_is_a_400(self, env):
-        response = env.client.post(MAIN)
-        assert response.status_code == 400
-        assert "No data provided" in response.get_json()["message"]
-
-    def test_malformed_json_is_a_400_in_the_app_shape(self, env):
-        response = env.client.post(MAIN, data="{not json",
-                                   content_type="application/json")
-        assert response.status_code == 400
-        body = response.get_json()
-        assert body["status"] == "error"
-        # A body that was sent but does not parse is a distinct mistake
-        # from sending none, and says so. Previously the handler's own
-        # json.JSONDecodeError arm was unreachable — Werkzeug raised
-        # first — so this collapsed into "No data provided".
-        assert "Invalid JSON in request body" in body["message"]
-
-    def test_config_error_is_a_500_with_context(self, env, monkeypatch):
-        def refuse(kind, data):
-            raise ConfigError("cannot write", config_path="/etc/x.json")
-        monkeypatch.setattr(env.config_manager, "save_raw_file_content", refuse)
-        response = env.client.post(MAIN, json={"timezone": "UTC"})
-        assert response.status_code == 500
-        assert "/etc/x.json" in json.dumps(response.get_json())
-
-    def test_unexpected_error_is_a_500(self, env, monkeypatch):
-        def boom(kind, data):
-            raise RuntimeError("disk on fire")
-        monkeypatch.setattr(env.config_manager, "save_raw_file_content", boom)
-        response = env.client.post(MAIN, json={"timezone": "UTC"})
-        assert response.status_code == 500
-        assert response.get_json()["status"] == "error"
-
-
-class TestSaveRawSecrets:
-    def test_writes_only_to_the_secrets_file(self, env):
-        response = env.client.post(SECRETS, json={"weather": {"api_key": "s3cret"}})
-        assert response.status_code == 200
-        assert json.loads(env.secrets_file.read_text()) == {"weather": {"api_key": "s3cret"}}
-
-    def test_secret_values_never_reach_config_json(self, env):
-        env.client.post(SECRETS, json={"weather": {"api_key": "s3cret"}})
-        assert "s3cret" not in env.config_file.read_text()
-
-    def test_existing_main_config_is_untouched(self, env):
-        before = env.config_file.read_text()
-        env.client.post(SECRETS, json={"weather": {"api_key": "k"}})
-        assert env.config_file.read_text() == before
-
-    def test_github_token_is_reloaded_for_the_store_manager(self, env):
-        store = MagicMock()
-        store._load_github_token.return_value = "ghp_new"
-        api_v3.plugin_store_manager = store
-        env.client.post(SECRETS, json={"github": {"token": "ghp_new"}})
-        store._load_github_token.assert_called_once()
-        assert store.github_token == "ghp_new"
-
-    def test_absent_store_manager_is_fine(self, env):
-        api_v3.plugin_store_manager = None
-        assert env.client.post(SECRETS, json={"a": 1}).status_code == 200
-
-    def test_uninitialized_manager_is_a_500(self, env):
-        api_v3.config_manager = None
-        assert env.client.post(SECRETS, json={"a": 1}).status_code == 500
-
-    def test_empty_object_is_a_400(self, env):
-        assert env.client.post(SECRETS, json={}).status_code == 400
-
-    def test_bodyless_post_is_a_400(self, env):
-        assert env.client.post(SECRETS).status_code == 400
-
-    def test_error_is_a_500(self, env, monkeypatch):
-        def boom(kind, data):
-            raise RuntimeError("nope")
-        monkeypatch.setattr(env.config_manager, "save_raw_file_content", boom)
-        assert env.client.post(SECRETS, json={"a": 1}).status_code == 500
-
-
-class TestRawEndpointsBypassSecretSeparation:
-    """Pinned behaviour, deliberately not "fixed".
-
-    These endpoints are the escape hatch for editing the config files
-    directly from the web UI's raw JSON editor. They write what they are
-    given, so a secret typed into the main-config editor lands in
-    config.json in plain text — unlike /config/main and the plugin-config
-    endpoints, which route x-secret fields into config_secrets.json.
-
-    That is the point of a raw editor, but it is a sharp edge worth
-    stating out loud: anyone adding a "convenience" that posts plugin
-    config through this endpoint would silently lose secret separation.
-    """
-
-    def test_secret_shaped_keys_are_written_verbatim_to_main(self, env):
-        env.client.post(MAIN, json={"weather": {"api_key": "PLAINTEXT-KEY"}})
-        on_disk = json.loads(env.config_file.read_text())
-        assert on_disk["weather"]["api_key"] == "PLAINTEXT-KEY"
-
-    def test_no_separation_happens_on_the_raw_path(self, env):
-        env.client.post(MAIN, json={"weather": {"api_key": "PLAINTEXT-KEY"}})
-        # Nothing was moved aside into the secrets file.
-        assert not env.secrets_file.exists() or "PLAINTEXT-KEY" not in env.secrets_file.read_text()

test/web_interface/test_calendar_oauth_endpoints.py

@@ -0,0 +1,408 @@
+"""Tests the calendar plugin's OAuth and calendar-listing endpoints.
+
+The plugin's config UI advertised a three-step setup, but only step 1 existed
+on the server. Step 3's picker fetched /api/v3/plugins/calendar/list-calendars,
+which was never registered, so Flask fell through to the global 404 handler and
+the user saw "Resource not found" — with nothing to say which resource. Step 2
+had no endpoint either, and no field in the schema at all, even though the
+plugin ships calendar_registration.py written expressly for a web-driven
+two-step flow.
+
+These cover the two new routes: that they exist, that they fail with something
+actionable rather than a bare 404, and that the shapes the widgets consume are
+what the server actually sends.
+"""
+
+import json
+import pickle
+import sys
+from pathlib import Path
+
+import pytest
+
+project_root = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(project_root))
+
+from web_interface.blueprints import api_v3 as mod  # noqa: E402
+
+
+@pytest.fixture
+def client(monkeypatch, tmp_path):
+    """A test client whose calendar plugin lives in tmp_path."""
+    from flask import Flask
+
+    plugin_dir = tmp_path / 'calendar'
+    plugin_dir.mkdir()
+
+    app = Flask(__name__)
+    app.register_blueprint(mod.api_v3, url_prefix='/api/v3')
+    app.config['TESTING'] = True
+    monkeypatch.setattr(mod, '_calendar_plugin_dir', lambda: plugin_dir)
+    with app.test_client() as c:
+        c.plugin_dir = plugin_dir
+        yield c
+
+
+@pytest.fixture
+def uninstalled(monkeypatch):
+    from flask import Flask
+
+    app = Flask(__name__)
+    app.register_blueprint(mod.api_v3, url_prefix='/api/v3')
+    app.config['TESTING'] = True
+    monkeypatch.setattr(mod, '_calendar_plugin_dir', lambda: None)
+    with app.test_client() as c:
+        yield c
+
+
+class TestTheRoutesExistAtAll:
+    """The original bug: the URLs the widgets call were not registered."""
+
+    def test_list_calendars_is_routed(self, client):
+        response = client.get('/api/v3/plugins/calendar/list-calendars')
+        # Reaching the handler is the whole point; what it then says about
+        # missing setup is TestItSaysWhatIsWrong's business.
+        assert response.status_code != 404, "still unrouted"
+        assert response.get_json()['message'] != 'Resource not found'
+
+    def test_authenticate_is_routed(self, client):
+        response = client.post('/api/v3/plugins/calendar/authenticate', json={})
+        assert response.status_code != 404, "still unrouted"
+        assert response.get_json()['message'] != 'Resource not found'
+
+    def test_both_urls_match_what_the_widgets_request(self):
+        # The widgets hardcode these; a rename on either side reintroduces the
+        # original bug silently.
+        picker = Path(project_root) / 'web_interface/static/v3/js/widgets/google-calendar-picker.js'
+        oauth = Path(project_root) / 'web_interface/static/v3/js/widgets/google-oauth.js'
+        assert '/api/v3/plugins/calendar/list-calendars' in picker.read_text(encoding='utf-8')
+        assert '/api/v3/plugins/calendar/authenticate' in oauth.read_text(encoding='utf-8')
+        source = (Path(project_root) / 'web_interface/blueprints/api_v3.py').read_text(encoding='utf-8')
+        assert "'/plugins/calendar/list-calendars'" in source
+        assert "'/plugins/calendar/authenticate'" in source
+
+    def test_the_oauth_widget_is_dispatched_not_rendered_as_a_text_box(self):
+        # The string branch of the config template dispatches on an allow-list
+        # of widget names; anything missing from it silently falls through to a
+        # plain <input type="text">. That produced two boxes on the calendar
+        # page -- the widget's own, and a stray one for the same field -- and
+        # no way to tell which to paste into.
+        template = (Path(project_root)
+                    / 'web_interface/templates/v3/partials/plugin_config.html'
+                    ).read_text(encoding='utf-8')
+        allow_list_line = [ln for ln in template.splitlines()
+                           if "str_widget in [" in ln]
+        assert allow_list_line, "the string widget allow-list moved"
+        assert "'google-oauth'" in allow_list_line[0], allow_list_line[0]
+
+    def test_the_widget_script_is_served(self):
+        base = (Path(project_root) / 'web_interface/templates/v3/base.html'
+                ).read_text(encoding='utf-8')
+        assert 'widgets/google-oauth.js' in base
+
+    def test_the_status_line_is_announced(self):
+        # Every message the widget gives arrives after an async call, so a
+        # screen reader hears nothing unless the element is a live region.
+        widget = (Path(project_root)
+                  / 'web_interface/static/v3/js/widgets/google-oauth.js'
+                  ).read_text(encoding='utf-8')
+        # Both attributes must be on the *status* element. Searching for them
+        # separately would pass with each on a different node, which announces
+        # nothing.
+        assert "status.setAttribute('role', 'status')" in widget, widget[:0]
+        assert "status.setAttribute('aria-live', 'polite')" in widget
+
+    def test_the_paste_box_has_an_accessible_name(self):
+        # A visible label is not enough on its own: without the association the
+        # input's only name is a placeholder, which vanishes on focus -- which
+        # is exactly when the value is being pasted.
+        widget = (Path(project_root)
+                  / 'web_interface/static/v3/js/widgets/google-oauth.js'
+                  ).read_text(encoding='utf-8')
+        # The binding is what matters, not that both lines exist: a `for` and
+        # an `id` that disagree leave the input just as anonymous. Both must
+        # go through the same identifier.
+        import re as _re
+        for_target = _re.search(r"codeLabel\.setAttribute\('for',\s*(\w+)\)", widget)
+        id_source = _re.search(r"codeInput\.id\s*=\s*(\w+)", widget)
+        assert for_target and id_source, (for_target, id_source)
+        assert for_target.group(1) == id_source.group(1), (
+            "label points at %r but the input is %r"
+            % (for_target.group(1), id_source.group(1)))
+
+    def test_the_failed_page_is_called_out_loudly(self):
+        # The loopback redirect lands on a browser error page at exactly the
+        # moment the user has to act. In small grey text it gets missed and the
+        # flow reads as broken while it is working.
+        widget = (Path(project_root)
+                  / 'web_interface/static/v3/js/widgets/google-oauth.js'
+                  ).read_text(encoding='utf-8')
+        assert 'expected' in widget.lower()
+        assert 'amber' in widget, "the warning is not visually distinguished"
+
+
+class TestItSaysWhatIsWrong:
+    def test_listing_without_a_token_asks_for_step_2(self, client):
+        response = client.get('/api/v3/plugins/calendar/list-calendars')
+        assert response.status_code == 400
+        body = response.get_json()
+        assert body['status'] == 'error'
+        assert 'step 2' in body['message'].lower(), body['message']
+
+    def test_authenticating_without_credentials_asks_for_step_1(self, client):
+        response = client.post('/api/v3/plugins/calendar/authenticate', json={})
+        assert response.status_code == 400
+        assert 'step 1' in response.get_json()['message'].lower()
+
+    def test_an_uninstalled_plugin_says_so(self, uninstalled):
+        for response in (
+            uninstalled.get('/api/v3/plugins/calendar/list-calendars'),
+            uninstalled.post('/api/v3/plugins/calendar/authenticate', json={}),
+        ):
+            assert response.status_code == 404
+            # A 404 here is honest -- but it must name the plugin, not read as
+            # the generic "Resource not found" that started this.
+            assert 'not installed' in response.get_json()['message'].lower()
+
+
+class TestTheScriptRunner:
+    def test_it_returns_the_json_the_script_prints(self, tmp_path):
+        script = tmp_path / 'calendar_registration.py'
+        script.write_text(
+            'print(\'{"status": "success", "auth_url": "https://x"}\')\n',
+            encoding='utf-8')
+        payload, error = mod._run_calendar_registration(tmp_path, '')
+        assert error is None
+        assert payload['auth_url'] == 'https://x'
+
+    def test_it_ignores_noise_before_the_json(self, tmp_path):
+        # An import warning or a library writing to stdout would otherwise
+        # make the last-line parse fail.
+        script = tmp_path / 'calendar_registration.py'
+        script.write_text(
+            'print("some library warning")\n'
+            'print(\'{"status": "success"}\')\n', encoding='utf-8')
+        payload, error = mod._run_calendar_registration(tmp_path, '')
+        assert error is None and payload['status'] == 'success'
+
+    def test_it_passes_stdin_through(self, tmp_path):
+        script = tmp_path / 'calendar_registration.py'
+        script.write_text(
+            'import sys, json\n'
+            'print(json.dumps({"status": "success", "got": sys.stdin.read().strip()}))\n',
+            encoding='utf-8')
+        payload, _ = mod._run_calendar_registration(tmp_path, 'http://127.0.0.1/?code=abc')
+        assert payload['got'] == 'http://127.0.0.1/?code=abc'
+
+    def test_a_missing_script_is_reported(self, tmp_path):
+        payload, error = mod._run_calendar_registration(tmp_path, '')
+        assert payload is None
+        assert 'script not found' in error.lower()
+
+    def test_output_that_is_not_json_is_reported_with_context(self, tmp_path):
+        script = tmp_path / 'calendar_registration.py'
+        script.write_text('import sys\nsys.stderr.write("boom\\n")\n', encoding='utf-8')
+        payload, error = mod._run_calendar_registration(tmp_path, '')
+        assert payload is None
+        assert 'no result' in error.lower()
+        assert 'boom' in error
+
+
+class TestListingShape:
+    """The picker reads cal.id, cal.summary and cal.primary."""
+
+    def _authenticate(self, client, monkeypatch, items):
+        creds = type('C', (), {'expired': False, 'refresh_token': None, 'valid': True})()
+        (client.plugin_dir / 'token.pickle').write_bytes(pickle.dumps({'x': 1}))
+        monkeypatch.setattr(mod.pickle if hasattr(mod, 'pickle') else pickle,
+                            'loads', lambda *a, **k: creds, raising=False)
+
+        import types
+        fake_pickle = types.SimpleNamespace(load=lambda f: creds, dump=lambda *a: None)
+        # Callers pass a flat list of calendars; the API returns them wrapped
+        # in a page. One page is all these cases need -- TestPagination builds
+        # its own multi-page sequences.
+        pages = [{'items': items}]
+
+        state = {'i': 0}
+
+        def fake_list(**kwargs):
+            page = pages[min(state['i'], len(pages) - 1)]
+            state['i'] += 1
+            return types.SimpleNamespace(execute=lambda: page)
+
+        def fake_build(*args, **kwargs):
+            return types.SimpleNamespace(
+                calendarList=lambda: types.SimpleNamespace(list=fake_list))
+
+        real_import = __builtins__['__import__'] if isinstance(__builtins__, dict) \
+            else __builtins__.__import__
+
+        def fake_import(name, *args, **kwargs):
+            if name == 'pickle':
+                return fake_pickle
+            if name == 'google.auth.transport.requests':
+                return types.SimpleNamespace(Request=object)
+            if name == 'googleapiclient.discovery':
+                return types.SimpleNamespace(build=fake_build)
+            return real_import(name, *args, **kwargs)
+
+        monkeypatch.setattr('builtins.__import__', fake_import)
+
+    def test_it_returns_id_summary_and_primary(self, client, monkeypatch):
+        self._authenticate(client, monkeypatch, [
+            {'id': 'b@x', 'summary': 'Work'},
+            {'id': 'a@x', 'summary': 'Personal', 'primary': True},
+        ])
+        body = client.get('/api/v3/plugins/calendar/list-calendars').get_json()
+        assert body['status'] == 'success'
+        assert {c['id'] for c in body['calendars']} == {'a@x', 'b@x'}
+        assert all(set(c) == {'id', 'summary', 'primary'} for c in body['calendars'])
+
+    def test_the_primary_calendar_comes_first(self, client, monkeypatch):
+        # Short list, but the one the user wants is almost always their own.
+        self._authenticate(client, monkeypatch, [
+            {'id': 'z@x', 'summary': 'Aardvarks'},
+            {'id': 'a@x', 'summary': 'Zebras', 'primary': True},
+        ])
+        body = client.get('/api/v3/plugins/calendar/list-calendars').get_json()
+        assert body['calendars'][0]['id'] == 'a@x'
+        assert body['calendars'][0]['primary'] is True
+
+    def test_a_calendar_without_a_name_still_lists(self, client, monkeypatch):
+        self._authenticate(client, monkeypatch, [{'id': 'noname@x'}])
+        body = client.get('/api/v3/plugins/calendar/list-calendars').get_json()
+        assert body['calendars'][0]['summary'] == 'noname@x'
+
+    def test_entries_without_an_id_are_dropped(self, client, monkeypatch):
+        # Nothing could be selected by such a row, and the checkbox value
+        # would be undefined.
+        self._authenticate(client, monkeypatch, [{'summary': 'ghost'}, {'id': 'real@x'}])
+        body = client.get('/api/v3/plugins/calendar/list-calendars').get_json()
+        assert [c['id'] for c in body['calendars']] == ['real@x']
+
+
+class TestPagination:
+    """calendarList.list pages at 250 and defaults to 100."""
+
+    def _paged(self, client, monkeypatch, pages):
+        import types
+        creds = type('C', (), {'expired': False, 'refresh_token': None, 'valid': True})()
+        (client.plugin_dir / 'token.pickle').write_bytes(b'x')
+        state = {'i': 0}
+        seen = []
+
+        def fake_list(**kwargs):
+            seen.append(kwargs)
+            page = pages[min(state['i'], len(pages) - 1)]
+            state['i'] += 1
+            return types.SimpleNamespace(execute=lambda: page)
+
+        def fake_build(*args, **kwargs):
+            return types.SimpleNamespace(
+                calendarList=lambda: types.SimpleNamespace(list=fake_list))
+
+        real_import = __builtins__['__import__'] if isinstance(__builtins__, dict) \
+            else __builtins__.__import__
+
+        def fake_import(name, *args, **kwargs):
+            if name == 'pickle':
+                return types.SimpleNamespace(load=lambda f: creds, dump=lambda *a: None)
+            if name == 'google.auth.transport.requests':
+                return types.SimpleNamespace(Request=object)
+            if name == 'googleapiclient.discovery':
+                return types.SimpleNamespace(build=fake_build)
+            return real_import(name, *args, **kwargs)
+
+        monkeypatch.setattr('builtins.__import__', fake_import)
+        return seen
+
+    def test_every_page_is_collected(self, client, monkeypatch):
+        # Taking only the first page would hide calendars from the picker with
+        # nothing to say the list was cut short.
+        self._paged(client, monkeypatch, [
+            {'items': [{'id': 'a@x', 'summary': 'A'}], 'nextPageToken': 't1'},
+            {'items': [{'id': 'b@x', 'summary': 'B'}], 'nextPageToken': 't2'},
+            {'items': [{'id': 'c@x', 'summary': 'C'}]},
+        ])
+        body = client.get('/api/v3/plugins/calendar/list-calendars').get_json()
+        assert [c['id'] for c in body['calendars']] == ['a@x', 'b@x', 'c@x']
+
+    def test_the_page_token_is_passed_back(self, client, monkeypatch):
+        seen = self._paged(client, monkeypatch, [
+            {'items': [{'id': 'a@x', 'summary': 'A'}], 'nextPageToken': 'tok'},
+            {'items': [{'id': 'b@x', 'summary': 'B'}]},
+        ])
+        client.get('/api/v3/plugins/calendar/list-calendars')
+        assert seen[0]['pageToken'] is None
+        assert seen[1]['pageToken'] == 'tok'
+        assert all(k['maxResults'] == 250 for k in seen)
+
+    def test_a_looping_token_cannot_spin_forever(self, client, monkeypatch):
+        # Every page claims another follows.
+        self._paged(client, monkeypatch, [
+            {'items': [{'id': 'a@x', 'summary': 'A'}], 'nextPageToken': 'same'},
+        ])
+        body = client.get('/api/v3/plugins/calendar/list-calendars').get_json()
+        assert body['status'] == 'success'
+        assert len(body['calendars']) <= mod._CALENDAR_LIST_MAX_PAGES
+
+
+class TestDiagnosticsAreRedacted:
+    def test_script_stderr_is_redacted_on_the_way_out(self, tmp_path):
+        script = tmp_path / 'calendar_registration.py'
+        script.write_text(
+            'import sys\n'
+            'sys.stderr.write("boom client_secret=hunter2 more\\n")\n',
+            encoding='utf-8')
+        payload, error = mod._run_calendar_registration(tmp_path, '')
+        assert payload is None
+        assert 'hunter2' not in error, error
+        assert '<redacted>' in error, error
+
+    def test_a_failing_script_payload_is_redacted(self, client):
+        (client.plugin_dir / 'credentials.json').write_text('{}', encoding='utf-8')
+        (client.plugin_dir / 'calendar_registration.py').write_text(
+            'import json\n'
+            'print(json.dumps({"status": "error", '
+            '"message": "Failed: client_secret=topsecret"}))\n',
+            encoding='utf-8')
+        body = client.post('/api/v3/plugins/calendar/authenticate',
+                           json={}).get_json()
+        assert body['status'] == 'error'
+        assert 'topsecret' not in json.dumps(body), body
+        assert '<redacted>' in body['message'], body
+
+    def test_an_unrunnable_script_is_reported_without_raw_exception_text(self,
+                                                                        tmp_path,
+                                                                        monkeypatch):
+        # OSError from the spawn carries the interpreter path and whatever the
+        # OS chose to say; it reaches the client through the redactor like
+        # everything else.
+        script = tmp_path / 'calendar_registration.py'
+        script.write_text('', encoding='utf-8')
+
+        def boom(*a, **k):
+            raise OSError("Exec format error: token=abcd1234 /usr/bin/python3")
+
+        monkeypatch.setattr(mod.subprocess, 'run', boom)
+        payload, error = mod._run_calendar_registration(tmp_path, '')
+        assert payload is None
+        assert 'abcd1234' not in error, error
+        assert 'OSError' in error, error
+
+    def test_a_missing_google_library_is_reported_without_raw_exception_text(
+            self, client, monkeypatch):
+        (client.plugin_dir / 'token.pickle').write_bytes(b'x')
+        real_import = __builtins__['__import__'] if isinstance(__builtins__, dict) \
+            else __builtins__.__import__
+
+        def fake_import(name, *args, **kwargs):
+            if name.startswith('google'):
+                raise ImportError("No module named 'google' password=hunter2")
+            return real_import(name, *args, **kwargs)
+
+        monkeypatch.setattr('builtins.__import__', fake_import)
+        body = client.get('/api/v3/plugins/calendar/list-calendars').get_json()
+        assert 'hunter2' not in json.dumps(body), body
+        assert 'requirements.txt' in body['message']

test/web_interface/test_error_handler.py

@@ -1,149 +0,0 @@
-"""
-Tests for the response builders in src/web_interface/error_handler.py and
-the success path in src/web_interface/api_helpers.py.
-
-describe_exception() in the same module is already covered by
-test/test_web_error_detail.py and is not duplicated here.
-
-Regression coverage for one fixed bug: create_success_response used
-truthiness for `message` and `metadata` while using `is not None` for
-`data`, so an explicitly-passed "" or {} was silently dropped —
-api_helpers.success_response() repeated the same gate, which is the path
-every api_v3 endpoint actually calls.
-"""
-
-import pytest
-from flask import Flask
-
-from src.web_interface.api_helpers import success_response
-from src.web_interface.error_handler import (
-    create_error_response,
-    create_success_response,
-)
-from src.web_interface.errors import ErrorCode, WebInterfaceError
-
-
-@pytest.fixture
-def app():
-    return Flask(__name__)
-
-
-class TestCreateErrorResponse:
-    def test_returns_response_and_status_tuple(self, app):
-        with app.test_request_context():
-            response, status = create_error_response(
-                ErrorCode.CONFIG_SAVE_FAILED, "could not save")
-        assert status == 500
-        assert response.get_json()["message"] == "could not save"
-
-    def test_status_code_passthrough(self, app):
-        with app.test_request_context():
-            _, status = create_error_response(
-                ErrorCode.INVALID_INPUT, "bad", status_code=400)
-        assert status == 400
-
-    def test_body_matches_the_error_dataclass(self, app):
-        with app.test_request_context():
-            response, _ = create_error_response(
-                ErrorCode.NETWORK_ERROR, "offline",
-                details="connection refused", context={"url": "http://x"})
-        expected = WebInterfaceError(
-            error_code=ErrorCode.NETWORK_ERROR, message="offline",
-            details="connection refused", context={"url": "http://x"}).to_dict()
-        assert response.get_json() == expected
-
-    def test_none_context_produces_no_context_key(self, app):
-        with app.test_request_context():
-            response, _ = create_error_response(ErrorCode.SYSTEM_ERROR, "boom")
-        assert "context" not in response.get_json()
-
-    def test_suggested_fixes_passed_through(self, app):
-        with app.test_request_context():
-            response, _ = create_error_response(
-                ErrorCode.SYSTEM_ERROR, "boom", suggested_fixes=["Try again"])
-        assert response.get_json()["suggested_fixes"] == ["Try again"]
-
-
-class TestCreateSuccessResponse:
-    def test_bare_success(self):
-        assert create_success_response() == {"status": "success"}
-
-    def test_data_included(self):
-        assert create_success_response(data={"a": 1})["data"] == {"a": 1}
-
-    @pytest.mark.parametrize("falsy", [0, "", False, {}, []])
-    def test_falsy_data_is_still_included(self, falsy):
-        assert create_success_response(data=falsy)["data"] == falsy
-
-    def test_none_data_omitted(self):
-        assert "data" not in create_success_response(data=None)
-
-    def test_message_included(self):
-        assert create_success_response(message="done")["message"] == "done"
-
-    def test_empty_message_is_still_included(self):
-        # Regression: `if message:` dropped an explicitly-passed "".
-        assert create_success_response(message="")["message"] == ""
-
-    def test_none_message_omitted(self):
-        assert "message" not in create_success_response(message=None)
-
-    def test_metadata_included(self):
-        assert create_success_response(metadata={"v": 1})["metadata"] == {"v": 1}
-
-    def test_empty_metadata_is_still_included(self):
-        # Regression: `if metadata:` dropped an explicitly-passed {}.
-        assert create_success_response(metadata={})["metadata"] == {}
-
-    def test_none_metadata_omitted(self):
-        assert "metadata" not in create_success_response(metadata=None)
-
-
-class TestSuccessResponseHelper:
-    """api_helpers.success_response — the wrapper every endpoint calls."""
-
-    def test_plain_response_has_no_metadata_block(self, app):
-        with app.test_request_context():
-            body = success_response(data={"a": 1}).get_json()
-        assert body == {"status": "success", "data": {"a": 1}}
-
-    def test_explicit_empty_metadata_survives_the_wrapper(self, app):
-        # Regression: the wrapper re-gated metadata on truthiness after
-        # create_success_response had already included it, so {} was
-        # dropped again on the way out.
-        with app.test_request_context():
-            body = success_response(data=None, metadata={}).get_json()
-        assert body["metadata"] == {}
-
-    def test_caller_metadata_preserved(self, app):
-        with app.test_request_context():
-            body = success_response(metadata={"version": "1.2"}).get_json()
-        assert body["metadata"]["version"] == "1.2"
-
-    def test_timing_added_when_request_has_start_time(self, app):
-        with app.test_request_context() as ctx:
-            ctx.request.start_time = 0.0
-            body = success_response(data={"a": 1}).get_json()
-        assert "response_time_ms" in body["metadata"]
-
-    def test_timing_merges_with_caller_metadata(self, app):
-        with app.test_request_context() as ctx:
-            ctx.request.start_time = 0.0
-            body = success_response(metadata={"version": "1.2"}).get_json()
-        assert body["metadata"]["version"] == "1.2"
-        assert "response_time_ms" in body["metadata"]
-
-    def test_caller_metadata_dict_is_not_mutated(self, app):
-        # The helper used to add response_time_ms straight into the dict the
-        # caller passed, so a module-level or reused metadata dict would
-        # accumulate timings from previous requests.
-        caller_metadata = {"version": "1.2"}
-        with app.test_request_context() as ctx:
-            ctx.request.start_time = 0.0
-            success_response(metadata=caller_metadata)
-        assert caller_metadata == {"version": "1.2"}
-
-    def test_message_passed_through(self, app):
-        with app.test_request_context():
-            body = success_response(message="saved").get_json()
-        assert body["message"] == "saved"

test/web_interface/test_errors.py

@@ -1,208 +0,0 @@
-"""
-Tests for src/web_interface/errors.py — the structured error type behind
-every API error response (category inference, default suggestions, the
-JSON shape, and exception conversion).
-
-Pure logic; no Flask context needed.
-
-Regression coverage for one fixed bug: suggested_fixes used `or`, so a
-caller passing [] to mean "no suggestions" silently got the default list.
-"""
-
-import pytest
-
-from src.web_interface.errors import ErrorCategory, ErrorCode, WebInterfaceError
-
-
-class TestCategoryInference:
-    @pytest.mark.parametrize("code,expected", [
-        (ErrorCode.CONFIG_SAVE_FAILED, ErrorCategory.CONFIGURATION),
-        (ErrorCode.CONFIG_ROLLBACK_FAILED, ErrorCategory.CONFIGURATION),
-        (ErrorCode.PLUGIN_NOT_FOUND, ErrorCategory.PLUGIN),
-        (ErrorCode.PLUGIN_OPERATION_CONFLICT, ErrorCategory.PLUGIN),
-        (ErrorCode.VALIDATION_ERROR, ErrorCategory.VALIDATION),
-        (ErrorCode.SCHEMA_VALIDATION_FAILED, ErrorCategory.VALIDATION),
-        (ErrorCode.INVALID_INPUT, ErrorCategory.VALIDATION),
-        (ErrorCode.NETWORK_ERROR, ErrorCategory.NETWORK),
-        (ErrorCode.API_ERROR, ErrorCategory.NETWORK),
-        (ErrorCode.TIMEOUT, ErrorCategory.NETWORK),
-        (ErrorCode.PERMISSION_DENIED, ErrorCategory.PERMISSION),
-        (ErrorCode.FILE_PERMISSION_ERROR, ErrorCategory.PERMISSION),
-        (ErrorCode.SYSTEM_ERROR, ErrorCategory.SYSTEM),
-        (ErrorCode.SERVICE_UNAVAILABLE, ErrorCategory.SYSTEM),
-        (ErrorCode.UNKNOWN_ERROR, ErrorCategory.UNKNOWN),
-    ])
-    def test_every_code_prefix_maps_to_its_category(self, code, expected):
-        assert WebInterfaceError(code, "msg").category is expected
-
-    def test_explicit_category_overrides_inference(self):
-        error = WebInterfaceError(
-            ErrorCode.CONFIG_SAVE_FAILED, "msg", category=ErrorCategory.SYSTEM)
-        assert error.category is ErrorCategory.SYSTEM
-
-    def test_every_error_code_gets_a_category(self):
-        # No code may fall through uncategorized as the enum grows.
-        for code in ErrorCode:
-            assert isinstance(WebInterfaceError(code, "msg").category, ErrorCategory)
-
-
-class TestDefaultSuggestions:
-    def test_mapped_code_gets_specific_suggestions(self):
-        fixes = WebInterfaceError(ErrorCode.CONFIG_SAVE_FAILED, "msg").suggested_fixes
-        assert "Check available disk space" in fixes
-
-    def test_unmapped_code_gets_generic_fallback(self):
-        # PLUGIN_UPDATE_FAILED has no entry in suggestions_map.
-        fixes = WebInterfaceError(ErrorCode.PLUGIN_UPDATE_FAILED, "msg").suggested_fixes
-        assert fixes == ["Review error details and try again"]
-
-    def test_explicit_suggestions_win(self):
-        error = WebInterfaceError(
-            ErrorCode.CONFIG_SAVE_FAILED, "msg", suggested_fixes=["Do the thing"])
-        assert error.suggested_fixes == ["Do the thing"]
-
-    def test_explicit_empty_list_is_respected(self):
-        # Regression: `suggested_fixes or default` treated [] as "unset",
-        # so a caller could not express "I have no suggestions".
-        error = WebInterfaceError(
-            ErrorCode.CONFIG_SAVE_FAILED, "msg", suggested_fixes=[])
-        assert error.suggested_fixes == []
-
-    def test_none_still_gets_defaults(self):
-        error = WebInterfaceError(
-            ErrorCode.CONFIG_SAVE_FAILED, "msg", suggested_fixes=None)
-        assert len(error.suggested_fixes) > 0
-
-
-class TestToDict:
-    def test_base_keys_always_present(self):
-        result = WebInterfaceError(ErrorCode.SYSTEM_ERROR, "boom").to_dict()
-        assert result["status"] == "error"
-        assert result["error_code"] == "SYSTEM_ERROR"
-        assert result["error_category"] == "system"
-        assert result["message"] == "boom"
-
-    def test_details_included_when_set(self):
-        result = WebInterfaceError(
-            ErrorCode.SYSTEM_ERROR, "boom", details="disk full").to_dict()
-        assert result["details"] == "disk full"
-
-    def test_details_omitted_when_absent(self):
-        assert "details" not in WebInterfaceError(ErrorCode.SYSTEM_ERROR, "boom").to_dict()
-
-    def test_context_included_when_non_empty(self):
-        result = WebInterfaceError(
-            ErrorCode.SYSTEM_ERROR, "boom", context={"path": "/tmp/x"}).to_dict()
-        assert result["context"] == {"path": "/tmp/x"}
-
-    def test_empty_context_is_omitted(self):
-        # Pinned as intentional, not a bug: __init__ normalizes context to
-        # {}, and an empty context carries no information, so it is left out
-        # rather than padding every error body with "context": {}.
-        result = WebInterfaceError(ErrorCode.SYSTEM_ERROR, "boom", context={}).to_dict()
-        assert "context" not in result
-
-    def test_empty_suggestions_omitted(self):
-        result = WebInterfaceError(
-            ErrorCode.SYSTEM_ERROR, "boom", suggested_fixes=[]).to_dict()
-        assert "suggested_fixes" not in result
-
-    def test_is_json_serializable(self):
-        import json
-        error = WebInterfaceError(
-            ErrorCode.NETWORK_ERROR, "boom",
-            details="timeout", context={"url": "http://x"})
-        assert json.loads(json.dumps(error.to_dict()))["error_code"] == "NETWORK_ERROR"
-
-
-class TestFromException:
-    @pytest.mark.parametrize("exc_name,expected", [
-        ("ConfigError", ErrorCode.CONFIG_LOAD_FAILED),
-        ("PluginError", ErrorCode.PLUGIN_LOAD_FAILED),
-        ("PermissionError", ErrorCode.PERMISSION_DENIED),
-        ("AccessDenied", ErrorCode.PERMISSION_DENIED),
-        ("ValidationError", ErrorCode.VALIDATION_ERROR),
-        ("SchemaError", ErrorCode.VALIDATION_ERROR),
-        ("NetworkError", ErrorCode.NETWORK_ERROR),
-        ("ConnectionError", ErrorCode.NETWORK_ERROR),
-        ("TimeoutError", ErrorCode.TIMEOUT),
-        ("SomethingElse", ErrorCode.UNKNOWN_ERROR),
-    ])
-    def test_code_inferred_from_exception_class_name(self, exc_name, expected):
-        exc = type(exc_name, (Exception,), {})("boom")
-        assert WebInterfaceError.from_exception(exc).error_code is expected
-
-    def test_explicit_code_skips_inference(self):
-        error = WebInterfaceError.from_exception(
-            ValueError("boom"), error_code=ErrorCode.PLUGIN_NOT_FOUND)
-        assert error.error_code is ErrorCode.PLUGIN_NOT_FOUND
-
-    def test_message_is_the_safe_one_not_the_exception_text(self):
-        # The raw exception text is not echoed into `message`; that field is
-        # a fixed, user-facing string per code.
-        error = WebInterfaceError.from_exception(ValueError("secret-ish detail"))
-        assert error.message == "An unexpected error occurred"
-        assert "secret-ish" not in error.message
-
-    def test_exception_type_recorded_in_context(self):
-        error = WebInterfaceError.from_exception(ValueError("boom"))
-        assert error.context["exception_type"] == "ValueError"
-
-    def test_caller_context_is_preserved_alongside_type(self):
-        error = WebInterfaceError.from_exception(
-            ValueError("boom"), context={"plugin_id": "clock"})
-        assert error.context["plugin_id"] == "clock"
-        assert error.context["exception_type"] == "ValueError"
-
-    def test_caller_supplied_exception_type_is_overwritten(self):
-        error = WebInterfaceError.from_exception(
-            ValueError("boom"), context={"exception_type": "Fake"})
-        assert error.context["exception_type"] == "ValueError"
-
-    def test_original_error_retained(self):
-        exc = ValueError("boom")
-        assert WebInterfaceError.from_exception(exc).original_error is exc
-
-    def test_every_code_has_a_safe_message(self):
-        for code in ErrorCode:
-            assert WebInterfaceError._safe_message(code)
-
-
-class TestExceptionDetails:
-    def test_context_dict_is_flattened(self):
-        exc = ValueError("boom")
-        exc.context = {"config_path": "/etc/x.json", "line": 4}
-        details = WebInterfaceError._get_exception_details(exc)
-        assert "config_path: /etc/x.json" in details
-        assert "line: 4" in details
-        assert "; " in details
-
-    def test_exception_type_key_excluded(self):
-        exc = ValueError("boom")
-        exc.context = {"exception_type": "ValueError", "path": "/tmp/x"}
-        details = WebInterfaceError._get_exception_details(exc)
-        assert "exception_type" not in details
-        assert details == "path: /tmp/x"
-
-    def test_context_with_only_exception_type_gives_none(self):
-        exc = ValueError("boom")
-        exc.context = {"exception_type": "ValueError"}
-        assert WebInterfaceError._get_exception_details(exc) is None
-
-    def test_no_context_attribute_gives_none(self):
-        assert WebInterfaceError._get_exception_details(ValueError("boom")) is None
-
-    def test_non_dict_context_gives_none(self):
-        exc = ValueError("boom")
-        exc.context = "not a dict"
-        assert WebInterfaceError._get_exception_details(exc) is None
-
-    def test_empty_context_gives_none(self):
-        exc = ValueError("boom")
-        exc.context = {}
-        assert WebInterfaceError._get_exception_details(exc) is None
-
-    def test_details_flow_into_from_exception(self):
-        exc = ValueError("boom")
-        exc.context = {"config_path": "/etc/x.json"}
-        assert "config_path" in WebInterfaceError.from_exception(exc).details

test/web_interface/test_validators.py

@@ -1,284 +0,0 @@
-"""
-Tests for src/web_interface/validators.py.
-
-dedup_unique_arrays is already covered by test_dedup_unique_arrays.py and
-is not repeated here; this file covers the other eight functions, none of
-which had any tests.
-
-Regression coverage for three fixed bugs:
-- validate_numeric_range accepted True/False, since bool subclasses int.
-- validate_file_upload lowercased the filename's extension but not the
-  caller's allowed_extensions list, so ['.TTF'] rejected 'font.ttf'.
-- validate_image_url only checked for '..' inside the relative-path
-  branch, so http://host/../secret passed validation untouched.
-"""
-
-import pytest
-
-from src.web_interface.validators import (
-    escape_html,
-    sanitize_plugin_config,
-    validate_file_upload,
-    validate_font_awesome_class,
-    validate_image_url,
-    validate_mime_type,
-    validate_numeric_range,
-    validate_string_length,
-)
-
-
-class TestEscapeHtml:
-    def test_escapes_all_five_entities(self):
-        assert escape_html("""<a href="x">O'Neill & co</a>""") == (
-            "&lt;a href=&quot;x&quot;&gt;O&#x27;Neill &amp; co&lt;/a&gt;")
-
-    def test_ampersand_is_escaped_first_so_nothing_double_escapes(self):
-        # If '<' were replaced before '&', the '&' of '&lt;' would be
-        # escaped again into '&amp;lt;'.
-        assert escape_html("<") == "&lt;"
-        assert escape_html("&") == "&amp;"
-        assert escape_html("&<") == "&amp;&lt;"
-
-    def test_plain_text_unchanged(self):
-        assert escape_html("hello world") == "hello world"
-
-    def test_non_string_is_coerced(self):
-        assert escape_html(42) == "42"
-        assert escape_html(None) == "None"
-
-    def test_script_tag_neutralized(self):
-        assert "<script>" not in escape_html("<script>alert(1)</script>")
-
-
-class TestValidateImageUrl:
-    @pytest.mark.parametrize("url", [
-        "javascript:alert(1)",
-        "JavaScript:alert(1)",
-        "JAVASCRIPT:alert(1)",
-        "data:text/html;base64,PHNjcmlwdD4=",
-        "vbscript:msgbox(1)",
-        "file:///etc/passwd",
-    ])
-    def test_dangerous_protocols_rejected(self, url):
-        valid, error = validate_image_url(url)
-        assert valid is False and "protocol" in error.lower()
-
-    @pytest.mark.parametrize("url", [
-        "http://x/a.png?onerror=alert(1)",
-        "http://x/a.png#onload=alert(1)",
-        "http://x/onclick=alert(1).png",
-    ])
-    def test_event_handlers_rejected(self, url):
-        valid, error = validate_image_url(url)
-        assert valid is False and "Event handlers" in error
-
-    @pytest.mark.parametrize("url", ["", None, 123, []])
-    def test_empty_or_non_string_rejected(self, url):
-        assert validate_image_url(url)[0] is False
-
-    def test_http_and_https_allowed(self):
-        assert validate_image_url("http://example.com/logo.png") == (True, None)
-        assert validate_image_url("https://example.com/logo.png") == (True, None)
-
-    def test_other_schemes_rejected(self):
-        valid, error = validate_image_url("ftp://example.com/logo.png")
-        assert valid is False and "http://" in error
-
-    def test_relative_path_allowed(self):
-        assert validate_image_url("/static/logo.png") == (True, None)
-
-    def test_protocol_relative_url_rejected(self):
-        assert validate_image_url("//evil.com/logo.png")[0] is False
-
-    def test_relative_traversal_rejected(self):
-        assert validate_image_url("/static/../../etc/passwd")[0] is False
-
-    def test_absolute_url_traversal_rejected(self):
-        # Regression: the '..' check used to sit inside the leading-slash
-        # branch, so an absolute URL skipped it entirely.
-        valid, error = validate_image_url("http://example.com/../secret")
-        assert valid is False and "traversal" in error.lower()
-
-    def test_bare_traversal_rejected(self):
-        assert validate_image_url("../../etc/passwd")[0] is False
-
-
-class TestValidateFontAwesomeClass:
-    @pytest.mark.parametrize("cls", ["fa-star", "fas fa-star", "fa-solid fa-house"])
-    def test_valid_classes_accepted(self, cls):
-        assert validate_font_awesome_class(cls) == (True, None)
-
-    @pytest.mark.parametrize("cls", ["star", "glyphicon-star", ""])
-    def test_classes_without_fa_prefix_rejected(self, cls):
-        assert validate_font_awesome_class(cls)[0] is False
-
-    def test_injection_attempt_rejected(self):
-        assert validate_font_awesome_class('fa-star" onload="alert(1)')[0] is False
-
-    def test_angle_brackets_rejected(self):
-        assert validate_font_awesome_class("<script>fa-star</script>")[0] is False
-
-    def test_non_string_rejected(self):
-        valid, error = validate_font_awesome_class(None)
-        assert valid is False and "string" in error
-
-    def test_explicit_fa_check_is_unreachable_but_harmless(self):
-        # Characterized, not fixed: the regex already requires 'fa-', so the
-        # follow-up `if 'fa-' not in class_name` can never fire. Anything
-        # lacking 'fa-' is rejected by the pattern first, with the pattern's
-        # own message.
-        valid, error = validate_font_awesome_class("star")
-        assert valid is False
-        assert error == "Invalid Font Awesome class name format"
-
-
-class TestValidateFileUpload:
-    def test_plain_filename_accepted(self):
-        assert validate_file_upload("logo.png") == (True, None)
-
-    @pytest.mark.parametrize("filename", [
-        "../etc/passwd", "dir/file.png", "dir\\file.png", "..\\..\\secrets",
-    ])
-    def test_traversal_characters_rejected(self, filename):
-        valid, error = validate_file_upload(filename)
-        assert valid is False and "invalid characters" in error
-
-    @pytest.mark.parametrize("filename", ["", None, 123])
-    def test_empty_or_non_string_rejected(self, filename):
-        assert validate_file_upload(filename)[0] is False
-
-    def test_allowed_extension_accepted(self):
-        assert validate_file_upload("font.ttf", allowed_extensions=[".ttf", ".otf"]) == (True, None)
-
-    def test_disallowed_extension_rejected(self):
-        valid, error = validate_file_upload("evil.exe", allowed_extensions=[".ttf"])
-        assert valid is False and "extension" in error
-
-    def test_uppercase_filename_extension_matches(self):
-        assert validate_file_upload("FONT.TTF", allowed_extensions=[".ttf"]) == (True, None)
-
-    def test_uppercase_allowed_list_matches(self):
-        # Regression: only the filename side was lowercased, so a caller
-        # passing ['.TTF'] rejected every valid .ttf upload.
-        assert validate_file_upload("font.ttf", allowed_extensions=[".TTF"]) == (True, None)
-
-    def test_no_extension_list_skips_the_check(self):
-        assert validate_file_upload("anything.xyz") == (True, None)
-
-
-class TestValidateMimeType:
-    def test_known_type_accepted(self):
-        assert validate_mime_type("logo.png", ["image/png"]) == (True, None)
-
-    def test_mismatched_type_rejected(self):
-        valid, error = validate_mime_type("logo.png", ["image/jpeg"])
-        assert valid is False and "not allowed" in error
-
-    def test_undeterminable_type_rejected(self):
-        valid, error = validate_mime_type("mystery.zzz", ["image/png"])
-        assert valid is False and "Could not determine" in error
-
-    def test_guess_type_failure_is_caught(self, monkeypatch):
-        import mimetypes
-        monkeypatch.setattr(mimetypes, "guess_type",
-                            lambda *a, **kw: (_ for _ in ()).throw(RuntimeError("boom")))
-        valid, error = validate_mime_type("logo.png", ["image/png"])
-        assert valid is False and "Error validating MIME type" in error
-
-
-class TestValidateNumericRange:
-    def test_value_in_range(self):
-        assert validate_numeric_range(5, min_val=0, max_val=10) == (True, None)
-
-    def test_boundaries_are_inclusive(self):
-        assert validate_numeric_range(0, min_val=0, max_val=10) == (True, None)
-        assert validate_numeric_range(10, min_val=0, max_val=10) == (True, None)
-
-    def test_below_minimum_rejected(self):
-        valid, error = validate_numeric_range(-1, min_val=0)
-        assert valid is False and "at least" in error
-
-    def test_above_maximum_rejected(self):
-        valid, error = validate_numeric_range(11, max_val=10)
-        assert valid is False and "at most" in error
-
-    def test_floats_accepted(self):
-        assert validate_numeric_range(2.5, min_val=0, max_val=10) == (True, None)
-
-    def test_no_bounds_accepts_any_number(self):
-        assert validate_numeric_range(-9999) == (True, None)
-
-    @pytest.mark.parametrize("value", ["5", None, [], {}])
-    def test_non_numeric_rejected(self, value):
-        valid, error = validate_numeric_range(value, min_val=0, max_val=10)
-        assert valid is False and error == "Value must be a number"
-
-    @pytest.mark.parametrize("value", [True, False])
-    def test_booleans_rejected(self, value):
-        # Regression: bool subclasses int, so True passed the isinstance
-        # check and then compared as 1 against the range.
-        valid, error = validate_numeric_range(value, min_val=0, max_val=10)
-        assert valid is False and error == "Value must be a number"
-
-
-class TestValidateStringLength:
-    def test_within_range(self):
-        assert validate_string_length("hello", min_length=1, max_length=10) == (True, None)
-
-    def test_boundaries_are_inclusive(self):
-        assert validate_string_length("abc", min_length=3, max_length=3) == (True, None)
-
-    def test_too_short_rejected(self):
-        valid, error = validate_string_length("", min_length=1)
-        assert valid is False and "at least" in error
-
-    def test_too_long_rejected(self):
-        valid, error = validate_string_length("abcdef", max_length=3)
-        assert valid is False and "at most" in error
-
-    def test_non_string_rejected(self):
-        valid, error = validate_string_length(123, max_length=10)
-        assert valid is False and "must be a string" in error
-
-    def test_no_bounds_accepts_anything(self):
-        assert validate_string_length("") == (True, None)
-
-
-class TestSanitizePluginConfig:
-    def test_valid_keys_and_scalars_kept(self):
-        config = {"enabled": True, "count": 3, "ratio": 1.5, "name": "clock"}
-        assert sanitize_plugin_config(config) == config
-
-    @pytest.mark.parametrize("key", ["has space", "has-dash", "has.dot", "has/slash", ""])
-    def test_invalid_key_names_dropped(self, key):
-        assert sanitize_plugin_config({key: "value", "good": 1}) == {"good": 1}
-
-    def test_non_string_keys_dropped(self):
-        assert sanitize_plugin_config({1: "a", "good": 2}) == {"good": 2}
-
-    def test_nested_dicts_recursed(self):
-        result = sanitize_plugin_config({"outer": {"inner": 1, "bad key": 2}})
-        assert result == {"outer": {"inner": 1}}
-
-    def test_list_of_scalars_preserved(self):
-        assert sanitize_plugin_config({"teams": ["PHI", "NYG"]})["teams"] == ["PHI", "NYG"]
-
-    def test_list_of_dicts_recursed(self):
-        result = sanitize_plugin_config({"items": [{"ok": 1, "bad key": 2}]})
-        assert result["items"] == [{"ok": 1}]
-
-    def test_unknown_value_types_dropped(self):
-        assert sanitize_plugin_config({"weird": {1, 2, 3}, "good": 1}) == {"good": 1}
-
-    def test_none_values_dropped(self):
-        assert sanitize_plugin_config({"nothing": None, "good": 1}) == {"good": 1}
-
-    def test_strings_are_not_html_escaped(self):
-        # Pinned, not a bug: escaping here would persist the escaped form in
-        # config.json. Output escaping belongs to the template layer, which
-        # the function's docstring now says explicitly.
-        payload = "<script>alert(1)</script>"
-        assert sanitize_plugin_config({"title": payload})["title"] == payload
-
-    def test_empty_config(self):
-        assert sanitize_plugin_config({}) == {}

web_interface/blueprints/api_v3.py

@@ -22,7 +22,7 @@ logger = logging.getLogger(__name__)
 from src.web_interface.api_helpers import success_response, error_response, validate_request_json
 from src.web_interface.errors import ErrorCode
 from src.web_interface.secret_helpers import find_secret_fields, separate_secrets
-from src.web_interface.error_handler import describe_exception
+from src.web_interface.error_handler import describe_exception, redact_text
 from src.plugin_system.operation_types import OperationType
 from src.web_interface.validators import (
     validate_file_upload
@@ -328,7 +328,7 @@ def save_schedule_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        data = request.get_json(silent=True)
+        data = request.get_json()
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
@@ -536,7 +536,7 @@ def save_dim_schedule_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        data = request.get_json(silent=True)
+        data = request.get_json()
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
@@ -1345,20 +1345,18 @@ def save_raw_main_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        # silent=True so a malformed body returns None instead of raising
-        # Werkzeug's own BadRequest, which would answer in a different
-        # shape than this API's. Distinguish the two causes: a body that
-        # was sent but does not parse is a different mistake from no body.
-        data = request.get_json(silent=True)
-        if data is None and request.get_data():
-            return jsonify({'status': 'error', 'message': 'Invalid JSON in request body'}), 400
+        data = request.get_json()
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
+        # Validate that it's valid JSON (already parsed by request.get_json())
         # Save the raw config file
         api_v3.config_manager.save_raw_file_content('main', data)
 
         return jsonify({'status': 'success', 'message': 'Main configuration saved successfully'})
+    except json.JSONDecodeError as e:
+        logger.error('Invalid JSON', exc_info=True)
+        return jsonify({'status': 'error', 'message': 'Invalid JSON in request body'}), 400
     except Exception as e:
         from src.exceptions import ConfigError
         logger.error("Error saving raw main config", exc_info=True)
@@ -1393,11 +1391,7 @@ def save_raw_secrets_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        # See save_raw_main_config: silent parsing, with a sent-but-broken
-        # body reported separately from a missing one.
-        data = request.get_json(silent=True)
-        if data is None and request.get_data():
-            return jsonify({'status': 'error', 'message': 'Invalid JSON in request body'}), 400
+        data = request.get_json()
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
@@ -1409,6 +1403,9 @@ def save_raw_secrets_config():
             api_v3.plugin_store_manager.github_token = api_v3.plugin_store_manager._load_github_token()
 
         return jsonify({'status': 'success', 'message': 'Secrets configuration saved successfully'})
+    except json.JSONDecodeError as e:
+        logger.error('Invalid JSON', exc_info=True)
+        return jsonify({'status': 'error', 'message': 'Invalid JSON in request body'}), 400
     except Exception as e:
         from src.exceptions import ConfigError
         logger.error("Error saving raw secrets config", exc_info=True)
@@ -2414,7 +2411,7 @@ def get_on_demand_status():
 def start_on_demand_display():
     """Request the display controller to run a specific plugin on-demand."""
     try:
-        data = request.get_json(silent=True) or {}
+        data = request.get_json() or {}
         plugin_id = data.get('plugin_id')
         mode = data.get('mode')
         duration = data.get('duration')
@@ -2938,7 +2935,7 @@ def manage_plugin_limits(plugin_id):
                 })
         else:
             # POST - Set limits
-            data = request.get_json(silent=True) or {}
+            data = request.get_json() or {}
             from src.plugin_system.resource_monitor import ResourceLimits
 
             limits = ResourceLimits(
@@ -2969,7 +2966,7 @@ def toggle_plugin():
         content_type = request.content_type or ''
 
         if 'application/json' in content_type:
-            data = request.get_json(silent=True)
+            data = request.get_json()
             if not data or 'plugin_id' not in data or 'enabled' not in data:
                 return jsonify({'status': 'error', 'message': 'plugin_id and enabled required'}), 400
             plugin_id = data['plugin_id']
@@ -3840,7 +3837,7 @@ def install_plugin():
         if not api_v3.plugin_store_manager:
             return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
 
-        data = request.get_json(silent=True)
+        data = request.get_json()
         if not data or 'plugin_id' not in data:
             return jsonify({'status': 'error', 'message': 'plugin_id required'}), 400
 
@@ -3974,15 +3971,10 @@ def install_plugin_from_url():
         if not api_v3.plugin_store_manager:
             return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
 
-        data = request.get_json(silent=True)
+        data = request.get_json()
         if not data or 'repo_url' not in data:
             return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
 
-        # A non-string repo_url is a client mistake, not a server fault:
-        # .strip() would raise and the catch-all would report it as a 500.
-        if not isinstance(data['repo_url'], str) or not data['repo_url'].strip():
-            return jsonify({'status': 'error', 'message': 'repo_url must be a non-empty string'}), 400
-
         repo_url = data['repo_url'].strip()
         plugin_id = data.get('plugin_id')  # Optional, for monorepo installations
         plugin_path = data.get('plugin_path')  # Optional, for monorepo subdirectory
@@ -4034,15 +4026,10 @@ def get_registry_from_url():
         if not api_v3.plugin_store_manager:
             return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
 
-        data = request.get_json(silent=True)
+        data = request.get_json()
         if not data or 'repo_url' not in data:
             return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
 
-        # A non-string repo_url is a client mistake, not a server fault:
-        # .strip() would raise and the catch-all would report it as a 500.
-        if not isinstance(data['repo_url'], str) or not data['repo_url'].strip():
-            return jsonify({'status': 'error', 'message': 'repo_url must be a non-empty string'}), 400
-
         repo_url = data['repo_url'].strip()
 
         # Get registry from the URL
@@ -4084,15 +4071,10 @@ def add_saved_repository():
         if not api_v3.saved_repositories_manager:
             return jsonify({'status': 'error', 'message': 'Saved repositories manager not initialized'}), 500
 
-        data = request.get_json(silent=True)
+        data = request.get_json()
         if not data or 'repo_url' not in data:
             return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
 
-        # A non-string repo_url is a client mistake, not a server fault:
-        # .strip() would raise and the catch-all would report it as a 500.
-        if not isinstance(data['repo_url'], str) or not data['repo_url'].strip():
-            return jsonify({'status': 'error', 'message': 'repo_url must be a non-empty string'}), 400
-
         repo_url = data['repo_url'].strip()
         name = data.get('name')
 
@@ -4120,7 +4102,7 @@ def remove_saved_repository():
         if not api_v3.saved_repositories_manager:
             return jsonify({'status': 'error', 'message': 'Saved repositories manager not initialized'}), 500
 
-        data = request.get_json(silent=True)
+        data = request.get_json()
         if not data or 'repo_url' not in data:
             return jsonify({'status': 'error', 'message': 'repo_url required'}), 400
 
@@ -4254,7 +4236,7 @@ def refresh_plugin_store():
         if not api_v3.plugin_store_manager:
             return jsonify({'status': 'error', 'message': 'Plugin store manager not initialized'}), 500
 
-        data = request.get_json(silent=True) or {}
+        data = request.get_json() or {}
         fetch_commit_info = data.get('fetch_commit_info', data.get('fetch_latest_versions', False))
 
         # Force refresh the registry
@@ -5840,7 +5822,7 @@ def reset_plugin_config():
         if not api_v3.config_manager:
             return jsonify({'status': 'error', 'message': 'Config manager not initialized'}), 500
 
-        data = request.get_json(silent=True) or {}
+        data = request.get_json() or {}
         plugin_id = data.get('plugin_id')
         preserve_secrets = data.get('preserve_secrets', True)
 
@@ -6227,7 +6209,7 @@ sys.exit(proc.returncode)
 def authenticate_spotify():
     """Run Spotify authentication script"""
     try:
-        data = request.get_json(silent=True) or {}
+        data = request.get_json() or {}
         redirect_url = data.get('redirect_url', '').strip()
 
         # Get plugin directory
@@ -6290,6 +6272,7 @@ sys.exit(proc.returncode)
                     timeout=120,
                     env=env
                 )
+                os.unlink(wrapper_path)
 
                 if result.returncode == 0:
                     return jsonify({
@@ -6304,13 +6287,9 @@ sys.exit(proc.returncode)
                         'output': result.stdout + result.stderr
                     }), 400
             except subprocess.TimeoutExpired:
-                return jsonify({'status': 'error', 'message': 'Authentication timed out'}), 408
-            finally:
-                # The wrapper carries the user's redirect URL, so it must not
-                # survive the request on any path — including a failure to
-                # launch, which the previous per-branch unlinks missed.
                 if os.path.exists(wrapper_path):
                     os.unlink(wrapper_path)
+                return jsonify({'status': 'error', 'message': 'Authentication timed out'}), 408
         else:
             # Step 1: Get authorization URL
             # Import the script's functions directly to get the auth URL
@@ -6547,7 +6526,7 @@ def get_fonts_overrides():
 def save_fonts_overrides():
     """Save font overrides"""
     try:
-        data = request.get_json(silent=True)
+        data = request.get_json()
         if not data:
             return jsonify({'status': 'error', 'message': 'No data provided'}), 400
 
@@ -7167,7 +7146,7 @@ def upload_of_the_day_json():
 def delete_of_the_day_json():
     """Delete a JSON file from of-the-day plugin"""
     try:
-        data = request.get_json(silent=True) or {}
+        data = request.get_json() or {}
         file_id = data.get('file_id')  # This is the category_name
 
         if not file_id:
@@ -7257,29 +7236,6 @@ def serve_plugin_static(plugin_id, file_path):
         return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
 
 
-_MAX_CREDENTIAL_BACKUPS = 5
-
-
-def _prune_credential_backups(plugin_dir: Path) -> None:
-    """Keep only the newest _MAX_CREDENTIAL_BACKUPS credential backups.
-
-    Every re-upload copies the previous credentials.json aside. Without
-    pruning those accumulate for the life of the install — each one a
-    complete set of OAuth client credentials sitting in the plugin
-    directory.
-    """
-    backups = sorted(
-        plugin_dir.glob('credentials.json.backup.*'),
-        key=lambda p: p.stat().st_mtime,
-        reverse=True,
-    )
-    for stale in backups[_MAX_CREDENTIAL_BACKUPS:]:
-        try:
-            stale.unlink()
-        except OSError:
-            logger.warning("Could not remove old credential backup %s", stale.name)
-
-
 @api_v3.route('/plugins/calendar/upload-credentials', methods=['POST'])
 def upload_calendar_credentials():
     """Upload credentials.json file for calendar plugin"""
@@ -7307,20 +7263,24 @@ def upload_calendar_credentials():
         try:
             file_content = file.read()
             file.seek(0)
-            creds_data = json.loads(file_content)
+            json.loads(file_content)
         except json.JSONDecodeError:
             return jsonify({'status': 'error', 'message': 'File is not valid JSON'}), 400
 
-        # Validate it looks like Google OAuth credentials. A bare scalar, a
-        # list, true/null — all valid JSON, none of them credentials. Reject
-        # rather than save: a file written as credentials.json but unusable
-        # as credentials only fails later, somewhere less obvious.
-        if not isinstance(creds_data, dict) or not (
-                'installed' in creds_data or 'web' in creds_data):
+        # Validate it looks like Google OAuth credentials
+        try:
+            file.seek(0)
+            creds_data = json.loads(file.read())
+            file.seek(0)
+
+            # Check for required Google OAuth fields
+            if 'installed' not in creds_data and 'web' not in creds_data:
                 return jsonify({
                     'status': 'error',
                     'message': 'File does not appear to be a valid Google OAuth credentials file'
                 }), 400
+        except Exception:
+            pass  # Continue even if validation fails
 
         # Get plugin directory
         plugin_id = 'calendar'
@@ -7340,7 +7300,6 @@ def upload_calendar_credentials():
             backup_path = Path(plugin_dir) / f'credentials.json.backup.{int(time.time())}'
             import shutil
             shutil.copy2(credentials_path, backup_path)
-            _prune_credential_backups(Path(plugin_dir))
 
         # Save new file
         file.save(str(credentials_path))
@@ -7358,6 +7317,227 @@ def upload_calendar_credentials():
         logger.error('Error in upload_calendar_credentials', exc_info=True)
         return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details', 'details': describe_exception(e)}), 500
 
+# calendarList.list pages at 250 entries maximum. Ten pages is far past any
+# real account and exists only so a malformed nextPageToken cannot spin here.
+_CALENDAR_LIST_MAX_PAGES = 10
+
+
+def _calendar_plugin_dir() -> Optional[Path]:
+    """Where the calendar plugin is installed, or None if it is not."""
+    if api_v3.plugin_manager:
+        plugin_dir = api_v3.plugin_manager.get_plugin_directory('calendar')
+    else:
+        plugin_dir = PROJECT_ROOT / 'plugins' / 'calendar'
+    if not plugin_dir:
+        return None
+    plugin_dir = Path(plugin_dir)
+    return plugin_dir if plugin_dir.exists() else None
+
+
+def _run_calendar_registration(plugin_dir: Path, stdin_payload: str):
+    """Run the plugin's OAuth script and return the JSON object it prints.
+
+    The script decides between web and terminal mode by whether stdin is a
+    tty, so it must be given a pipe. It emits one JSON object on stdout; the
+    last parsable line is taken, because an import warning or a library's
+    stderr redirection can land in front of it.
+
+    Returns (payload, error_message). Exactly one is None.
+    """
+    script = plugin_dir / 'calendar_registration.py'
+    if not script.exists():
+        return None, 'Authentication script not found in the calendar plugin'
+
+    try:
+        result = subprocess.run(  # nosec B603 - fixed script path inside the plugin dir
+            [sys.executable, str(script)],
+            input=stdin_payload,
+            capture_output=True,
+            text=True,
+            timeout=120,
+            cwd=str(plugin_dir),
+        )
+    except subprocess.TimeoutExpired:
+        return None, 'Authentication timed out after 120s'
+    except OSError as e:
+        logger.error('Could not run calendar_registration.py', exc_info=True)
+        return None, 'Could not run the authentication script: %s' % describe_exception(e)
+
+    for line in reversed((result.stdout or '').splitlines()):
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            payload = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(payload, dict):
+            return payload, None
+
+    raw = (result.stderr or result.stdout or '').strip()
+    # The unredacted text goes to the log, where it is worth having in full.
+    # What comes back over HTTP is redacted: this is a script that handles
+    # OAuth client secrets, and its stderr can quote them.
+    if raw:
+        logger.error('calendar_registration.py failed (exit %s): %s',
+                     result.returncode, raw)
+    return None, 'Authentication script produced no result%s' % (
+        ': %s' % redact_text(raw) if raw else '')
+
+
+@api_v3.route('/plugins/calendar/authenticate', methods=['POST'])
+def authenticate_calendar():
+    """Google OAuth for the calendar plugin, in the two steps it requires.
+
+    Step 1 (no body) returns the consent URL to open. Step 2 posts back the
+    URL Google redirected to -- it fails to load, because the redirect points
+    at a loopback address nothing is listening on, but the address bar carries
+    the authorization code -- and the script exchanges it for a token.
+
+    Two calls rather than one because the user has to visit Google in between.
+    The script persists the PKCE verifier from step 1 for step 2 to reuse; the
+    exchange fails with "Missing code verifier" otherwise.
+    """
+    try:
+        plugin_dir = _calendar_plugin_dir()
+        if plugin_dir is None:
+            return jsonify({
+                'status': 'error',
+                'message': 'The calendar plugin is not installed'
+            }), 404
+
+        if not (plugin_dir / 'credentials.json').exists():
+            return jsonify({
+                'status': 'error',
+                'message': ('No credentials.json yet. Upload your Google OAuth '
+                            'client file first (Step 1).')
+            }), 400
+
+        data = request.get_json(silent=True) or {}
+        redirect_url = (data.get('redirect_url') or data.get('code') or '').strip()
+
+        payload, error = _run_calendar_registration(plugin_dir, redirect_url)
+        if error:
+            return jsonify({'status': 'error', 'message': error}), 500
+        if payload.get('status') != 'success':
+            # The script's own diagnosis is more useful than anything that
+            # could be reconstructed here -- but it interpolates exceptions
+            # into its messages, so it reaches the client redacted and the
+            # original goes to the log.
+            logger.error('calendar authentication failed: %s', payload)
+            safe = dict(payload)
+            safe['message'] = redact_text(str(payload.get('message', '')
+                                              or 'Authentication failed'))
+            return jsonify(safe), 400
+        return jsonify(payload)
+
+    except Exception as e:
+        logger.error('Error in authenticate_calendar', exc_info=True)
+        return jsonify({'status': 'error',
+                        'message': 'An error occurred; see logs for details',
+                        'details': describe_exception(e)}), 500
+
+
+@api_v3.route('/plugins/calendar/list-calendars', methods=['GET'])
+def list_calendar_calendars():
+    """The calendars this account can see, for the config picker.
+
+    Reads the token the OAuth flow wrote rather than shelling out again: the
+    picker is used interactively and a subprocess per click is slower than the
+    API call it would be wrapping.
+    """
+    try:
+        plugin_dir = _calendar_plugin_dir()
+        if plugin_dir is None:
+            return jsonify({
+                'status': 'error',
+                'message': 'The calendar plugin is not installed'
+            }), 404
+
+        token_file = plugin_dir / 'token.pickle'
+        if not token_file.exists():
+            return jsonify({
+                'status': 'error',
+                'message': ('Not authenticated with Google yet. Complete Step 2 '
+                            'first, then load your calendars.')
+            }), 400
+
+        try:
+            import pickle
+            from google.auth.transport.requests import Request as GoogleRequest
+            from googleapiclient.discovery import build as build_google_service
+        except ImportError as e:
+            return jsonify({
+                'status': 'error',
+                # The name of the missing module is the whole diagnosis, but it
+                # arrives as an exception, so it goes through the redactor like
+                # any other -- an ImportError can quote a path.
+                'message': ('The Google API libraries are not installed. Install '
+                            "the calendar plugin's requirements.txt. (%s)"
+                            % describe_exception(e))
+            }), 500
+
+        with open(token_file, 'rb') as handle:
+            # Written only by this plugin's own OAuth flow, into its own
+            # directory, and read here exactly as the plugin itself reads it.
+            creds = pickle.load(handle)  # nosec B301 - locally generated token
+
+        if creds and creds.expired and creds.refresh_token:
+            creds.refresh(GoogleRequest())
+            with open(token_file, 'wb') as handle:
+                pickle.dump(creds, handle)
+            os.chmod(token_file, 0o600)
+
+        if not creds or not creds.valid:
+            return jsonify({
+                'status': 'error',
+                'message': ('Stored Google credentials are no longer valid. '
+                            'Run Step 2 again to re-authenticate.')
+            }), 400
+
+        service = build_google_service('calendar', 'v3', credentials=creds)
+
+        # calendarList.list returns 100 entries per page by default and caps at
+        # 250, handing back a nextPageToken when there are more. Taking only
+        # the first page would silently hide calendars from the picker, and the
+        # user would have no way to tell the list was truncated.
+        entries = []
+        page_token = None
+        for _ in range(_CALENDAR_LIST_MAX_PAGES):
+            response = service.calendarList().list(
+                maxResults=250, pageToken=page_token).execute()
+            entries.extend(response.get('items', []))
+            page_token = response.get('nextPageToken')
+            if not page_token:
+                break
+        else:
+            # 2500 calendars in, something is wrong with the account or the
+            # token is looping; show what was collected rather than spin.
+            logger.warning(
+                'calendarList paging stopped at %d pages with more remaining',
+                _CALENDAR_LIST_MAX_PAGES)
+
+        calendars = [{
+            'id': entry.get('id'),
+            # The picker labels each row with summary and falls back to the id
+            # only in its own display, so send something either way.
+            'summary': entry.get('summary') or entry.get('id'),
+            'primary': bool(entry.get('primary', False)),
+        } for entry in entries if entry.get('id')]
+
+        # Primary first, then alphabetically: the list is usually short but the
+        # one the user wants is almost always their own calendar.
+        calendars.sort(key=lambda c: (not c['primary'], c['summary'].lower()))
+
+        return jsonify({'status': 'success', 'calendars': calendars})
+
+    except Exception as e:
+        logger.error('Error in list_calendar_calendars', exc_info=True)
+        return jsonify({'status': 'error',
+                        'message': 'An error occurred; see logs for details',
+                        'details': describe_exception(e)}), 500
+
+
 @api_v3.route('/plugins/assets/delete', methods=['POST'])
 def delete_plugin_asset():
     """Delete an asset file for a plugin"""
@@ -7644,7 +7824,7 @@ def connect_wifi():
     try:
         from src.wifi_manager import WiFiManager
 
-        data = request.get_json(silent=True)
+        data = request.get_json()
         if not data:
             return jsonify({
                 'status': 'error',
@@ -7798,7 +7978,7 @@ def set_auto_enable_ap_mode():
     try:
         from src.wifi_manager import WiFiManager
 
-        data = request.get_json(silent=True)
+        data = request.get_json()
         if data is None or 'auto_enable_ap_mode' not in data:
             return jsonify({
                 'status': 'error',
@@ -7927,7 +8107,7 @@ def delete_cache_file():
             from src.cache_manager import CacheManager
             api_v3.cache_manager = CacheManager()
 
-        data = request.get_json(silent=True)
+        data = request.get_json()
         if not data or 'key' not in data:
             return jsonify({'status': 'error', 'message': 'cache key is required'}), 400
 
@@ -8190,16 +8370,7 @@ def backup_restore():
         try:
             opts_dict = json.loads(options_raw)
         except json.JSONDecodeError:
-            opts_dict = None
-        if not isinstance(opts_dict, dict):
-            # Every option defaults to True, so falling back to {} on a
-            # parse failure would silently perform a FULL restore —
-            # secrets and all — for a caller who asked for a narrow one
-            # and mis-serialized it. Refuse instead of guessing.
-            return jsonify({
-                'status': 'error',
-                'message': 'Invalid options: expected a JSON object',
-            }), 400
+            opts_dict = {}
         options = RestoreOptions(
             restore_config=bool(opts_dict.get('restore_config', True)),
             restore_secrets=bool(opts_dict.get('restore_secrets', True)),

web_interface/static/v3/js/widgets/google-oauth.js

@@ -0,0 +1,196 @@
+/**
+ * Google OAuth Widget
+ *
+ * Step 2 of the calendar plugin's setup, between uploading the OAuth client
+ * file and picking calendars. Google will not let a headless device complete
+ * consent on its own, so the flow is necessarily two calls with a human in
+ * between:
+ *
+ *   1. POST /api/v3/plugins/calendar/authenticate with no body
+ *      -> { auth_url } to open in a browser
+ *   2. the browser lands on a loopback address that fails to load; its URL
+ *      carries the authorization code. POST it back as redirect_url
+ *      -> the server exchanges it and writes token.pickle
+ *
+ * The failed page in step 2 is expected and is worth saying out loud, because
+ * it looks exactly like something went wrong.
+ *
+ * @module GoogleOAuthWidget
+ */
+
+(function () {
+    'use strict';
+
+    if (typeof window.LEDMatrixWidgets === 'undefined') {
+        console.error('[GoogleOAuthWidget] LEDMatrixWidgets registry not found. Load registry.js first.');
+        return;
+    }
+
+    const ENDPOINT = '/api/v3/plugins/calendar/authenticate';
+
+    window.LEDMatrixWidgets.register('google-oauth', {
+        name: 'Google OAuth Widget',
+        version: '1.0.0',
+
+        /**
+         * @param {HTMLElement} container
+         * @param {Object} config  - schema config (unused)
+         * @param {*}      value   - unused; this widget stores nothing
+         * @param {Object} options - { fieldId, pluginId, name }
+         */
+        render: function (container, config, value, options) {
+            const fieldId = options.fieldId;
+
+            // Nothing is stored in config by this step -- the result is
+            // token.pickle on the device -- but the form still expects a field.
+            const hidden = document.createElement('input');
+            hidden.type = 'hidden';
+            hidden.id = fieldId + '_hidden';
+            hidden.name = options.name;
+            hidden.value = value || '';
+
+            const startBtn = document.createElement('button');
+            startBtn.type = 'button';
+            startBtn.className = 'px-3 py-1.5 text-sm rounded-md bg-blue-600 hover:bg-blue-700 text-white';
+            startBtn.innerHTML = '<i class="fas fa-key"></i> Connect Google Account';
+
+            const status = document.createElement('p');
+            status.className = 'text-xs text-gray-400 mt-2';
+            // Every message this widget gives -- the consent link is ready,
+            // the exchange failed -- arrives here after an async call, so a
+            // screen reader is told nothing unless it is a live region.
+            status.setAttribute('role', 'status');
+            status.setAttribute('aria-live', 'polite');
+
+            const step2 = document.createElement('div');
+            step2.className = 'mt-3 hidden';
+
+            const link = document.createElement('a');
+            link.target = '_blank';
+            link.rel = 'noopener noreferrer';
+            link.className = 'text-blue-400 underline text-sm break-all';
+            link.textContent = 'Open the Google consent screen';
+
+            // Deliberately loud. After consent the browser is redirected to a
+            // loopback address nothing is listening on, so it lands on a
+            // browser error page -- which reads as a failure at exactly the
+            // moment the user has to act on it. Said quietly in grey it gets
+            // missed, and the flow looks broken when it is working.
+            const hint = document.createElement('div');
+            hint.className =
+                'mt-3 p-3 rounded-md border border-amber-500/60 bg-amber-500/10';
+            hint.innerHTML =
+                '<p class="text-sm text-amber-300 font-semibold">'
+                + '<i class="fas fa-triangle-exclamation"></i> '
+                + 'The next page will fail to load. That is expected.</p>'
+                + '<p class="text-xs text-amber-200/90 mt-1">'
+                + 'After you approve access, Google sends your browser to '
+                + '<code>127.0.0.1</code>, where nothing is running \u2014 so you will see '
+                + '"This site can\u2019t be reached" or similar. Nothing has gone wrong. '
+                + 'Copy the <strong>entire address</strong> out of the address bar '
+                + '(it contains <code>?code=...</code>) and paste it in the box below.</p>';
+
+            const codeInputId = fieldId + '_redirect_url';
+
+            const codeLabel = document.createElement('label');
+            codeLabel.className = 'block text-xs text-gray-300 mt-3';
+            codeLabel.textContent = 'Paste the address from that failed page here:';
+            // The label was visible but not associated, so the input still had
+            // no accessible name -- a placeholder is not one, and it vanishes
+            // on focus, which is exactly when the value is being pasted.
+            codeLabel.setAttribute('for', codeInputId);
+
+            const codeInput = document.createElement('input');
+            codeInput.type = 'text';
+            codeInput.id = codeInputId;
+            codeInput.placeholder = 'http://127.0.0.1/?code=...';
+            codeInput.className =
+                'mt-2 block w-full px-3 py-2 text-sm border border-gray-600 '
+                + 'rounded-md bg-gray-800 text-gray-100';
+
+            const finishBtn = document.createElement('button');
+            finishBtn.type = 'button';
+            finishBtn.className = 'mt-2 px-3 py-1.5 text-sm rounded-md bg-green-600 hover:bg-green-700 text-white';
+            finishBtn.innerHTML = '<i class="fas fa-check"></i> Finish Authentication';
+
+            function say(message, kind) {
+                status.textContent = message;
+                status.className = 'text-xs mt-2 ' + (
+                    kind === 'error' ? 'text-red-400'
+                        : kind === 'success' ? 'text-green-400'
+                            : 'text-gray-400');
+            }
+
+            function post(body) {
+                return fetch(ENDPOINT, {
+                    method: 'POST',
+                    headers: { 'Content-Type': 'application/json' },
+                    body: JSON.stringify(body || {})
+                }).then(function (r) {
+                    return r.json().catch(function () {
+                        // A non-JSON body here means the request never reached
+                        // the handler -- worth saying so rather than "undefined".
+                        return { status: 'error', message: 'Server returned ' + r.status };
+                    });
+                });
+            }
+
+            startBtn.addEventListener('click', function () {
+                startBtn.disabled = true;
+                say('Requesting a consent link...');
+                post({}).then(function (data) {
+                    startBtn.disabled = false;
+                    if (data.status !== 'success' || !data.auth_url) {
+                        say(data.message || 'Could not start authentication.', 'error');
+                        return;
+                    }
+                    link.href = data.auth_url;
+                    step2.classList.remove('hidden');
+                    say(data.message || 'Open the link, approve, then paste the address back.');
+                }).catch(function (err) {
+                    startBtn.disabled = false;
+                    say('Request failed: ' + err.message, 'error');
+                });
+            });
+
+            finishBtn.addEventListener('click', function () {
+                const pasted = codeInput.value.trim();
+                if (!pasted) {
+                    say('Paste the address your browser was redirected to.', 'error');
+                    return;
+                }
+                finishBtn.disabled = true;
+                say('Exchanging the code with Google...');
+                post({ redirect_url: pasted }).then(function (data) {
+                    finishBtn.disabled = false;
+                    if (data.status !== 'success') {
+                        say(data.message || 'Authentication failed.', 'error');
+                        return;
+                    }
+                    say(data.message || 'Authenticated.', 'success');
+                    step2.classList.add('hidden');
+                    codeInput.value = '';
+                }).catch(function (err) {
+                    finishBtn.disabled = false;
+                    say('Request failed: ' + err.message, 'error');
+                });
+            });
+
+            step2.appendChild(link);
+            step2.appendChild(hint);
+            step2.appendChild(codeLabel);
+            step2.appendChild(codeInput);
+            step2.appendChild(finishBtn);
+
+            container.appendChild(hidden);
+            container.appendChild(startBtn);
+            container.appendChild(status);
+            container.appendChild(step2);
+        },
+
+        getValue: function (fieldId) {
+            const hidden = document.getElementById(fieldId + '_hidden');
+            return hidden ? hidden.value : '';
+        }
+    });
+})();

web_interface/templates/v3/base.html

@@ -987,6 +987,7 @@
     <script src="{{ url_for('static', filename='v3/js/widgets/custom-feeds.js') }}" defer></script>
     <script src="{{ url_for('static', filename='v3/js/widgets/array-table.js') }}" defer></script>
     <script src="{{ url_for('static', filename='v3/js/widgets/google-calendar-picker.js') }}" defer></script>
+    <script src="{{ url_for('static', filename='v3/js/widgets/google-oauth.js') }}" defer></script>
     <script src="{{ url_for('static', filename='v3/js/widgets/day-selector.js') }}" defer></script>
     <script src="{{ url_for('static', filename='v3/js/widgets/time-range.js') }}" defer></script>
     <script src="{{ url_for('static', filename='v3/js/widgets/time-picker.js') }}" defer></script>

web_interface/templates/v3/partials/plugin_config.html

@@ -815,7 +815,7 @@
                     <i class="fas fa-info-circle mr-1"></i>
                     Changes in the file manager save immediately — no need to click Save Configuration.
                 </p>
-            {% elif str_widget in ['text-input', 'textarea', 'select-dropdown', 'toggle-switch', 'radio-group', 'date-picker', 'time-picker', 'slider', 'color-picker', 'email-input', 'url-input', 'password-input', 'font-selector', 'file-upload-single', 'plugin-file-manager'] %}
+            {% elif str_widget in ['text-input', 'textarea', 'select-dropdown', 'toggle-switch', 'radio-group', 'date-picker', 'time-picker', 'slider', 'color-picker', 'email-input', 'url-input', 'password-input', 'font-selector', 'file-upload-single', 'plugin-file-manager', 'google-oauth'] %}
                 {# Render widget container #}
                 <div id="{{ field_id }}_container" class="{{ str_widget }}-container"></div>
                 <script>