perf: optimize resolve workflow and plugin listing (#857)

## Summary
- **Skip update checks in resolve_workflow**: `resolve_workflow()` only
needs plugin `name` and `version`, not update availability. Added
`skip_update_check` parameter to `list_plugins_sync()` — resolve goes
from N × (up to 60s) subprocess calls to near-instant.
- **TTL cache on `_check_plugin_update()`**: Results are cached for 10
minutes so repeated Nodes panel opens or plugin listings reuse cached
results instead of spawning `uv pip compile` subprocesses.
- **Pre-warm cache at startup**: A background task calls
`list_plugins_sync()` during app startup so the cache is warm by the
time the user first opens Nodes.
- **Cache invalidation**: `_invalidate_plugin_caches()` (called on
install/uninstall) also clears the update check TTL cache.

## Test plan
- [ ] Open the app, click "Nodes" — should load quickly (cache warmed at
startup)
- [ ] Import a workflow via "Resolve Workflow" — should resolve
near-instantly
- [ ] Install/uninstall a plugin, then reopen Nodes — should show fresh
update info
- [ ] Wait 10+ minutes and reopen Nodes — should re-fetch update info
(TTL expired)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Signed-off-by: Rafał Leszko <rafal@livepeer.org>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: leszko

src/scope/core/plugins/manager.py

@@ -8,6 +8,7 @@
 import subprocess
 import sys
 import threading
+import time
 from dataclasses import dataclass
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
@@ -113,6 +114,17 @@ def __init__(self):
         # Cache for bundled plugin names (file is immutable at runtime)
         self._bundled_package_names: set[str] | None = None
 
+        # TTL cache for plugin update checks: {name: (result_dict, timestamp)}
+        self._update_check_cache: dict[str, tuple[dict[str, Any], float]] = {}
+        self._update_check_ttl: float = 600.0  # 10 minutes
+
+    def clear_update_check_cache(self) -> None:
+        """Clear the TTL cache for plugin update checks.
+
+        Should be called after plugin install/uninstall/upgrade.
+        """
+        self._update_check_cache.clear()
+
     def _read_plugins_file(self) -> list[str]:
         """Read plugin specifiers from plugins.txt."""
         plugins_file = get_plugins_file()
@@ -665,16 +677,22 @@ def _get_plugin_source(self, dist: Any) -> tuple[str, bool, str | None, str | No
         # Default to PyPI
         return ("pypi", False, None, None)
 
-    async def list_plugins_async(self) -> list[dict[str, Any]]:
+    async def list_plugins_async(
+        self, *, skip_update_check: bool = False
+    ) -> list[dict[str, Any]]:
         """Get all installed plugins with metadata.
 
         Returns:
             List of plugin info dictionaries
         """
         loop = asyncio.get_event_loop()
-        return await loop.run_in_executor(None, self.list_plugins_sync)
+        return await loop.run_in_executor(
+            None, lambda: self.list_plugins_sync(skip_update_check=skip_update_check)
+        )
 
-    def list_plugins_sync(self) -> list[dict[str, Any]]:
+    def list_plugins_sync(
+        self, *, skip_update_check: bool = False
+    ) -> list[dict[str, Any]]:
         """Synchronous implementation of list_plugins."""
         from importlib.metadata import distributions
 
@@ -722,7 +740,7 @@ def list_plugins_sync(self) -> list[dict[str, Any]]:
                     )
 
                     # Check for updates (skip local/editable plugins)
-                    if source == "local" or editable:
+                    if skip_update_check or source == "local" or editable:
                         latest_version = None
                         update_available = None
                     else:
@@ -779,6 +797,9 @@ def _check_plugin_update(
         Compares current resolved.txt with a fresh compile using --upgrade-package
         to find if a newer version is available that respects project constraints.
 
+        Results are cached for ``_update_check_ttl`` seconds to avoid repeated
+        expensive subprocess calls.
+
         Args:
             name: Package name (used for version lookup)
             package_spec: Package specifier (not used in compile approach, kept for API compat)
@@ -788,14 +809,25 @@ def _check_plugin_update(
         """
         import tempfile
 
+        # Return cached result if still fresh
+        cached = self._update_check_cache.get(name)
+        if cached is not None:
+            result_dict, timestamp = cached
+            if time.monotonic() - timestamp < self._update_check_ttl:
+                return result_dict
+
         resolved_file = get_resolved_file()
 
         # Get current version from resolved.txt (if it exists)
         current_version = self._get_version_from_resolved(name, str(resolved_file))
 
+        def _cache_and_return(r: dict[str, Any]) -> dict[str, Any]:
+            self._update_check_cache[name] = (r, time.monotonic())
+            return r
+
         # If no resolved file exists, we can't check for updates via compile
         if not resolved_file.exists():
-            return {"latest_version": None, "update_available": None}
+            return _cache_and_return({"latest_version": None, "update_available": None})
 
         # Create temp file for upgrade check
         try:
@@ -809,7 +841,9 @@ def _check_plugin_update(
             pyproject = project_root / "pyproject.toml"
 
             if not pyproject.exists():
-                return {"latest_version": None, "update_available": None}
+                return _cache_and_return(
+                    {"latest_version": None, "update_available": None}
+                )
 
             args = [
                 "uv",
@@ -841,19 +875,25 @@ def _check_plugin_update(
             )
 
             if result.returncode != 0:
-                return {"latest_version": None, "update_available": None}
+                return _cache_and_return(
+                    {"latest_version": None, "update_available": None}
+                )
 
             # Get new version from temp resolved file
             new_version = self._get_version_from_resolved(name, temp_resolved)
 
             if new_version and new_version != current_version:
-                return {"latest_version": new_version, "update_available": True}
+                return _cache_and_return(
+                    {"latest_version": new_version, "update_available": True}
+                )
 
-            return {"latest_version": None, "update_available": False}
+            return _cache_and_return(
+                {"latest_version": None, "update_available": False}
+            )
 
         except Exception as e:
             logger.warning(f"Failed to check updates for {name}: {e}")
-            return {"latest_version": None, "update_available": None}
+            return _cache_and_return({"latest_version": None, "update_available": None})
         finally:
             Path(temp_resolved).unlink(missing_ok=True)
 

src/scope/core/workflows/resolve.py

@@ -285,7 +285,7 @@ def resolve_workflow(
     all_pipelines_ok = True
 
     try:
-        plugins = plugin_manager.list_plugins_sync()
+        plugins = plugin_manager.list_plugins_sync(skip_update_check=True)
     except Exception:
         logger.warning(
             "Failed to list plugins; treating all plugins as missing", exc_info=True

src/scope/server/app.py

@@ -131,6 +131,14 @@ def _invalidate_plugin_caches():
     _pipeline_schemas_cache = None
     _plugins_list_cache = None
 
+    # Also clear the plugin manager's per-plugin update check TTL cache
+    try:
+        from scope.core.plugins import get_plugin_manager
+
+        get_plugin_manager().clear_update_check_cache()
+    except Exception:
+        pass
+
 
 class STUNErrorFilter(logging.Filter):
     """Filter to suppress STUN/TURN connection errors that are not critical."""
@@ -328,6 +336,19 @@ async def prewarm_pipeline(pipeline_id: str):
         logger.error(f"Error pre-warming pipeline {pipeline_id} in background: {e}")
 
 
+async def _prewarm_plugin_update_cache():
+    """Background task to warm the plugin update check cache at startup."""
+    try:
+        from scope.core.plugins import get_plugin_manager
+
+        pm = get_plugin_manager()
+        loop = asyncio.get_event_loop()
+        await loop.run_in_executor(None, pm.list_plugins_sync)
+        logger.info("Plugin update check cache warmed")
+    except Exception as e:
+        logger.debug(f"Plugin update cache warm-up skipped: {e}")
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """Lifespan handler for startup and shutdown events."""
@@ -384,6 +405,10 @@ async def lifespan(app: FastAPI):
     if PIPELINE is not None:
         asyncio.create_task(prewarm_pipeline(PIPELINE))
 
+    # Pre-warm the plugin update check cache in the background so the first
+    # "Nodes" / resolve-workflow call doesn't block on PyPI lookups.
+    asyncio.create_task(_prewarm_plugin_update_cache())
+
     webrtc_manager = WebRTCManager()
     logger.info("WebRTC manager initialized")