fix(web): custom-feed logo upload uses the wrong request/response contract

Every custom-feed logo upload has been failing: handleCustomFeedLogoUpload
posts the file under field name "file" and reads the response from
data.data.files, but the backend endpoint it calls
(api_v3.upload_plugin_asset, /api/v3/plugins/assets/upload) requires the
field name "files" (checks 'files' not in request.files, 400s "No files
provided" otherwise) and returns the result in a top-level "uploaded_files"
key - there is no nested "data" wrapper in the response at all. Confirmed
by reading the endpoint directly, and cross-checked against
file-upload-single.js, a sibling widget that uses the correct contract
against the same endpoint.

- formData.append('file', file) -> formData.append('files', file)
- data.data.files / data.data.files[0] -> data.uploaded_files /
  data.uploaded_files[0]

No other call sites in this file used the stale contract (grepped for both
patterns after the fix - zero remaining). The response entries' 'path' and
'id' fields (both read further down in the same handler) are unaffected -
only the wrapper shape was wrong.

Found incidentally while re-verifying a CodeRabbit review on an unrelated
PR (#417) that had deleted a differently-named dead file
(custom-feeds-helpers.js) with the same bug; this widget (custom-feeds.js)
is the live code path and was never touched by that PR.

Validation: brace/paren balance check; explicit assertions that the old
field name and response shape no longer appear anywhere in the file. No
Python changed, no existing tests cover this endpoint's client flow.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KEZK1P1Q1fu5pcuVrkrCFZ

bandit.yaml

@@ -0,0 +1,29 @@
+# bandit.yaml — LEDMatrix bandit configuration
+# https://bandit.readthedocs.io/en/latest/config.html
+#
+# Skips are justified by the specific codebase context documented below.
+# Do not remove skips without updating the justification comment.
+
+skips:
+  # B104: Binding to all interfaces (0.0.0.0)
+  # Intentional — the Flask server binds 0.0.0.0 for LAN access on a Raspberry Pi.
+  # This is not internet-facing and is documented in web_interface/app.py.
+  - B104
+
+  # B603: subprocess call without shell=True
+  # All subprocess.run() calls in this codebase use list arguments (confirmed by
+  # grep — zero uses of shell=True in src/ or web_interface/). List args prevent
+  # shell injection. See src/common/permission_utils.py for the primary usage.
+  - B603
+
+  # B607: Starting a process with a partial executable path
+  # The subprocess calls invoke system utilities (systemctl, sudo, git) by name.
+  # These are fixed-list invocations, not user-controlled, and rely on PATH.
+  - B607
+
+exclude_dirs:
+  - tests
+  - test
+  - venv
+  - .venv
+  - rpi-rgb-led-matrix-master

docs/HOW_TO_RUN_TESTS.md

@@ -248,7 +248,6 @@ test/
 ├── test_config_service.py               # Config service tests
 ├── test_config_validation_edge_cases.py # Config edge cases
 ├── test_font_manager.py                 # Font manager tests
-├── test_layout_manager.py               # Layout manager tests
 ├── test_text_helper.py                  # Text helper tests
 ├── test_error_handling.py               # Error handling tests
 ├── test_error_aggregator.py             # Error aggregation tests

requirements.txt

@@ -8,16 +8,11 @@ numpy>=1.24.0  # For fast array operations in ScrollHelper (compatible with 2.x)
 
 # Timezone handling
 pytz>=2024.2,<2025.0  # Updated for latest timezone data
-timezonefinder>=6.5.0,<7.0.0  # Updated for better performance and accuracy
-geopy>=2.4.1,<3.0.0
 
 # HTTP requests
 requests>=2.33.0,<3.0.0
 
 # Google API integration
-google-auth-oauthlib>=1.2.0,<2.0.0
-google-auth-httplib2>=0.2.0,<1.0.0
-google-api-python-client>=2.147.0,<3.0.0
 
 # Font rendering
 freetype-py>=2.5.1,<3.0.0
@@ -29,10 +24,8 @@ spotipy>=2.25.2,<3.0.0
 Flask>=3.1.3,<4.0.0
 
 # Text processing
-unidecode>=1.3.8,<2.0.0
 
 # Calendar integration
-icalevents>=0.1.27,<1.0.0
 
 # WebSocket support
 python-socketio>=5.14.0,<6.0.0

scripts/audit_plugins.py

@@ -0,0 +1,344 @@
+#!/usr/bin/env python3
+"""
+LEDMatrix Plugin Security Auditor
+
+Performs AST-based security analysis of all Python files in plugin directories.
+Designed to run in CI — exits non-zero on CRITICAL findings only.
+
+Usage:
+    python scripts/audit_plugins.py
+    python scripts/audit_plugins.py --verbose
+    python scripts/audit_plugins.py --plugin hello-world
+    python scripts/audit_plugins.py --output results.json
+"""
+
+import ast
+import argparse
+import json
+import sys
+from dataclasses import dataclass, asdict
+from pathlib import Path
+from datetime import datetime, timezone
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+
+PLUGIN_BASE_DIRS = [
+    PROJECT_ROOT / "plugins",
+    PROJECT_ROOT / "plugin-repos",
+]
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Finding dataclass
+# ─────────────────────────────────────────────────────────────────────────────
+
+@dataclass
+class Finding:
+    plugin_id: str
+    file: str
+    line: int
+    severity: str   # CRITICAL | WARNING | INFO
+    rule: str
+    message: str
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# AST visitor
+# ─────────────────────────────────────────────────────────────────────────────
+
+class _PluginVisitor(ast.NodeVisitor):
+    """Collect security findings from a single plugin Python file."""
+
+    def __init__(self, filepath: Path, plugin_id: str):
+        self.filepath = filepath
+        self.plugin_id = plugin_id
+        self.findings: list[Finding] = []
+        # Local name -> real dotted path, so aliased imports and from-imports
+        # of dangerous APIs (import subprocess as sp; from builtins import
+        # eval as e) are still recognized in visit_Call below.
+        self._aliases: dict[str, str] = {}
+
+    def _add(self, node: ast.AST, severity: str, rule: str, message: str) -> None:
+        self.findings.append(Finding(
+            plugin_id=self.plugin_id,
+            file=str(self.filepath.relative_to(PROJECT_ROOT)),
+            line=getattr(node, "lineno", 0),
+            severity=severity,
+            rule=rule,
+            message=message,
+        ))
+
+    def _resolve(self, local_name: str) -> str:
+        """Resolve a local name through recorded import aliases to its real
+        dotted path (e.g. "sp" -> "subprocess"); unresolved names pass through
+        unchanged."""
+        return self._aliases.get(local_name, local_name)
+
+    def _resolve_call_target(self, func: ast.expr) -> str | None:
+        """Resolve a Call's func node to a fully-qualified dotted target,
+        covering a direct name (bare builtin, aliased import, or
+        from-import: from builtins import eval as e; from subprocess
+        import run; from os import system as s) and module-attribute
+        access (subprocess.run, sp.run, os.system, o.system) uniformly.
+        Returns None for call shapes this doesn't attempt to resolve."""
+        if isinstance(func, ast.Name):
+            return self._resolve(func.id)
+        if isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name):
+            base = self._resolve(func.value.id)
+            return f"{base}.{func.attr}"
+        return None
+
+    def visit_Call(self, node: ast.Call) -> None:
+        target = self._resolve_call_target(node.func)
+        if target is None:
+            self.generic_visit(node)
+            return
+
+        leaf = target.rsplit(".", 1)[-1]
+
+        # eval() / exec() / compile() — arbitrary code execution, whether a
+        # bare call, an aliased import, or a from-import
+        # (from builtins import eval as e; e(...))
+        if leaf == "eval":
+            self._add(node, "CRITICAL", "PLUGIN-001",
+                      "eval() call — arbitrary code execution risk")
+        elif leaf == "exec":
+            self._add(node, "CRITICAL", "PLUGIN-002",
+                      "exec() call — arbitrary code execution risk")
+        elif leaf == "compile":
+            self._add(node, "WARNING", "PLUGIN-003",
+                      "compile() call — dynamic code compilation")
+
+        # subprocess.*(shell=True), whether subprocess.run(...), sp.run(...),
+        # or a from-import (from subprocess import run; run(..., shell=True))
+        if target in {
+            "subprocess.run", "subprocess.call", "subprocess.Popen",
+            "subprocess.check_call", "subprocess.check_output",
+        }:
+            for kw in node.keywords:
+                if (kw.arg == "shell" and
+                        isinstance(kw.value, ast.Constant) and
+                        kw.value.value is True):
+                    self._add(node, "WARNING", "PLUGIN-004",
+                              f"subprocess.{leaf}(shell=True) — "
+                              f"shell injection risk if args include user input")
+
+        # os.system(), whether os.system(...), o.system(...), or a
+        # from-import (from os import system as s; s(...))
+        if target == "os.system":
+            self._add(node, "WARNING", "PLUGIN-005",
+                      "os.system() call — prefer subprocess with list args")
+
+        self.generic_visit(node)
+
+    def visit_Import(self, node: ast.Import) -> None:
+        for alias in node.names:
+            if alias.asname:
+                local, real = alias.asname, alias.name
+            else:
+                # `import os.path` binds the top-level name `os`, not `os.path`
+                local = real = alias.name.split(".")[0]
+            self._aliases[local] = real
+            self._check_import(node, alias.name)
+        self.generic_visit(node)
+
+    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
+        if node.module:
+            for alias in node.names:
+                local = alias.asname or alias.name
+                self._aliases[local] = f"{node.module}.{alias.name}"
+            self._check_import(node, node.module)
+        self.generic_visit(node)
+
+    def _check_import(self, node: ast.AST, module_name: str) -> None:
+        dangerous = {
+            "ctypes": ("WARNING", "PLUGIN-010", "ctypes import — native code execution"),
+            "cffi": ("WARNING", "PLUGIN-011", "cffi import — native code execution"),
+            "pickle": ("WARNING", "PLUGIN-012",
+                       "pickle import — deserialization can execute arbitrary code"),
+            "marshal": ("WARNING", "PLUGIN-013",
+                        "marshal import — deserialization risk"),
+        }
+        for mod, (severity, rule, msg) in dangerous.items():
+            if module_name == mod or module_name.startswith(mod + "."):
+                self._add(node, severity, rule, msg)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Per-plugin audit
+# ─────────────────────────────────────────────────────────────────────────────
+
+def audit_plugin(plugin_dir: Path) -> list[Finding]:
+    """Audit a single plugin directory. Returns all findings."""
+    findings: list[Finding] = []
+    plugin_id = plugin_dir.name
+
+    # Check for required files
+    for required_file, rule, msg in [
+        ("manifest.json", "PLUGIN-020",
+         "manifest.json missing — plugin may be incomplete"),
+        ("config_schema.json", "PLUGIN-021",
+         "config_schema.json missing — no input validation schema declared"),
+    ]:
+        if not (plugin_dir / required_file).exists():
+            findings.append(Finding(
+                plugin_id=plugin_id,
+                file=str((plugin_dir / required_file).relative_to(PROJECT_ROOT)),
+                line=0,
+                severity="WARNING",
+                rule=rule,
+                message=msg,
+            ))
+
+    # AST analysis of all Python files
+    for py_file in sorted(plugin_dir.rglob("*.py")):
+        try:
+            source = py_file.read_text(encoding="utf-8")
+            tree = ast.parse(source, filename=str(py_file))
+            visitor = _PluginVisitor(py_file, plugin_id)
+            visitor.visit(tree)
+            findings.extend(visitor.findings)
+        except SyntaxError as exc:
+            # A file the visitor can't even parse is a file we can't verify
+            # is safe -- this must block the audit, not just warn.
+            findings.append(Finding(
+                plugin_id=plugin_id,
+                file=str(py_file.relative_to(PROJECT_ROOT)),
+                line=getattr(exc, "lineno", 0) or 0,
+                severity="CRITICAL",
+                rule="PLUGIN-030",
+                message=f"Python syntax error — cannot be parsed: {exc}",
+            ))
+        except OSError as exc:
+            # Same reasoning as SyntaxError: an unreadable file was never
+            # actually scanned, so it must block rather than pass silently.
+            findings.append(Finding(
+                plugin_id=plugin_id,
+                file=str(py_file.relative_to(PROJECT_ROOT)),
+                line=0,
+                severity="CRITICAL",
+                rule="PLUGIN-031",
+                message=f"Could not read file: {exc}",
+            ))
+
+    return findings
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Main
+# ─────────────────────────────────────────────────────────────────────────────
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="LEDMatrix plugin security auditor",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument("--plugin", "-p", default=None,
+                        help="Audit a specific plugin ID only")
+    parser.add_argument("--output", "-o", default=None,
+                        help="Write JSON results to this file")
+    parser.add_argument("--verbose", "-v", action="store_true",
+                        help="Show all findings, not just summary")
+    args = parser.parse_args()
+
+    print("=" * 60)
+    print("LEDMatrix Plugin Security Audit")
+    print(f"Project root: {PROJECT_ROOT}")
+    print("=" * 60)
+
+    all_findings: list[Finding] = []
+    plugins_scanned = 0
+    plugin_found = args.plugin is None
+
+    for base_dir in PLUGIN_BASE_DIRS:
+        if not base_dir.exists():
+            if args.verbose:
+                print(f"  ⏭️  Skipping {base_dir.name}/ (directory not found)")
+            continue
+
+        base_label = base_dir.relative_to(PROJECT_ROOT)
+        print(f"\n  Scanning {base_label}/")
+
+        for plugin_dir in sorted(base_dir.iterdir()):
+            if not plugin_dir.is_dir():
+                continue
+            if plugin_dir.name.startswith((".", "_")):
+                continue
+            if args.plugin and plugin_dir.name != args.plugin:
+                continue
+            if args.plugin:
+                plugin_found = True
+
+            findings = audit_plugin(plugin_dir)
+            all_findings.extend(findings)
+            plugins_scanned += 1
+
+            critical = [f for f in findings if f.severity == "CRITICAL"]
+            warnings = [f for f in findings if f.severity == "WARNING"]
+
+            if critical:
+                icon, label = "🚨", "CRITICAL"
+            elif warnings:
+                icon, label = "⚠️ ", "WARN   "
+            else:
+                icon, label = "✅", "PASS   "
+
+            print(f"    {icon} [{label}] {plugin_dir.name}"
+                  f" — {len(critical)} critical, {len(warnings)} warnings")
+
+            if args.verbose:
+                for f in findings:
+                    severity_icon = {"CRITICAL": "🚨", "WARNING": "⚠️ ", "INFO": "ℹ️ "}.get(
+                        f.severity, "  "
+                    )
+                    print(f"           {severity_icon} {f.rule} {f.file}:{f.line} — {f.message}")
+
+    if args.plugin and not plugin_found:
+        print(f"\n  🚨 Plugin '{args.plugin}' not found in any of "
+              f"{[str(d.relative_to(PROJECT_ROOT)) for d in PLUGIN_BASE_DIRS]} — "
+              f"nothing was audited")
+        return 1
+
+    # Summary
+    critical_findings = [f for f in all_findings if f.severity == "CRITICAL"]
+    warning_findings = [f for f in all_findings if f.severity == "WARNING"]
+
+    print(f"\n{'=' * 60}")
+    print(f"  Plugins scanned : {plugins_scanned}")
+    print(f"  CRITICAL        : {len(critical_findings)}")
+    print(f"  WARNING         : {len(warning_findings)}")
+
+    if critical_findings:
+        print("\n  🚨 CRITICAL findings:")
+        for f in critical_findings:
+            print(f"    {f.plugin_id} | {Path(f.file).name}:{f.line} | {f.message}")
+
+    # Write JSON output
+    if args.output:
+        output_data = {
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+            "plugins_scanned": plugins_scanned,
+            "summary": {
+                "critical": len(critical_findings),
+                "warnings": len(warning_findings),
+            },
+            "findings": [f.to_dict() for f in all_findings],
+        }
+        Path(args.output).write_text(
+            json.dumps(output_data, indent=2), encoding="utf-8"
+        )
+        print(f"\n  Results written to: {args.output}")
+
+    if critical_findings:
+        print("\n  🚨 Blocking — CRITICAL issues must be resolved")
+        return 1
+
+    print("\n  ✅ No critical issues found")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/check_plugin.py

@@ -37,7 +37,7 @@ os.environ['EMULATOR'] = 'true'
 
 from src.logging_config import get_logger  # noqa: E402
 from src.plugin_system.testing.loading import (  # noqa: E402
-    find_plugin_dir, load_config_defaults, load_harness_spec, load_manifest,
+    build_full_config, find_plugin_dir, load_harness_spec, load_manifest,
 )
 from src.plugin_system.testing.harness import (  # noqa: E402
     RenderResult, render_plugin_matrix, compare_to_goldens, write_goldens,
@@ -97,12 +97,11 @@ def check_one(plugin_id: str, search_dirs: List[str], sizes, mock_data: Dict,
     # matrix path does; explicit CLI flags still override the file.
     spec = load_harness_spec(plugin_dir)
 
-    # config_schema defaults (real-install behavior), then harness.json config,
-    # then CLI --config — most specific wins.
-    full_config = {"enabled": True}
-    full_config.update(load_config_defaults(plugin_dir))
-    full_config.update(spec.get("config", {}))
-    full_config.update(config)
+    # config_schema defaults (real-install behavior, with enabled forced True
+    # so a plugin's own enabled:false default can't accidentally disable
+    # testing), then harness.json config, then CLI --config — most specific
+    # wins.
+    full_config = build_full_config(plugin_dir, spec, config)
 
     # Precedence: CLI flag > LEDMATRIX_TEST_SIZES env > harness.json > default.
     effective_sizes = sizes if sizes else resolve_test_sizes(spec.get("sizes"))

scripts/dev/test_pillow_compat.py

@@ -55,7 +55,7 @@ def main():
     failures += not check("draw.textbbox",
         lambda: draw.textbbox((0, 0), "Test", font=font))
 
-    print("\nResampling (used in logo_helper, image_utils, sports base):")
+    print("\nResampling (used in logo_helper, sports base):")
     logo = Image.new('RGBA', (200, 200), (255, 128, 0, 200))
     failures += not check("Image.Resampling.LANCZOS exists",
         lambda: str(Image.Resampling.LANCZOS))

scripts/generate_report.py

@@ -0,0 +1,356 @@
+#!/usr/bin/env python3
+"""
+Security Report Generator
+
+Aggregates JSON output from all CI security audit jobs into a single
+Markdown report suitable for PR comments and artifact storage.
+
+Expected artifact layout (from actions/download-artifact@v4):
+    <artifact-dir>/
+        sast-results/
+            bandit-results.json
+            semgrep-results.json
+        dependency-audit-results/
+            pip-audit-results.json
+            safety-results.json
+        secrets-scan-results/
+            gitleaks-results.json
+        security-proofs-results/
+            security-proofs-results.json
+        plugin-audit-results/
+            plugin-audit-results.json
+
+Usage:
+    python scripts/generate_report.py --artifact-dir audit-artifacts/ --output report.md
+    python scripts/generate_report.py --artifact-dir audit-artifacts/ --output report.md --verbose
+"""
+
+import argparse
+import json
+import sys
+from pathlib import Path
+from datetime import datetime, timezone
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+
+# Gitleaks matches exactly equal to one of these (not a substring match -- a
+# real secret that merely contains one of these words as part of its actual
+# value must still be reported) are known template placeholders.
+_GITLEAKS_SUPPRESS_EXACT_VALUES = {
+    "YOUR_YOUTUBE_API_KEY",
+    "YOUR_YOUTUBE_CHANNEL_ID",
+    "YOUR_GITHUB_PERSONAL_ACCESS_TOKEN",
+}
+
+# Findings in these files are suppressed regardless of value -- they are
+# template/example files that are expected to only ever contain placeholders.
+_GITLEAKS_SUPPRESS_PATHS = [
+    "config_secrets.template.json",
+    "config.template.json",
+]
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Helpers
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _load(path: Path) -> tuple[dict | list | None, str | None]:
+    """Load a JSON artifact file.
+
+    Returns (data, error): error is None on success (data is whatever was
+    parsed, which may legitimately be an empty list/dict for a clean scan);
+    otherwise error is a human-readable reason the artifact is unavailable,
+    distinguishing "missing/malformed artifact" from "valid empty result" so
+    callers don't silently treat a broken CI job as a clean pass.
+    """
+    if not path.exists():
+        return None, f"artifact not found: {path}"
+    try:
+        return json.loads(path.read_text(encoding="utf-8")), None
+    except (json.JSONDecodeError, OSError) as exc:
+        return None, f"could not read/parse {path}: {exc}"
+
+
+def _md_sanitize_cell(value: object) -> str:
+    """Escape/normalize a value so scanner-controlled content (a matched
+    secret, a bandit issue_text, a file path) can't alter the Markdown
+    table's structure: pipes would add bogus columns, newlines would break
+    out of the row (or forge a fake header/separator line)."""
+    text = str(value)
+    text = text.replace("\\", "\\\\").replace("|", "\\|")
+    text = text.replace("\r\n", " ").replace("\n", " ").replace("\r", " ")
+    return text
+
+
+def _md_table_row(*cells: str) -> str:
+    return "| " + " | ".join(_md_sanitize_cell(c) for c in cells) + " |"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Per-tool summarizers
+# Returns: (markdown_lines: list[str], critical_count: int, available: bool)
+# `available=False` means the artifact was missing or malformed -- distinct
+# from a valid scan that simply found nothing -- so the caller can report
+# INCOMPLETE instead of silently counting it as a clean pass.
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _summarize_bandit(artifact_dir: Path) -> tuple[list[str], int, bool]:
+    data, error = _load(artifact_dir / "sast-results" / "bandit-results.json")
+    if error:
+        return [f"_bandit results unavailable: {error}_"], 0, False
+
+    results = data.get("results", [])
+    high = [r for r in results if r.get("issue_severity") == "HIGH"]
+    medium = [r for r in results if r.get("issue_severity") == "MEDIUM"]
+    low = [r for r in results if r.get("issue_severity") == "LOW"]
+
+    lines = [
+        f"**Bandit**: {len(high)} HIGH · {len(medium)} MEDIUM · {len(low)} LOW"
+    ]
+
+    if high:
+        lines += [
+            "",
+            "| Severity | File | Line | Issue |",
+            "| --- | --- | --- | --- |",
+        ]
+        for r in high[:10]:
+            fname = Path(r.get("filename", "")).name
+            lines.append(_md_table_row(
+                "HIGH", f"`{fname}`",
+                str(r.get("line_number", "?")),
+                r.get("issue_text", "")
+            ))
+        if len(high) > 10:
+            lines.append(f"_… and {len(high) - 10} more HIGH findings_")
+
+    return lines, len(high), True
+
+
+def _summarize_pip_audit(artifact_dir: Path) -> tuple[list[str], int, bool]:
+    data, error = _load(artifact_dir / "dependency-audit-results" / "pip-audit-results.json")
+    if error:
+        return [f"_pip-audit results unavailable: {error}_"], 0, False
+
+    # pip-audit JSON format: {"dependencies": [{"name": ..., "vulns": [...]}]}
+    vulns: list[dict] = []
+    for dep in data.get("dependencies", []):
+        for v in dep.get("vulns", []):
+            vulns.append({"package": dep.get("name", "?"), **v})
+
+    lines = [f"**pip-audit**: {len(vulns)} vulnerabilities found"]
+
+    if vulns:
+        lines += ["", "| Package | ID | Fix |", "| --- | --- | --- |"]
+        for v in vulns[:10]:
+            fix = v.get("fix_versions", ["none"])
+            fix_str = ", ".join(fix) if fix else "none"
+            lines.append(_md_table_row(
+                v.get("package", "?"),
+                v.get("id", "?"),
+                fix_str,
+            ))
+
+    # Treat known vulnerabilities as warnings, not critical (they may be unavoidable)
+    return lines, 0, True
+
+
+def _summarize_gitleaks(artifact_dir: Path) -> tuple[list[str], int, bool]:
+    data, error = _load(artifact_dir / "secrets-scan-results" / "gitleaks-results.json")
+    if error:
+        return [f"_gitleaks results unavailable: {error}_"], 0, False
+
+    if not isinstance(data, list):
+        data = []
+
+    real_findings = []
+    suppressed = 0
+    for finding in data:
+        secret_val = str(finding.get("Secret", "") or finding.get("Match", ""))
+        file_name = Path(finding.get("File", "")).name
+        if (secret_val in _GITLEAKS_SUPPRESS_EXACT_VALUES
+                or file_name in _GITLEAKS_SUPPRESS_PATHS):
+            suppressed += 1
+        else:
+            real_findings.append(finding)
+
+    lines = [
+        f"**Gitleaks**: {len(real_findings)} finding(s) "
+        f"({suppressed} suppressed as template placeholders)"
+    ]
+
+    if real_findings:
+        lines += ["", "| Rule | File | Line | Description |", "| --- | --- | --- | --- |"]
+        for f in real_findings[:10]:
+            fname = Path(f.get("File", "")).name
+            lines.append(_md_table_row(
+                f.get("RuleID", "?"),
+                f"`{fname}`",
+                str(f.get("StartLine", "?")),
+                f.get("Description", ""),
+            ))
+
+    critical = len(real_findings)  # any real secret is critical
+    return lines, critical, True
+
+
+def _summarize_security_proofs(artifact_dir: Path) -> tuple[list[str], int, bool]:
+    data, error = _load(artifact_dir / "security-proofs-results" / "security-proofs-results.json")
+    if error:
+        return [f"_security proofs results unavailable: {error}_"], 0, False
+
+    if not isinstance(data, list):
+        data = []
+
+    critical = [r for r in data if r.get("severity") == "CRITICAL"]
+    warnings = [r for r in data if r.get("severity") == "WARNING"]
+    passed = [r for r in data if r.get("severity") == "PASS"]
+    skipped = [r for r in data if r.get("severity") == "SKIP"]
+
+    lines = [
+        f"**Security Proofs**: "
+        f"{len(passed)} PASS · {len(warnings)} WARN · "
+        f"{len(critical)} CRITICAL · {len(skipped)} SKIP",
+        "",
+    ]
+
+    _icon = {"PASS": "✅", "INFO": "ℹ️", "WARNING": "⚠️",  # nosec B105 - severity labels, not credentials
+             "CRITICAL": "🚨", "SKIP": "⏭️"}
+    for r in data:
+        icon = _icon.get(r.get("severity", ""), "❓")
+        lines.append(
+            f"- {icon} **{r.get('test_id', '?')}**: {r.get('message', '')}"
+        )
+        if r.get("details") and r.get("severity") in ("CRITICAL", "WARNING"):
+            lines.append(f"  - _{r['details']}_")
+
+    return lines, len(critical), True
+
+
+def _summarize_plugin_audit(artifact_dir: Path) -> tuple[list[str], int, bool]:
+    data, error = _load(artifact_dir / "plugin-audit-results" / "plugin-audit-results.json")
+    if error:
+        return [f"_plugin audit results unavailable: {error}_"], 0, False
+
+    summary = data.get("summary", {})
+    findings = data.get("findings", [])
+    critical_findings = [f for f in findings if f.get("severity") == "CRITICAL"]
+    warning_findings = [f for f in findings if f.get("severity") == "WARNING"]
+
+    lines = [
+        f"**Plugin Audit**: {data.get('plugins_scanned', '?')} plugins scanned — "
+        f"{summary.get('critical', 0)} CRITICAL · {summary.get('warnings', 0)} WARNINGS"
+    ]
+
+    if critical_findings:
+        lines += ["", "| Plugin | File | Line | Rule | Message |",
+                  "| --- | --- | --- | --- | --- |"]
+        for f in critical_findings[:10]:
+            fname = Path(f.get("file", "")).name
+            lines.append(_md_table_row(
+                f.get("plugin_id", "?"),
+                f"`{fname}`",
+                str(f.get("line", "?")),
+                f.get("rule", "?"),
+                f.get("message", ""),
+            ))
+
+    if warning_findings and not critical_findings:
+        lines.append(f"\n_{len(warning_findings)} warning(s) found — see artifact for details_")
+
+    return lines, summary.get("critical", 0), True
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Main
+# ─────────────────────────────────────────────────────────────────────────────
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="Generate consolidated security audit report",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument("--artifact-dir", required=True,
+                        help="Directory containing downloaded CI artifacts")
+    parser.add_argument("--output", "-o", required=True,
+                        help="Output Markdown file path")
+    parser.add_argument("--verbose", "-v", action="store_true")
+    args = parser.parse_args()
+
+    artifact_dir = Path(args.artifact_dir)
+    timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")
+
+    bandit_lines, bandit_crit, bandit_ok = _summarize_bandit(artifact_dir)
+    pip_audit_lines, pip_audit_crit, pip_audit_ok = _summarize_pip_audit(artifact_dir)
+    gitleaks_lines, gitleaks_crit, gitleaks_ok = _summarize_gitleaks(artifact_dir)
+    proofs_lines, proofs_crit, proofs_ok = _summarize_security_proofs(artifact_dir)
+    plugins_lines, plugins_crit, plugins_ok = _summarize_plugin_audit(artifact_dir)
+
+    unavailable_tools = [
+        name for name, ok in [
+            ("bandit", bandit_ok), ("pip-audit", pip_audit_ok),
+            ("gitleaks", gitleaks_ok), ("security-proofs", proofs_ok),
+            ("plugin-audit", plugins_ok),
+        ] if not ok
+    ]
+
+    total_critical = bandit_crit + pip_audit_crit + gitleaks_crit + proofs_crit + plugins_crit
+    if unavailable_tools:
+        # A missing/malformed artifact means that tool's checks never
+        # actually ran -- this must not be reported as a clean PASS just
+        # because the *artifacts that did load* found nothing.
+        overall = "INCOMPLETE ⚠️"
+    elif total_critical > 0:
+        overall = "ACTION REQUIRED 🚨"
+    else:
+        overall = "PASSED ✅"
+
+    def section(title: str, lines: list[str]) -> str:
+        return f"### {title}\n\n" + "\n".join(lines) + "\n"
+
+    incomplete_note = (
+        f"\n_⚠️ Incomplete: results unavailable for {', '.join(unavailable_tools)} "
+        f"— see the corresponding section(s) below for details_\n"
+        if unavailable_tools else ""
+    )
+
+    report = f"""## 🔒 Security Audit — {overall}
+
+_Generated: {timestamp}_
+{incomplete_note}
+| Critical | High/Warn | Overall |
+| :---: | :---: | :---: |
+| {'🚨 ' + str(total_critical) if total_critical else '✅ 0'} | ⚠️ see below | {overall} |
+
+---
+
+{section('SAST — Bandit', bandit_lines)}
+{section('Dependencies — pip-audit', pip_audit_lines)}
+{section('Secrets — Gitleaks', gitleaks_lines)}
+{section('LEDMatrix Security Proofs', proofs_lines)}
+{section('Plugin Security Audit', plugins_lines)}
+---
+
+_Total critical findings: **{total_critical}**_
+"""
+
+    output_path = Path(args.output)
+    output_path.write_text(report, encoding="utf-8")
+
+    if args.verbose:
+        print(f"  Report written to: {output_path}")
+        print(f"  Status: {overall}")
+        print(f"  Critical findings: {total_critical}")
+        print(f"    bandit={bandit_crit}  pip-audit={pip_audit_crit}  "
+              f"gitleaks={gitleaks_crit}  proofs={proofs_crit}  plugins={plugins_crit}")
+        if unavailable_tools:
+            print(f"  Unavailable: {', '.join(unavailable_tools)}")
+
+    if unavailable_tools:
+        return 1
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/prove_security.py

@@ -0,0 +1,593 @@
+#!/usr/bin/env python3
+"""
+LEDMatrix Security Proof Tests
+
+Automated proofs that run in CI to verify security properties hold on every
+commit. Inspired by the Huntarr security review approach of using standard
+tooling to confirm specific vulnerability classes are absent.
+
+Usage:
+    python scripts/prove_security.py
+    python scripts/prove_security.py --verbose
+    python scripts/prove_security.py --output results.json
+
+Exit code: 1 only if CRITICAL findings are detected. Warnings are reported
+but do not block CI.
+"""
+
+import ast
+import argparse
+import hashlib
+import json
+import re
+import sys
+from dataclasses import dataclass, asdict
+from pathlib import Path
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Result dataclass
+# ─────────────────────────────────────────────────────────────────────────────
+
+@dataclass
+class TestResult:
+    test_id: str
+    severity: str   # PASS | INFO | WARNING | CRITICAL | SKIP
+    message: str
+    details: str = ""
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    @property
+    def icon(self) -> str:
+        return {
+            "PASS": "✅",  # nosec B105 - severity label, not a credential
+            "INFO": "ℹ️ ",
+            "WARNING": "⚠️ ",
+            "CRITICAL": "🚨",
+            "SKIP": "⏭️ ",
+        }.get(self.severity, "❓")
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# T1: Plugin Loading / Zip Slip
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_t1a_zip_slip_protection() -> TestResult:
+    """
+    Verify that zip-slip protection actually guards zip extraction in
+    store_manager.py.
+
+    A whole-file substring check for "is_relative_to"/"Zip-slip detected"
+    would pass even if the guard existed somewhere unrelated, or covered
+    only one of several extract()/extractall() call sites. Instead, this
+    walks the AST: for every extract()/extractall() call, it confirms an
+    is_relative_to() check (and the "Zip-slip detected" log) appears
+    earlier in that same enclosing function -- validate-then-bulk-extract
+    (validate every member, then call extractall() only after all passed)
+    counts as protecting the call, since it covers the same member list.
+    """
+    store_manager = PROJECT_ROOT / "src" / "plugin_system" / "store_manager.py"
+    if not store_manager.exists():
+        return TestResult("T1a", "CRITICAL",
+                          "store_manager.py not found",
+                          f"Expected at {store_manager}")
+
+    content = store_manager.read_text(encoding="utf-8")
+    try:
+        tree = ast.parse(content, filename=str(store_manager))
+    except SyntaxError as exc:
+        return TestResult("T1a", "CRITICAL",
+                          "store_manager.py could not be parsed",
+                          str(exc))
+
+    extraction_sites = 0
+    unprotected: list[str] = []
+
+    for func in ast.walk(tree):
+        if not isinstance(func, (ast.FunctionDef, ast.AsyncFunctionDef)):
+            continue
+
+        extract_calls = [
+            node for node in ast.walk(func)
+            if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
+            and node.func.attr in ("extract", "extractall")
+        ]
+        if not extract_calls:
+            continue
+        extraction_sites += len(extract_calls)
+
+        guard_lines = [
+            n.lineno for n in ast.walk(func)
+            if isinstance(n, ast.Attribute) and n.attr == "is_relative_to"
+        ]
+        has_zip_slip_log = any(
+            isinstance(n, ast.Constant) and isinstance(n.value, str)
+            and "Zip-slip detected" in n.value
+            for n in ast.walk(func)
+        )
+
+        for call in extract_calls:
+            guarded = has_zip_slip_log and any(g < call.lineno for g in guard_lines)
+            if not guarded:
+                unprotected.append(
+                    f"{func.name}() line {call.lineno}: {call.func.attr}() call not "
+                    f"clearly preceded by an is_relative_to() guard + Zip-slip log "
+                    f"in the same function"
+                )
+
+    if extraction_sites == 0:
+        return TestResult("T1a", "WARNING",
+                          "No zipfile extract()/extractall() calls found in store_manager.py",
+                          "Verify plugin installation no longer extracts zip archives, "
+                          "or that this check still targets the right file")
+
+    if unprotected:
+        return TestResult("T1a", "CRITICAL",
+                          f"{len(unprotected)} of {extraction_sites} zip extraction "
+                          f"call(s) not clearly guarded",
+                          "; ".join(unprotected))
+
+    return TestResult("T1a", "PASS",
+                      "Zip-slip protection verified",
+                      f"All {extraction_sites} extract()/extractall() call(s) in "
+                      f"store_manager.py are preceded by an is_relative_to() guard "
+                      f"with a Zip-slip log in the same function")
+
+
+def test_t1b_dangerous_plugin_calls() -> list[TestResult]:
+    """
+    Scan plugin directories for dangerous function calls (eval, exec).
+    These represent arbitrary code execution risks in plugin code.
+    """
+    results = []
+    plugin_dirs = [
+        PROJECT_ROOT / "plugins",
+        PROJECT_ROOT / "plugin-repos",
+    ]
+
+    violations: list[str] = []
+    files_scanned = 0
+
+    scan_errors: list[str] = []
+
+    for base in plugin_dirs:
+        if not base.exists():
+            continue
+        for plugin_dir in sorted(base.iterdir()):
+            if not plugin_dir.is_dir() or plugin_dir.name.startswith(('.', '_')):
+                continue
+            for py_file in plugin_dir.rglob("*.py"):
+                files_scanned += 1
+                try:
+                    source = py_file.read_text(encoding="utf-8")
+                    tree = ast.parse(source, filename=str(py_file))
+                    for node in ast.walk(tree):
+                        if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
+                            if node.func.id in ("eval", "exec"):
+                                rel = py_file.relative_to(PROJECT_ROOT)
+                                violations.append(
+                                    f"{rel}:{node.lineno} — {node.func.id}() call")
+                except (SyntaxError, OSError) as exc:
+                    # A file we couldn't parse/read was never actually
+                    # scanned for eval()/exec() -- that must block this
+                    # test, not silently pass as if it were clean.
+                    rel = py_file.relative_to(PROJECT_ROOT)
+                    scan_errors.append(f"{rel} — {type(exc).__name__}: {exc}")
+
+    if scan_errors:
+        results.append(TestResult(
+            "T1b", "CRITICAL",
+            f"{len(scan_errors)} plugin file(s) could not be scanned for eval()/exec()",
+            "; ".join(scan_errors[:10])
+        ))
+
+    if violations:
+        results.append(TestResult(
+            "T1b", "CRITICAL",
+            f"Dangerous function calls found in plugins ({len(violations)} instance(s))",
+            "; ".join(violations[:10])
+        ))
+    elif not scan_errors:
+        results.append(TestResult(
+            "T1b", "PASS",
+            "No eval()/exec() calls found in plugins",
+            f"{files_scanned} plugin Python files scanned"
+        ))
+
+    return results
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# T2: API Surface Inventory
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_t2a_api_surface_inventory() -> TestResult:
+    """
+    Document the API surface area.
+
+    This app intentionally has no authentication (local-only Raspberry Pi
+    design, documented in web_interface/app.py). This test produces an
+    inventory for audit purposes and warns only if the design-intent comment
+    is removed from app.py (which would indicate someone deleted the rationale
+    without adding auth, rather than a deliberate undocumented change).
+    """
+    api_file = PROJECT_ROOT / "web_interface" / "blueprints" / "api_v3.py"
+    app_file = PROJECT_ROOT / "web_interface" / "app.py"
+
+    if not api_file.exists():
+        return TestResult("T2a", "WARNING", "api_v3.py not found", str(api_file))
+
+    api_content = api_file.read_text(encoding="utf-8")
+    routes = re.findall(r"@api_v3\.route\('([^']+)'", api_content)
+
+    csrf_documented = False
+    if app_file.exists():
+        app_content = app_file.read_text(encoding="utf-8")
+        csrf_documented = "CSRF protection disabled for local-only" in app_content
+
+    summary = (
+        f"{len(routes)} API routes in api_v3.py. "
+        f"No auth decorators (intentional local-only design). "
+        f"CSRF disabled: {'YES — design intent documented in app.py' if csrf_documented else 'YES — but design intent comment NOT found in app.py'}. "
+        f"Rate limiting: 1000/min."
+    )
+
+    if not csrf_documented:
+        return TestResult(
+            "T2a", "WARNING",
+            "CSRF is disabled but the design-intent comment is missing from app.py",
+            "Add the rationale comment back, or add proper CSRF protection if "
+            "the app is now internet-facing"
+        )
+
+    # There is currently no config mechanism that actually enforces the
+    # local-only boundary the design-intent comment describes -- app.py
+    # hardcodes host='0.0.0.0' unconditionally, so nothing here can confirm
+    # this deployment is in fact LAN-only. Reporting this as mere INFO
+    # understates that: an unauthenticated, CSRF-disabled API surface is a
+    # real risk the moment this ever runs somewhere other than a home LAN,
+    # documented rationale or not.
+    return TestResult(
+        "T2a", "WARNING",
+        "API surface has no auth and CSRF disabled; enforcement of the "
+        "documented local-only boundary cannot be confirmed",
+        summary
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# T3: Secrets & Credential Handling
+# ─────────────────────────────────────────────────────────────────────────────
+
+# Patterns that suggest real credentials (must be >8 chars, not placeholders)
+_SECRET_PATTERNS = [
+    (r'(?i)password\s*=\s*["\'](?!none|empty|placeholder|example|test|default|""|'')[^"\']{8,}["\']', "WARNING", "password"),
+    (r'(?i)api[_-]?key\s*=\s*["\'](?!none|empty|placeholder|YOUR_|example|test)[^"\']{16,}["\']', "WARNING", "api_key"),
+    (r'(?i)secret\s*=\s*["\'](?!none|empty|placeholder|YOUR_|example|test)[^"\']{16,}["\']', "WARNING", "secret"),
+    # Real GitHub token pattern
+    (r'ghp_[a-zA-Z0-9]{36}', "CRITICAL", "github_token"),
+    # Generic long bearer tokens
+    (r'Bearer\s+[a-zA-Z0-9\-_\.]{32,}', "WARNING", "bearer_token"),
+]
+
+_TEMPLATE_SKIP_STRINGS = [
+    "YOUR_", "PLACEHOLDER", "_HERE", "example.com", "config_secrets.template",
+    "prove_security",  # this file itself
+]
+
+_SCAN_DIRS = ["src", "web_interface", "scripts"]
+
+
+def test_t3a_hardcoded_secrets() -> TestResult:
+    """Scan source code for hardcoded credentials."""
+    violations: list[str] = []
+
+    for dir_name in _SCAN_DIRS:
+        scan_dir = PROJECT_ROOT / dir_name
+        if not scan_dir.exists():
+            continue
+        for py_file in scan_dir.rglob("*.py"):
+            # Skip test files and this script
+            if "test" in str(py_file).lower() or "prove_security" in str(py_file):
+                continue
+            try:
+                content = py_file.read_text(encoding="utf-8")
+            except OSError:
+                continue
+
+            for pattern, severity, pattern_type in _SECRET_PATTERNS:
+                for match in re.finditer(pattern, content):
+                    line_content = match.group(0)
+                    # Skip lines containing template placeholder strings.
+                    # line_content is only used for this in-memory check --
+                    # it must never be stored or included in output below.
+                    if any(skip in line_content for skip in _TEMPLATE_SKIP_STRINGS):
+                        continue
+                    rel = py_file.relative_to(PROJECT_ROOT)
+                    line_no = content[: match.start()].count("\n") + 1
+                    # Redacted fingerprint lets the same finding be recognized
+                    # across scans without ever reporting the matched
+                    # credential itself (which would otherwise get published
+                    # into CI logs, JSON artifacts, and PR comments -- wider
+                    # exposure than the original leak).
+                    fingerprint = hashlib.sha256(line_content.encode()).hexdigest()[:12]
+                    violations.append(
+                        f"[{severity}] {rel}:{line_no} — {pattern_type} "
+                        f"(fingerprint {fingerprint})"
+                    )
+
+    critical_violations = [v for v in violations if "[CRITICAL]" in v]
+    if critical_violations:
+        return TestResult(
+            "T3a", "CRITICAL",
+            f"Hardcoded secrets found ({len(critical_violations)} critical)",
+            "; ".join(critical_violations[:5])
+        )
+    if violations:
+        return TestResult(
+            "T3a", "WARNING",
+            f"Potential hardcoded secrets found ({len(violations)} instance(s))",
+            "; ".join(violations[:5])
+        )
+
+    return TestResult("T3a", "PASS", "No hardcoded secrets detected",
+                      f"Scanned {', '.join(_SCAN_DIRS)}")
+
+
+def test_t3b_plaintext_password_storage() -> TestResult:
+    """
+    Check for user account password storage without hashing.
+
+    The LEDMatrix app has no user account system, so this should produce INFO.
+    It would only CRITICAL if someone added user auth and stored passwords without hashing.
+
+    We require all three of: a password *variable assignment or DB operation*,
+    a clear storage call (INSERT / db commit / ORM save), and no hashing lib present
+    — to avoid false positives from files that contain 'password' for WiFi handling
+    and '.save()' for image/file saving in unrelated functions.
+    """
+    hashing_libs = ["bcrypt", "argon2", "pbkdf2", "scrypt",
+                    "generate_password_hash", "hashpw", "make_password"]
+    # Patterns that indicate password being stored in a database / ORM context.
+    # Must be specific enough to avoid matching set.add(), file.save(), etc.
+    db_storage_patterns = ["INSERT INTO", "db.session", "session.add(", "session.commit(", "orm.save"]
+
+    password_storage_found = False
+
+    for dir_name in _SCAN_DIRS:
+        scan_dir = PROJECT_ROOT / dir_name
+        if not scan_dir.exists():
+            continue
+        for py_file in scan_dir.rglob("*.py"):
+            try:
+                content = py_file.read_text(encoding="utf-8")
+            except OSError:
+                continue
+            # Require DB/ORM context specifically — not just any .save() call
+            if ("password" in content.lower() and
+                    any(store in content for store in db_storage_patterns) and
+                    not any(h in content for h in hashing_libs)):
+                password_storage_found = True
+
+    if password_storage_found:
+        return TestResult(
+            "T3b", "CRITICAL",
+            "Potential plaintext password storage in database/ORM detected",
+            "Found password + database storage operations without a recognized hashing library"
+        )
+
+    return TestResult("T3b", "INFO",
+                      "No plaintext password storage detected",
+                      "App has no user account system — expected result")
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# T4: Path Traversal
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_t4a_path_traversal() -> TestResult:
+    """
+    Verify static file serving uses send_from_directory (safe) rather than
+    open() with user-supplied paths. Also checks for extractall() calls that
+    lack the is_relative_to() guard.
+    """
+    issues: list[str] = []
+
+    app_file = PROJECT_ROOT / "web_interface" / "app.py"
+    if app_file.exists():
+        content = app_file.read_text(encoding="utf-8")
+        # The file-serve route should use send_from_directory or commonpath
+        if "send_from_directory" not in content and "commonpath" not in content:
+            issues.append("app.py: file-serve routes may not use send_from_directory/commonpath")
+
+    # Check all extractall() calls have a preceding is_relative_to guard
+    for py_file in (PROJECT_ROOT / "src").rglob("*.py"):
+        try:
+            content = py_file.read_text(encoding="utf-8")
+        except OSError:
+            continue
+        if "extractall(" in content and "is_relative_to" not in content:
+            rel = py_file.relative_to(PROJECT_ROOT)
+            issues.append(f"{rel}: extractall() without is_relative_to() guard")
+
+    if issues:
+        return TestResult(
+            "T4a", "WARNING",
+            f"Potential path traversal patterns found ({len(issues)})",
+            "; ".join(issues)
+        )
+
+    return TestResult("T4a", "PASS",
+                      "Path traversal mitigations verified",
+                      "send_from_directory/commonpath used for file serving; "
+                      "extractall() calls have is_relative_to() guards")
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# T5: Auth Bypass Patterns
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_t5a_auth_bypass_patterns() -> TestResult:
+    """
+    Look for broken auth bypass patterns — not the intentional no-auth design
+    (T2a covers that), but patterns that suggest auth was INTENDED to exist
+    but has an exploitable bypass: broad substring matching, debug-mode skips,
+    or if-True conditions.
+    """
+    bypass_signals = [
+        (r'if\s+True\s*:', "if True: bypass"),
+        (r'if\s+debug\s*:', "debug-mode auth skip"),
+        (r'request\.path\s+in\s+', "substring path matching in auth (Huntarr pattern)"),
+        (r'EXEMPT_ROUTES\s*=', "exempt routes list"),
+    ]
+
+    findings: list[str] = []
+
+    for dir_name in ["src", "web_interface"]:
+        scan_dir = PROJECT_ROOT / dir_name
+        if not scan_dir.exists():
+            continue
+        for py_file in scan_dir.rglob("*.py"):
+            try:
+                content = py_file.read_text(encoding="utf-8")
+            except OSError:
+                continue
+            for pattern, label in bypass_signals:
+                if re.search(pattern, content):
+                    # Only flag if the file also contains auth-related terms
+                    if any(auth in content.lower() for auth in
+                           ["auth", "login", "authenticate", "token", "permission"]):
+                        rel = py_file.relative_to(PROJECT_ROOT)
+                        findings.append(f"{rel}: {label}")
+
+    if findings:
+        return TestResult(
+            "T5a", "WARNING",
+            f"Potential auth bypass patterns found ({len(findings)})",
+            "; ".join(findings[:5])
+        )
+
+    return TestResult("T5a", "PASS",
+                      "No auth bypass patterns detected",
+                      "Checked src/ and web_interface/ for bypass signals")
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# T6: Docker / Container Hardening
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_t6_docker_hardening() -> TestResult:
+    """Container security — skipped if no Dockerfile exists."""
+    dockerfile = PROJECT_ROOT / "Dockerfile"
+    if not dockerfile.exists():
+        return TestResult("T6", "SKIP",
+                          "No Dockerfile found — container security scan not applicable",
+                          "If Docker support is added in future, enable hadolint/trivy scanning "
+                          "in .github/workflows/security-audit.yml")
+
+    content = dockerfile.read_text(encoding="utf-8")
+    issues: list[str] = []
+
+    # Check for non-root USER directive
+    user_lines = [l for l in content.splitlines() if l.strip().startswith("USER")]
+    if not user_lines or user_lines[-1].strip() == "USER root":
+        issues.append("Container runs as root — use USER directive to drop privileges")
+
+    # Check for pinned base image tags. A tag (even a specific version, not
+    # just :latest) is mutable -- the same tag can point to a different
+    # image later. Only a @sha256 digest is truly immutable/reproducible.
+    from_lines = [line for line in content.splitlines() if line.strip().startswith("FROM")]
+    for from_line in from_lines:
+        parts = from_line.split()
+        # FROM [--platform=<platform>] <image> [AS <name>] -- skip an
+        # optional --platform= flag so it's never mistaken for the image
+        # token itself (which would falsely report it as unpinned).
+        image_parts = [p for p in parts[1:] if not p.startswith("--platform=")]
+        if image_parts:
+            image = image_parts[0]
+            if "@sha256:" not in image:
+                issues.append(f"Base image not pinned to a digest: {image}")
+
+    if issues:
+        return TestResult("T6", "WARNING",
+                          f"Dockerfile hardening issues ({len(issues)})",
+                          "; ".join(issues))
+
+    return TestResult("T6", "PASS", "Dockerfile hardening checks passed", "")
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Runner
+# ─────────────────────────────────────────────────────────────────────────────
+
+def main() -> int:
+    parser = argparse.ArgumentParser(
+        description="LEDMatrix security proof tests",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    parser.add_argument("--output", "-o", default=None,
+                        help="Write JSON results to this file")
+    parser.add_argument("--verbose", "-v", action="store_true",
+                        help="Show details for each check")
+    args = parser.parse_args()
+
+    print("=" * 60)
+    print("LEDMatrix Security Proof Tests")
+    print(f"Project root: {PROJECT_ROOT}")
+    print("=" * 60)
+
+    all_results: list[TestResult] = []
+
+    # Run all test groups
+    all_results.append(test_t1a_zip_slip_protection())
+    all_results.extend(test_t1b_dangerous_plugin_calls())
+    all_results.append(test_t2a_api_surface_inventory())
+    all_results.append(test_t3a_hardcoded_secrets())
+    all_results.append(test_t3b_plaintext_password_storage())
+    all_results.append(test_t4a_path_traversal())
+    all_results.append(test_t5a_auth_bypass_patterns())
+    all_results.append(test_t6_docker_hardening())
+
+    # Print results
+    print()
+    for r in all_results:
+        line = f"  {r.icon} [{r.severity:<8}] {r.test_id}: {r.message}"
+        print(line)
+        if args.verbose and r.details:
+            print(f"             {r.details}")
+
+    # Tally
+    critical = [r for r in all_results if r.severity == "CRITICAL"]
+    warnings = [r for r in all_results if r.severity == "WARNING"]
+    passed = [r for r in all_results if r.severity == "PASS"]
+    skipped = [r for r in all_results if r.severity == "SKIP"]
+
+    print()
+    print(f"  Results: {len(passed)} PASS  {len(warnings)} WARN  "
+          f"{len(critical)} CRITICAL  {len(skipped)} SKIP")
+
+    # Write JSON output
+    if args.output:
+        output_data = [r.to_dict() for r in all_results]
+        Path(args.output).write_text(
+            json.dumps(output_data, indent=2), encoding="utf-8"
+        )
+        print(f"  Results written to: {args.output}")
+
+    if critical:
+        print(f"\n  🚨 {len(critical)} CRITICAL issue(s) found — blocking")
+        return 1
+
+    if warnings:
+        print(f"\n  ⚠️  {len(warnings)} warning(s) found — non-blocking")
+
+    print("\n  ✅ All checks passed (warnings are non-blocking)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/render_plugin.py

@@ -28,7 +28,7 @@ os.environ['EMULATOR'] = 'true'
 # Import logger after path setup so src.logging_config is importable
 from src.logging_config import get_logger  # noqa: E402
 from src.plugin_system.testing.loading import (  # noqa: E402
-    find_plugin_dir, load_manifest, load_config_defaults,
+    build_full_config, find_plugin_dir, load_manifest,
 )
 logger = get_logger("[Render Plugin]")
 
@@ -83,16 +83,13 @@ def main() -> int:
     manifest = load_manifest(Path(plugin_dir))
 
     # Parse config: start with schema defaults, then apply overrides
-    config_defaults = load_config_defaults(Path(plugin_dir))
     try:
         user_config = json.loads(args.config)
     except json.JSONDecodeError as e:
         logger.error("Invalid JSON config: %s", e)
         return 1
 
-    config = {'enabled': True}
-    config.update(config_defaults)
-    config.update(user_config)
+    config = build_full_config(Path(plugin_dir), cli_config=user_config)
 
     # Load mock data if provided
     mock_data = {}

scripts/utils/wifi_monitor_daemon.py

@@ -78,21 +78,17 @@ class WiFiMonitorDaemon:
         
         while self.running:
             try:
-                # Get current status before checking
-                status = self.wifi_manager.get_wifi_status()
-                ethernet_connected = self.wifi_manager._is_ethernet_connected()
-                
-                # Check WiFi status and manage AP mode
-                state_changed = self.wifi_manager.check_and_manage_ap_mode()
-                
-                # Get updated status after check
-                updated_status = self.wifi_manager.get_wifi_status()
-                updated_ethernet = self.wifi_manager._is_ethernet_connected()
-                
+                # One combined check that also returns the state it observed —
+                # the previous flow fetched status before AND after the check
+                # on top of the check's own internal fetch, each one several
+                # nmcli subprocess forks, every 30s, forever.
+                (state_changed, updated_status, updated_ethernet,
+                 ap_active) = self.wifi_manager.check_and_manage_ap_mode_with_state()
+
                 current_state = {
                     'connected': updated_status.connected,
                     'ethernet_connected': updated_ethernet,
-                    'ap_active': updated_status.ap_mode_active,
+                    'ap_active': ap_active,
                     'ssid': updated_status.ssid
                 }
                 
@@ -109,7 +105,7 @@ class WiFiMonitorDaemon:
                     else:
                         logger.debug("Ethernet not connected")
                     
-                    if updated_status.ap_mode_active:
+                    if ap_active:
                         logger.info(f"AP mode ACTIVE - SSID: {ap_ssid} (IP: 192.168.4.1)")
                     else:
                         logger.debug("AP mode inactive")
@@ -123,16 +119,16 @@ class WiFiMonitorDaemon:
                     # Log periodic status (less verbose)
                     if updated_status.connected:
                         logger.debug(f"Status check: WiFi={updated_status.ssid} ({updated_status.signal}%), "
-                                   f"Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}")
+                                   f"Ethernet={updated_ethernet}, AP={ap_active}")
                     else:
-                        logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}")
+                        logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={ap_active}")
                 
                 # Escalating recovery: if nmcli reports connected but actual internet
                 # is unreachable for several consecutive checks, restart NetworkManager.
                 # This is done HERE (not inside check_and_manage_ap_mode) to keep the
                 # AP-enable trigger clean and avoid false-positive AP enables from
                 # transient packet loss on otherwise working WiFi.
-                if updated_status.connected and not updated_status.ap_mode_active:
+                if updated_status.connected and not ap_active:
                     if not self.wifi_manager.check_internet_connectivity():
                         self._consecutive_internet_failures += 1
                         logger.warning(

src/background_cache_mixin.py

@@ -1,134 +0,0 @@
-"""
-Background Cache Mixin for Sports Managers
-
-This mixin provides common caching functionality to eliminate code duplication
-across all sports managers. It implements the background service cache pattern
-where Recent/Upcoming managers consume data from the background service cache.
-"""
-
-import time
-from typing import Dict, Optional, Any, Callable
-
-
-class BackgroundCacheMixin:
-    """
-    Mixin class that provides background service cache functionality to sports managers.
-    
-    This mixin eliminates code duplication by providing a common implementation
-    for the background service cache pattern used across all sports managers.
-    
-    Note: For non-sports managers (weather, stocks, news, etc.), use
-    GenericCacheMixin instead. See src/generic_cache_mixin.py for details.
-    """
-    
-    def _fetch_data_with_background_cache(self, 
-                                        sport_key: str, 
-                                        api_fetch_method: Callable,
-                                        live_manager_class: type = None) -> Optional[Dict]:
-        """
-        Common logic for fetching data with background service cache support.
-        
-        This method implements the background service cache pattern:
-        1. Live managers always fetch fresh data
-        2. Recent/Upcoming managers try background cache first
-        3. Fallback to direct API call if background data unavailable
-        
-        Args:
-            sport_key: Sport identifier (e.g., 'nba', 'nfl', 'ncaa_fb')
-            api_fetch_method: Method to call for direct API fetch
-            live_manager_class: Class to check if this is a live manager
-            
-        Returns:
-            Cached or fresh data from API
-        """
-        start_time = time.time()
-        cache_hit = False
-        cache_source = None
-        
-        try:
-            # For Live managers, always fetch fresh data
-            if live_manager_class and isinstance(self, live_manager_class):
-                self.logger.info(f"[{sport_key.upper()}] Live manager - fetching fresh data")
-                result = api_fetch_method(use_cache=False)
-                cache_source = "live_fresh"
-            else:
-                # For Recent/Upcoming managers, try background service cache first
-                cache_key = self.cache_manager.generate_sport_cache_key(sport_key)
-                
-                # Check if background service has fresh data
-                if self.cache_manager.is_background_data_available(cache_key, sport_key):
-                    cached_data = self.cache_manager.get_background_cached_data(cache_key, sport_key)
-                    if cached_data:
-                        self.logger.info(f"[{sport_key.upper()}] Using background service cache for {cache_key}")
-                        result = cached_data
-                        cache_hit = True
-                        cache_source = "background_cache"
-                    else:
-                        self.logger.warning(f"[{sport_key.upper()}] Background cache check passed but no data returned for {cache_key}")
-                        result = None
-                        cache_source = "background_miss"
-                else:
-                    self.logger.info(f"[{sport_key.upper()}] Background data not available for {cache_key}")
-                    result = None
-                    cache_source = "background_unavailable"
-                
-                # Fallback to direct API call if background data not available
-                if result is None:
-                    self.logger.info(f"[{sport_key.upper()}] Fetching directly from API for {cache_key}")
-                    result = api_fetch_method(use_cache=True)
-                    cache_source = "api_fallback"
-            
-            # Record performance metrics
-            duration = time.time() - start_time
-            self.cache_manager.record_fetch_time(duration)
-            
-            # Log performance metrics
-            self._log_fetch_performance(sport_key, duration, cache_hit, cache_source)
-            
-            return result
-            
-        except Exception as e:
-            duration = time.time() - start_time
-            self.logger.error(f"[{sport_key.upper()}] Error in background cache fetch after {duration:.2f}s: {e}")
-            self.cache_manager.record_fetch_time(duration)
-            raise
-    
-    def _log_fetch_performance(self, sport_key: str, duration: float, cache_hit: bool, cache_source: str):
-        """
-        Log detailed performance metrics for fetch operations.
-        
-        Args:
-            sport_key: Sport identifier
-            duration: Fetch operation duration in seconds
-            cache_hit: Whether this was a cache hit
-            cache_source: Source of the data (background_cache, api_fallback, etc.)
-        """
-        # Log basic performance info
-        self.logger.info(f"[{sport_key.upper()}] Fetch completed in {duration:.2f}s "
-                        f"(cache_hit={cache_hit}, source={cache_source})")
-        
-        # Log detailed metrics every 10 operations
-        if hasattr(self, '_fetch_count'):
-            self._fetch_count += 1
-        else:
-            self._fetch_count = 1
-            
-        if self._fetch_count % 10 == 0:
-            metrics = self.cache_manager.get_cache_metrics()
-            self.logger.info(f"[{sport_key.upper()}] Cache Performance Summary - "
-                           f"Hit Rate: {metrics['cache_hit_rate']:.2%}, "
-                           f"Background Hit Rate: {metrics['background_hit_rate']:.2%}, "
-                           f"API Calls Saved: {metrics['api_calls_saved']}")
-    
-    def get_cache_performance_summary(self) -> Dict[str, Any]:
-        """
-        Get cache performance summary for this manager.
-        
-        Returns:
-            Dictionary containing cache performance metrics
-        """
-        return self.cache_manager.get_cache_metrics()
-    
-    def log_cache_performance(self):
-        """Log current cache performance metrics."""
-        self.cache_manager.log_cache_metrics()

src/common/basketball_plugin_example.py

@@ -1,328 +0,0 @@
-"""
-Example: Basketball Plugin using LEDMatrix Common Helpers
-
-This example shows how to refactor the basketball plugin to use the
-ledmatrix-common package for cleaner, more maintainable code.
-"""
-
-from pathlib import Path
-from typing import Any, Dict, List, Optional
-
-
-# Import common helpers
-from src.common import (
-    LogoHelper, TextHelper, APIHelper, DisplayHelper, 
-    GameHelper, ConfigHelper
-)
-from src.plugin_system.base_plugin import BasePlugin
-
-
-class BasketballPluginManager(BasePlugin):
-    """
-    Basketball scoreboard plugin using LEDMatrix Common helpers.
-    
-    This version is much cleaner and more maintainable than the original
-    because it delegates common functionality to the shared helpers.
-    """
-    
-    def __init__(
-        self,
-        plugin_id: str,
-        config: Dict[str, Any],
-        display_manager,
-        cache_manager,
-        plugin_manager
-    ):
-        """Initialize the basketball plugin with common helpers."""
-        super().__init__(plugin_id, config, display_manager, cache_manager, plugin_manager)
-        
-        # Get display dimensions
-        self.display_width = display_manager.matrix.width
-        self.display_height = display_manager.matrix.height
-        
-        # Initialize common helpers
-        self._init_helpers()
-        
-        # Load configuration
-        self._load_config()
-        
-        # State tracking
-        self.current_games = []
-        self.current_game = None
-        
-        # Log initialization
-        enabled_leagues = [k for k, v in self.league_configs.items() if v['enabled']]
-        self.logger.info(f"Basketball plugin initialized with leagues: {enabled_leagues}")
-    
-    def _init_helpers(self):
-        """Initialize all common helpers."""
-        # Logo helper for team logos
-        self.logo_helper = LogoHelper(
-            display_width=self.display_width,
-            display_height=self.display_height,
-            logger=self.logger
-        )
-        
-        # Text helper for rendering
-        self.text_helper = TextHelper(logger=self.logger)
-        self.fonts = self.text_helper.load_fonts()
-        
-        # API helper for ESPN data
-        self.api_helper = APIHelper(
-            cache_manager=self.cache_manager,
-            logger=self.logger
-        )
-        
-        # Display helper for layouts
-        self.display_helper = DisplayHelper(
-            display_width=self.display_width,
-            display_height=self.display_height,
-            logger=self.logger
-        )
-        
-        # Game helper for data processing
-        self.game_helper = GameHelper(
-            timezone_str=self.config.get('timezone', 'UTC'),
-            logger=self.logger
-        )
-        
-        # Config helper for configuration management
-        self.config_helper = ConfigHelper(logger=self.logger)
-    
-    def _load_config(self):
-        """Load and validate configuration."""
-        # Get basketball-specific config
-        basketball_config = self.config_helper.get_sports_config(self.config, 'basketball')
-        
-        # Build league configurations
-        self.league_configs = {
-            'nba': {
-                'enabled': basketball_config.get('nba_enabled', True),
-                'url': 'https://site.api.espn.com/apis/site/v2/sports/basketball/nba/scoreboard',
-                'logo_dir': Path('assets/sports/nba_logos'),
-                'favorite_teams': basketball_config.get('nba_favorite_teams', []),
-                'display_modes': {
-                    'nba_live': basketball_config.get('nba_display_modes_live', True),
-                    'nba_recent': basketball_config.get('nba_display_modes_recent', True),
-                    'nba_upcoming': basketball_config.get('nba_display_modes_upcoming', True),
-                },
-            },
-            'wnba': {
-                'enabled': basketball_config.get('wnba_enabled', False),
-                'url': 'https://site.api.espn.com/apis/site/v2/sports/basketball/wnba/scoreboard',
-                'logo_dir': Path('assets/sports/wnba_logos'),
-                'favorite_teams': basketball_config.get('wnba_favorite_teams', []),
-                'display_modes': {
-                    'wnba_live': basketball_config.get('wnba_display_modes_live', True),
-                    'wnba_recent': basketball_config.get('wnba_display_modes_recent', True),
-                    'wnba_upcoming': basketball_config.get('wnba_display_modes_upcoming', True),
-                },
-            },
-            'ncaam': {
-                'enabled': basketball_config.get('ncaam_basketball_enabled', False),
-                'url': 'https://site.api.espn.com/apis/site/v2/sports/basketball/mens-college-basketball/scoreboard',
-                'logo_dir': Path('assets/sports/ncaa_logos'),
-                'favorite_teams': basketball_config.get('ncaam_basketball_favorite_teams', []),
-                'display_modes': {
-                    'ncaam_basketball_live': basketball_config.get('ncaam_basketball_display_modes_live', True),
-                    'ncaam_basketball_recent': basketball_config.get('ncaam_basketball_display_modes_recent', True),
-                    'ncaam_basketball_upcoming': basketball_config.get('ncaam_basketball_display_modes_upcoming', True),
-                },
-            },
-            'ncaaw': {
-                'enabled': basketball_config.get('ncaaw_basketball_enabled', False),
-                'url': 'https://site.api.espn.com/apis/site/v2/sports/basketball/womens-college-basketball/scoreboard',
-                'logo_dir': Path('assets/sports/ncaa_logos'),
-                'favorite_teams': basketball_config.get('ncaaw_basketball_favorite_teams', []),
-                'display_modes': {
-                    'ncaaw_basketball_live': basketball_config.get('ncaaw_basketball_display_modes_live', True),
-                    'ncaaw_basketball_recent': basketball_config.get('ncaaw_basketball_display_modes_recent', True),
-                    'ncaaw_basketball_upcoming': basketball_config.get('ncaaw_basketball_display_modes_upcoming', True),
-                },
-            },
-        }
-    
-    def update(self) -> None:
-        """Update game data for all enabled leagues."""
-        try:
-            all_games = []
-            
-            for league_key, league_config in self.league_configs.items():
-                if not league_config['enabled']:
-                    continue
-                
-                games = self._fetch_league_games(league_key, league_config)
-                for game in games:
-                    game['league_key'] = league_key
-                    game['league_config'] = league_config
-                all_games.extend(games)
-            
-            self.current_games = all_games
-            self.logger.debug(f"Updated basketball data: {len(all_games)} total games")
-            
-        except Exception as e:
-            self.logger.error(f"Error updating basketball data: {e}", exc_info=True)
-    
-    def _fetch_league_games(self, league_key: str, league_config: Dict) -> List[Dict]:
-        """Fetch games for a specific league using API helper."""
-        try:
-            # Use API helper to fetch ESPN data with caching
-            data = self.api_helper.fetch_espn_scoreboard(
-                sport='basketball',
-                league=league_key,
-                cache_key=f"basketball_{league_key}",
-                cache_ttl=300  # 5 minutes cache
-            )
-            
-            if not data:
-                return []
-            
-            # Use game helper to process events
-            events = data.get('events', [])
-            games = self.game_helper.process_games(events, sport='basketball')
-            
-            # Add logo paths to games
-            for game in games:
-                logo_dir = league_config['logo_dir']
-                game['home_logo_path'] = logo_dir / f"{game['home_abbr']}.png"
-                game['away_logo_path'] = logo_dir / f"{game['away_abbr']}.png"
-            
-            return games
-            
-        except Exception as e:
-            self.logger.error(f"Error fetching {league_key} games: {e}", exc_info=True)
-            return []
-    
-    def display(self, force_clear: bool = False, display_mode: str = None) -> None:
-        """Display basketball games using display helper."""
-        try:
-            mode = display_mode or self._determine_display_mode()
-            
-            if not mode:
-                self._display_no_games()
-                return
-            
-            # Filter games for mode
-            filtered_games = self._filter_games_for_mode(mode)
-            
-            if not filtered_games:
-                self._display_no_games()
-                return
-            
-            # Display first game
-            self.current_game = filtered_games[0]
-            self._draw_scorebug_layout(self.current_game, force_clear)
-            
-        except Exception as e:
-            self.logger.error(f"Error displaying game: {e}", exc_info=True)
-    
-    def _determine_display_mode(self) -> Optional[str]:
-        """Determine display mode based on available games."""
-        # Priority: live > recent > upcoming
-        for game in self.current_games:
-            if game.get('is_live'):
-                return f"{game['league_key']}_live"
-        for game in self.current_games:
-            if game.get('is_final'):
-                return f"{game['league_key']}_recent"
-        for game in self.current_games:
-            if game.get('is_upcoming'):
-                return f"{game['league_key']}_upcoming"
-        return None
-    
-    def _filter_games_for_mode(self, mode: str) -> List[Dict]:
-        """Filter games based on display mode."""
-        filtered = []
-        
-        for game in self.current_games:
-            league_config = game.get('league_config', {})
-            display_modes = league_config.get('display_modes', {})
-            
-            if mode in display_modes and display_modes[mode]:
-                if 'live' in mode and game.get('is_live'):
-                    filtered.append(game)
-                elif 'recent' in mode and game.get('is_final'):
-                    filtered.append(game)
-                elif 'upcoming' in mode and game.get('is_upcoming'):
-                    filtered.append(game)
-        
-        return filtered[:5]
-    
-    def _draw_scorebug_layout(self, game: Dict, force_clear: bool = False) -> None:
-        """Draw the basketball scorebug layout using display helper."""
-        try:
-            # Load logos using logo helper
-            home_logo = self.logo_helper.load_logo(
-                game['home_abbr'], 
-                game['home_logo_path']
-            )
-            away_logo = self.logo_helper.load_logo(
-                game['away_abbr'], 
-                game['away_logo_path']
-            )
-            
-            if not home_logo or not away_logo:
-                self.logger.error("Failed to load logos")
-                self._display_error("Logo Error")
-                return
-            
-            # Use display helper to create scorebug layout
-            final_img = self.display_helper.draw_scorebug_layout(
-                game_data=game,
-                fonts=self.fonts,
-                home_logo=home_logo,
-                away_logo=away_logo
-            )
-            
-            # Display the image
-            self.display_manager.image.paste(final_img, (0, 0))
-            self.display_manager.update_display()
-            
-        except Exception as e:
-            self.logger.error(f"Error drawing scorebug: {e}", exc_info=True)
-    
-    def _display_no_games(self) -> None:
-        """Display 'no games' message using display helper."""
-        try:
-            img = self.display_helper.draw_no_data_message("No Games")
-            self.display_manager.image = img.copy()
-            self.display_manager.update_display()
-        except Exception as e:
-            self.logger.error(f"Error displaying no games: {e}", exc_info=True)
-    
-    def _display_error(self, message: str) -> None:
-        """Display error message using display helper."""
-        try:
-            img = self.display_helper.draw_error_message(message)
-            self.display_manager.image = img.copy()
-            self.display_manager.update_display()
-        except Exception as e:
-            self.logger.error(f"Error displaying error message: {e}", exc_info=True)
-    
-    def get_display_duration(self) -> float:
-        """Get display duration."""
-        return self.config.get('display_duration', 15)
-    
-    def cleanup(self) -> None:
-        """Cleanup resources."""
-        self.current_games = []
-        self.logger.info("Basketball plugin cleaned up")
-
-
-# Example usage and benefits:
-"""
-Benefits of using LEDMatrix Common helpers:
-
-1. **Cleaner Code**: The plugin is much shorter and more readable
-2. **Reusable Components**: Common functionality is shared across plugins
-3. **Better Testing**: Each helper can be tested independently
-4. **Easier Maintenance**: Bug fixes in helpers benefit all plugins
-5. **Consistent Behavior**: All plugins use the same underlying logic
-6. **Reduced Dependencies**: Plugins don't need to import LEDMatrix core
-7. **Better Error Handling**: Centralized error handling in helpers
-8. **Configuration Management**: Consistent config handling across plugins
-
-The original basketball plugin was 326 lines. This version is much cleaner
-and delegates most functionality to the common helpers, making it easier to
-maintain and extend.
-"""

src/common/permission_utils.py

@@ -146,6 +146,60 @@ def ensure_file_permissions(path: Path, mode: int = 0o644) -> None:
         raise
 
 
+_shared_group_gid_cache: Optional[int] = None
+
+
+def get_shared_group_gid() -> Optional[int]:
+    """
+    Return the gid that should own config/secrets files shared between the
+    root-run ``ledmatrix.service`` (main display) and the non-root user that
+    ``ledmatrix-web.service`` runs as (see install_web_service.sh, which sets
+    ``User=$SUDO_USER``).
+
+    Resolved once from the project root directory's current group (normally
+    the login user's group from the initial ``git clone``), since that user
+    is stable across reinstalls unlike any single file's ownership.
+
+    Returns:
+        The gid, or None if it cannot be determined.
+    """
+    global _shared_group_gid_cache
+    if _shared_group_gid_cache is not None:
+        return _shared_group_gid_cache
+    try:
+        project_root = Path(__file__).resolve().parent.parent.parent
+        _shared_group_gid_cache = project_root.stat().st_gid
+        return _shared_group_gid_cache
+    except OSError:
+        return None
+
+
+def ensure_shared_group_ownership(path: Path) -> None:
+    """
+    Best-effort chgrp of ``path`` to the shared group (see
+    :func:`get_shared_group_gid`) when running as root.
+
+    Only root can change a file's group to one the calling process isn't a
+    member of, which is exactly the case that causes the web interface
+    (running as a non-root user) to get ``PermissionError`` reading files
+    the root-run display service just wrote with a 0o640/2775 mode: the mode
+    is group-readable, but without this the group is root's, not the web
+    user's. Silently does nothing if not running as root or on any error —
+    this is a hardening step, not a required one.
+    """
+    if os.geteuid() != 0:
+        return
+    gid = get_shared_group_gid()
+    if gid is None:
+        return
+    try:
+        if path.exists() and path.stat().st_gid != gid:
+            os.chown(path, -1, gid)
+            logger.debug(f"Set shared group ownership (gid {gid}) on {path}")
+    except OSError as e:
+        logger.debug(f"Could not set shared group ownership on {path}: {e}")
+
+
 def get_config_file_mode(file_path: Path) -> int:
     """
     Return appropriate permission mode for config files.

src/config_manager.py

@@ -38,6 +38,7 @@ from src.config_manager_atomic import (
 from src.common.permission_utils import (
     ensure_directory_permissions,
     ensure_file_permissions,
+    ensure_shared_group_ownership,
     get_config_file_mode,
     get_config_dir_mode
 )
@@ -56,6 +57,13 @@ class ConfigManager:
         self.secrets_path: str = secrets_path or "config/config_secrets.json"
         self.template_path: str = "config/config.template.json"
         self.config: Dict[str, Any] = {}
+        # (mtime_ns, size) signature of (config, secrets, template) at the
+        # last successful load. load_config() skips the full re-read (3 file
+        # parses + recursive template migration) when nothing changed —
+        # ~30 web request handlers call it, some 2-3x per request. Cross-
+        # process freshness is preserved: another process's save bumps the
+        # mtime, so the next load here re-reads.
+        self._loaded_sig: Optional[tuple] = None
         self.logger: logging.Logger = get_logger(__name__)
         
         # Initialize atomic config manager
@@ -122,6 +130,14 @@ class ConfigManager:
         # Update in-memory config if save was successful
         if result.status == SaveResultStatus.SUCCESS:
             self.config = new_config_data
+            # In-memory config now matches what was just written; refresh
+            # the load signature so the fast path stays valid. NOTE: the
+            # in-memory copy includes merged secrets; the on-disk file has
+            # them stripped — the fast path returning self.config preserves
+            # exactly the pre-cache behavior (load-after-save also returned
+            # the secret-merged self.config only after re-reading secrets;
+            # here secrets file is unchanged, so contents are equivalent).
+            self._loaded_sig = self._files_signature()
             self.logger.info(f"Configuration successfully saved atomically to {os.path.abspath(self.config_path)}")
         elif result.status == SaveResultStatus.ROLLED_BACK:
             # Reload config from file after rollback
@@ -179,13 +195,36 @@ class ConfigManager:
         atomic_mgr = self._get_atomic_manager()
         return atomic_mgr.validate_config_file(config_path)
 
+    def _files_signature(self) -> tuple:
+        """(mtime_ns, size) of config/secrets/template, None for missing —
+        cheap staleness probe (3 stats) for the load_config fast path."""
+        sig = []
+        for path in (self.config_path, self.secrets_path, self.template_path):
+            try:
+                st = os.stat(path)
+                sig.append((st.st_mtime_ns, st.st_size))
+            except OSError:
+                sig.append(None)
+        return tuple(sig)
+
     def load_config(self) -> Dict[str, Any]:
-        """Load configuration from JSON files."""
+        """Load configuration from JSON files.
+
+        Fast path: when config.json, config_secrets.json and the template
+        are all unchanged since the last successful load (mtime_ns + size),
+        the already-parsed self.config is returned without touching the
+        files — same aliasing semantics as the full path, which also
+        returns self.config.
+        """
         try:
+            current_sig = self._files_signature()
+            if self.config and self._loaded_sig == current_sig:
+                return self.config
+
             # Check if config file exists, if not create from template
             if not os.path.exists(self.config_path):
                 self._create_config_from_template()
-            
+
             # Load main config
             self.logger.info(f"Attempting to load config from: {os.path.abspath(self.config_path)}")
             with open(self.config_path, 'r') as f:
@@ -196,6 +235,11 @@ class ConfigManager:
 
             # Load and merge secrets if they exist (be permissive on errors)
             if os.path.exists(self.secrets_path):
+                # Self-heal stale group ownership (e.g. the root-run display
+                # service wrote this file before the web user was granted
+                # group access) before every load attempt; no-op unless
+                # running as root and the group is already wrong.
+                ensure_shared_group_ownership(Path(self.secrets_path))
                 try:
                     with open(self.secrets_path, 'r') as f:
                         secrets = json.load(f)
@@ -205,7 +249,10 @@ class ConfigManager:
                     self.logger.warning(f"Secrets file not readable ({self.secrets_path}): {e}. Continuing without secrets.")
                 except (json.JSONDecodeError, OSError) as e:
                     self.logger.warning(f"Error reading secrets file ({self.secrets_path}): {e}. Continuing without secrets.")
-            
+
+            # Signature taken AFTER load + migration (migration may write the
+            # config back), so it reflects exactly what was read/written.
+            self._loaded_sig = self._files_signature()
             return self.config
             
         except FileNotFoundError as e:
@@ -264,7 +311,8 @@ class ConfigManager:
                 json.dump(config_to_write, f, indent=4)
             
             # Update the in-memory config to the new state (which includes secrets for runtime)
-            self.config = new_config_data 
+            self.config = new_config_data
+            self._loaded_sig = self._files_signature()
             self.logger.info(f"Configuration successfully saved to {os.path.abspath(self.config_path)}")
             if secrets_content:
                  self.logger.info("Secret values were preserved in memory and not written to the main config file.")
@@ -321,6 +369,7 @@ class ConfigManager:
         # Set proper file permissions after creation
         config_path_obj = Path(self.config_path)
         ensure_file_permissions(config_path_obj, get_config_file_mode(config_path_obj))
+        ensure_shared_group_ownership(config_path_obj)
         
         self.logger.info(f"Created config.json from template at {os.path.abspath(self.config_path)}")
 
@@ -433,6 +482,11 @@ class ConfigManager:
             self.logger.error(error_msg)
             raise ConfigError(error_msg, config_path=path_to_load)
 
+        if file_type == "secrets":
+            # Best-effort self-heal: no-op unless running as root and the
+            # group is stale (see load_config for why this can happen).
+            ensure_shared_group_ownership(Path(path_to_load))
+
         try:
             with open(path_to_load, 'r') as f:
                 return json.load(f)
@@ -440,7 +494,18 @@ class ConfigManager:
             error_msg = f"Error parsing {file_type} configuration file: {path_to_load}"
             self.logger.error(error_msg, exc_info=True)
             raise ConfigError(error_msg, config_path=path_to_load) from e
-        except (IOError, OSError, PermissionError) as e:
+        except PermissionError as e:
+            if file_type == "secrets":
+                # Match load_config()'s tolerance: a secrets file the web
+                # process can't read (e.g. written 0640 by the root-run
+                # display service before the group was fixed up) shouldn't
+                # 500 the settings page — degrade to "no secrets" instead.
+                self.logger.warning(f"Secrets file not readable ({path_to_load}): {e}. Returning empty secrets.")
+                return {}
+            error_msg = f"Error loading {file_type} configuration file {path_to_load}: {str(e)}"
+            self.logger.error(error_msg, exc_info=True)
+            raise ConfigError(error_msg, config_path=path_to_load) from e
+        except (IOError, OSError) as e:
             error_msg = f"Error loading {file_type} configuration file {path_to_load}: {str(e)}"
             self.logger.error(error_msg, exc_info=True)
             raise ConfigError(error_msg, config_path=path_to_load) from e
@@ -497,6 +562,7 @@ class ConfigManager:
                 # Ensure final file has correct permissions
                 try:
                     ensure_file_permissions(path_obj, file_mode)
+                    ensure_shared_group_ownership(path_obj)
                 except OSError as perm_error:
                     # If we can't set permissions but file was written, log warning but don't fail
                     self.logger.warning(

src/config_manager_atomic.py

@@ -17,6 +17,7 @@ from enum import Enum
 
 from src.exceptions import ConfigError
 from src.logging_config import get_logger
+from src.common.permission_utils import ensure_shared_group_ownership
 
 
 class SaveResultStatus(Enum):
@@ -410,6 +411,13 @@ class AtomicConfigManager:
             # This is important because temp files may have different permissions
             # and we need root service to be able to read config.json
             os.chmod(destination, target_mode)
+
+            # Also fix group ownership when this save is running as root
+            # (the display service): 0o640 alone only helps the non-root web
+            # user read a root-written secrets file if its group already
+            # matches the web user's group, which isn't guaranteed. See
+            # permission_utils.ensure_shared_group_ownership for why.
+            ensure_shared_group_ownership(destination)
             
         except Exception as e:
             raise ConfigError(f"Error during atomic move: {e}") from e

src/display_controller.py

@@ -23,6 +23,9 @@ Entry point: :func:`main` — instantiates :class:`DisplayController` and calls
 import time
 import os
 import json
+import threading
+import types
+from contextlib import contextmanager
 from pathlib import Path
 from typing import Dict, Any, List, Optional, Callable
 from datetime import datetime
@@ -892,6 +895,30 @@ class DisplayController:
             except Exception:  # pylint: disable=broad-except
                 logger.exception("Error running scheduled plugin updates")
 
+    @contextmanager
+    def _display_lock_or_skip(self, plugin_id):
+        """Try-lock guard keeping a plugin's display() off its in-flight update().
+
+        Yields True when display may run (lock held, released on exit) or
+        when no lock support exists (older plugin manager). Yields False when
+        the plugin's update() is currently executing on the background
+        worker — the caller should treat the frame as displayed (the panel
+        holds the last pushed frame) rather than as a plugin failure, so a
+        mid-update skip never advances the rotation.
+        """
+        pm = self.plugin_manager
+        if not pm or not hasattr(pm, 'get_plugin_lock') or not plugin_id:
+            yield True
+            return
+        lock = pm.get_plugin_lock(plugin_id)
+        if not lock.acquire(blocking=False):
+            yield False
+            return
+        try:
+            yield True
+        finally:
+            lock.release()
+
     _FOLLOWER_SEND_INTERVAL = 1.0 / 90  # raw bytes are cheap; 90fps > follower render rate
 
     def _follower_rebuild_scroll_image(self) -> None:
@@ -1879,6 +1906,7 @@ class DisplayController:
                     plugin_id = getattr(manager_to_display, 'plugin_id', active_mode)
                     try:
                         logger.debug(f"Calling display() for {active_mode} with force_clear={self.force_change}")
+                        can_display = False
                         if hasattr(manager_to_display, 'display'):
                             # Opt #1: look up (or compute once) whether display() accepts display_mode
                             _cache_key = plugin_id
@@ -1889,14 +1917,64 @@ class DisplayController:
                                 )
                             _accepts_display_mode = self._plugin_accepts_display_mode[_cache_key]
 
-                            # Use PluginExecutor for safe execution with timeout
-                            if self.plugin_manager and hasattr(self.plugin_manager, 'plugin_executor'):
-                                result = self.plugin_manager.plugin_executor.execute_display(
-                                    manager_to_display,
-                                    plugin_id,
-                                    force_clear=self.force_change,
-                                    display_mode=active_mode if _accepts_display_mode else None
-                                )
+                            pm = self.plugin_manager
+                            display_lock = None
+                            can_display = True
+                            if pm and hasattr(pm, 'get_plugin_lock'):
+                                display_lock = pm.get_plugin_lock(plugin_id)
+                                can_display = display_lock.acquire(blocking=False)
+
+                            if not can_display:
+                                # update() in flight on the worker — hold
+                                # the last frame; not a plugin failure
+                                result = True
+                            elif pm and hasattr(pm, 'plugin_executor'):
+                                # PluginExecutor's own thread.join(timeout) can
+                                # return before the real display() call
+                                # finishes (a lingering daemon thread keeps
+                                # running it) -- so the lock is released from
+                                # inside the wrapped call itself, whichever
+                                # thread actually finishes it, rather than
+                                # here when this dispatch merely returns.
+                                release_guard = threading.Lock()
+                                released = {'done': False}
+
+                                def _release_display_lock():
+                                    with release_guard:
+                                        if released['done']:
+                                            return
+                                        released['done'] = True
+                                    if display_lock is not None:
+                                        display_lock.release()
+
+                                if _accepts_display_mode:
+                                    def _display_target(display_mode=None, force_clear=False):
+                                        try:
+                                            return manager_to_display.display(
+                                                display_mode=display_mode, force_clear=force_clear)
+                                        finally:
+                                            _release_display_lock()
+                                else:
+                                    def _display_target(force_clear=False):
+                                        try:
+                                            return manager_to_display.display(force_clear=force_clear)
+                                        finally:
+                                            _release_display_lock()
+
+                                try:
+                                    result = self.plugin_manager.plugin_executor.execute_display(
+                                        types.SimpleNamespace(display=_display_target),
+                                        plugin_id,
+                                        force_clear=self.force_change,
+                                        display_mode=active_mode if _accepts_display_mode else None
+                                    )
+                                except Exception:  # pragma: no cover - defensive;
+                                    # execute_display catches everything
+                                    # internally, but guarantee the lock is
+                                    # never leaked if something unexpected
+                                    # slips through.
+                                    _release_display_lock()
+                                    raise
                                 # execute_display returns bool, convert to expected format
                                 if result:
                                     result = True  # Success
@@ -1904,10 +1982,14 @@ class DisplayController:
                                     result = False  # Failed
                             else:
                                 # Fallback to direct call if executor not available
-                                if _accepts_display_mode:
-                                    result = manager_to_display.display(display_mode=active_mode, force_clear=self.force_change)
-                                else:
-                                    result = manager_to_display.display(force_clear=self.force_change)
+                                try:
+                                    if _accepts_display_mode:
+                                        result = manager_to_display.display(display_mode=active_mode, force_clear=self.force_change)
+                                    else:
+                                        result = manager_to_display.display(force_clear=self.force_change)
+                                finally:
+                                    if display_lock is not None:
+                                        display_lock.release()
                             
                             logger.debug(f"display() returned: {result} (type: {type(result)})")
                             # Check if display() returned a boolean (new behavior)
@@ -1916,11 +1998,15 @@ class DisplayController:
                                 if not display_result:
                                     logger.info("Plugin %s display() returned False for mode %s", plugin_id, active_mode)
                         
-                        # Record success if display completed without exception
-                        if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
-                            self.plugin_manager.health_tracker.record_success(plugin_id)
-                        
-                        self.force_change = False
+                        # Record success only when display() actually ran this
+                        # frame -- a skipped frame (lock busy) held the last
+                        # frame, not a real success, and must not clear
+                        # force_change or the pending mode-switch clear will
+                        # be lost when display() finally does run.
+                        if can_display:
+                            if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
+                                self.plugin_manager.health_tracker.record_success(plugin_id)
+                            self.force_change = False
                     except Exception as exc:  # pylint: disable=broad-except
                         logger.exception("Error displaying %s", self.current_display_mode)
                         # Record failure
@@ -2125,10 +2211,23 @@ class DisplayController:
 
                     # For plugins, call display multiple times to allow game rotation
                     if manager_to_display and hasattr(manager_to_display, 'display'):
-                        # Check if plugin needs high FPS (like stock ticker)
-                        # Always enable high-FPS for static-image plugin (for GIF animation support)
+                        # High-FPS decision, in precedence order:
+                        # 1. A plugin that declares needs_high_fps knows best
+                        #    (e.g. static-image sets it False for still PNGs,
+                        #    True for animated GIFs).
+                        # 2. Back-compat: older static-image versions without
+                        #    the attribute keep the historical forced high-FPS
+                        #    (GIF support).
+                        # 3. Otherwise scrolling plugins get high FPS.
                         plugin_id = getattr(manager_to_display, 'plugin_id', None)
-                        if plugin_id == 'static-image':
+                        declared = getattr(manager_to_display, 'needs_high_fps', None)
+                        if declared is not None:
+                            needs_high_fps = bool(declared)
+                            logger.debug(
+                                "[DisplayController] FPS check for %s (plugin=%s) - "
+                                "plugin declares needs_high_fps=%s",
+                                active_mode, plugin_id, needs_high_fps)
+                        elif plugin_id == 'static-image':
                             needs_high_fps = True
                             logger.debug("FPS check - static-image plugin: forcing high-FPS mode for GIF support")
                         else:
@@ -2192,11 +2291,16 @@ class DisplayController:
 
                             while True:
                                 try:
-                                    # Pass display_mode to maintain sticky manager state
-                                    if _accepts_display_mode:
-                                        result = manager_to_display.display(display_mode=active_mode, force_clear=False)
-                                    else:
-                                        result = manager_to_display.display(force_clear=False)
+                                    with self._display_lock_or_skip(plugin_id) as can_display:
+                                        if can_display:
+                                            # Pass display_mode to maintain sticky manager state
+                                            if _accepts_display_mode:
+                                                result = manager_to_display.display(display_mode=active_mode, force_clear=False)
+                                            else:
+                                                result = manager_to_display.display(force_clear=False)
+                                        else:
+                                            # update() in flight — hold the last frame
+                                            result = True
                                     if isinstance(result, bool) and not result:
                                         logger.debug("Display returned False, breaking early")
                                         break
@@ -2256,11 +2360,16 @@ class DisplayController:
                                     break
 
                                 try:
-                                    # Pass display_mode to maintain sticky manager state
-                                    if _accepts_display_mode:
-                                        result = manager_to_display.display(display_mode=active_mode, force_clear=False)
-                                    else:
-                                        result = manager_to_display.display(force_clear=False)
+                                    with self._display_lock_or_skip(plugin_id) as can_display:
+                                        if can_display:
+                                            # Pass display_mode to maintain sticky manager state
+                                            if _accepts_display_mode:
+                                                result = manager_to_display.display(display_mode=active_mode, force_clear=False)
+                                            else:
+                                                result = manager_to_display.display(force_clear=False)
+                                        else:
+                                            # update() in flight — hold the last frame
+                                            result = True
                                     if isinstance(result, bool) and not result:
                                         # For dynamic duration plugins, don't exit on False - keep looping
                                         # until cycle is complete or max duration is reached
@@ -2778,6 +2887,14 @@ class DisplayController:
 
     def cleanup(self):
         """Clean up resources."""
+        # Stop the async update worker first so no in-flight update() call
+        # is still touching display/cache-backed resources while they're
+        # torn down below.
+        if self.plugin_manager and hasattr(self.plugin_manager, 'stop_update_worker'):
+            try:
+                self.plugin_manager.stop_update_worker()
+            except Exception as e:
+                logger.warning("Error stopping plugin update worker: %s", e)
         # Shutdown config service if it exists
         if hasattr(self, 'config_service'):
             try:

src/display_manager.py

@@ -33,6 +33,7 @@ else:
 from contextlib import contextmanager
 from pathlib import Path
 from PIL import Image, ImageDraw, ImageFont
+import threading
 import time
 from collections import OrderedDict
 from typing import Dict, Any, List, Optional, Tuple
@@ -219,6 +220,17 @@ class DisplayManager:
         # but is never silent: the snapshot's mtime doubles as the web UI's
         # hardware-liveness signal, so a quiet failure makes health checks lie.
         self._snapshot_fail_log_ts = 0.0
+        # Dirty tracking: (image digest, brightness) of the last frame pushed
+        # to the panel; update_display() skips identical pushes. Kill switch:
+        # display.dirty_tracking: false.
+        self._dirty_tracking_enabled = bool(
+            self.config.get('display', {}).get('dirty_tracking', True))
+        self._last_pushed_digest = None
+        # Serializes update_display(): plugins can call it directly from
+        # background threads (see docstring on update_display), not just the
+        # render loop. RLock in case a caller within the critical section
+        # ever re-enters (e.g. via a nested draw callback).
+        self._update_lock = threading.RLock()
         
         # Scrolling state tracking for graceful updates
         self._scrolling_state = {
@@ -449,6 +461,10 @@ class DisplayManager:
         try:
             # RGBMatrix accepts brightness as a property
             self.matrix.brightness = brightness
+            # Brightness applies on the next swap — force a re-push even if
+            # the image itself is unchanged (belt-and-braces: brightness is
+            # also part of the dirty-tracking digest when readable).
+            self._last_pushed_digest = None
             logger.info(f"[BRIGHTNESS] Display brightness set to {brightness}%")
             return True
         except AttributeError as e:
@@ -540,33 +556,70 @@ class DisplayManager:
         return phys
 
     def update_display(self):
-        """Update the display using double buffering with proper sync."""
+        """Update the display using double buffering with proper sync.
+
+        Skips the panel push entirely when the frame is byte-identical to
+        the last pushed one (same image digest AND same brightness) — static
+        content re-rendered every second, and 125 fps loops between actual
+        scroll steps, otherwise re-walk the full framebuffer for nothing.
+        The panel keeps refreshing the current frame from its own thread,
+        so skipping a swap never blanks or freezes the hardware.
+
+        Correctness hinges on invalidation: clear() resets the digest (it
+        writes to the matrix directly), and brightness is PART of the digest
+        so a dim-schedule change is never skipped. Disable via config
+        ``display.dirty_tracking: false`` if a redraw issue is ever suspected.
+
+        Serialized via ``_update_lock``: plugins can call this directly from
+        background threads (e.g. sports base classes push an immediate
+        "live" refresh from inside update()), so without a lock two callers
+        could both pass the digest check before either writes it back,
+        double-pushing a frame, or interleave the offscreen/current canvas
+        swap below. The lock is scoped to this method, so callers never
+        need to know about it.
+        """
         try:
-            if self.matrix is None:
-                # Fallback mode - no actual hardware to update
-                logger.debug("Update display called in fallback mode (no hardware)")
-                # Still write a snapshot so the web UI can preview
+            with self._update_lock:
+                if self.matrix is None:
+                    # Fallback mode - no actual hardware to update
+                    logger.debug("Update display called in fallback mode (no hardware)")
+                    # Still write a snapshot so the web UI can preview
+                    self._write_snapshot_if_due()
+                    return
+
+                if self._capture_mode_active:
+                    return  # Skip hardware write — content is being captured off-screen
+
+                digest = None
+                if self._dirty_tracking_enabled:
+                    try:
+                        brightness = getattr(self.matrix, 'brightness', None)
+                    except AttributeError:
+                        brightness = None
+                    digest = (zlib.adler32(self.image.tobytes()), brightness)
+                    if digest == self._last_pushed_digest:
+                        # Nothing changed since the last push — the panel is
+                        # already showing exactly this frame.
+                        self._write_snapshot_if_due()
+                        return
+
+                # Copy the current image to the offscreen canvas. In double-sided
+                # mode the logical screen is first tiled across the full chain.
+                if self._double_sided is not None:
+                    self.offscreen_canvas.SetImage(self._composite_double_sided())
+                else:
+                    self.offscreen_canvas.SetImage(self.image)
+
+                # Swap buffers immediately
+                self.matrix.SwapOnVSync(self.offscreen_canvas)
+
+                # Swap our canvas references
+                self.offscreen_canvas, self.current_canvas = self.current_canvas, self.offscreen_canvas
+
+                self._last_pushed_digest = digest
+
+                # Write a snapshot for the web preview (throttled)
                 self._write_snapshot_if_due()
-                return
-
-            if self._capture_mode_active:
-                return  # Skip hardware write — content is being captured off-screen
-
-            # Copy the current image to the offscreen canvas. In double-sided
-            # mode the logical screen is first tiled across the full chain.
-            if self._double_sided is not None:
-                self.offscreen_canvas.SetImage(self._composite_double_sided())
-            else:
-                self.offscreen_canvas.SetImage(self.image)
-
-            # Swap buffers immediately
-            self.matrix.SwapOnVSync(self.offscreen_canvas)
-            
-            # Swap our canvas references
-            self.offscreen_canvas, self.current_canvas = self.current_canvas, self.offscreen_canvas
-
-            # Write a snapshot for the web preview (throttled)
-            self._write_snapshot_if_due()
         except Exception as e:
             logger.error(f"Error updating display: {e}")
 
@@ -600,6 +653,9 @@ class DisplayManager:
                 # Clear both canvases and the underlying matrix to ensure no artifacts.
                 # Failures are non-fatal — the image buffer is already black above, so
                 # the next update_display() call will push clean content regardless.
+                # The matrix content no longer matches the last pushed digest,
+                # so dirty tracking must not skip the next push.
+                self._last_pushed_digest = None
                 try:
                     self.offscreen_canvas.Clear()
                 except (RuntimeError, OSError) as e:

src/font_test_manager.py

@@ -1,135 +0,0 @@
-import os
-import freetype
-from PIL import ImageDraw, ImageFont
-import logging
-from typing import Dict, Any
-from src.display_manager import DisplayManager
-
-# Configure logging
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-
-class FontTestManager:
-    """Manager for testing fonts with easy BDF/TTF switching."""
-    
-    def __init__(self, config: Dict[str, Any], display_manager: DisplayManager):
-        self.display_manager = display_manager
-        self.config = config
-        self.logger = logging.getLogger('FontTest')
-        
-        # FONT CONFIGURATION - EASY SWITCHING
-        # Set to 'bdf' or 'ttf' to switch font types
-        self.font_type = 'bdf'  # Change this to 'ttf' to use TTF font
-        
-        # Font configurations
-        self.font_configs = {
-            'bdf': {
-                'path': "assets/fonts/cozette.bdf",
-                'display_name': "Cozette BTF",
-                'description': "BTF font Test"
-            },
-            'ttf': {
-                'path': "assets/fonts/5by7.regular.ttf",
-                'display_name': "5by7 TTF",
-                'description': "TTF font test"
-            }
-        }
-        
-        # Get current font configuration
-        self.current_config = self.font_configs[self.font_type]
-        self.font_path = self.current_config['path']
-        
-        # Verify font exists
-        if not os.path.exists(self.font_path):
-            self.logger.error(f"Font file not found: {self.font_path}")
-            raise FileNotFoundError(f"Font file not found: {self.font_path}")
-        
-        # Load the font based on type
-        if self.font_type == 'bdf':
-            self._load_bdf_font()
-        else:
-            self._load_ttf_font()
-        
-        self.logger.info(f"Initialized FontTestManager with {self.current_config['description']}")
-    
-    def _load_bdf_font(self):
-        """Load BDF font using freetype."""
-        try:
-            self.face = freetype.Face(self.font_path)
-            self.logger.info(f"Successfully loaded BDF font from {self.font_path}")
-        except Exception as e:
-            self.logger.error(f"Failed to load BDF font: {e}")
-            raise
-    
-    def _load_ttf_font(self):
-        """Load TTF font using PIL."""
-        try:
-            self.font = ImageFont.truetype(self.font_path, 8)  # Size 8 for 5x7 font
-            self.logger.info(f"Successfully loaded TTF font from {self.font_path}")
-        except Exception as e:
-            self.logger.error(f"Failed to load TTF font: {e}")
-            raise
-    
-    def update(self):
-        """No update needed for static display."""
-    
-    def display(self, force_clear: bool = False):
-        """Display the font with sample text."""
-        try:
-            # Clear the display
-            self.display_manager.clear()
-            
-            # Draw font name at the top
-            self.display_manager.draw_text(self.current_config['display_name'], y=2, color=(255, 255, 255))
-            
-            # Draw sample text
-            draw = ImageDraw.Draw(self.display_manager.image)
-            sample_text = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
-            
-            # Calculate starting position
-            x = 10  # Start 10 pixels from the left
-            y = 10  # Start 10 pixels from the top
-            
-            # Draw text based on font type
-            if self.font_type == 'bdf':
-                self._draw_bdf_text(draw, sample_text, x, y)
-            else:
-                self._draw_ttf_text(draw, sample_text, x, y)
-            
-            # Update the display once
-            self.display_manager.update_display()
-            
-            # Log that display is complete
-            self.logger.info("Font test display complete.")
-            
-        except Exception as e:
-            self.logger.error(f"Error displaying font test: {e}", exc_info=True)
-    
-    def _draw_bdf_text(self, draw, text, x, y):
-        """Draw text using BDF font."""
-        for char in text:
-            # Load the glyph
-            self.face.load_char(char)
-            bitmap = self.face.glyph.bitmap
-            
-            # Draw the glyph
-            for i in range(bitmap.rows):
-                for j in range(bitmap.width):
-                    try:
-                        # Get the byte containing the pixel
-                        byte_index = i * bitmap.pitch + (j // 8)
-                        if byte_index < len(bitmap.buffer):
-                            byte = bitmap.buffer[byte_index]
-                            # Check if the specific bit is set
-                            if byte & (1 << (7 - (j % 8))):
-                                draw.point((x + j, y + i), fill=(255, 255, 255))
-                    except IndexError:
-                        self.logger.warning(f"Index out of range for char '{char}' at position ({i}, {j})")
-                        continue
-            
-            # Move to next character position
-            x += self.face.glyph.advance.x >> 6
-    
-    def _draw_ttf_text(self, draw, text, x, y):
-        """Draw text using TTF font."""
-        draw.text((x, y), text, font=self.font, fill=(255, 255, 255)) 

src/generic_cache_mixin.py

@@ -1,150 +0,0 @@
-"""
-Generic Cache Mixin for Any Manager
-
-This mixin provides caching functionality that can be used by any manager
-that needs to cache data, not just sports managers. It's a more general
-version of BackgroundCacheMixin that works for weather, stocks, news, etc.
-"""
-
-import time
-from typing import Dict, Optional, Any, Callable
-
-
-class GenericCacheMixin:
-    """
-    Generic mixin class that provides caching functionality to any manager.
-    
-    This mixin can be used by weather, stock, news, or any other manager
-    that needs to cache data with performance monitoring.
-    
-    Note: For sports managers that need background service cache integration,
-    use BackgroundCacheMixin instead. See src/background_cache_mixin.py for details.
-    """
-    
-    def _fetch_data_with_cache(self, 
-                             cache_key: str,
-                             api_fetch_method: Callable,
-                             cache_ttl: int = 300,
-                             force_refresh: bool = False) -> Optional[Dict]:
-        """
-        Generic caching pattern for any manager.
-        
-        Args:
-            cache_key: Unique cache key for this data
-            api_fetch_method: Method to call for fresh data
-            cache_ttl: Time-to-live in seconds (default: 5 minutes)
-            force_refresh: Skip cache and fetch fresh data
-            
-        Returns:
-            Cached or fresh data from API
-        """
-        start_time = time.time()
-        cache_hit = False
-        cache_source = None
-        
-        try:
-            # Check cache first (unless forcing refresh)
-            if not force_refresh:
-                cached_data = self.cache_manager.get_cached_data(cache_key, cache_ttl)
-                if cached_data:
-                    self.logger.info(f"Using cached data for {cache_key}")
-                    cache_hit = True
-                    cache_source = "cache"
-                    self.cache_manager.record_cache_hit('regular')
-                    
-                    # Record performance metrics
-                    duration = time.time() - start_time
-                    self.cache_manager.record_fetch_time(duration)
-                    self._log_fetch_performance(cache_key, duration, cache_hit, cache_source)
-                    
-                    return cached_data
-            
-            # Fetch fresh data
-            self.logger.info(f"Fetching fresh data for {cache_key}")
-            result = api_fetch_method()
-            cache_source = "api_fresh"
-            
-            # Store in cache if we got data
-            if result:
-                self.cache_manager.save_cache(cache_key, result)
-                self.cache_manager.record_cache_miss('regular')
-            else:
-                self.logger.warning(f"No data returned for {cache_key}")
-            
-            # Record performance metrics
-            duration = time.time() - start_time
-            self.cache_manager.record_fetch_time(duration)
-            
-            # Log performance
-            self._log_fetch_performance(cache_key, duration, cache_hit, cache_source)
-            
-            return result
-            
-        except Exception as e:
-            duration = time.time() - start_time
-            self.logger.error(f"Error fetching data for {cache_key} after {duration:.2f}s: {e}")
-            self.cache_manager.record_fetch_time(duration)
-            raise
-    
-    def _log_fetch_performance(self, cache_key: str, duration: float, cache_hit: bool, cache_source: str):
-        """
-        Log detailed performance metrics for fetch operations.
-        
-        Args:
-            cache_key: Cache key that was accessed
-            duration: Fetch operation duration in seconds
-            cache_hit: Whether this was a cache hit
-            cache_source: Source of the data (cache, api_fresh, etc.)
-        """
-        # Log basic performance info
-        self.logger.info(f"Fetch completed for {cache_key} in {duration:.2f}s "
-                        f"(cache_hit={cache_hit}, source={cache_source})")
-        
-        # Log detailed metrics every 10 operations
-        if hasattr(self, '_fetch_count'):
-            self._fetch_count += 1
-        else:
-            self._fetch_count = 1
-            
-        if self._fetch_count % 10 == 0:
-            metrics = self.cache_manager.get_cache_metrics()
-            self.logger.info(f"Cache Performance Summary - "
-                           f"Hit Rate: {metrics['cache_hit_rate']:.2%}, "
-                           f"API Calls Saved: {metrics['api_calls_saved']}, "
-                           f"Avg Fetch Time: {metrics['average_fetch_time']:.2f}s")
-    
-    def get_cache_performance_summary(self) -> Dict[str, Any]:
-        """
-        Get cache performance summary for this manager.
-        
-        Returns:
-            Dictionary containing cache performance metrics
-        """
-        return self.cache_manager.get_cache_metrics()
-    
-    def log_cache_performance(self):
-        """Log current cache performance metrics."""
-        self.cache_manager.log_cache_metrics()
-    
-    def clear_cache_for_key(self, cache_key: str):
-        """Clear cache for a specific key."""
-        self.cache_manager.clear_cache(cache_key)
-        self.logger.info(f"Cleared cache for {cache_key}")
-    
-    def get_cache_info(self, cache_key: str) -> Dict[str, Any]:
-        """
-        Get information about a cached item.
-        
-        Args:
-            cache_key: Cache key to check
-            
-        Returns:
-            Dictionary with cache information
-        """
-        # This would need to be implemented in CacheManager
-        # For now, just return basic info
-        return {
-            'key': cache_key,
-            'exists': self.cache_manager.get_cached_data(cache_key, 0) is not None,
-            'ttl': 'unknown'  # Would need to be implemented
-        }

src/image_utils.py

@@ -1,22 +0,0 @@
-"""Deprecated: use src/adaptive_images.py (fit_image) instead.
-
-This module predates the adaptive image system and has no known callers.
-It is kept only so any out-of-tree code importing it keeps working.
-"""
-
-import logging
-from PIL import Image
-
-logger = logging.getLogger(__name__)
-
-def scale_to_max_dimensions(img, max_width, max_height):
-    h_to_w_ratio = img.height / img.width
-    w_to_h_ratio = img.width / img.height
-
-    if img.height > max_height:
-        img = img.resize((int(max_height * w_to_h_ratio), max_height), Image.Resampling.LANCZOS)
-    
-    if img.width > max_width:
-        img = img.resize((max_width, int(max_width * h_to_w_ratio)), Image.Resampling.LANCZOS)
-    
-    return img

src/layout_manager.py

@@ -1,409 +0,0 @@
-"""
-Layout Manager for LED Matrix Display
-Handles custom layouts, element positioning, and display composition.
-"""
-
-import json
-import os
-import logging
-from typing import Dict, List, Any
-from datetime import datetime
-
-logger = logging.getLogger(__name__)
-
-class LayoutManager:
-    def __init__(self, display_manager=None, config_path="config/custom_layouts.json"):
-        self.display_manager = display_manager
-        self.config_path = config_path
-        self.layouts = self.load_layouts()
-        self.current_layout = None
-        
-    def load_layouts(self) -> Dict[str, Any]:
-        """Load saved layouts from file."""
-        try:
-            if os.path.exists(self.config_path):
-                with open(self.config_path, 'r') as f:
-                    return json.load(f)
-            return {}
-        except Exception as e:
-            logger.error(f"Error loading layouts: {e}")
-            return {}
-    
-    def save_layouts(self) -> bool:
-        """Save layouts to file."""
-        try:
-            from pathlib import Path
-            from src.common.permission_utils import (
-                ensure_directory_permissions,
-                get_config_dir_mode
-            )
-            config_path_obj = Path(self.config_path)
-            ensure_directory_permissions(config_path_obj.parent, get_config_dir_mode())
-            with open(self.config_path, 'w') as f:
-                json.dump(self.layouts, f, indent=2)
-            return True
-        except Exception as e:
-            logger.error(f"Error saving layouts: {e}")
-            return False
-    
-    def create_layout(self, name: str, elements: List[Dict], description: str = "") -> bool:
-        """Create a new layout."""
-        try:
-            self.layouts[name] = {
-                'elements': elements,
-                'description': description,
-                'created': datetime.now().isoformat(),
-                'modified': datetime.now().isoformat()
-            }
-            return self.save_layouts()
-        except Exception as e:
-            logger.error(f"Error creating layout '{name}': {e}")
-            return False
-    
-    def update_layout(self, name: str, elements: List[Dict], description: str = None) -> bool:
-        """Update an existing layout."""
-        try:
-            if name not in self.layouts:
-                return False
-            
-            self.layouts[name]['elements'] = elements
-            self.layouts[name]['modified'] = datetime.now().isoformat()
-            
-            if description is not None:
-                self.layouts[name]['description'] = description
-                
-            return self.save_layouts()
-        except Exception as e:
-            logger.error(f"Error updating layout '{name}': {e}")
-            return False
-    
-    def delete_layout(self, name: str) -> bool:
-        """Delete a layout."""
-        try:
-            if name in self.layouts:
-                del self.layouts[name]
-                return self.save_layouts()
-            return False
-        except Exception as e:
-            logger.error(f"Error deleting layout '{name}': {e}")
-            return False
-    
-    def get_layout(self, name: str) -> Dict[str, Any]:
-        """Get a specific layout."""
-        return self.layouts.get(name, {})
-    
-    def list_layouts(self) -> List[str]:
-        """Get list of all layout names."""
-        return list(self.layouts.keys())
-    
-    def set_current_layout(self, name: str) -> bool:
-        """Set the current active layout."""
-        if name in self.layouts:
-            self.current_layout = name
-            return True
-        return False
-    
-    def render_layout(self, layout_name: str = None, data_context: Dict = None) -> bool:
-        """Render a layout to the display."""
-        if not self.display_manager:
-            logger.error("No display manager available")
-            return False
-        
-        layout_name = layout_name or self.current_layout
-        if not layout_name or layout_name not in self.layouts:
-            logger.error(f"Layout '{layout_name}' not found")
-            return False
-        
-        try:
-            # Clear the display
-            self.display_manager.clear()
-            
-            # Get layout elements
-            elements = self.layouts[layout_name]['elements']
-            
-            # Render each element
-            for element in elements:
-                self.render_element(element, data_context or {})
-            
-            # Update the display
-            self.display_manager.update_display()
-            return True
-            
-        except Exception as e:
-            logger.error(f"Error rendering layout '{layout_name}': {e}")
-            return False
-    
-    def render_element(self, element: Dict, data_context: Dict) -> None:
-        """Render a single element."""
-        element_type = element.get('type')
-        x = element.get('x', 0)
-        y = element.get('y', 0)
-        properties = element.get('properties', {})
-        
-        try:
-            if element_type == 'text':
-                self._render_text_element(x, y, properties, data_context)
-            elif element_type == 'weather_icon':
-                self._render_weather_icon_element(x, y, properties, data_context)
-            elif element_type == 'rectangle':
-                self._render_rectangle_element(x, y, properties)
-            elif element_type == 'line':
-                self._render_line_element(x, y, properties)
-            elif element_type == 'clock':
-                self._render_clock_element(x, y, properties)
-            elif element_type == 'data_text':
-                self._render_data_text_element(x, y, properties, data_context)
-            else:
-                logger.warning(f"Unknown element type: {element_type}")
-                
-        except Exception as e:
-            logger.error(f"Error rendering element {element_type}: {e}")
-    
-    def _render_text_element(self, x: int, y: int, properties: Dict, data_context: Dict) -> None:
-        """Render a text element."""
-        text = properties.get('text', 'Sample Text')
-        color = tuple(properties.get('color', [255, 255, 255]))
-        font_size = properties.get('font_size', 'normal')
-        
-        # Support template variables in text
-        text = self._process_template_text(text, data_context)
-        
-        # Select font
-        if font_size == 'small':
-            font = self.display_manager.small_font
-        elif font_size == 'large':
-            font = self.display_manager.regular_font
-        else:
-            font = self.display_manager.regular_font
-        
-        self.display_manager.draw_text(text, x, y, color, font=font)
-    
-    def _render_weather_icon_element(self, x: int, y: int, properties: Dict, data_context: Dict) -> None:
-        """Render a weather icon element."""
-        condition = properties.get('condition', 'sunny')
-        size = properties.get('size', 16)
-        
-        # Use weather data from context if available
-        if 'weather' in data_context and 'condition' in data_context['weather']:
-            condition = data_context['weather']['condition'].lower()
-        
-        self.display_manager.draw_weather_icon(condition, x, y, size)
-    
-    def _render_rectangle_element(self, x: int, y: int, properties: Dict) -> None:
-        """Render a rectangle element."""
-        width = properties.get('width', 10)
-        height = properties.get('height', 10)
-        color = tuple(properties.get('color', [255, 255, 255]))
-        filled = properties.get('filled', False)
-        
-        if filled:
-            self.display_manager.draw.rectangle(
-                [x, y, x + width, y + height], 
-                fill=color
-            )
-        else:
-            self.display_manager.draw.rectangle(
-                [x, y, x + width, y + height], 
-                outline=color
-            )
-    
-    def _render_line_element(self, x: int, y: int, properties: Dict) -> None:
-        """Render a line element."""
-        x2 = properties.get('x2', x + 10)
-        y2 = properties.get('y2', y)
-        color = tuple(properties.get('color', [255, 255, 255]))
-        width = properties.get('width', 1)
-        
-        self.display_manager.draw.line([x, y, x2, y2], fill=color, width=width)
-    
-    def _render_clock_element(self, x: int, y: int, properties: Dict) -> None:
-        """Render a clock element."""
-        format_str = properties.get('format', '%H:%M')
-        color = tuple(properties.get('color', [255, 255, 255]))
-        
-        current_time = datetime.now().strftime(format_str)
-        self.display_manager.draw_text(current_time, x, y, color)
-    
-    def _render_data_text_element(self, x: int, y: int, properties: Dict, data_context: Dict) -> None:
-        """Render a data-driven text element."""
-        data_key = properties.get('data_key', '')
-        format_str = properties.get('format', '{value}')
-        color = tuple(properties.get('color', [255, 255, 255]))
-        default_value = properties.get('default', 'N/A')
-        
-        # Extract data from context
-        value = self._get_nested_value(data_context, data_key, default_value)
-        
-        # Format the text
-        try:
-            text = format_str.format(value=value)
-        except (ValueError, TypeError, KeyError, IndexError):
-            text = str(value)
-        
-        self.display_manager.draw_text(text, x, y, color)
-    
-    def _process_template_text(self, text: str, data_context: Dict) -> str:
-        """Process template variables in text."""
-        try:
-            # Simple template processing - replace {key} with values from context
-            for key, value in data_context.items():
-                placeholder = f"{{{key}}}"
-                if placeholder in text:
-                    text = text.replace(placeholder, str(value))
-            return text
-        except Exception as e:
-            logger.error(f"Error processing template text: {e}")
-            return text
-    
-    def _get_nested_value(self, data: Dict, key: str, default=None):
-        """Get a nested value from a dictionary using dot notation."""
-        try:
-            keys = key.split('.')
-            value = data
-            for k in keys:
-                value = value[k]
-            return value
-        except (KeyError, TypeError):
-            return default
-    
-    def create_preset_layouts(self) -> None:
-        """Create some preset layouts for common use cases."""
-        # Basic clock layout
-        clock_layout = [
-            {
-                'type': 'clock',
-                'x': 10,
-                'y': 10,
-                'properties': {
-                    'format': '%H:%M',
-                    'color': [255, 255, 255]
-                }
-            },
-            {
-                'type': 'clock',
-                'x': 10,
-                'y': 20,
-                'properties': {
-                    'format': '%m/%d',
-                    'color': [100, 100, 255]
-                }
-            }
-        ]
-        self.create_layout('basic_clock', clock_layout, 'Simple clock with date')
-        
-        # Weather layout
-        weather_layout = [
-            {
-                'type': 'weather_icon',
-                'x': 5,
-                'y': 5,
-                'properties': {
-                    'condition': 'sunny',
-                    'size': 20
-                }
-            },
-            {
-                'type': 'data_text',
-                'x': 30,
-                'y': 8,
-                'properties': {
-                    'data_key': 'weather.temperature',
-                    'format': '{value}°',
-                    'color': [255, 200, 0],
-                    'default': '--°'
-                }
-            },
-            {
-                'type': 'data_text',
-                'x': 30,
-                'y': 18,
-                'properties': {
-                    'data_key': 'weather.condition',
-                    'format': '{value}',
-                    'color': [200, 200, 200],
-                    'default': 'Unknown'
-                }
-            }
-        ]
-        self.create_layout('weather_display', weather_layout, 'Weather icon with temperature and condition')
-        
-        # Mixed dashboard layout
-        dashboard_layout = [
-            {
-                'type': 'clock',
-                'x': 2,
-                'y': 2,
-                'properties': {
-                    'format': '%H:%M',
-                    'color': [255, 255, 255]
-                }
-            },
-            {
-                'type': 'weather_icon',
-                'x': 50,
-                'y': 2,
-                'properties': {
-                    'size': 16
-                }
-            },
-            {
-                'type': 'data_text',
-                'x': 70,
-                'y': 5,
-                'properties': {
-                    'data_key': 'weather.temperature',
-                    'format': '{value}°',
-                    'color': [255, 200, 0],
-                    'default': '--°'
-                }
-            },
-            {
-                'type': 'line',
-                'x': 0,
-                'y': 15,
-                'properties': {
-                    'x2': 128,
-                    'y2': 15,
-                    'color': [100, 100, 100]
-                }
-            },
-            {
-                'type': 'data_text',
-                'x': 2,
-                'y': 18,
-                'properties': {
-                    'data_key': 'stocks.AAPL.price',
-                    'format': 'AAPL: ${value}',
-                    'color': [0, 255, 0],
-                    'default': 'AAPL: N/A'
-                }
-            }
-        ]
-        self.create_layout('dashboard', dashboard_layout, 'Mixed dashboard with clock, weather, and stocks')
-        
-        logger.info("Created preset layouts")
-
-    def get_layout_preview(self, layout_name: str) -> Dict[str, Any]:
-        """Get a preview representation of a layout."""
-        if layout_name not in self.layouts:
-            return {}
-        
-        layout = self.layouts[layout_name]
-        elements = layout['elements']
-        
-        # Create a simple preview representation
-        preview = {
-            'name': layout_name,
-            'description': layout.get('description', ''),
-            'element_count': len(elements),
-            'elements': []
-        }
-        
-        for element in elements:
-            preview['elements'].append({
-                'type': element.get('type'),
-                'position': f"({element.get('x', 0)}, {element.get('y', 0)})",
-                'properties': list(element.get('properties', {}).keys())
-            })
-        
-        return preview

src/plugin_system/plugin_manager.py

@@ -8,12 +8,13 @@ API Version: 1.0.0
 """
 
 import json
+import queue
 import sys
 import time
 import threading
 import types
 from pathlib import Path
-from typing import Dict, List, Optional, Any
+from typing import Dict, List, Optional, Any, Tuple
 import logging
 from src.exceptions import PluginError, ConfigError
 from src.logging_config import get_logger
@@ -97,6 +98,50 @@ class PluginManager:
         # Health tracking (optional, set by display_controller if available)
         self.health_tracker = None
         self.resource_monitor = None
+
+        # --- Asynchronous plugin updates -------------------------------
+        # update() used to run inline in the render loop (execute_update's
+        # internal thread.join(timeout=30) blocked it), so one slow plugin
+        # HTTP fetch froze scrolling for the whole fetch. Scheduling still
+        # happens on the render thread (run_scheduled_updates), but
+        # execution moves to this single background worker. Per-plugin
+        # locks keep a plugin's update() and display() mutually exclusive —
+        # today's implicit guarantee, now explicit (and, unlike today,
+        # also held across the post-timeout window).
+        # Kill switch: plugin_system.synchronous_updates: true restores the
+        # inline path.
+        self._update_queue: "queue.Queue[Optional[Tuple[str, float]]]" = queue.Queue()
+        self._pending_updates: set = set()
+        self._pending_lock = threading.Lock()
+        self._plugin_locks: Dict[str, threading.Lock] = {}
+        self._plugin_locks_guard = threading.Lock()
+        self._update_worker: Optional[threading.Thread] = None
+        self._synchronous_updates = False
+        if self.config_manager is not None:
+            try:
+                cfg = self.config_manager.get_config() or {}
+            except (OSError, ValueError) as exc:
+                self.logger.warning(
+                    "Could not load config to check plugin_system.synchronous_updates "
+                    "(%s: %s); defaulting to synchronous updates", type(exc).__name__, exc)
+                self._synchronous_updates = True
+            else:
+                plugin_system_cfg = cfg.get('plugin_system', {})
+                if not isinstance(plugin_system_cfg, dict):
+                    self.logger.warning(
+                        "config plugin_system must be a mapping, got %s; "
+                        "defaulting to synchronous updates",
+                        type(plugin_system_cfg).__name__)
+                    self._synchronous_updates = True
+                else:
+                    sync_value = plugin_system_cfg.get('synchronous_updates', False)
+                    if not isinstance(sync_value, bool):
+                        self.logger.warning(
+                            "config plugin_system.synchronous_updates must be a boolean, "
+                            "got %r; defaulting to synchronous updates", sync_value)
+                        self._synchronous_updates = True
+                    else:
+                        self._synchronous_updates = sync_value
         
         # Ensure plugins directory exists with proper permissions
         try:
@@ -744,47 +789,189 @@ class PluginManager:
                 last_update = self.plugin_last_update.get(plugin_id, 0.0)
 
             if last_update == 0.0 or (current_time - last_update) >= interval:
-                # Update state to RUNNING
-                self.state_manager.set_state(plugin_id, PluginState.RUNNING)
-                
-                try:
-                    # Use PluginExecutor for safe execution
-                    success = False
-                    if self.resource_monitor:
-                        # If resource monitor exists, wrap the call
-                        def monitored_update():
-                            self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
-                        # SimpleNamespace stores `update` as an *instance*
-                        # attribute, so attribute lookup returns the plain
-                        # function object as-is. A dynamically-built class
-                        # (`type(..., {'update': monitored_update})`) instead
-                        # stores it as a *class* attribute, which the
-                        # descriptor protocol turns into a bound method on
-                        # access -- silently prepending the instance as an
-                        # implicit first argument to a function that takes
-                        # none, raising "monitored_update() takes 0
-                        # positional arguments but 1 was given" on every call.
-                        success = self.plugin_executor.execute_update(
-                            types.SimpleNamespace(update=monitored_update),
-                            plugin_id
-                        )
-                    else:
-                        success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
-                    
-                    if success:
-                        with self._plugin_last_update_lock:
-                            self.plugin_last_update[plugin_id] = current_time
-                        self.state_manager.record_update(plugin_id)
-                        # Update state back to ENABLED
-                        self.state_manager.set_state(plugin_id, PluginState.ENABLED)
-                        # Record success
-                        if self.health_tracker:
-                            self.health_tracker.record_success(plugin_id)
-                    else:
-                        self._record_update_failure(plugin_id)
-                except Exception as exc:  # pylint: disable=broad-except
-                    self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
+                if self._synchronous_updates:
+                    # Kill-switch path: the original inline execution
+                    # (blocks the caller until update() completes/times out)
+                    self.state_manager.set_state(plugin_id, PluginState.RUNNING)
+                    self._execute_update_now(plugin_id, plugin_instance, current_time)
+                else:
+                    self._enqueue_update(plugin_id, current_time)
+
+    def get_plugin_lock(self, plugin_id: str) -> threading.Lock:
+        """Per-plugin lock keeping update() and display() mutually exclusive.
+
+        The update worker holds it for the duration of a plugin's update();
+        the display side acquires it non-blocking and skips that frame's
+        display() call when the plugin is mid-update.
+        """
+        with self._plugin_locks_guard:
+            lock = self._plugin_locks.get(plugin_id)
+            if lock is None:
+                lock = threading.Lock()
+                self._plugin_locks[plugin_id] = lock
+            return lock
+
+    def _enqueue_update(self, plugin_id: str, scheduled_time: float) -> None:
+        """Queue a due update for the background worker (dedup while pending)."""
+        with self._pending_lock:
+            if plugin_id in self._pending_updates:
+                return
+            self._pending_updates.add(plugin_id)
+        # RUNNING is set at enqueue time so can_execute() blocks re-entry and
+        # the web UI shows the truthful state while the item waits its turn.
+        self.state_manager.set_state(plugin_id, PluginState.RUNNING)
+        self._ensure_update_worker()
+        self._update_queue.put((plugin_id, scheduled_time))
+
+    def _ensure_update_worker(self) -> None:
+        if self._update_worker is not None and self._update_worker.is_alive():
+            return
+        self._update_worker = threading.Thread(
+            target=self._update_worker_loop, name='plugin-update-worker',
+            daemon=True)
+        self._update_worker.start()
+
+    def _update_worker_loop(self) -> None:
+        """Single worker: dispatches queued updates off the render thread
+        (matching the old inline behavior — no thundering herd of
+        concurrent fetches).
+
+        The plugin's lock is acquired here, before its instance is looked
+        up, and the instance is re-fetched under the lock — a concurrent
+        unload_plugin() can't leave this loop about to run update() on an
+        instance that's already been torn down. The lock — and RUNNING/
+        pending lifecycle state — is released by the update itself once the
+        real update() call genuinely finishes (see _execute_update_now),
+        which can be after this dispatch returns if PluginExecutor's own
+        timeout elapses first.
+        """
+        while True:
+            item = self._update_queue.get()
+            if item is None:  # shutdown sentinel
+                return
+            plugin_id, scheduled_time = item
+            lock = self.get_plugin_lock(plugin_id)
+            lock.acquire()
+            plugin_instance = self.plugins.get(plugin_id)
+            if plugin_instance is None:  # unloaded while queued; its
+                # lifecycle state was already cleared by unload_plugin —
+                # leave it alone rather than resurrecting it to ENABLED
+                lock.release()
+                with self._pending_lock:
+                    self._pending_updates.discard(plugin_id)
+                continue
+            try:
+                self._execute_update_now(plugin_id, plugin_instance,
+                                         scheduled_time, lock=lock)
+            except Exception:  # pylint: disable=broad-except
+                # _execute_update_now guarantees the lock/pending bookkeeping
+                # is released via its own _finish() before returning or
+                # raising; this is a last-resort log only.
+                self.logger.exception("update worker: unexpected error for %s",
+                                      plugin_id)
+
+    def stop_update_worker(self, timeout: float = 5.0) -> None:
+        """Signal the worker to exit (used by cleanup; thread is a daemon)."""
+        if self._update_worker is not None and self._update_worker.is_alive():
+            self._update_queue.put(None)
+            self._update_worker.join(timeout=timeout)
+            if self._update_worker.is_alive():
+                self.logger.warning(
+                    "Update worker did not stop within %.1fs; it is a daemon "
+                    "thread and will be abandoned on shutdown", timeout)
+
+    def _execute_update_now(self, plugin_id: str, plugin_instance: Any,
+                            scheduled_time: float,
+                            lock: Optional[threading.Lock] = None) -> None:
+        """Execute a plugin's update() via PluginExecutor, then bookkeep.
+
+        Caller is responsible for having set RUNNING state.
+
+        On the synchronous path (``lock=None``) this is the original,
+        unchanged inline behavior. On the async worker path, PluginExecutor's
+        internal thread.join(timeout) blocks only the calling thread -- on
+        timeout the lingering daemon update-thread keeps running the real
+        plugin.update() call unkillable in the background. So that the
+        plugin's lock (and its RUNNING/pending lifecycle state) stays held
+        for that real duration rather than just this bounded wait, ownership
+        of both is carried by the wrapped update callable itself, released
+        from whichever thread actually finishes it -- see _finish() below.
+        """
+        finish_guard = threading.Lock()
+        finished = {'done': False}
+
+        def _finish(success: bool, exc: Optional[Exception] = None) -> None:
+            with finish_guard:
+                if finished['done']:
+                    return
+                finished['done'] = True
+            try:
+                if success:
+                    with self._plugin_last_update_lock:
+                        self.plugin_last_update[plugin_id] = scheduled_time
+                    self.state_manager.record_update(plugin_id)
+                    self.state_manager.set_state(plugin_id, PluginState.ENABLED)
+                    if self.health_tracker:
+                        self.health_tracker.record_success(plugin_id)
+                else:
                     self._record_update_failure(plugin_id, exc=exc)
+            finally:
+                if lock is not None:
+                    lock.release()
+                    with self._pending_lock:
+                        self._pending_updates.discard(plugin_id)
+
+        if lock is None:
+            # Synchronous / no-lock path: unchanged behavior.
+            try:
+                if self.resource_monitor:
+                    def monitored_update():
+                        self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
+                    # SimpleNamespace stores `update` as an *instance*
+                    # attribute, so attribute lookup returns the plain
+                    # function object as-is. A dynamically-built class
+                    # (`type(..., {'update': monitored_update})`) instead
+                    # stores it as a *class* attribute, which the
+                    # descriptor protocol turns into a bound method on
+                    # access -- silently prepending the instance as an
+                    # implicit first argument to a function that takes
+                    # none, raising "monitored_update() takes 0
+                    # positional arguments but 1 was given" on every call.
+                    success = self.plugin_executor.execute_update(
+                        types.SimpleNamespace(update=monitored_update),
+                        plugin_id
+                    )
+                else:
+                    success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
+                _finish(success)
+            except Exception as exc:  # pylint: disable=broad-except
+                self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
+                _finish(False, exc=exc)
+            return
+
+        # Async worker path: the real update() call -- through the resource
+        # monitor, if configured -- owns finishing the lock/lifecycle
+        # bookkeeping, from whichever thread actually runs it to completion.
+        def _target_update() -> None:
+            try:
+                if self.resource_monitor:
+                    self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
+                else:
+                    plugin_instance.update()
+            except Exception as exc:
+                _finish(False, exc=exc)
+                raise
+            else:
+                _finish(True)
+
+        try:
+            self.plugin_executor.execute_update(
+                types.SimpleNamespace(update=_target_update), plugin_id)
+        except Exception as exc:  # pragma: no cover - defensive; execute_update
+            # catches everything internally, but guarantee _finish still
+            # runs (releasing the lock) if something unexpected slips through.
+            self.logger.exception("Unexpected error dispatching update for %s: %s", plugin_id, exc)
+            _finish(False, exc=exc)
 
     def run_scheduled_updates_with_changes(self, current_time: Optional[float] = None) -> List[str]:
         """

src/plugin_system/testing/loading.py

@@ -88,3 +88,27 @@ def load_harness_spec(plugin_dir: Union[str, Path]) -> Dict[str, Any]:
         with open(mock_path, 'r') as mf:
             spec['mock_data_contents'] = json.load(mf)
     return spec
+
+
+def build_full_config(
+    plugin_dir: Union[str, Path],
+    spec: Optional[Dict[str, Any]] = None,
+    cli_config: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    """Build the config a plugin sees under test.
+
+    Merge order: config_schema.json defaults, then a forced ``enabled: True``,
+    then harness.json's config overlay, then the caller's explicit config --
+    most specific wins. `enabled` is re-asserted *after* the schema defaults
+    so a plugin that reasonably ships `enabled: false` (e.g. a seasonal or
+    opt-in plugin) can't silently make every harness run test "disabled, do
+    nothing" by accident -- callers that genuinely want to test the disabled
+    path can still do so via `cli_config={"enabled": False}`.
+    """
+    spec = spec or {}
+    config: Dict[str, Any] = {}
+    config.update(load_config_defaults(plugin_dir))
+    config["enabled"] = True
+    config.update(spec.get("config", {}))
+    config.update(cli_config or {})
+    return config

src/plugin_system/testing/mocks.py

@@ -71,6 +71,7 @@ class MockCacheManager:
         self.get_calls = []
         self.set_calls = []
         self.delete_calls = []
+        self.get_cached_data_with_strategy_calls = []
         # Real temp dir for plugins that write/read files under cache_dir.
         # Registered for cleanup so each mock instance doesn't leak a tmp dir.
         self.cache_dir = tempfile.mkdtemp(prefix="ledmatrix-mock-cache-")
@@ -108,6 +109,24 @@ class MockCacheManager:
         self.delete_calls.append(key)
         if key in self._cache:
             del self._cache[key]
+
+    def get_cached_data_with_strategy(self, key: str, data_type: str = 'default') -> Optional[Any]:
+        """Mock of CacheManager.get_cached_data_with_strategy (src/cache_manager.py).
+
+        The real method picks a max_age/memory_ttl strategy per data_type
+        (and extends it during market-closed hours for market data) before
+        delegating to get_cached_data(). None of that timing nuance matters
+        for a mock -- plugins under test just need the method to exist and
+        return whatever was cached, so this delegates straight to get().
+        """
+        self.get_cached_data_with_strategy_calls.append({'key': key, 'data_type': data_type})
+        return self.get(key)
+
+    def save_cache(self, key: str, data: Any) -> None:
+        """Mock of CacheManager.save_cache (src/cache_manager.py) -- the
+        write-side counterpart to get_cached_data_with_strategy, used by the
+        same real-CacheManager-oriented plugins. Delegates to set()."""
+        self.set(key, data)
         if key in self._cache_timestamps:
             del self._cache_timestamps[key]
     
@@ -118,6 +137,7 @@ class MockCacheManager:
         self.get_calls = []
         self.set_calls = []
         self.delete_calls = []
+        self.get_cached_data_with_strategy_calls = []
 
 
 class MockConfigManager:

src/wifi_manager.py

@@ -2564,11 +2564,35 @@ address=/detectportal.firefox.com/192.168.4.1
         Returns:
             True if AP mode state changed, False otherwise
         """
+        changed, _status, _ethernet, _ap = self.check_and_manage_ap_mode_with_state()
+        return changed
+
+    def check_and_manage_ap_mode_with_state(self) -> Tuple[bool, WiFiStatus, bool, bool]:
+        """Like check_and_manage_ap_mode, but also returns the state it
+        observed, so callers (the wifi monitor daemon) don't have to re-run
+        the same nmcli subprocess battery before AND after the check —
+        each status fetch is several process forks.
+
+        Returns:
+            (state_changed, WiFiStatus, ethernet_connected, ap_active_after)
+        """
         try:
             # Get status with retry for more reliable detection
             status = self._get_wifi_status_with_retry()
             ethernet_connected = self._is_ethernet_connected()
             ap_active = self._is_ap_mode_active()
+            changed = self._manage_ap_mode(status, ethernet_connected, ap_active)
+            # State only ever changes via one enable or one disable, so the
+            # post-state is the inverse of the pre-state when changed.
+            ap_after = (not ap_active) if changed else ap_active
+            return changed, status, ethernet_connected, ap_after
+        except Exception as e:
+            logger.error(f"Error checking AP mode: {e}", exc_info=True)
+            return False, WiFiStatus(connected=False), False, False
+
+    def _manage_ap_mode(self, status: WiFiStatus, ethernet_connected: bool, ap_active: bool) -> bool:
+        """AP-mode decision logic against an already-fetched state snapshot."""
+        try:
             auto_enable = self.config.get("auto_enable_ap_mode", True)  # Default: True (safe due to grace period)
             
             # Log current state for debugging

test/plugins/test_harness.py

@@ -253,3 +253,61 @@ class TestCheckPluginHonorsHarnessJson:
         )
         assert captured["freeze_time"] == "2030-01-01 00:00:00"
         assert captured["config"]["timezone"] == "America/New_York"
+
+
+class TestBuildFullConfigForcesEnabled:
+    """Regression: a plugin's own config_schema.json may reasonably default
+    enabled to False (e.g. a seasonal or opt-in plugin) -- march-madness and
+    14 other real plugins do. The harness must still test it as enabled
+    unless a caller explicitly asks otherwise, or every render silently
+    becomes a same-shaped "disabled, do nothing" no-op."""
+
+    def _make_plugin_with_disabled_default(self, tmp_path):
+        pdir = tmp_path / "plugins" / "demo-seasonal"
+        pdir.mkdir(parents=True)
+        (pdir / "manifest.json").write_text(json.dumps({
+            "id": "demo-seasonal", "name": "Demo Seasonal", "version": "1.0.0",
+            "author": "test", "entry_point": "manager.py",
+            "class_name": "DemoSeasonal", "display_modes": ["demo-seasonal"],
+            "compatible_versions": ["*"],
+        }))
+        (pdir / "config_schema.json").write_text(json.dumps({
+            "type": "object",
+            "properties": {"enabled": {"type": "boolean", "default": False}},
+        }))
+        return pdir
+
+    def test_schema_disabled_default_does_not_win(self, tmp_path):
+        from src.plugin_system.testing.loading import build_full_config
+        plugin_dir = self._make_plugin_with_disabled_default(tmp_path)
+        config = build_full_config(plugin_dir)
+        assert config["enabled"] is True
+
+    def test_harness_json_config_can_still_disable(self, tmp_path):
+        from src.plugin_system.testing.loading import build_full_config
+        plugin_dir = self._make_plugin_with_disabled_default(tmp_path)
+        config = build_full_config(plugin_dir, spec={"config": {"enabled": False}})
+        assert config["enabled"] is False
+
+    def test_explicit_cli_config_can_still_disable(self, tmp_path):
+        from src.plugin_system.testing.loading import build_full_config
+        plugin_dir = self._make_plugin_with_disabled_default(tmp_path)
+        config = build_full_config(plugin_dir, cli_config={"enabled": False})
+        assert config["enabled"] is False
+
+    def test_check_one_renders_a_schema_disabled_plugin_as_enabled(self, tmp_path, monkeypatch):
+        """End-to-end: check_plugin.py's check_one() must not blank-render a
+        plugin just because its own schema defaults enabled to False."""
+        mod = _load_check_plugin_cli()
+        plugin_dir = self._make_plugin_with_disabled_default(tmp_path)
+        captured = {}
+        monkeypatch.setattr(mod, "render_plugin_matrix",
+                            lambda **kw: captured.update(kw) or [])
+        monkeypatch.setattr(mod, "compare_to_goldens", lambda *a, **k: [])
+        mod.check_one(
+            plugin_id="demo-seasonal", search_dirs=[str(tmp_path / "plugins")],
+            sizes=None, mock_data={}, config={}, run_update=True,
+            out_dir=None, update_golden=False, golden_dir_override=None,
+            freeze_time=None,
+        )
+        assert captured["config"]["enabled"] is True

test/plugins/test_plugin_matrix.py

@@ -22,7 +22,7 @@ import pytest
 from src.plugin_system.testing.harness import (
     render_plugin_matrix, compare_to_goldens,
 )
-from src.plugin_system.testing.loading import load_config_defaults, load_harness_spec
+from src.plugin_system.testing.loading import build_full_config, load_harness_spec
 from src.plugin_system.testing.sizes import resolve_test_sizes
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
@@ -81,9 +81,7 @@ def test_plugin_renders_across_sizes_and_screens(plugin_id: str) -> None:
     plugin_dir = _PLUGINS[plugin_id]
     spec = load_harness_spec(plugin_dir)
 
-    config = {"enabled": True}
-    config.update(load_config_defaults(plugin_dir))
-    config.update(spec.get("config", {}))
+    config = build_full_config(plugin_dir, spec)
 
     # Sizes: LEDMATRIX_TEST_SIZES env (test on real hardware) wins, then the
     # plugin's own harness.json "sizes", else the default representative sample.

test/test_async_plugin_updates.py

@@ -0,0 +1,270 @@
+"""Tests for asynchronous plugin updates (plugin_manager background worker).
+
+The invariants that keep this change safe:
+1. run_scheduled_updates returns immediately — a slow update() can never
+   again freeze the render loop (the original defect: 30s scroll freezes).
+2. A plugin's update() and display() are NEVER concurrent — the per-plugin
+   lock makes the old implicit no-overlap guarantee explicit, and it stays
+   held through PluginExecutor's own timeout: a lingering, still-running
+   update() keeps the lock even after PluginExecutor gives up waiting on it.
+3. Failure/timeout bookkeeping is unchanged (same executor, same
+   _record_update_failure path, same last-update stamping).
+4. The kill switch (plugin_system.synchronous_updates) restores the
+   inline path exactly.
+"""
+
+import os
+import sys
+import threading
+import time
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.plugin_system.plugin_manager import PluginManager  # noqa: E402
+from src.plugin_system.plugin_state import PluginState  # noqa: E402
+
+
+class SlowPlugin:
+    """Fake plugin whose update() sleeps and records overlap violations."""
+
+    def __init__(self, update_seconds=0.5):
+        self.enabled = True
+        self.update_seconds = update_seconds
+        self.update_calls = 0
+        self.display_calls = 0
+        self.in_update = False
+        self.overlap_detected = False
+
+    def update(self):
+        self.in_update = True
+        self.update_calls += 1
+        time.sleep(self.update_seconds)
+        self.in_update = False
+        return True
+
+    def display(self, force_clear=False):
+        if self.in_update:
+            self.overlap_detected = True
+        self.display_calls += 1
+        return True
+
+
+@pytest.fixture
+def pm(tmp_path):
+    manager = PluginManager(plugins_dir=str(tmp_path), config_manager=None,
+                            display_manager=None, cache_manager=None)
+    yield manager
+    manager.stop_update_worker()
+
+
+def _install(pm, plugin, plugin_id="slow-plugin"):
+    pm.plugins[plugin_id] = plugin
+    pm._update_interval_cache[plugin_id] = 0.01  # always due
+    # load_plugin normally registers state; can_execute() gates on it
+    pm.state_manager.set_state(plugin_id, PluginState.ENABLED)
+    return plugin_id
+
+
+class TestSchedulerNonBlocking:
+    def test_run_scheduled_updates_returns_immediately(self, pm):
+        plugin_id = _install(pm, SlowPlugin(update_seconds=2.0))
+        start = time.monotonic()
+        pm.run_scheduled_updates()
+        elapsed = time.monotonic() - start
+        assert elapsed < 0.1, f"scheduler blocked for {elapsed:.2f}s"
+        # the update actually runs in the background
+        deadline = time.monotonic() + 5
+        while pm.plugins[plugin_id].update_calls == 0 and time.monotonic() < deadline:
+            time.sleep(0.05)
+        assert pm.plugins[plugin_id].update_calls == 1
+
+    def test_no_double_enqueue_while_pending(self, pm):
+        plugin_id = _install(pm, SlowPlugin(update_seconds=0.8))
+        for _ in range(20):
+            pm.run_scheduled_updates()
+            time.sleep(0.01)
+        time.sleep(1.5)  # let the single queued update finish
+        assert pm.plugins[plugin_id].update_calls == 1
+
+
+class TestUpdateDisplayExclusion:
+    def test_display_lock_held_during_update(self, pm):
+        plugin_id = _install(pm, SlowPlugin(update_seconds=0.6))
+        pm.run_scheduled_updates()
+        # give the worker a moment to take the lock and enter update()
+        deadline = time.monotonic() + 2
+        while not pm.plugins[plugin_id].in_update and time.monotonic() < deadline:
+            time.sleep(0.01)
+        lock = pm.get_plugin_lock(plugin_id)
+        assert lock.acquire(blocking=False) is False, \
+            "lock must be held while update() runs"
+        # and released afterwards
+        deadline = time.monotonic() + 3
+        while pm.plugins[plugin_id].in_update and time.monotonic() < deadline:
+            time.sleep(0.05)
+        time.sleep(0.1)
+        assert lock.acquire(blocking=False) is True
+        lock.release()
+
+    def test_no_overlap_under_hammering_display_loop(self, pm):
+        """Simulate the render loop's try-lock display pattern at high rate
+        while updates fire — the plugin itself asserts no overlap."""
+        plugin = SlowPlugin(update_seconds=0.15)
+        plugin_id = _install(pm, plugin)
+        stop = threading.Event()
+
+        def render_loop():
+            while not stop.is_set():
+                lock = pm.get_plugin_lock(plugin_id)
+                if lock.acquire(blocking=False):
+                    try:
+                        plugin.display()
+                    finally:
+                        lock.release()
+                time.sleep(0.002)
+
+        renderer = threading.Thread(target=render_loop, daemon=True)
+        renderer.start()
+        try:
+            for _ in range(6):
+                pm.plugin_last_update.pop(plugin_id, None)  # force due
+                pm.run_scheduled_updates()
+                time.sleep(0.3)
+        finally:
+            stop.set()
+            renderer.join(timeout=2)
+        assert plugin.update_calls >= 3
+        assert plugin.display_calls > 10
+        assert plugin.overlap_detected is False
+
+    def test_lock_held_through_timeout_until_real_update_finishes(self, pm):
+        """PluginExecutor's join(timeout) can return before the real
+        update() call does -- the lingering daemon thread keeps running it.
+        The plugin's lock must stay held for that whole real duration, not
+        just PluginExecutor's bounded wait, or display() could run
+        concurrently with a still-executing update()."""
+        plugin = SlowPlugin(update_seconds=0.3)
+        plugin_id = _install(pm, plugin)
+        pm.plugin_executor.default_timeout = 0.05  # times out well before
+                                                    # update_seconds elapses
+
+        pm.run_scheduled_updates()
+
+        deadline = time.monotonic() + 2
+        while not plugin.in_update and time.monotonic() < deadline:
+            time.sleep(0.01)
+        assert plugin.in_update is True
+
+        # PluginExecutor's own timeout has now elapsed, but the real
+        # update() (0.3s) is still running in its lingering daemon thread.
+        time.sleep(0.15)
+        assert plugin.in_update is True, "test setup: update should still be running"
+        lock = pm.get_plugin_lock(plugin_id)
+        assert lock.acquire(blocking=False) is False, \
+            "lock must stay held through PluginExecutor's timeout while the real update() runs"
+
+        # Once the real update() genuinely finishes, the lock is released.
+        deadline = time.monotonic() + 2
+        while plugin.in_update and time.monotonic() < deadline:
+            time.sleep(0.02)
+        time.sleep(0.1)
+        assert lock.acquire(blocking=False) is True
+        lock.release()
+
+    def test_state_returns_to_enabled_after_update(self, pm):
+        """RUNNING is set at enqueue (blocks re-entry via can_execute) and
+        must return to an executable state once the update finishes."""
+        plugin_id = _install(pm, SlowPlugin(update_seconds=0.1))
+        pm.run_scheduled_updates()
+        # while queued/running, re-entry is blocked
+        assert pm.state_manager.can_execute(plugin_id) is False
+        deadline = time.monotonic() + 3
+        while time.monotonic() < deadline:
+            if (pm.plugins[plugin_id].update_calls
+                    and pm.state_manager.can_execute(plugin_id)):
+                break
+            time.sleep(0.05)
+        assert pm.plugins[plugin_id].update_calls == 1
+        assert pm.state_manager.can_execute(plugin_id) is True
+
+
+class TestFailurePaths:
+    def test_update_failure_routes_through_failure_bookkeeping(self, pm):
+        class FailingPlugin(SlowPlugin):
+            def update(self):
+                self.update_calls += 1
+                raise RuntimeError("boom")
+
+        plugin_id = _install(pm, FailingPlugin())
+        pm.run_scheduled_updates()
+        deadline = time.monotonic() + 3
+        while pm.plugins[plugin_id].update_calls == 0 and time.monotonic() < deadline:
+            time.sleep(0.05)
+        time.sleep(0.2)
+        # failure stamped so the interval gate holds (no hot retry loop)
+        assert pm.plugin_last_update.get(plugin_id, 0) > 0
+        # lock released after failure
+        assert pm.get_plugin_lock(plugin_id).acquire(blocking=False) is True
+        pm.get_plugin_lock(plugin_id).release()
+
+    def test_unloaded_while_queued_is_harmless(self, pm):
+        """Exercise the public unload_plugin() lifecycle rather than
+        deleting pm.plugins directly: queue the target's update behind a
+        deterministic blocker (occupying the single worker), unload the
+        target while its item still sits queued, then release the blocker
+        and confirm the target's update never ran and its state stayed
+        unloaded rather than being resurrected to ENABLED."""
+        blocker_event = threading.Event()
+
+        class BlockerPlugin(SlowPlugin):
+            def update(self):
+                self.update_calls += 1
+                blocker_event.wait(timeout=5)
+                return True
+
+        blocker_id = _install(pm, BlockerPlugin(), plugin_id="blocker-plugin")
+        target = SlowPlugin(update_seconds=0.05)
+        target_id = _install(pm, target, plugin_id="slow-plugin")
+
+        # Dispatch the blocker first so it occupies the single worker
+        # thread, then enqueue the target behind it -- deterministically
+        # queued, not yet started.
+        pm._enqueue_update(blocker_id, time.time())
+        deadline = time.monotonic() + 2
+        while pm.plugins[blocker_id].update_calls == 0 and time.monotonic() < deadline:
+            time.sleep(0.01)
+        assert pm.plugins[blocker_id].update_calls == 1
+
+        pm._enqueue_update(target_id, time.time())
+        assert target_id in pm._pending_updates
+
+        assert pm.unload_plugin(target_id) is True
+        assert target_id not in pm.plugins
+
+        blocker_event.set()  # let the blocker finish; worker moves on to
+                              # the target's queued item
+
+        deadline = time.monotonic() + 3
+        while target_id in pm._pending_updates and time.monotonic() < deadline:
+            time.sleep(0.02)
+
+        assert target.update_calls == 0, "update() must not run for an unloaded plugin"
+        assert target_id not in pm._pending_updates
+        assert pm.state_manager.get_state(target_id) == PluginState.UNLOADED
+
+
+class TestKillSwitch:
+    def test_synchronous_mode_blocks_like_before(self, pm):
+        pm._synchronous_updates = True
+        plugin_id = _install(pm, SlowPlugin(update_seconds=0.4))
+        start = time.monotonic()
+        pm.run_scheduled_updates()
+        elapsed = time.monotonic() - start
+        assert elapsed >= 0.4, "synchronous mode must run inline"
+        assert pm.plugins[plugin_id].update_calls == 1
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-v"]))

test/test_config_load_cache.py

@@ -0,0 +1,129 @@
+"""Tests for load_config's mtime fast path (src/config_manager.py).
+
+load_config used to re-read + re-parse config.json, the template (with a
+recursive migration diff) and secrets on EVERY call — ~30 web request
+handlers call it, some 2-3x per request. The fast path skips all of it
+when the three files' (mtime_ns, size) signatures are unchanged.
+
+The invariant that matters most: cross-process freshness — a save from
+the web process must be picked up by the display process's next load.
+That's guaranteed because the signature is re-stat'd on every call.
+"""
+
+import json
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.config_manager import ConfigManager  # noqa: E402
+
+
+@pytest.fixture
+def mgr(tmp_path):
+    config = tmp_path / "config.json"
+    secrets = tmp_path / "secrets.json"
+    template = tmp_path / "template.json"
+    config.write_text(json.dumps({"display": {"brightness": 90}, "timezone": "UTC"}))
+    secrets.write_text(json.dumps({"weather": {"api_key": "sek"}}))
+    template.write_text(json.dumps({"display": {"brightness": 90}, "timezone": "UTC"}))
+    m = ConfigManager(config_path=str(config), secrets_path=str(secrets))
+    m.template_path = str(template)
+    return m, config, secrets, template
+
+
+def _count_opens(monkeypatch, mgr_paths):
+    """Count open() calls hitting the config files."""
+    counts = {"n": 0}
+    real_open = open
+
+    def counting_open(file, *args, **kwargs):
+        if str(file) in mgr_paths:
+            counts["n"] += 1
+        return real_open(file, *args, **kwargs)
+
+    import builtins
+    monkeypatch.setattr(builtins, "open", counting_open)
+    return counts
+
+
+class TestFastPath:
+    def test_unchanged_files_are_not_reread(self, mgr, monkeypatch):
+        m, config, secrets, template = mgr
+        first = m.load_config()
+        assert first["weather"]["api_key"] == "sek"  # secrets merged
+        counts = _count_opens(monkeypatch, {str(config), str(secrets), str(template)})
+        for _ in range(10):
+            again = m.load_config()
+        assert counts["n"] == 0, "fast path must not re-open any config file"
+        assert again is first  # same aliasing semantics as the full path
+
+    def test_config_change_triggers_reload(self, mgr):
+        m, config, secrets, template = mgr
+        m.load_config()
+        data = json.loads(config.read_text())
+        data["display"]["brightness"] = 55
+        config.write_text(json.dumps(data))
+        os.utime(config, (os.stat(config).st_atime, os.stat(config).st_mtime + 2))
+        assert m.load_config()["display"]["brightness"] == 55
+
+    def test_secrets_change_triggers_reload(self, mgr):
+        m, config, secrets, template = mgr
+        m.load_config()
+        secrets.write_text(json.dumps({"weather": {"api_key": "NEW"}}))
+        os.utime(secrets, (os.stat(secrets).st_atime, os.stat(secrets).st_mtime + 2))
+        assert m.load_config()["weather"]["api_key"] == "NEW"
+
+    def test_template_change_triggers_reload_and_migration(self, mgr):
+        m, config, secrets, template = mgr
+        m.load_config()
+        template.write_text(json.dumps({
+            "display": {"brightness": 90}, "timezone": "UTC",
+            "brand_new_key": {"added": True}}))
+        os.utime(template, (os.stat(template).st_atime, os.stat(template).st_mtime + 2))
+        reloaded = m.load_config()
+        assert reloaded.get("brand_new_key") == {"added": True}
+
+    def test_same_second_edit_detected_via_mtime_ns_or_size(self, mgr):
+        """Coarse-mtime same-second edits: size difference still busts it."""
+        m, config, secrets, template = mgr
+        m.load_config()
+        st = os.stat(config)
+        data = json.loads(config.read_text())
+        data["timezone"] = "America/New_York"  # different byte length
+        config.write_text(json.dumps(data))
+        os.utime(config, (st.st_atime, st.st_mtime))  # force same mtime
+        assert m.load_config()["timezone"] == "America/New_York"
+
+
+class TestSaveCoherence:
+    def test_save_config_then_load_returns_saved_data(self, mgr, monkeypatch):
+        m, config, secrets, template = mgr
+        m.load_config()
+        new = {"display": {"brightness": 42}, "timezone": "UTC",
+               "weather": {"api_key": "sek"}}
+        m.save_config(new)
+        counts = _count_opens(monkeypatch, {str(config), str(secrets), str(template)})
+        loaded = m.load_config()
+        assert loaded["display"]["brightness"] == 42
+        assert loaded["weather"]["api_key"] == "sek"  # secrets survive in memory
+        assert counts["n"] == 0  # signature refreshed by save; no re-read
+
+    def test_cross_process_save_is_picked_up(self, mgr):
+        """Another process writing config.json (different mtime) must bust
+        this process's fast path — the core cross-process guarantee."""
+        m, config, secrets, template = mgr
+        m.load_config()
+        other = ConfigManager(config_path=str(config), secrets_path=str(secrets))
+        other.template_path = str(template)
+        other.load_config()
+        other.save_config({"display": {"brightness": 11}, "timezone": "UTC",
+                           "weather": {"api_key": "sek"}})
+        os.utime(config, (os.stat(config).st_atime, os.stat(config).st_mtime + 2))
+        assert m.load_config()["display"]["brightness"] == 11
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-v"]))

test/test_display_dirty_tracking.py

@@ -0,0 +1,162 @@
+"""Tests for update_display dirty tracking (src/display_manager.py).
+
+Runs against RGBMatrixEmulator (EMULATOR=true), exercising the REAL
+DisplayManager — not a mock — so the skip logic, its invalidation hooks,
+and the kill switch are verified off-Pi.
+
+The invariants:
+- identical frames are pushed exactly once (SwapOnVSync not re-called)
+- ANY pixel change pushes
+- clear() and set_brightness() invalidate (the two paths that alter panel
+  state outside the digest's view)
+- the kill switch (display.dirty_tracking: false) restores always-push
+"""
+
+import os
+import sys
+import time
+
+os.environ["EMULATOR"] = "true"
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+
+@pytest.fixture(scope="module")
+def dm():
+    """One real DisplayManager on the emulator (it's a process singleton)."""
+    from src.display_manager import DisplayManager
+    DisplayManager._instance = None
+    DisplayManager._initialized = False
+    manager = DisplayManager({
+        "display": {
+            "hardware": {"rows": 32, "cols": 64, "chain_length": 2,
+                         "parallel": 1, "brightness": 90},
+            "runtime": {"gpio_slowdown": 0},
+        },
+    }, suppress_test_pattern=True)
+    yield manager
+    DisplayManager._instance = None
+    DisplayManager._initialized = False
+
+
+class _SwapSpy:
+    """Counts SwapOnVSync calls through the real matrix object."""
+
+    def __init__(self, matrix):
+        self.matrix = matrix
+        self.count = 0
+        self._orig = matrix.SwapOnVSync
+
+    def __enter__(self):
+        def counting(canvas):
+            self.count += 1
+            return self._orig(canvas)
+        self.matrix.SwapOnVSync = counting
+        return self
+
+    def __exit__(self, *exc):
+        self.matrix.SwapOnVSync = self._orig
+
+
+class TestDirtyTracking:
+    def test_identical_frames_push_once(self, dm):
+        dm.draw.rectangle([0, 0, 10, 10], fill=(255, 0, 0))
+        with _SwapSpy(dm.matrix) as spy:
+            dm.update_display()
+            dm.update_display()
+            dm.update_display()
+        assert spy.count == 1
+
+    def test_pixel_change_pushes(self, dm):
+        dm.update_display()
+        with _SwapSpy(dm.matrix) as spy:
+            dm.draw.point((5, 5), fill=(0, 255, 0))
+            dm.update_display()
+            dm.update_display()  # unchanged again
+        assert spy.count == 1
+
+    def test_clear_invalidates(self, dm):
+        dm.draw.rectangle([0, 0, 20, 20], fill=(0, 0, 255))
+        dm.update_display()
+        dm.clear()  # writes to the matrix directly; digest must reset
+        with _SwapSpy(dm.matrix) as spy:
+            dm.update_display()  # black frame after clear must still push
+        assert spy.count == 1
+
+    def test_brightness_change_forces_push(self, dm):
+        dm.draw.rectangle([0, 0, 20, 20], fill=(200, 200, 200))
+        dm.update_display()
+        with _SwapSpy(dm.matrix) as spy:
+            dm.update_display()          # identical -> skipped
+            assert spy.count == 0
+            dm.set_brightness(40)        # dim schedule scenario
+            dm.update_display()          # same image, new brightness -> push
+        assert spy.count == 1
+        dm.set_brightness(90)
+
+    def test_snapshot_still_written_on_skip(self, dm, tmp_path):
+        """The web preview mirror must keep working through skipped panel
+        pushes: _write_snapshot_if_due() still runs on the dirty-tracking
+        skip path and applies its own write/touch policy rather than being
+        bypassed entirely (see src/common/snapshot_policy.py — an unchanged
+        frame is touched, not re-encoded, once TOUCH_INTERVAL elapses)."""
+        dm._snapshot_path = str(tmp_path / "snap.png")
+        dm._last_snapshot_ts = 0.0
+        dm._last_snapshot_touch_ts = 0.0
+        dm._last_snapshot_digest = None
+        dm.draw.rectangle([0, 0, 30, 8], fill=(255, 255, 0))
+        dm.update_display()   # push + snapshot write (first frame)
+        assert os.path.exists(dm._snapshot_path)
+        first_mtime = os.path.getmtime(dm._snapshot_path)
+
+        # Age the write/touch bookkeeping past TOUCH_INTERVAL so the next
+        # identical frame is due for a touch, then push it again: dirty
+        # tracking must skip the panel write, but the snapshot mirror must
+        # still get its mtime bumped so the health check doesn't go stale.
+        from src.common import snapshot_policy
+        stale_ts = time.time() - snapshot_policy.TOUCH_INTERVAL - 1.0
+        dm._last_snapshot_ts = stale_ts
+        dm._last_snapshot_touch_ts = stale_ts
+        with _SwapSpy(dm.matrix) as spy:
+            dm.update_display()  # identical frame -> panel push skipped
+        assert spy.count == 0
+        assert os.path.getmtime(dm._snapshot_path) > first_mtime
+
+
+class TestKillSwitch:
+    def test_dirty_tracking_can_be_disabled(self, dm):
+        dm._dirty_tracking_enabled = False
+        try:
+            dm.draw.rectangle([0, 0, 10, 10], fill=(1, 2, 3))
+            with _SwapSpy(dm.matrix) as spy:
+                dm.update_display()
+                dm.update_display()
+                dm.update_display()
+            assert spy.count == 3  # always-push, exactly the old behavior
+        finally:
+            dm._dirty_tracking_enabled = True
+            dm._last_pushed_digest = None
+
+    def test_config_flag_wires_through(self):
+        from src.display_manager import DisplayManager
+        DisplayManager._instance = None
+        DisplayManager._initialized = False
+        try:
+            manager = DisplayManager({
+                "display": {
+                    "hardware": {"rows": 32, "cols": 64, "chain_length": 1,
+                                 "parallel": 1},
+                    "runtime": {"gpio_slowdown": 0},
+                    "dirty_tracking": False,
+                },
+            }, suppress_test_pattern=True)
+            assert manager._dirty_tracking_enabled is False
+        finally:
+            DisplayManager._instance = None
+            DisplayManager._initialized = False
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-v"]))

test/test_layout_manager.py

@@ -1,392 +0,0 @@
-"""
-Tests for LayoutManager.
-
-Tests layout creation, management, rendering, and element positioning.
-"""
-
-import pytest
-import json
-from unittest.mock import MagicMock
-from src.layout_manager import LayoutManager
-
-
-class TestLayoutManager:
-    """Test LayoutManager functionality."""
-    
-    @pytest.fixture
-    def tmp_layout_file(self, tmp_path):
-        """Create a temporary layout file."""
-        layout_file = tmp_path / "custom_layouts.json"
-        return str(layout_file)
-    
-    @pytest.fixture
-    def mock_display_manager(self):
-        """Create a mock display manager."""
-        dm = MagicMock()
-        dm.clear = MagicMock()
-        dm.update_display = MagicMock()
-        dm.draw_text = MagicMock()
-        dm.draw_weather_icon = MagicMock()
-        dm.small_font = MagicMock()
-        dm.regular_font = MagicMock()
-        return dm
-    
-    @pytest.fixture
-    def layout_manager(self, tmp_layout_file, mock_display_manager):
-        """Create a LayoutManager instance."""
-        return LayoutManager(
-            display_manager=mock_display_manager,
-            config_path=tmp_layout_file
-        )
-    
-    def test_init(self, tmp_layout_file, mock_display_manager):
-        """Test LayoutManager initialization."""
-        lm = LayoutManager(
-            display_manager=mock_display_manager,
-            config_path=tmp_layout_file
-        )
-        
-        assert lm.display_manager == mock_display_manager
-        assert lm.config_path == tmp_layout_file
-        assert lm.layouts == {}
-        assert lm.current_layout is None
-    
-    def test_load_layouts_file_exists(self, tmp_path, mock_display_manager):
-        """Test loading layouts from existing file."""
-        layout_file = tmp_path / "custom_layouts.json"
-        layout_data = {
-            "test_layout": {
-                "elements": [{"type": "text", "x": 0, "y": 0}],
-                "description": "Test layout"
-            }
-        }
-        with open(layout_file, 'w') as f:
-            json.dump(layout_data, f)
-        
-        lm = LayoutManager(
-            display_manager=mock_display_manager,
-            config_path=str(layout_file)
-        )
-        
-        assert "test_layout" in lm.layouts
-        assert lm.layouts["test_layout"]["description"] == "Test layout"
-    
-    def test_load_layouts_file_not_exists(self, tmp_layout_file, mock_display_manager):
-        """Test loading layouts when file doesn't exist."""
-        lm = LayoutManager(
-            display_manager=mock_display_manager,
-            config_path=tmp_layout_file
-        )
-        
-        assert lm.layouts == {}
-    
-    def test_create_layout(self, layout_manager):
-        """Test creating a new layout."""
-        elements = [{"type": "text", "x": 10, "y": 20, "properties": {"text": "Hello"}}]
-        
-        result = layout_manager.create_layout("test_layout", elements, "Test description")
-        
-        assert result is True
-        assert "test_layout" in layout_manager.layouts
-        assert layout_manager.layouts["test_layout"]["elements"] == elements
-        assert layout_manager.layouts["test_layout"]["description"] == "Test description"
-        assert "created" in layout_manager.layouts["test_layout"]
-        assert "modified" in layout_manager.layouts["test_layout"]
-    
-    def test_update_layout(self, layout_manager):
-        """Test updating an existing layout."""
-        # Create a layout first
-        elements1 = [{"type": "text", "x": 0, "y": 0}]
-        layout_manager.create_layout("test_layout", elements1, "Original")
-        
-        # Update it
-        elements2 = [{"type": "text", "x": 10, "y": 20}]
-        result = layout_manager.update_layout("test_layout", elements2, "Updated")
-        
-        assert result is True
-        assert layout_manager.layouts["test_layout"]["elements"] == elements2
-        assert layout_manager.layouts["test_layout"]["description"] == "Updated"
-        assert "modified" in layout_manager.layouts["test_layout"]
-    
-    def test_update_layout_not_exists(self, layout_manager):
-        """Test updating a non-existent layout."""
-        elements = [{"type": "text", "x": 0, "y": 0}]
-        result = layout_manager.update_layout("nonexistent", elements)
-        
-        assert result is False
-    
-    def test_delete_layout(self, layout_manager):
-        """Test deleting a layout."""
-        elements = [{"type": "text", "x": 0, "y": 0}]
-        layout_manager.create_layout("test_layout", elements)
-        
-        result = layout_manager.delete_layout("test_layout")
-        
-        assert result is True
-        assert "test_layout" not in layout_manager.layouts
-    
-    def test_delete_layout_not_exists(self, layout_manager):
-        """Test deleting a non-existent layout."""
-        result = layout_manager.delete_layout("nonexistent")
-        
-        assert result is False
-    
-    def test_get_layout(self, layout_manager):
-        """Test getting a specific layout."""
-        elements = [{"type": "text", "x": 0, "y": 0}]
-        layout_manager.create_layout("test_layout", elements)
-        
-        layout = layout_manager.get_layout("test_layout")
-        
-        assert layout is not None
-        assert layout["elements"] == elements
-    
-    def test_get_layout_not_exists(self, layout_manager):
-        """Test getting a non-existent layout."""
-        layout = layout_manager.get_layout("nonexistent")
-        
-        assert layout == {}
-    
-    def test_list_layouts(self, layout_manager):
-        """Test listing all layouts."""
-        layout_manager.create_layout("layout1", [])
-        layout_manager.create_layout("layout2", [])
-        layout_manager.create_layout("layout3", [])
-        
-        layouts = layout_manager.list_layouts()
-        
-        assert len(layouts) == 3
-        assert "layout1" in layouts
-        assert "layout2" in layouts
-        assert "layout3" in layouts
-    
-    def test_set_current_layout(self, layout_manager):
-        """Test setting the current layout."""
-        layout_manager.create_layout("test_layout", [])
-        
-        result = layout_manager.set_current_layout("test_layout")
-        
-        assert result is True
-        assert layout_manager.current_layout == "test_layout"
-    
-    def test_set_current_layout_not_exists(self, layout_manager):
-        """Test setting a non-existent layout as current."""
-        result = layout_manager.set_current_layout("nonexistent")
-        
-        assert result is False
-        assert layout_manager.current_layout is None
-    
-    def test_render_layout(self, layout_manager, mock_display_manager):
-        """Test rendering a layout."""
-        elements = [
-            {"type": "text", "x": 0, "y": 0, "properties": {"text": "Hello"}},
-            {"type": "text", "x": 10, "y": 10, "properties": {"text": "World"}}
-        ]
-        layout_manager.create_layout("test_layout", elements)
-        
-        result = layout_manager.render_layout("test_layout")
-        
-        assert result is True
-        mock_display_manager.clear.assert_called_once()
-        mock_display_manager.update_display.assert_called_once()
-        assert mock_display_manager.draw_text.call_count == 2
-    
-    def test_render_layout_no_display_manager(self, tmp_layout_file):
-        """Test rendering without display manager."""
-        lm = LayoutManager(display_manager=None, config_path=tmp_layout_file)
-        lm.create_layout("test_layout", [])
-        
-        result = lm.render_layout("test_layout")
-        
-        assert result is False
-    
-    def test_render_layout_not_exists(self, layout_manager):
-        """Test rendering a non-existent layout."""
-        result = layout_manager.render_layout("nonexistent")
-        
-        assert result is False
-    
-    def test_render_element_text(self, layout_manager, mock_display_manager):
-        """Test rendering a text element."""
-        element = {
-            "type": "text",
-            "x": 10,
-            "y": 20,
-            "properties": {
-                "text": "Hello",
-                "color": [255, 0, 0],
-                "font_size": "small"
-            }
-        }
-        
-        layout_manager.render_element(element, {})
-        
-        mock_display_manager.draw_text.assert_called_once()
-        call_args = mock_display_manager.draw_text.call_args
-        assert call_args[0][0] == "Hello"  # text
-        assert call_args[0][1] == 10  # x
-        assert call_args[0][2] == 20  # y
-    
-    def test_render_element_weather_icon(self, layout_manager, mock_display_manager):
-        """Test rendering a weather icon element."""
-        element = {
-            "type": "weather_icon",
-            "x": 10,
-            "y": 20,
-            "properties": {
-                "condition": "sunny",
-                "size": 16
-            }
-        }
-        
-        layout_manager.render_element(element, {})
-        
-        mock_display_manager.draw_weather_icon.assert_called_once_with("sunny", 10, 20, 16)
-    
-    def test_render_element_weather_icon_from_context(self, layout_manager, mock_display_manager):
-        """Test rendering weather icon with data from context."""
-        element = {
-            "type": "weather_icon",
-            "x": 10,
-            "y": 20,
-            "properties": {"size": 16}
-        }
-        data_context = {
-            "weather": {
-                "condition": "cloudy"
-            }
-        }
-        
-        layout_manager.render_element(element, data_context)
-        
-        mock_display_manager.draw_weather_icon.assert_called_once_with("cloudy", 10, 20, 16)
-    
-    def test_render_element_rectangle(self, layout_manager, mock_display_manager):
-        """Test rendering a rectangle element."""
-        element = {
-            "type": "rectangle",
-            "x": 10,
-            "y": 20,
-            "properties": {
-                "width": 50,
-                "height": 30,
-                "color": [255, 0, 0],
-                "filled": True
-            }
-        }
-        
-        # Mock the draw object and rectangle method
-        mock_draw = MagicMock()
-        mock_display_manager.draw = mock_draw
-        
-        layout_manager.render_element(element, {})
-        
-        # Verify rectangle was drawn
-        mock_draw.rectangle.assert_called_once()
-    
-    def test_render_element_unknown_type(self, layout_manager):
-        """Test rendering an unknown element type."""
-        element = {
-            "type": "unknown_type",
-            "x": 0,
-            "y": 0,
-            "properties": {}
-        }
-        
-        # Should not raise an exception
-        layout_manager.render_element(element, {})
-    
-    def test_process_template_text(self, layout_manager):
-        """Test template text processing."""
-        text = "Hello {name}, temperature is {temp}°F"
-        data_context = {
-            "name": "World",
-            "temp": 72
-        }
-        
-        result = layout_manager._process_template_text(text, data_context)
-        
-        assert result == "Hello World, temperature is 72°F"
-    
-    def test_process_template_text_no_context(self, layout_manager):
-        """Test template text with missing context."""
-        text = "Hello {name}"
-        data_context = {}
-        
-        result = layout_manager._process_template_text(text, data_context)
-        
-        # Should leave template as-is or handle gracefully
-        assert "{name}" in result or result == "Hello "
-    
-    def test_save_layouts_error_handling(self, layout_manager):
-        """Test error handling when saving layouts."""
-        # Create a layout
-        layout_manager.create_layout("test", [])
-        
-        # Make save fail by using invalid path
-        layout_manager.config_path = "/nonexistent/directory/layouts.json"
-        
-        result = layout_manager.save_layouts()
-        
-        # Should handle error gracefully
-        assert result is False
-    
-    def test_render_element_line(self, layout_manager, mock_display_manager):
-        """Test rendering a line element."""
-        element = {
-            "type": "line",
-            "x": 10,
-            "y": 20,
-            "properties": {
-                "x2": 50,
-                "y2": 30,
-                "color": [255, 0, 0],
-                "width": 2
-            }
-        }
-        
-        mock_draw = MagicMock()
-        mock_display_manager.draw = mock_draw
-        
-        layout_manager.render_element(element, {})
-        
-        mock_draw.line.assert_called_once()
-    
-    def test_render_element_clock(self, layout_manager, mock_display_manager):
-        """Test rendering a clock element."""
-        element = {
-            "type": "clock",
-            "x": 10,
-            "y": 20,
-            "properties": {
-                "format": "%H:%M",
-                "color": [255, 255, 255]
-            }
-        }
-        
-        layout_manager.render_element(element, {})
-        
-        mock_display_manager.draw_text.assert_called_once()
-    
-    def test_render_element_data_text(self, layout_manager, mock_display_manager):
-        """Test rendering a data text element."""
-        element = {
-            "type": "data_text",
-            "x": 10,
-            "y": 20,
-            "properties": {
-                "data_key": "weather.temperature",
-                "format": "Temp: {value}°F",
-                "color": [255, 255, 255],
-                "default": "N/A"
-            }
-        }
-        data_context = {
-            "weather": {
-                "temperature": 72
-            }
-        }
-        
-        layout_manager.render_element(element, data_context)
-        
-        mock_display_manager.draw_text.assert_called_once()

test/test_plugin_system.py

@@ -123,6 +123,14 @@ class TestPluginManager:
 
             pm.run_scheduled_updates(current_time=time.time())
 
+            # Updates now execute on the background worker (the scheduler
+            # returns immediately) — wait for completion before asserting.
+            deadline = time.time() + 5
+            while (plugin_instance.update.call_count == 0
+                   and time.time() < deadline):
+                time.sleep(0.02)
+            pm.stop_update_worker()
+
             plugin_instance.update.assert_called_once()
             assert "test_plugin" in pm.plugin_last_update
             assert pm.state_manager.get_state("test_plugin") == PluginState.ENABLED

test/test_testing_mocks.py

@@ -0,0 +1,45 @@
+"""
+Unit tests for src/plugin_system/testing/mocks.py.
+
+MockCacheManager/MockPluginManager stand in for the real production
+managers under the plugin safety harness -- a missing method here isn't a
+harness bug in the abstract, it's a plugin silently failing to render
+under test (confirmed on ledmatrix-leaderboard, which calls
+get_cached_data_with_strategy() and previously hit an AttributeError that
+its own broad except swallowed, producing an empty-but-green render).
+"""
+
+from src.plugin_system.testing.mocks import MockCacheManager
+
+
+class TestMockCacheManagerStrategyMethod:
+    def test_get_cached_data_with_strategy_returns_cached_value(self):
+        cm = MockCacheManager()
+        cm.set("standings_nfl", {"teams": ["KC", "BUF"]})
+        result = cm.get_cached_data_with_strategy("standings_nfl", "sports_live")
+        assert result == {"teams": ["KC", "BUF"]}
+
+    def test_get_cached_data_with_strategy_returns_none_when_missing(self):
+        cm = MockCacheManager()
+        assert cm.get_cached_data_with_strategy("missing_key") is None
+
+    def test_get_cached_data_with_strategy_defaults_data_type(self):
+        cm = MockCacheManager()
+        cm.set("k", "v")
+        assert cm.get_cached_data_with_strategy("k") == "v"
+
+    def test_calls_are_tracked(self):
+        cm = MockCacheManager()
+        cm.get_cached_data_with_strategy("k", "sports_live")
+        assert cm.get_cached_data_with_strategy_calls == [{"key": "k", "data_type": "sports_live"}]
+
+    def test_save_cache_is_readable_via_strategy_lookup(self):
+        cm = MockCacheManager()
+        cm.save_cache("standings_nfl", {"teams": ["KC", "BUF"]})
+        assert cm.get_cached_data_with_strategy("standings_nfl") == {"teams": ["KC", "BUF"]}
+
+    def test_reset_clears_strategy_call_tracking(self):
+        cm = MockCacheManager()
+        cm.get_cached_data_with_strategy("k", "sports_live")
+        cm.reset()
+        assert cm.get_cached_data_with_strategy_calls == []

test/test_wifi_check_state.py

@@ -0,0 +1,92 @@
+"""Tests for check_and_manage_ap_mode_with_state (src/wifi_manager.py).
+
+The wifi monitor daemon used to fetch WiFi status before AND after each
+check on top of the check's own internal fetch — every fetch is several
+nmcli subprocess forks, every 30s, forever. The new API returns the state
+the check observed, so the daemon runs exactly one fetch battery per tick.
+"""
+
+import os
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.wifi_manager import WiFiManager, WiFiStatus  # noqa: E402
+
+
+@pytest.fixture
+def wm(tmp_path):
+    with patch.object(WiFiManager, "_load_config", return_value={}, create=True):
+        manager = WiFiManager.__new__(WiFiManager)
+        # minimal attribute setup without running the real __init__
+        manager.config = {"auto_enable_ap_mode": True}
+        manager._disconnected_checks = 0
+        manager._disconnected_checks_required = 3
+        manager._ap_enabled_at = None
+        return manager
+
+
+def _wire(wm, connected, ethernet, ap_active):
+    wm._get_wifi_status_with_retry = MagicMock(
+        return_value=WiFiStatus(connected=connected, ssid="net" if connected else None))
+    wm._is_ethernet_connected = MagicMock(return_value=ethernet)
+    wm._is_ap_mode_active = MagicMock(return_value=ap_active)
+    wm.enable_ap_mode = MagicMock(return_value=(True, "ok"))
+    wm.disable_ap_mode = MagicMock(return_value=(True, "ok"))
+    wm.scan_networks = MagicMock(return_value=([], False))
+    wm._save_cached_scan = MagicMock()
+    wm._FORCE_AP_FLAG_PATH = MagicMock()
+    wm._FORCE_AP_FLAG_PATH.exists.return_value = False
+
+
+class TestWithState:
+    def test_single_fetch_per_call(self, wm):
+        _wire(wm, connected=True, ethernet=False, ap_active=False)
+        wm.check_and_manage_ap_mode_with_state()
+        assert wm._get_wifi_status_with_retry.call_count == 1
+        assert wm._is_ethernet_connected.call_count == 1
+        assert wm._is_ap_mode_active.call_count == 1
+
+    def test_returns_observed_state(self, wm):
+        _wire(wm, connected=True, ethernet=True, ap_active=False)
+        changed, status, ethernet, ap_after = wm.check_and_manage_ap_mode_with_state()
+        assert changed is False
+        assert status.connected is True
+        assert ethernet is True
+        assert ap_after is False
+
+    def test_ap_after_inverts_on_disable(self, wm):
+        """WiFi reconnects while AP is up -> auto-disable -> ap_after False."""
+        _wire(wm, connected=True, ethernet=False, ap_active=True)
+        changed, _status, _ethernet, ap_after = wm.check_and_manage_ap_mode_with_state()
+        assert changed is True
+        assert ap_after is False
+        wm.disable_ap_mode.assert_called_once()
+
+    def test_ap_after_inverts_on_enable(self, wm):
+        """Grace period exhausted with nothing connected -> enable -> True."""
+        _wire(wm, connected=False, ethernet=False, ap_active=False)
+        wm._disconnected_checks = 2  # this call is the 3rd
+        changed, _status, _ethernet, ap_after = wm.check_and_manage_ap_mode_with_state()
+        assert changed is True
+        assert ap_after is True
+        wm.enable_ap_mode.assert_called_once()
+
+    def test_bool_wrapper_is_back_compatible(self, wm):
+        _wire(wm, connected=True, ethernet=False, ap_active=False)
+        assert wm.check_and_manage_ap_mode() is False
+        _wire(wm, connected=True, ethernet=False, ap_active=True)
+        assert wm.check_and_manage_ap_mode() is True
+
+    def test_exception_path_never_raises(self, wm):
+        wm._get_wifi_status_with_retry = MagicMock(side_effect=RuntimeError("nmcli gone"))
+        changed, status, _ethernet, _ap_after = wm.check_and_manage_ap_mode_with_state()
+        assert changed is False
+        assert status.connected is False
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-v"]))

web_interface/static/v3/js/widgets/custom-feeds.js

@@ -404,9 +404,15 @@
         if (!file) return;
         
         const formData = new FormData();
-        formData.append('file', file);
+        // Backend contract (see api_v3.upload_plugin_asset): the request
+        // field must be named "files" (it does request.files.getlist('files')
+        // and 400s with "No files provided" otherwise), and the response
+        // carries the result in a top-level "uploaded_files" key, not nested
+        // under "data". file-upload-single.js's working upload flow uses this
+        // same contract.
+        formData.append('files', file);
         formData.append('plugin_id', pluginId);
-        
+
         fetch('/api/v3/plugins/assets/upload', {
             method: 'POST',
             body: formData
@@ -421,8 +427,8 @@
             return response.json();
         })
         .then(data => {
-            if (data.status === 'success' && data.data && data.data.files && data.data.files.length > 0) {
-                const uploadedFile = data.data.files[0];
+            if (data.status === 'success' && data.uploaded_files && data.uploaded_files.length > 0) {
+                const uploadedFile = data.uploaded_files[0];
                 const row = document.querySelector(`#${fieldId}_tbody tr[data-index="${index}"]`);
                 if (row) {
                     const logoCell = row.querySelector('td:nth-child(3)');