test: sync mock cache manager signature with CacheManager.get

test_circuit_breaker has been failing on main: plugin_health passes
memory_ttl= to cache_manager.get(), and the conftest mock's signature
was never updated — the same component/double drift class as the
monitored_update bug (#392).

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

src/common/permission_utils.py

@@ -8,6 +8,7 @@ files that need to be accessible by both root service and web user.
 
 import os
 import logging
+import re
 import shutil as _shutil
 import subprocess
 import sys
@@ -16,6 +17,25 @@ from typing import Optional
 
 logger = logging.getLogger(__name__)
 
+# Matches the credentials portion of a "scheme://user:pass@host" URL, so pip's
+# own error output can be logged/displayed without echoing back a private
+# index URL's embedded basic-auth secret verbatim (e.g. from a
+# requirements.txt --index-url line or the PIP_INDEX_URL env var).
+_URL_CREDENTIALS_RE = re.compile(r'://[^/\s@:]+:[^/\s@]+@')
+
+
+def _redact_url_credentials(text: Optional[str]) -> str:
+    """Replace embedded user:pass@ URL credentials in text with a placeholder.
+
+    Safe to call on any subprocess output destined for logs: it only ever
+    shortens/replaces the credential substring, never changes the presence
+    or absence of the specific fixed phrases callers check for
+    (e.g. "a password is required"), so it can't affect control flow.
+    """
+    if not text:
+        return text or ""
+    return _URL_CREDENTIALS_RE.sub('://***:***@', text)
+
 # System directories that should never have their permissions modified
 # These directories have special system-level permissions that must be preserved
 PROTECTED_SYSTEM_DIRECTORIES = {  # nosec B108 - these are checked to PREVENT permission changes, not to use as temp paths
@@ -338,6 +358,13 @@ def install_requirements_file(req_file: Path, timeout: int = 300) -> subprocess.
                 ["sudo", "-n", bash_path, str(wrapper), str(req_file)],
                 capture_output=True, text=True, timeout=timeout, cwd=str(project_root)
             )
+            # Redact immediately: pip can echo a private index URL's embedded
+            # basic-auth credentials back in its own error/progress output
+            # (e.g. from a requirements.txt --index-url line). Doesn't affect
+            # the fixed-phrase "denied" check below -- those phrases never
+            # overlap with URL syntax.
+            result.stderr = _redact_url_credentials(result.stderr)
+            result.stdout = _redact_url_credentials(result.stdout)
             if result.returncode == 0:
                 return result
             # Distinguish "sudo rejected this exact command line" (worth
@@ -348,16 +375,24 @@ def install_requirements_file(req_file: Path, timeout: int = 300) -> subprocess.
                 for phrase in ("a password is required", "is not allowed to run", "no tty present")
             )
             if not denied:
+                # Deliberately don't interpolate req_file or the pip output here:
+                # this log line is scanner-visible, and a static analyzer can't
+                # tell "already redacted above" from "still raw" just by looking
+                # at this call in isolation. The full (redacted) text is still
+                # available to callers via the returned CompletedProcess.
                 logger.warning(
-                    "Root pip install failed (rc=%s) for %s: %s",
-                    result.returncode, req_file, result.stderr.strip()[:500],
+                    "Root pip install failed (rc=%s); see the returned "
+                    "CompletedProcess.stderr for details.",
+                    result.returncode,
                 )
                 return result
 
+        # Same reasoning as above: no req_file / pip-output interpolation in
+        # this log line, only in the returned note/CompletedProcess.
         logger.warning(
-            "Root pip install wrapper denied via sudo for %s; falling back to "
-            "user-level install: %s",
-            req_file, result.stderr.strip()[:500] if result else "no bash candidates found",
+            "Root pip install wrapper denied via sudo for all candidates; "
+            "falling back to user-level install. See the returned "
+            "CompletedProcess.stderr for details."
         )
         note = (
             f"[Root install unavailable ({(result.stderr.strip() if result else 'sudo denied') or 'sudo denied'}); "
@@ -367,8 +402,7 @@ def install_requirements_file(req_file: Path, timeout: int = 300) -> subprocess.
         )
     else:
         logger.warning(
-            "safe_pip_install.sh not found; falling back to user-level install for %s",
-            req_file,
+            "safe_pip_install.sh not found; falling back to user-level install."
         )
         note = (
             "[safe_pip_install.sh not found; installed for the current process's "
@@ -386,6 +420,7 @@ def install_requirements_file(req_file: Path, timeout: int = 300) -> subprocess.
         [sys.executable, "-m", "pip", "install", "--break-system-packages", "--ignore-installed", "-r", str(req_file)],
         capture_output=True, text=True, timeout=timeout, cwd=str(project_root)
     )
-    result.stdout = note + (result.stdout or "")
+    result.stderr = _redact_url_credentials(result.stderr)
+    result.stdout = note + _redact_url_credentials(result.stdout)
     return result
 

src/display_manager.py

@@ -33,7 +33,8 @@ else:
 from contextlib import contextmanager
 from PIL import Image, ImageDraw, ImageFont
 import time
-from typing import Dict, Any, List, Optional
+from collections import OrderedDict
+from typing import Dict, Any, List, Optional, Tuple
 import logging
 import math
 import freetype
@@ -180,14 +181,25 @@ class DisplayManager:
         # the logical image is blitted to the matrix unchanged.
         self._double_sided = None  # dict {copies, axis, logical_width, logical_height} or None
         self._physical_image = None  # full-chain buffer reused each frame when tiling
-        # Text-width measurement cache: (text, id(font)) -> pixel_width
+        # Text-width measurement cache: (text, id(font)) -> (width, font_ref)
         # Avoids re-measuring the same string+font on every display() call.
+        # LRU-bounded: keys embed the TEXT, so changing strings (a clock, a
+        # live score) would otherwise grow it forever on a 24/7 service.
+        # Entries hold a strong reference to the font so its id() can't be
+        # recycled by a different font object — an id-keyed cache without
+        # the reference can return the WRONG width after garbage collection.
         # Cleared on _load_fonts() so stale entries don't survive a font reload.
-        self._text_width_cache: Dict[tuple, int] = {}
+        self._text_width_cache: "OrderedDict[tuple, Tuple[int, Any]]" = OrderedDict()
+        self._TEXT_WIDTH_CACHE_MAX = 1024
         # Snapshot settings for web preview integration (service writes, web reads)
         self._snapshot_path = "/tmp/led_matrix_preview.png"  # nosec B108 - fixed path intentional; web UI reads same path
         self._snapshot_min_interval_sec = 0.2  # max ~5 fps
         self._last_snapshot_ts = 0.0
