feat(trigger): add context caching for trigger providers

- Add plugin_trigger_providers and plugin_trigger_providers_lock to contexts module
- Implement caching mechanism in TriggerManager.get_trigger_provider() method
- Cache fetched trigger providers to reduce repeated daemon calls
- Use double-check locking pattern for thread-safe cache access

This follows the same pattern as ToolManager.get_plugin_provider() to improve performance
by avoiding redundant requests to the daemon when accessing trigger providers.

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

Co-Authored-By: Claude <noreply@anthropic.com>

api/contexts/__init__.py

@@ -8,6 +8,7 @@ if TYPE_CHECKING:
     from core.model_runtime.entities.model_entities import AIModelEntity
     from core.plugin.entities.plugin_daemon import PluginModelProviderEntity
     from core.tools.plugin_tool.provider import PluginToolProviderController
+    from core.trigger.provider import PluginTriggerProviderController
     from core.workflow.entities.variable_pool import VariablePool
 
 
@@ -33,3 +34,11 @@ plugin_model_schema_lock: RecyclableContextVar[Lock] = RecyclableContextVar(Cont
 plugin_model_schemas: RecyclableContextVar[dict[str, "AIModelEntity"]] = RecyclableContextVar(
     ContextVar("plugin_model_schemas")
 )
+
+plugin_trigger_providers: RecyclableContextVar[dict[str, "PluginTriggerProviderController"]] = RecyclableContextVar(
+    ContextVar("plugin_trigger_providers")
+)
+
+plugin_trigger_providers_lock: RecyclableContextVar[Lock] = RecyclableContextVar(
+    ContextVar("plugin_trigger_providers_lock")
+)

api/core/trigger/trigger_manager.py

@@ -4,10 +4,12 @@ Trigger Manager for loading and managing trigger providers and triggers
 
 import logging
 from collections.abc import Mapping
+from threading import Lock
 from typing import Any, Optional
 
 from flask import Request
 
+import contexts
 from core.plugin.entities.plugin import TriggerProviderID
 from core.plugin.entities.plugin_daemon import CredentialType
 from core.plugin.entities.request import TriggerInvokeResponse
@@ -50,8 +52,8 @@ class TriggerManager:
                     tenant_id=tenant_id,
                 )
                 controllers.append(controller)
-            except Exception as e:
-                logger.exception("Failed to load trigger provider {provider.plugin_id}")
+            except Exception:
+                logger.exception("Failed to load trigger provider %s", provider.plugin_id)
                 continue
 
         return controllers
@@ -65,23 +67,43 @@ class TriggerManager:
         :param provider_id: Provider ID
         :return: Trigger provider controller or None
         """
-        manager = PluginTriggerManager()
-        provider = manager.fetch_trigger_provider(tenant_id, provider_id)
-
-        if not provider:
-            raise ValueError(f"Trigger provider {provider_id} not found")
-
+        # check if context is set
         try:
-            return PluginTriggerProviderController(
-                entity=provider.declaration,
-                plugin_id=provider.plugin_id,
-                plugin_unique_identifier=provider.plugin_unique_identifier,
-                provider_id=provider_id,
-                tenant_id=tenant_id,
-            )
-        except Exception as e:
-            logger.exception("Failed to load trigger provider")
-            raise e
+            contexts.plugin_trigger_providers.get()
+        except LookupError:
+            contexts.plugin_trigger_providers.set({})
+            contexts.plugin_trigger_providers_lock.set(Lock())
+
+        plugin_trigger_providers = contexts.plugin_trigger_providers.get()
+        provider_id_str = str(provider_id)
+        if provider_id_str in plugin_trigger_providers:
+            return plugin_trigger_providers[provider_id_str]
+
+        with contexts.plugin_trigger_providers_lock.get():
+            # double check
+            plugin_trigger_providers = contexts.plugin_trigger_providers.get()
+            if provider_id_str in plugin_trigger_providers:
+                return plugin_trigger_providers[provider_id_str]
+
+            manager = PluginTriggerManager()
+            provider = manager.fetch_trigger_provider(tenant_id, provider_id)
+
+            if not provider:
+                raise ValueError(f"Trigger provider {provider_id} not found")
+
+            try:
+                controller = PluginTriggerProviderController(
+                    entity=provider.declaration,
+                    plugin_id=provider.plugin_id,
+                    plugin_unique_identifier=provider.plugin_unique_identifier,
+                    provider_id=provider_id,
+                    tenant_id=tenant_id,
+                )
+                plugin_trigger_providers[provider_id_str] = controller
+                return controller
+            except Exception as e:
+                logger.exception("Failed to load trigger provider")
+                raise e
 
     @classmethod
     def list_all_trigger_providers(cls, tenant_id: str) -> list[PluginTriggerProviderController]: