Fix LLM validation and dashboard health hot paths

backend/src/services/__tests__/test_health_service.py

@@ -1,6 +1,6 @@
 import pytest
 from datetime import datetime, timedelta
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 from src.services.health_service import HealthService
 from src.models.llm import ValidationRecord
 
@@ -20,6 +20,7 @@ async def test_get_health_summary_aggregation():
     
     # Dashboard 1: Old FAIL, New PASS
     rec1_old = ValidationRecord(
+        id="rec-old",
         dashboard_id="dash_1",
         environment_id="env_1",
         status="FAIL",
@@ -28,6 +29,7 @@ async def test_get_health_summary_aggregation():
         issues=[]
     )
     rec1_new = ValidationRecord(
+        id="rec-new",
         dashboard_id="dash_1",
         environment_id="env_1",
         status="PASS",
@@ -38,6 +40,7 @@ async def test_get_health_summary_aggregation():
     
     # Dashboard 2: Single WARN
     rec2 = ValidationRecord(
+        id="rec-warn",
         dashboard_id="dash_2",
         environment_id="env_1",
         status="WARN",
@@ -69,6 +72,8 @@ async def test_get_health_summary_aggregation():
     dash_1_item = next(item for item in summary.items if item.dashboard_id == "dash_1")
     assert dash_1_item.status == "PASS"
     assert dash_1_item.summary == "New pass"
+    assert dash_1_item.record_id == rec1_new.id
+    assert dash_1_item.dashboard_slug == "dash_1"
 
 @pytest.mark.asyncio
 async def test_get_health_summary_empty():
@@ -84,4 +89,160 @@ async def test_get_health_summary_empty():
     assert summary.pass_count == 0
     assert len(summary.items) == 0
 
-# [/DEF:test_health_service:Module]
+
+@pytest.mark.asyncio
+async def test_get_health_summary_resolves_slug_and_title_from_superset():
+    db = MagicMock()
+    config_manager = MagicMock()
+    config_manager.get_environments.return_value = [MagicMock(id="env_1")]
+
+    record = ValidationRecord(
+        id="rec-1",
+        dashboard_id="42",
+        environment_id="env_1",
+        status="PASS",
+        timestamp=datetime.utcnow(),
+        summary="Healthy",
+        issues=[],
+    )
+    db.query.return_value.join.return_value.all.return_value = [record]
+
+    with patch("src.services.health_service.SupersetClient") as mock_client_cls:
+        mock_client = MagicMock()
+        mock_client.get_dashboards_summary.return_value = [
+            {"id": 42, "slug": "ops-overview", "title": "Ops Overview"}
+        ]
+        mock_client_cls.return_value = mock_client
+
+        service = HealthService(db, config_manager=config_manager)
+        summary = await service.get_health_summary(environment_id="env_1")
+
+    assert summary.items[0].dashboard_slug == "ops-overview"
+    assert summary.items[0].dashboard_title == "Ops Overview"
+    mock_client.get_dashboards_summary.assert_called_once_with()
+
+
+def test_delete_validation_report_deletes_dashboard_scope_and_linked_tasks():
+    db = MagicMock()
+    config_manager = MagicMock()
+    task_manager = MagicMock()
+    task_manager.tasks = {"task-1": object(), "task-2": object(), "task-3": object()}
+
+    target_record = ValidationRecord(
+        id="rec-1",
+        task_id="task-1",
+        dashboard_id="42",
+        environment_id="env_1",
+        status="PASS",
+        timestamp=datetime.utcnow(),
+        summary="Healthy",
+        issues=[],
+        screenshot_path=None,
+    )
+    older_peer = ValidationRecord(
+        id="rec-2",
+        task_id="task-2",
+        dashboard_id="42",
+        environment_id="env_1",
+        status="FAIL",
+        timestamp=datetime.utcnow() - timedelta(hours=1),
+        summary="Older",
+        issues=[],
+        screenshot_path=None,
+    )
+    other_environment = ValidationRecord(
+        id="rec-3",
+        task_id="task-3",
+        dashboard_id="42",
+        environment_id="env_2",
+        status="WARN",
+        timestamp=datetime.utcnow(),
+        summary="Other environment",
+        issues=[],
+        screenshot_path=None,
+    )
+
+    first_query = MagicMock()
+    first_query.first.return_value = target_record
+
+    peer_query = MagicMock()
+    peer_query.filter.return_value = peer_query
+    peer_query.all.return_value = [target_record, older_peer]
+
+    db.query.side_effect = [first_query, peer_query]
+
+    with patch("src.services.health_service.TaskCleanupService") as cleanup_cls:
+        cleanup_instance = MagicMock()
+        cleanup_cls.return_value = cleanup_instance
+
+        service = HealthService(db, config_manager=config_manager)
+        deleted = service.delete_validation_report("rec-1", task_manager=task_manager)
+
+    assert deleted is True
+    assert db.delete.call_count == 2
+    db.delete.assert_any_call(target_record)
+    db.delete.assert_any_call(older_peer)
+    db.commit.assert_called_once()
+    cleanup_instance.delete_task_with_logs.assert_any_call("task-1")
+    cleanup_instance.delete_task_with_logs.assert_any_call("task-2")
+    cleanup_instance.delete_task_with_logs.call_count == 2
+    assert "task-1" not in task_manager.tasks
+    assert "task-2" not in task_manager.tasks
+    assert "task-3" in task_manager.tasks
+
+
+def test_delete_validation_report_returns_false_for_unknown_record():
+    db = MagicMock()
+    db.query.return_value.filter.return_value.first.return_value = None
+
+    service = HealthService(db, config_manager=MagicMock())
+
+    assert service.delete_validation_report("missing") is False
+
+
+def test_delete_validation_report_swallows_linked_task_cleanup_failure():
+    db = MagicMock()
+    config_manager = MagicMock()
+    task_manager = MagicMock()
+    task_manager.tasks = {"task-1": object()}
+
+    record = ValidationRecord(
+        id="rec-1",
+        task_id="task-1",
+        dashboard_id="42",
+        environment_id="env_1",
+        status="PASS",
+        timestamp=datetime.utcnow(),
+        summary="Healthy",
+        issues=[],
+        screenshot_path=None,
+    )
+
+    first_query = MagicMock()
+    first_query.first.return_value = record
+
+    peer_query = MagicMock()
+    peer_query.filter.return_value = peer_query
+    peer_query.all.return_value = [record]
+
+    db.query.side_effect = [first_query, peer_query]
+
+    with patch("src.services.health_service.TaskCleanupService") as cleanup_cls, patch(
+        "src.services.health_service.logger"
+    ) as mock_logger:
+        cleanup_instance = MagicMock()
+        cleanup_instance.delete_task_with_logs.side_effect = RuntimeError("cleanup exploded")
+        cleanup_cls.return_value = cleanup_instance
+
+        service = HealthService(db, config_manager=config_manager)
+        deleted = service.delete_validation_report("rec-1", task_manager=task_manager)
+
+    assert deleted is True
+    db.delete.assert_called_once_with(record)
+    db.commit.assert_called_once()
+    cleanup_instance.delete_task_with_logs.assert_called_once_with("task-1")
+    mock_logger.warning.assert_called_once()
+    assert "task-1" not in task_manager.tasks
+
+
+# [/DEF:test_health_service:Module]

backend/src/services/__tests__/test_llm_provider.py

@@ -7,9 +7,15 @@ import pytest
 import os
 from unittest.mock import MagicMock
 from sqlalchemy.orm import Session
+from cryptography.fernet import Fernet
 from src.services.llm_provider import EncryptionManager, LLMProviderService
 from src.models.llm import LLMProvider
-from src.plugins.llm_analysis.models import LLMProviderConfig, ProviderType
+from src.plugins.llm_analysis.models import LLMProviderConfig, LLMProviderType
+
+# [DEF:_test_encryption_key_fixture:Global]
+# @PURPOSE: Ensure encryption-dependent provider tests run with a valid Fernet key.
+os.environ.setdefault("ENCRYPTION_KEY", Fernet.generate_key().decode())
+# [/DEF:_test_encryption_key_fixture:Global]
 
 # @TEST_CONTRACT: EncryptionManagerModel -> Invariants
 # @TEST_INVARIANT: symmetric_encryption
@@ -50,7 +56,7 @@ def test_get_all_providers(service, mock_db):
 
 def test_create_provider(service, mock_db):
     config = LLMProviderConfig(
-        provider_type=ProviderType.OPENAI,
+        provider_type=LLMProviderType.OPENAI,
         name="Test OpenAI",
         base_url="https://api.openai.com",
         api_key="sk-test",
@@ -79,3 +85,32 @@ def test_get_decrypted_api_key(service, mock_db):
 def test_get_decrypted_api_key_not_found(service, mock_db):
     mock_db.query().filter().first.return_value = None
     assert service.get_decrypted_api_key("missing") is None
+
+def test_update_provider_ignores_masked_placeholder_api_key(service, mock_db):
+    existing_encrypted = EncryptionManager().encrypt("secret-value")
+    mock_provider = LLMProvider(
+        id="p1",
+        provider_type="openai",
+        name="Existing",
+        base_url="https://api.openai.com/v1",
+        api_key=existing_encrypted,
+        default_model="gpt-4o",
+        is_active=True,
+    )
+    mock_db.query().filter().first.return_value = mock_provider
+    config = LLMProviderConfig(
+        id="p1",
+        provider_type=LLMProviderType.OPENAI,
+        name="Existing",
+        base_url="https://api.openai.com/v1",
+        api_key="********",
+        default_model="gpt-4o",
+        is_active=False,
+    )
+
+    updated = service.update_provider("p1", config)
+
+    assert updated is mock_provider
+    assert updated.api_key == existing_encrypted
+    assert EncryptionManager().decrypt(updated.api_key) == "secret-value"
+    assert updated.is_active is False

backend/src/services/health_service.py

@@ -4,21 +4,138 @@
 # @PURPOSE: Business logic for aggregating dashboard health status from validation records.
 # @LAYER: Domain/Service
 # @RELATION: DEPENDS_ON -> ValidationRecord
+# @RELATION: DEPENDS_ON -> backend.src.core.superset_client.SupersetClient
+# @RELATION: DEPENDS_ON -> backend.src.core.task_manager.cleanup.TaskCleanupService
 
-from typing import List, Dict, Any
+from typing import List, Dict, Any, Optional, Tuple
 from sqlalchemy.orm import Session
 from sqlalchemy import func, desc
+import os
 from ..models.llm import ValidationRecord
 from ..schemas.health import DashboardHealthItem, HealthSummaryResponse
 from ..core.logger import logger
+from ..core.superset_client import SupersetClient
+from ..core.task_manager.cleanup import TaskCleanupService
+from ..core.task_manager import TaskManager
 
+# [DEF:HealthService:Class]
+# @TIER: STANDARD
+# @PURPOSE: Aggregate latest dashboard validation state and manage persisted health report lifecycle.
+# @RELATION: CALLS -> backend.src.core.superset_client.SupersetClient
+# @RELATION: CALLS -> backend.src.core.task_manager.cleanup.TaskCleanupService
 class HealthService:
     """
     @PURPOSE: Service for managing and querying dashboard health data.
     """
-    def __init__(self, db: Session):
+    # [DEF:HealthService.__init__:Function]
+    # @PURPOSE: Initialize health service with DB session and optional config access for dashboard metadata resolution.
+    # @PRE: db is a valid SQLAlchemy session.
+    # @POST: Service is ready to aggregate summaries and delete health reports.
+    def __init__(self, db: Session, config_manager = None):
         self.db = db
+        self.config_manager = config_manager
+        self._dashboard_meta_cache: Dict[Tuple[str, str], Dict[str, Optional[str]]] = {}
+    # [/DEF:HealthService.__init__:Function]
 
+    # [DEF:HealthService._prime_dashboard_meta_cache:Function]
+    # @PURPOSE: Warm dashboard slug/title cache with one Superset list fetch per environment.
+    # @PRE: records may contain mixed numeric and slug dashboard identifiers.
+    # @POST: Numeric dashboard ids for known environments are cached when discoverable.
+    # @SIDE_EFFECT: May call Superset dashboard list API once per referenced environment.
+    def _prime_dashboard_meta_cache(self, records: List[ValidationRecord]) -> None:
+        if not self.config_manager or not records:
+            return
+
+        numeric_ids_by_env: Dict[str, set[str]] = {}
+        for record in records:
+            environment_id = str(record.environment_id or "").strip()
+            dashboard_id = str(record.dashboard_id or "").strip()
+            if not environment_id or not dashboard_id or not dashboard_id.isdigit():
+                continue
+            cache_key = (environment_id, dashboard_id)
+            if cache_key in self._dashboard_meta_cache:
+                continue
+            numeric_ids_by_env.setdefault(environment_id, set()).add(dashboard_id)
+
+        if not numeric_ids_by_env:
+            return
+
+        environments = {
+            str(getattr(env, "id", "")).strip(): env
+            for env in self.config_manager.get_environments()
+            if str(getattr(env, "id", "")).strip()
+        }
+
+        for environment_id, dashboard_ids in numeric_ids_by_env.items():
+            env = environments.get(environment_id)
+            if not env:
+                for dashboard_id in dashboard_ids:
+                    self._dashboard_meta_cache[(environment_id, dashboard_id)] = {
+                        "slug": None,
+                        "title": None,
+                    }
+                continue
+
+            try:
+                dashboards = SupersetClient(env).get_dashboards_summary()
+                dashboard_meta_map = {
+                    str(item.get("id")): {
+                        "slug": item.get("slug"),
+                        "title": item.get("title"),
+                    }
+                    for item in dashboards
+                    if str(item.get("id") or "").strip()
+                }
+                for dashboard_id in dashboard_ids:
+                    self._dashboard_meta_cache[(environment_id, dashboard_id)] = dashboard_meta_map.get(
+                        dashboard_id,
+                        {"slug": None, "title": None},
+                    )
+            except Exception as exc:
+                logger.warning(
+                    "[HealthService][_prime_dashboard_meta_cache] Failed to preload dashboard metadata for env=%s: %s",
+                    environment_id,
+                    exc,
+                )
+                for dashboard_id in dashboard_ids:
+                    self._dashboard_meta_cache[(environment_id, dashboard_id)] = {
+                        "slug": None,
+                        "title": None,
+                    }
+    # [/DEF:HealthService._prime_dashboard_meta_cache:Function]
+
+    # [DEF:HealthService._resolve_dashboard_meta:Function]
+    # @PURPOSE: Resolve slug/title for a dashboard referenced by persisted validation record.
+    # @PRE: dashboard_id may be numeric or slug-like; environment_id may be empty.
+    # @POST: Returns dict with `slug` and `title` keys, using cache when possible.
+    # @SIDE_EFFECT: May call Superset API through SupersetClient.
+    def _resolve_dashboard_meta(self, dashboard_id: str, environment_id: Optional[str]) -> Dict[str, Optional[str]]:
+        normalized_dashboard_id = str(dashboard_id or "").strip()
+        normalized_environment_id = str(environment_id or "").strip()
+        if not normalized_dashboard_id:
+            return {"slug": None, "title": None}
+
+        if not normalized_dashboard_id.isdigit():
+            return {"slug": normalized_dashboard_id, "title": None}
+
+        if not self.config_manager or not normalized_environment_id:
+            return {"slug": None, "title": None}
+
+        cache_key = (normalized_environment_id, normalized_dashboard_id)
+        cached = self._dashboard_meta_cache.get(cache_key)
+        if cached is not None:
+            return cached
+
+        meta = {"slug": None, "title": None}
+        self._dashboard_meta_cache[cache_key] = meta
+        return meta
+    # [/DEF:HealthService._resolve_dashboard_meta:Function]
+
+    # [DEF:HealthService.get_health_summary:Function]
+    # @PURPOSE: Aggregate latest validation status per dashboard and enrich rows with dashboard slug/title.
+    # @PRE: environment_id may be omitted to aggregate across all environments.
+    # @POST: Returns HealthSummaryResponse with counts and latest record row per dashboard.
+    # @SIDE_EFFECT: May call Superset API to resolve dashboard metadata.
     async def get_health_summary(self, environment_id: str = None) -> HealthSummaryResponse:
         """
         @PURPOSE: Aggregates the latest validation status for all dashboards.
@@ -44,6 +161,8 @@ class HealthService:
         
         records = query.all()
         
+        self._prime_dashboard_meta_cache(records)
+
         items = []
         pass_count = 0
         warn_count = 0
@@ -62,8 +181,12 @@ class HealthService:
                 unknown_count += 1
                 status = "UNKNOWN"
 
+            meta = self._resolve_dashboard_meta(rec.dashboard_id, rec.environment_id)
             items.append(DashboardHealthItem(
+                record_id=rec.id,
                 dashboard_id=rec.dashboard_id,
+                dashboard_slug=meta.get("slug"),
+                dashboard_title=meta.get("title"),
                 environment_id=rec.environment_id or "unknown",
                 status=status,
                 last_check=rec.timestamp,
@@ -80,5 +203,82 @@ class HealthService:
             fail_count=fail_count,
             unknown_count=unknown_count
         )
+    # [/DEF:HealthService.get_health_summary:Function]
 
-# [/DEF:health_service:Module]
+    # [DEF:HealthService.delete_validation_report:Function]
+    # @PURPOSE: Delete one persisted health report and optionally clean linked task/log artifacts.
+    # @PRE: record_id is a validation record identifier.
+    # @POST: Returns True only when a matching record was deleted.
+    # @SIDE_EFFECT: Deletes DB row, optional screenshot file, and optional task/log persistence.
+    def delete_validation_report(self, record_id: str, task_manager: Optional[TaskManager] = None) -> bool:
+        record = self.db.query(ValidationRecord).filter(ValidationRecord.id == record_id).first()
+        if not record:
+            return False
+
+        peer_query = self.db.query(ValidationRecord).filter(
+            ValidationRecord.dashboard_id == record.dashboard_id
+        )
+        if record.environment_id is None:
+            peer_query = peer_query.filter(ValidationRecord.environment_id.is_(None))
+        else:
+            peer_query = peer_query.filter(ValidationRecord.environment_id == record.environment_id)
+
+        records_to_delete = peer_query.all()
+        screenshot_paths = [
+            str(item.screenshot_path or "").strip()
+            for item in records_to_delete
+            if str(item.screenshot_path or "").strip()
+        ]
+        task_ids = {
+            str(item.task_id or "").strip()
+            for item in records_to_delete
+            if str(item.task_id or "").strip()
+        }
+
+        logger.info(
+            "[HealthService][delete_validation_report] Removing %s validation record(s) for dashboard=%s environment=%s triggered_by_record=%s",
+            len(records_to_delete),
+            record.dashboard_id,
+            record.environment_id,
+            record_id,
+        )
+
+        for item in records_to_delete:
+            self.db.delete(item)
+        self.db.commit()
+
+        for screenshot_path in screenshot_paths:
+            try:
+                if os.path.exists(screenshot_path):
+                    os.remove(screenshot_path)
+            except OSError as exc:
+                logger.warning(
+                    "[HealthService][delete_validation_report] Failed to remove screenshot %s: %s",
+                    screenshot_path,
+                    exc,
+                )
+
+        if task_ids and task_manager and self.config_manager:
+            try:
+                cleanup_service = TaskCleanupService(
+                    task_manager.persistence_service,
+                    task_manager.log_persistence_service,
+                    self.config_manager,
+                )
+                for task_id in task_ids:
+                    task_manager.tasks.pop(task_id, None)
+                    cleanup_service.delete_task_with_logs(task_id)
+            except Exception as exc:
+                logger.warning(
+                    "[HealthService][delete_validation_report] Failed to cleanup linked task/logs for dashboard=%s environment=%s: %s",
+                    record.dashboard_id,
+                    record.environment_id,
+                    exc,
+                )
+
+        return True
+    # [/DEF:HealthService.delete_validation_report:Function]
+    
+# [/DEF:HealthService:Class]
+
+# [/DEF:health_service:Module]

backend/src/services/llm_provider.py

@@ -16,6 +16,8 @@ import os
 if TYPE_CHECKING:
     from ..plugins.llm_analysis.models import LLMProviderConfig
 
+MASKED_API_KEY_PLACEHOLDER = "********"
+
 # [DEF:_require_fernet_key:Function]
 # @TIER: CRITICAL
 # @PURPOSE: Load and validate the Fernet key used for secret encryption.
@@ -145,8 +147,12 @@ class LLMProviderService:
             db_provider.provider_type = config.provider_type.value
             db_provider.name = config.name
             db_provider.base_url = config.base_url
-            # Only update API key if provided (not None and not empty)
-            if config.api_key is not None and config.api_key != "":
+            # Ignore masked placeholder values; they are display-only and must not overwrite secrets.
+            if (
+                config.api_key is not None
+                and config.api_key != ""
+                and config.api_key != MASKED_API_KEY_PLACEHOLDER
+            ):
                 db_provider.api_key = self.encryption.encrypt(config.api_key)
             db_provider.default_model = config.default_model
             db_provider.is_active = config.is_active

backend/src/services/profile_service.py

@@ -117,6 +117,33 @@ class ProfileService:
             )
     # [/DEF:get_my_preference:Function]
 
+    # [DEF:get_dashboard_filter_binding:Function]
+    # @PURPOSE: Return only dashboard-filter fields required by dashboards listing hot path.
+    # @PRE: current_user is authenticated.
+    # @POST: Returns normalized username and profile-default filter toggles without security summary expansion.
+    def get_dashboard_filter_binding(self, current_user: User) -> dict:
+        with belief_scope("ProfileService.get_dashboard_filter_binding", f"user_id={current_user.id}"):
+            preference = self._get_preference_row(current_user.id)
+            if preference is None:
+                return {
+                    "superset_username": None,
+                    "superset_username_normalized": None,
+                    "show_only_my_dashboards": False,
+                    "show_only_slug_dashboards": True,
+                }
+
+            return {
+                "superset_username": self._sanitize_username(preference.superset_username),
+                "superset_username_normalized": self._normalize_username(preference.superset_username),
+                "show_only_my_dashboards": bool(preference.show_only_my_dashboards),
+                "show_only_slug_dashboards": bool(
+                    preference.show_only_slug_dashboards
+                    if preference.show_only_slug_dashboards is not None
+                    else True
+                ),
+            }
+    # [/DEF:get_dashboard_filter_binding:Function]
+
     # [DEF:update_my_preference:Function]
     # @PURPOSE: Validate and persist current user's profile preference in self-scoped mode.
     # @PRE: current_user is authenticated and payload is provided.

backend/src/services/rbac_permission_catalog.py

@@ -10,6 +10,7 @@
 
 # [SECTION: IMPORTS]
 import re
+from functools import lru_cache
 from pathlib import Path
 from typing import Iterable, Set, Tuple
 
@@ -80,6 +81,17 @@ def _discover_route_permissions() -> Set[Tuple[str, str]]:
 # [/DEF:_discover_route_permissions:Function]
 
 
+# [DEF:_discover_route_permissions_cached:Function]
+# @PURPOSE: Cache route permission discovery because route source files are static during normal runtime.
+# @PRE: None.
+# @POST: Returns stable discovered route permission pairs without repeated filesystem scans.
+@lru_cache(maxsize=1)
+def _discover_route_permissions_cached() -> Tuple[Tuple[str, str], ...]:
+    with belief_scope("rbac_permission_catalog._discover_route_permissions_cached"):
+        return tuple(sorted(_discover_route_permissions()))
+# [/DEF:_discover_route_permissions_cached:Function]
+
+
 # [DEF:_discover_plugin_execute_permissions:Function]
 # @PURPOSE: Derives dynamic task permissions of form plugin:{plugin_id}:EXECUTE from plugin registry.
 # @PRE: plugin_loader is optional and may expose get_all_plugin_configs.
@@ -108,6 +120,19 @@ def _discover_plugin_execute_permissions(plugin_loader=None) -> Set[Tuple[str, s
 # [/DEF:_discover_plugin_execute_permissions:Function]
 
 
+# [DEF:_discover_plugin_execute_permissions_cached:Function]
+# @PURPOSE: Cache dynamic plugin EXECUTE permission pairs by normalized plugin id tuple.
+# @PRE: plugin_ids is a deterministic tuple of plugin ids.
+# @POST: Returns stable permission tuple without repeated plugin catalog expansion.
+@lru_cache(maxsize=8)
+def _discover_plugin_execute_permissions_cached(
+    plugin_ids: Tuple[str, ...],
+) -> Tuple[Tuple[str, str], ...]:
+    with belief_scope("rbac_permission_catalog._discover_plugin_execute_permissions_cached"):
+        return tuple((f"plugin:{plugin_id}", "EXECUTE") for plugin_id in plugin_ids)
+# [/DEF:_discover_plugin_execute_permissions_cached:Function]
+
+
 # [DEF:discover_declared_permissions:Function]
 # @PURPOSE: Builds canonical RBAC permission catalog from routes and plugin registry.
 # @PRE: plugin_loader may be provided for dynamic task plugin permission discovery.
@@ -115,8 +140,17 @@ def _discover_plugin_execute_permissions(plugin_loader=None) -> Set[Tuple[str, s
 # @RETURN: Set[Tuple[str, str]] - Complete discovered permission set.
 def discover_declared_permissions(plugin_loader=None) -> Set[Tuple[str, str]]:
     with belief_scope("rbac_permission_catalog.discover_declared_permissions"):
-        permissions = _discover_route_permissions()
-        permissions.update(_discover_plugin_execute_permissions(plugin_loader))
+        permissions = set(_discover_route_permissions_cached())
+        plugin_ids = tuple(
+            sorted(
+                {
+                    str(getattr(plugin_config, "id", "") or "").strip()
+                    for plugin_config in (plugin_loader.get_all_plugin_configs() if plugin_loader else [])
+                    if str(getattr(plugin_config, "id", "") or "").strip()
+                }
+            )
+        )
+        permissions.update(_discover_plugin_execute_permissions_cached(plugin_ids))
         return permissions
 # [/DEF:discover_declared_permissions:Function]
 
@@ -153,4 +187,4 @@ def sync_permission_catalog(
         return len(missing_pairs)
 # [/DEF:sync_permission_catalog:Function]
 
-# [/DEF:backend.src.services.rbac_permission_catalog:Module]
+# [/DEF:backend.src.services.rbac_permission_catalog:Module]