+        # Snapshot failures are logged as warnings, rate-limited so a
+        # persistent failure (e.g. an unwritable file) can't spam the log —
+        # 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
         
         # Scrolling state tracking for graceful updates
         self._scrolling_state = {
@@ -699,12 +711,15 @@ class DisplayManager:
 
         Results are cached by (text, font identity) so plugins that measure
         the same string every frame (e.g. to centre a score) pay only one
-        measurement per unique (text, font) pair.
+        measurement per unique (text, font) pair. The entry keeps the font
+        alive so its id() can't be recycled, and the cache is LRU-bounded so
+        ever-changing text (clocks, tickers) can't grow it without limit.
         """
         cache_key = (text, id(font))
         cached = self._text_width_cache.get(cache_key)
         if cached is not None:
-            return cached
+            self._text_width_cache.move_to_end(cache_key)
+            return cached[0]
 
         try:
             if isinstance(font, freetype.Face):
@@ -719,7 +734,9 @@ class DisplayManager:
             logger.error("Error getting text width: %s", e)
             return 0
 
-        self._text_width_cache[cache_key] = width
+        self._text_width_cache[cache_key] = (width, font)
+        while len(self._text_width_cache) > self._TEXT_WIDTH_CACHE_MAX:
+            self._text_width_cache.popitem(last=False)
         return width
 
     def get_font_height(self, font):
@@ -1164,5 +1181,15 @@ class DisplayManager:
                 pass
             self._last_snapshot_ts = now
         except Exception as e:
-            # Snapshot failures should never break display; log at debug to avoid noise
-            logger.debug(f"Snapshot write skipped: {e}")
+            # Snapshot failures must never break display — but they must not
+            # be silent either: the snapshot's mtime is the web UI's display
+            # mirror AND its hardware-liveness proxy, so a quietly failing
+            # write freezes the mirror and makes health checks lie (seen in
+            # the field: a stale root-owned /tmp file froze it for a day).
+            # Warn at most once per 5 minutes to avoid log spam.
+            if (now - self._snapshot_fail_log_ts) > 300:
+                self._snapshot_fail_log_ts = now
+                logger.warning("Snapshot write failing (web preview/health "
+                               "mirror is stale): %s", e)
+            else:
+                logger.debug(f"Snapshot write skipped: {e}")

src/font_manager.py

@@ -35,6 +35,7 @@ import urllib.request
 import zipfile
 import tempfile
 import time
+from collections import OrderedDict
 from pathlib import Path
 from PIL import ImageFont
 from typing import Dict, Tuple, Optional, Union, Any, List
@@ -58,7 +59,13 @@ class FontManager:
         # Font discovery and catalog
         self.font_catalog: Dict[str, str] = {}  # family_name -> file_path
         self.font_cache: Dict[str, Union[ImageFont.FreeTypeFont, freetype.Face]] = {}  # (family, size) -> font
-        self.metrics_cache: Dict[str, Tuple[int, int, int]] = {}  # (text, font_id) -> (width, height, baseline)
+        # (text, id(font)) -> ((width, height, baseline), font_ref).
+        # LRU-bounded — keys embed the measured TEXT, so changing strings
+        # (clocks, live scores) would otherwise grow it forever. Entries
+        # keep the font alive so its id() can't be recycled by a different
+        # font object (which would silently return wrong metrics).
+        self.metrics_cache: "OrderedDict[Any, Tuple[Tuple[int, int, int], Any]]" = OrderedDict()
+        self._METRICS_CACHE_MAX = 1024
 
         # Plugin font management
         self.plugin_fonts: Dict[str, Dict[str, Any]] = {}  # plugin_id -> font_manifest
@@ -507,10 +514,14 @@ class FontManager:
         Returns:
             Tuple of (width, height, baseline_offset)
         """
-        cache_key = f"{hash(text)}_{id(font)}"
+        # Key on the text itself (hash(text) could collide) + font identity;
+        # the entry below keeps the font referenced so the id stays valid.
+        cache_key = (text, id(font))
 
-        if cache_key in self.metrics_cache:
-            return self.metrics_cache[cache_key]
+        cached = self.metrics_cache.get(cache_key)
+        if cached is not None:
+            self.metrics_cache.move_to_end(cache_key)
+            return cached[0]
 
         try:
             if isinstance(font, freetype.Face):
@@ -547,7 +558,9 @@ class FontManager:
             baseline = 10
 
         result = (width, height, baseline)
-        self.metrics_cache[cache_key] = result
+        self.metrics_cache[cache_key] = (result, font)
+        while len(self.metrics_cache) > self._METRICS_CACHE_MAX:
+            self.metrics_cache.popitem(last=False)
         return result
 
     def get_font_height(self, font: Union[ImageFont.FreeTypeFont, freetype.Face]) -> int:

src/plugin_system/plugin_loader.py

@@ -93,6 +93,27 @@ def requirements_are_satisfied(requirements_file: str) -> bool:
     return True
 
 
+def find_trusted_subdir(trusted_dir: str, name: str) -> Optional[str]:
+    """Return `name` if it names an actual subdirectory of trusted_dir, else None.
+
+    Used as a containment check for a directory name derived from untrusted
+    input (a manifest-declared plugin id, an externally-supplied plugin
+    path): the returned value always comes from enumerating trusted_dir
+    itself via os.scandir(), so a caller that builds a path by joining
+    trusted_dir with this return value is joining against a name the
+    filesystem produced under a trusted root -- not the caller's original
+    string, which could otherwise smuggle a traversal sequence through.
+    """
+    try:
+        with os.scandir(trusted_dir) as entries:
+            for entry in entries:
+                if entry.name == name and entry.is_dir():
+                    return entry.name
+    except OSError:
+        pass
+    return None
+
+
 class PluginLoader:
     """Handles plugin module loading and class instantiation."""
 
@@ -200,14 +221,14 @@ class PluginLoader:
                 except (json.JSONDecodeError, Exception) as e:
                     self.logger.debug("Skipping %s due to manifest error: %s", item.name, e)
                     continue
-        
+
         return None
-    
+
     def install_dependencies(
         self,
         plugin_dir: Path,
         plugin_id: str,
-        plugins_dir: Optional[Path] = None,
+        plugins_dir: Path,
         timeout: int = 300
     ) -> bool:
         """
@@ -216,7 +237,12 @@ class PluginLoader:
         Args:
             plugin_dir: Plugin directory path
             plugin_id: Plugin identifier
-            plugins_dir: Trusted base plugins directory for path containment check
+            plugins_dir: Trusted base plugins directory for path containment check.
+                Required (not optional) so every caller reconstructs the plugin
+                path through the sanitiser below rather than trusting plugin_dir
+                directly -- CodeQL's path-injection query (and a malicious
+                manifest/plugin_id in practice) can't tell a legitimate
+                plugin_dir from one crafted to traverse outside plugins_dir.
             timeout: Installation timeout in seconds
 
         Returns:
@@ -228,31 +254,24 @@ class PluginLoader:
 
         # Resolve to a canonical absolute path (normalises .. and symlinks)
         plugin_dir_real = os.path.realpath(str(plugin_dir))
+        plugins_dir_real = os.path.realpath(str(plugins_dir))
+        requested_name = os.path.basename(plugin_dir_real)
 
-        if plugins_dir is not None:
-            # Reconstruct the plugin path from a trusted base + a sanitised
-            # directory name.  os.path.basename() is CodeQL's recognised
-            # py/path-injection sanitiser: it strips all directory components
-            # so the result cannot contain traversal sequences.  Joining it
-            # with the resolved, trusted plugins_dir produces a path that
-            # CodeQL considers untainted.
-            plugins_dir_real = os.path.realpath(str(plugins_dir))
-            safe_dir_name = os.path.basename(plugin_dir_real)
-            if not safe_dir_name:
-                self.logger.error("Could not determine plugin directory name for %s", plugin_id)
-                return False
-            safe_plugin_dir = os.path.join(plugins_dir_real, safe_dir_name)
-            if not os.path.isdir(safe_plugin_dir):
-                self.logger.error(
-                    "Plugin directory for %s not found inside plugins dir", plugin_id
-                )
-                return False
-        else:
-            safe_plugin_dir = plugin_dir_real
-            if not os.path.isdir(safe_plugin_dir):
-                self.logger.error("Plugin directory does not exist: %s", plugin_dir)
-                return False
+        # Match the requested directory against an entry actually enumerated
+        # from the trusted plugins_dir, and build the path from that entry --
+        # not from requested_name. A name that came out of os.scandir() on a
+        # trusted root carries no taint regardless of what the caller asked
+        # for, so this is a real containment guarantee (an allowlist check
+        # against a trusted source), not a string-sanitisation of untrusted
+        # input that a static analyzer has to trust blindly.
+        matched_name = find_trusted_subdir(plugins_dir_real, requested_name)
+        if matched_name is None:
+            self.logger.error(
+                "Plugin directory for %s not found inside plugins dir", plugin_id
+            )
+            return False
 
+        safe_plugin_dir = os.path.join(plugins_dir_real, matched_name)
         requirements_file = os.path.join(safe_plugin_dir, "requirements.txt")
 
         if not os.path.isfile(requirements_file):
@@ -698,6 +717,14 @@ class PluginLoader:
         """
         # Install dependencies if needed
         if install_deps:
+            if plugins_dir is None:
+                raise PluginError(
+                    f"plugins_dir is required to install dependencies for plugin {plugin_id} "
+                    "(needed for path containment; pass install_deps=False if the caller "
+                    "doesn't have a trusted plugins directory to supply)",
+                    plugin_id=plugin_id,
+                    context={'plugin_dir': str(plugin_dir)},
+                )
             if not self.install_dependencies(plugin_dir, plugin_id, plugins_dir=plugins_dir):
                 raise PluginError(
                     f"Dependency installation failed for plugin {plugin_id} in {plugin_dir}",

src/plugin_system/plugin_manager.py

@@ -11,6 +11,7 @@ import json
 import sys
 import time
 import threading
+import types
 from pathlib import Path
 from typing import Dict, List, Optional, Any
 import logging
@@ -743,8 +744,18 @@ class PluginManager:
                         # 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(
-                            type('obj', (object,), {'update': monitored_update})(),
+                            types.SimpleNamespace(update=monitored_update),
                             plugin_id
                         )
                     else:

src/plugin_system/store_manager.py

@@ -25,7 +25,9 @@ import logging
 from urllib.parse import urlparse
 
 from src.common.permission_utils import sudo_remove_directory, install_requirements_file
-from src.plugin_system.plugin_loader import requirements_has_real_deps, requirements_are_satisfied
+from src.plugin_system.plugin_loader import (
+    requirements_has_real_deps, requirements_are_satisfied, find_trusted_subdir
+)
 
 try:
     from jsonschema import Draft7Validator, ValidationError
@@ -1900,15 +1902,33 @@ class PluginStoreManager:
     def _install_dependencies(self, plugin_path: Path) -> bool:
         """
         Install Python dependencies from requirements.txt.
-        
+
         Args:
             plugin_path: Path to plugin directory
-            
+
         Returns:
             True if successful or no requirements file
         """
-        requirements_file = plugin_path / "requirements.txt"
-        
+        # Reconstruct the plugin path from the trusted self.plugins_dir base +
+        # an entry actually enumerated from it, rather than trusting
+        # plugin_path directly -- callers ultimately derive it from a
+        # plugin-supplied manifest "id" field (see install_plugin_from_url),
+        # so without this a malicious manifest could point requirements_file
+        # outside plugins_dir. find_trusted_subdir()'s return value always
+        # comes from os.scandir() on the trusted root, so building the path
+        # from it (not from the caller's string) is a real containment
+        # guarantee, matching the pattern in PluginLoader.install_dependencies().
+        plugin_dir_real = os.path.realpath(str(plugin_path))
+        plugins_dir_real = os.path.realpath(str(self.plugins_dir))
+        requested_name = os.path.basename(plugin_dir_real)
+        matched_name = find_trusted_subdir(plugins_dir_real, requested_name)
+        if matched_name is None:
+            self.logger.error("Plugin directory not found inside plugins dir for dependency install")
+            return False
+        safe_plugin_path = Path(os.path.join(plugins_dir_real, matched_name))
+
+        requirements_file = safe_plugin_path / "requirements.txt"
+
         if not requirements_file.exists():
             self.logger.debug(f"No requirements.txt found in {plugin_path.name}")
             return True

src/plugin_system/testing/visual_display_manager.py

@@ -454,6 +454,18 @@ class VisualTestDisplayManager:
         """Check if display is currently scrolling."""
         return self._scrolling_state['is_scrolling']
 
+    def process_deferred_updates(self):
+        """Process any deferred updates (no-op for testing).
+
+        Several ticker-style plugins (news, odds-ticker, leaderboard,
+        stock-news, stocks) call this unconditionally between
+        set_scrolling_state() and their scroll-position update, mirroring the
+        real display_manager's deferred-update queue. This double has no such
+        queue, so there is nothing to process — the no-op just lets those
+        plugins render under the harness instead of raising AttributeError.
+        """
+        pass
+
     # ------------------------------------------------------------------
     # Utility methods
     # ------------------------------------------------------------------

test/conftest.py

@@ -38,7 +38,11 @@ def mock_cache_manager():
     mock._memory_cache_timestamps = {}
     mock.cache_dir = "/tmp/test_cache"
     
-    def mock_get(key: str, max_age: int = 300) -> Optional[Dict]:
+    def mock_get(key: str, max_age: Optional[int] = 300,
+                 memory_ttl: Optional[int] = None) -> Optional[Dict]:
+        # Signature mirrors CacheManager.get — keep in sync or callers
+        # passing keyword args (health tracker, resource monitor) break
+        # only in tests, hiding real-API compatibility.
         return mock._memory_cache.get(key)
     
     def mock_set(key: str, data: Dict, ttl: Optional[int] = None) -> None:

test/plugins/test_visual_rendering.py

@@ -172,6 +172,16 @@ class TestVisualDisplayManager:
         vdm.set_scrolling_state(False)
         assert vdm.is_currently_scrolling() is False
 
+    def test_process_deferred_updates_is_noop(self):
+        # Ticker-style plugins (news, odds-ticker, leaderboard, stock-news,
+        # stocks) call this unconditionally alongside set_scrolling_state();
+        # it must exist and be harmless so those plugins render under the
+        # harness instead of raising AttributeError.
+        vdm = VisualTestDisplayManager(width=128, height=32)
+        vdm.set_scrolling_state(True)
+        vdm.process_deferred_updates()  # should not raise
+        assert vdm.is_currently_scrolling() is True
+
     def test_format_date_with_ordinal(self):
         from datetime import datetime
         vdm = VisualTestDisplayManager(width=128, height=32)

test/test_permission_utils.py

@@ -0,0 +1,83 @@
+"""
+Tests for src.common.permission_utils's URL-credential redaction.
+
+Covers the fix for a CodeQL clear-text-logging-of-secrets alert:
+install_requirements_file() must never let a private index URL's embedded
+user:pass@ credentials reach logs or its returned CompletedProcess, since
+pip can echo that URL back verbatim in its own stderr/stdout on failure.
+"""
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from src.common.permission_utils import _redact_url_credentials, install_requirements_file
+
+
+class TestRedactUrlCredentials:
+    def test_redacts_embedded_basic_auth(self):
+        text = "Could not fetch URL https://alice:s3cr3t@pypi.example.com/simple/: 403"
+        redacted = _redact_url_credentials(text)
+        assert "s3cr3t" not in redacted
+        assert "alice" not in redacted
+        assert "https://***:***@pypi.example.com/simple/" in redacted
+
+    def test_leaves_credential_free_text_unchanged(self):
+        text = "ERROR: Could not find a version that satisfies the requirement foo==1.0"
+        assert _redact_url_credentials(text) == text
+
+    def test_handles_none_and_empty(self):
+        assert _redact_url_credentials(None) == ""
+        assert _redact_url_credentials("") == ""
+
+    def test_does_not_touch_denied_check_phrases(self):
+        """The fixed phrases install_requirements_file greps for must survive
+        redaction untouched -- they don't overlap with URL syntax, but this
+        pins that assumption so a regex change can't silently break it."""
+        text = "sudo: a password is required"
+        assert _redact_url_credentials(text) == text
+
+
+class TestInstallRequirementsFileRedaction:
+    @patch('src.common.permission_utils.subprocess.run')
+    def test_wrapper_path_redacts_stderr_and_stdout(self, mock_run, tmp_path):
+        """safe_pip_install.sh exists in this repo, so install_requirements_file
+        takes the sudo-wrapper branch; a failing result must come back
+        with any embedded index-URL credentials already redacted."""
+        req_file = tmp_path / "requirements.txt"
+        req_file.write_text("requests\n")
+
+        mock_run.return_value = MagicMock(
+            returncode=1,
+            stdout="Looking in indexes: https://bob:hunter2@pypi.internal/simple\n",
+            stderr="ERROR https://bob:hunter2@pypi.internal/simple/foo: 401",
+        )
+
+        result = install_requirements_file(req_file, timeout=5)
+
+        assert "hunter2" not in result.stdout
+        assert "hunter2" not in result.stderr
+        assert "https://***:***@pypi.internal" in result.stdout
+        assert "https://***:***@pypi.internal" in result.stderr
+
+    @patch('src.common.permission_utils.subprocess.run')
+    @patch('src.common.permission_utils.Path.exists', return_value=False)
+    def test_no_wrapper_fallback_path_redacts_stderr_and_stdout(self, mock_exists, mock_run, tmp_path):
+        """No safe_pip_install.sh wrapper -> falls straight to the
+        sys.executable pip fallback (the second subprocess.run call site);
+        its result must come back redacted too, independent of the wrapper
+        branch's own redaction above."""
+        req_file = tmp_path / "requirements.txt"
+        req_file.write_text("requests\n")
+
+        mock_run.return_value = MagicMock(
+            returncode=1,
+            stdout="Looking in indexes: https://carol:swordfish@pypi.internal/simple\n",
+            stderr="ERROR https://carol:swordfish@pypi.internal/simple/foo: 401",
+        )
+
+        result = install_requirements_file(req_file, timeout=5)
+
+        assert "swordfish" not in result.stdout
+        assert "swordfish" not in result.stderr
+        assert "https://***:***@pypi.internal" in result.stdout
+        assert "https://***:***@pypi.internal" in result.stderr

test/test_plugin_loader.py

@@ -193,7 +193,7 @@ class TestPluginLoader:
         
         mock_subprocess.return_value = MagicMock(returncode=0)
         
-        result = plugin_loader.install_dependencies(plugin_dir, "test_plugin")
+        result = plugin_loader.install_dependencies(plugin_dir, "test_plugin", plugins_dir=tmp_plugins_dir)
         
         assert result is True
         mock_subprocess.assert_called_once()
@@ -204,7 +204,7 @@ class TestPluginLoader:
         plugin_dir = tmp_plugins_dir / "test_plugin"
         plugin_dir.mkdir()
         
-        result = plugin_loader.install_dependencies(plugin_dir, "test_plugin")
+        result = plugin_loader.install_dependencies(plugin_dir, "test_plugin", plugins_dir=tmp_plugins_dir)
         
         assert result is True
         mock_subprocess.assert_not_called()
@@ -219,7 +219,7 @@ class TestPluginLoader:
 
         mock_subprocess.return_value = MagicMock(returncode=1)
 
-        result = plugin_loader.install_dependencies(plugin_dir, "test_plugin")
+        result = plugin_loader.install_dependencies(plugin_dir, "test_plugin", plugins_dir=tmp_plugins_dir)
 
         assert result is False
 
@@ -245,7 +245,7 @@ class TestPluginLoader:
         retry_attempt = MagicMock(returncode=0, stderr="")
         mock_subprocess.side_effect = [first_attempt, retry_attempt]
 
-        result = plugin_loader.install_dependencies(plugin_dir, "test_plugin")
+        result = plugin_loader.install_dependencies(plugin_dir, "test_plugin", plugins_dir=tmp_plugins_dir)
 
         assert result is True
         assert mock_subprocess.call_count == 2
@@ -271,7 +271,7 @@ class TestPluginLoader:
         retry_attempt = MagicMock(returncode=1, stderr="some other pip error")
         mock_subprocess.side_effect = [first_attempt, retry_attempt]
 
-        result = plugin_loader.install_dependencies(plugin_dir, "test_plugin")
+        result = plugin_loader.install_dependencies(plugin_dir, "test_plugin", plugins_dir=tmp_plugins_dir)
 
         assert result is True
         assert mock_subprocess.call_count == 2
@@ -298,7 +298,7 @@ class TestPluginLoader:
             subprocess.TimeoutExpired(cmd="pip", timeout=300),
         ]
 
-        result = plugin_loader.install_dependencies(plugin_dir, "test_plugin")
+        result = plugin_loader.install_dependencies(plugin_dir, "test_plugin", plugins_dir=tmp_plugins_dir)
 
         assert result is True
         assert mock_subprocess.call_count == 2
@@ -311,7 +311,34 @@ class TestPluginLoader:
         requirements_file = plugin_dir / "requirements.txt"
         requirements_file.write_text("pytest>=1.0\n")
 
-        result = plugin_loader.install_dependencies(plugin_dir, "test_plugin")
+        result = plugin_loader.install_dependencies(plugin_dir, "test_plugin", plugins_dir=tmp_plugins_dir)
 
         assert result is True
         mock_subprocess.assert_not_called()
+
+    def test_install_dependencies_requires_plugins_dir(self, plugin_loader, tmp_plugins_dir):
+        """plugins_dir is a required argument, not an optional trust-me flag --
+        calling without it must fail loudly (TypeError) rather than silently
+        falling back to trusting plugin_dir unchecked."""
+        plugin_dir = tmp_plugins_dir / "test_plugin"
+        plugin_dir.mkdir()
+
+        with pytest.raises(TypeError):
+            plugin_loader.install_dependencies(plugin_dir, "test_plugin")
+
+    @patch('subprocess.run')
+    def test_install_dependencies_rejects_path_outside_plugins_dir(
+        self, mock_subprocess, plugin_loader, tmp_path, tmp_plugins_dir
+    ):
+        """A plugin_dir that doesn't actually live inside plugins_dir (e.g. a
+        manifest-derived id crafted to traverse elsewhere) must be rejected
+        rather than read from -- this is the path-injection containment
+        check CodeQL flagged as missing."""
+        outside_dir = tmp_path / "outside"
+        outside_dir.mkdir()
+        (outside_dir / "requirements.txt").write_text("requests>=2.0\n")
+
+        result = plugin_loader.install_dependencies(outside_dir, "evil_plugin", plugins_dir=tmp_plugins_dir)
+
+        assert result is False
+        mock_subprocess.assert_not_called()

test/test_plugin_system.py

@@ -3,6 +3,7 @@ from unittest.mock import MagicMock, patch
 from pathlib import Path
 from src.plugin_system.plugin_manager import PluginManager
 from src.plugin_system.plugin_state import PluginState
+from src.plugin_system.resource_monitor import PluginResourceMonitor
 
 class TestPluginManager:
     """Test PluginManager functionality."""
@@ -74,10 +75,58 @@ class TestPluginManager:
             
             # No manifest in pm.plugin_manifests
             result = pm.load_plugin("non_existent_plugin")
-            
+
             assert result is False
             assert pm.state_manager.get_state("non_existent_plugin") == PluginState.ERROR
 
+    def test_run_scheduled_updates_calls_update_with_resource_monitor(
+        self, mock_config_manager, mock_display_manager, mock_cache_manager
+    ):
+        """Regression test: run_scheduled_updates() must actually call a
+        plugin's update() when self.resource_monitor is set (as it is in
+        every real deployment -- display_controller.py and web_interface/
+        app.py both assign a real PluginResourceMonitor after construction).
+
+        Previously, the resource_monitor branch wrapped the call in a
+        function stored as a *class* attribute on a dynamically-built type
+        (`type('obj', (object,), {'update': monitored_update})()`), which
+        the descriptor protocol turns into a bound method on access --
+        silently passing the synthetic instance as an implicit first
+        argument to monitored_update(), which takes none. Every plugin's
+        scheduled update failed with "monitored_update() takes 0 positional
+        arguments but 1 was given" and was silently swallowed into a
+        circuit-breaker retry loop that never succeeded, so plugin data
+        (scores, odds, etc.) never refreshed.
+        """
+        with patch('src.plugin_system.plugin_manager.ensure_directory_permissions'):
+            pm = PluginManager(
+                plugins_dir="plugins",
+                config_manager=mock_config_manager,
+                display_manager=mock_display_manager,
+                cache_manager=mock_cache_manager
+            )
+
+            plugin_instance = MagicMock()
+            plugin_instance.enabled = True
+            plugin_instance.update = MagicMock()
+
+            pm.plugins["test_plugin"] = plugin_instance
+            pm.plugin_manifests["test_plugin"] = {"update_interval": 10}
+            pm.state_manager.set_state("test_plugin", PluginState.ENABLED)
+            # Plain MagicMock, not the mock_cache_manager fixture: this test
+            # is about run_scheduled_updates() actually invoking update()
+            # through the resource-monitor wrapper, not about
+            # PluginResourceMonitor's own cache-backed metrics persistence
+            # (which calls cache_manager.get(..., memory_ttl=...) --
+            # a kwarg the fixture's mock_get() doesn't accept).
+            pm.resource_monitor = PluginResourceMonitor(MagicMock())
+
+            pm.run_scheduled_updates(current_time=time.time())
+
+            plugin_instance.update.assert_called_once()
+            assert "test_plugin" in pm.plugin_last_update
+            assert pm.state_manager.get_state("test_plugin") == PluginState.ENABLED
+
 
 class TestPluginLoader:
     """Test PluginLoader functionality."""