feat: default plugins for new user (#39167)

api/.env.example

@@ -667,6 +667,12 @@ PLUGIN_REMOTE_INSTALL_HOST=localhost
 PLUGIN_MAX_PACKAGE_SIZE=15728640
 PLUGIN_MODEL_SCHEMA_CACHE_TTL=3600
 PLUGIN_MODEL_PROVIDERS_CACHE_TTL=86400
+# Comma-separated marketplace plugin IDs whose latest versions are installed for newly registered users.
+# Example: langgenius/openai,langgenius/gemini
+NEW_USER_DEFAULT_PLUGIN_IDS=
+# Comma-separated model_type:provider:model entries assigned after default plugins finish installing.
+# Example: llm:langgenius/openai/openai:gpt-4o-mini,text-embedding:langgenius/openai/openai:text-embedding-3-small
+NEW_USER_DEFAULT_MODELS=
 INNER_API_KEY_FOR_PLUGIN=QaHbTe77CtuXmsfyhR7+vRjI/+XbV1AaFy691iy+kGDv2Jvy0/eAh8Y1
 
 # Dify Agent backend

api/configs/feature/__init__.py

@@ -275,6 +275,42 @@ class PluginConfig(BaseSettings):
         default=50 * 1024 * 1024,
     )
 
+    NEW_USER_DEFAULT_PLUGIN_IDS: str = Field(
+        description="Comma-separated marketplace plugin IDs whose latest versions are installed for new users",
+        default="",
+    )
+
+    @property
+    def NEW_USER_DEFAULT_PLUGIN_ID_LIST(self) -> list[str]:
+        return [item.strip() for item in self.NEW_USER_DEFAULT_PLUGIN_IDS.split(",") if item.strip()]
+
+    NEW_USER_DEFAULT_MODELS: str = Field(
+        description=("Comma-separated default models for new users in 'model_type:provider:model' format"),
+        default="",
+    )
+
+    @property
+    def NEW_USER_DEFAULT_MODEL_LIST(self) -> list[tuple[str, str, str]]:
+        default_models: list[tuple[str, str, str]] = []
+        configured_model_types: set[str] = set()
+
+        for item in self.NEW_USER_DEFAULT_MODELS.split(","):
+            if not item.strip():
+                continue
+
+            parts = tuple(part.strip() for part in item.split(":", 2))
+            if len(parts) != 3 or not all(parts):
+                raise ValueError("NEW_USER_DEFAULT_MODELS entries must use 'model_type:provider:model' format")
+
+            model_type, provider, model = parts
+            if model_type in configured_model_types:
+                raise ValueError(f"NEW_USER_DEFAULT_MODELS contains duplicate model type: {model_type}")
+
+            configured_model_types.add(model_type)
+            default_models.append((model_type, provider, model))
+
+        return default_models
+
 
 class MarketplaceConfig(BaseSettings):
     """

api/events/event_handlers/__init__.py

@@ -7,6 +7,9 @@ from .delete_tool_parameters_cache_when_sync_draft_workflow import (
     handle as handle_delete_tool_parameters_cache_when_sync_draft_workflow,
 )
 from .queue_credential_sync_when_tenant_created import handle as handle_queue_credential_sync_when_tenant_created
+from .queue_default_plugin_install_when_tenant_created import (
+    handle as handle_queue_default_plugin_install_when_tenant_created,
+)
 from .sync_plugin_trigger_when_app_created import handle as handle_sync_plugin_trigger_when_app_created
 from .sync_webhook_when_app_created import handle as handle_sync_webhook_when_app_created
 from .sync_workflow_schedule_when_app_published import handle as handle_sync_workflow_schedule_when_app_published
@@ -32,6 +35,7 @@ __all__ = [
     "handle_create_site_record_when_app_created",
     "handle_delete_tool_parameters_cache_when_sync_draft_workflow",
     "handle_queue_credential_sync_when_tenant_created",
+    "handle_queue_default_plugin_install_when_tenant_created",
     "handle_sync_plugin_trigger_when_app_created",
     "handle_sync_webhook_when_app_created",
     "handle_sync_workflow_schedule_when_app_published",

api/events/event_handlers/queue_default_plugin_install_when_tenant_created.py

@@ -0,0 +1,22 @@
+"""Queue default marketplace plugin installation after tenant creation."""
+
+import logging
+
+from configs import dify_config
+from events.tenant_event import tenant_was_created
+from tasks.install_default_plugins_task import install_default_plugins_task
+
+logger = logging.getLogger(__name__)
+
+
+@tenant_was_created.connect
+def handle(sender, **kwargs) -> None:
+    """Keep tenant creation non-blocking while installing configured plugins asynchronously."""
+    plugin_ids = dify_config.NEW_USER_DEFAULT_PLUGIN_ID_LIST
+    if not plugin_ids:
+        return
+
+    try:
+        install_default_plugins_task.delay(sender.id, plugin_ids)
+    except Exception:
+        logger.exception("Failed to queue default plugin installation for tenant %s", sender.id)

api/extensions/ext_celery.py

@@ -156,6 +156,7 @@ def init_app(app: DifyApp) -> Celery:
         "tasks.generate_summary_index_task",  # summary index generation
         "tasks.regenerate_summary_index_task",  # summary index regeneration
         "tasks.initialize_created_app_rbac_access_task",  # app access initialization
+        "tasks.install_default_plugins_task",  # tenant default plugin installation
         "tasks.app_generate.resume_agent_app_task",  # ENG-635: Agent v2 chat ask_human resume
         "tasks.workflow_run_archive_download_tasks",  # workflow-run archive download preparation
     ]

api/tasks/install_default_plugins_task.py

@@ -0,0 +1,99 @@
+"""Install configured marketplace plugins for a newly created tenant."""
+
+import logging
+
+from celery import shared_task
+
+from configs import dify_config
+from core.helper import marketplace
+from core.plugin.entities.plugin_daemon import PluginInstallTaskStatus
+from core.plugin.plugin_service import PluginService
+from services.model_provider_service import ModelProviderService
+
+logger = logging.getLogger(__name__)
+
+
+@shared_task(queue="plugin", bind=True, max_retries=60, default_retry_delay=5)
+def configure_default_models_task(self, tenant_id: str, plugin_install_task_id: str | None) -> None:
+    """Set explicitly configured default models after default plugins finish installing."""
+    if not dify_config.NEW_USER_DEFAULT_MODELS:
+        return
+
+    plugin_install_failed = False
+    if plugin_install_task_id:
+        try:
+            install_task = PluginService.fetch_install_task(tenant_id, plugin_install_task_id)
+        except Exception as exc:
+            logger.warning(
+                "Failed to fetch default plugin installation task for tenant %s; retrying",
+                tenant_id,
+            )
+            raise self.retry(exc=exc)
+
+        if install_task.status in (PluginInstallTaskStatus.Pending, PluginInstallTaskStatus.Running):
+            raise self.retry()
+
+        plugin_install_failed = install_task.status == PluginInstallTaskStatus.Failed
+        if plugin_install_failed:
+            failed_plugin_ids = [
+                plugin.plugin_id for plugin in install_task.plugins if plugin.status == PluginInstallTaskStatus.Failed
+            ]
+            logger.error(
+                "Default plugin installation failed for tenant %s: %s",
+                tenant_id,
+                ", ".join(failed_plugin_ids),
+            )
+
+    model_provider_service = ModelProviderService()
+    failed_model_types: list[str] = []
+    for model_type, provider, model in dify_config.NEW_USER_DEFAULT_MODEL_LIST:
+        try:
+            model_provider_service.update_default_model_of_model_type(
+                tenant_id=tenant_id,
+                model_type=model_type,
+                provider=provider,
+                model=model,
+            )
+        except Exception:
+            failed_model_types.append(model_type)
+            logger.exception(
+                "Failed to configure default model for tenant %s: model_type=%s provider=%s model=%s",
+                tenant_id,
+                model_type,
+                provider,
+                model,
+            )
+
+    if plugin_install_failed or failed_model_types:
+        raise RuntimeError(
+            f"Failed to initialize defaults for tenant {tenant_id}; "
+            f"model types: {', '.join(failed_model_types) or 'none'}"
+        )
+
+
+@shared_task(queue="plugin")
+def install_default_plugins_task(tenant_id: str, plugin_ids: list[str]) -> None:
+    """Install the latest marketplace versions of the configured plugins."""
+    if not plugin_ids:
+        return
+
+    try:
+        manifests = {manifest.plugin_id: manifest for manifest in marketplace.batch_fetch_plugin_manifests(plugin_ids)}
+        plugin_identifiers = [
+            manifests[plugin_id].latest_package_identifier for plugin_id in plugin_ids if plugin_id in manifests
+        ]
+        missing_plugin_ids = [plugin_id for plugin_id in plugin_ids if plugin_id not in manifests]
+        if missing_plugin_ids:
+            logger.warning("Default plugins not found in marketplace: %s", ", ".join(missing_plugin_ids))
+        if not plugin_identifiers:
+            return
+
+        response = PluginService.install_from_marketplace_pkg(tenant_id, plugin_identifiers)
+        if dify_config.NEW_USER_DEFAULT_MODELS:
+            configure_default_models_task.delay(
+                tenant_id,
+                None if response.all_installed else response.task_id,
+            )
+    except Exception:
+        logger.exception("Failed to install default plugins for tenant %s", tenant_id)
+        raise

api/tests/unit_tests/configs/test_dify_config.py

@@ -89,6 +89,54 @@ def test_dify_config(monkeypatch: pytest.MonkeyPatch):
     assert Version(config.project.version) >= Version("1.0.0")
 
 
+def test_new_user_default_plugin_ids_are_parsed_from_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    _set_basic_config_env(monkeypatch)
+    monkeypatch.setenv(
+        "NEW_USER_DEFAULT_PLUGIN_IDS",
+        "langgenius/openai, langgenius/gemini",
+    )
+
+    config = DifyConfig(_env_file=None)
+
+    assert config.NEW_USER_DEFAULT_PLUGIN_ID_LIST == [
+        "langgenius/openai",
+        "langgenius/gemini",
+    ]
+
+
+def test_new_user_default_models_are_parsed_from_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    _set_basic_config_env(monkeypatch)
+    monkeypatch.setenv(
+        "NEW_USER_DEFAULT_MODELS",
+        (
+            "llm:langgenius/openai/openai:gpt-4o-mini, "
+            "text-embedding:langgenius/openai/openai:text-embedding-3-small, "
+            "rerank:langgenius/ollama/ollama:reranker:latest"
+        ),
+    )
+
+    config = DifyConfig(_env_file=None)
+
+    assert config.NEW_USER_DEFAULT_MODEL_LIST == [
+        ("llm", "langgenius/openai/openai", "gpt-4o-mini"),
+        ("text-embedding", "langgenius/openai/openai", "text-embedding-3-small"),
+        ("rerank", "langgenius/ollama/ollama", "reranker:latest"),
+    ]
+
+
+def test_new_user_default_models_reject_duplicate_model_types(monkeypatch: pytest.MonkeyPatch) -> None:
+    _set_basic_config_env(monkeypatch)
+    monkeypatch.setenv(
+        "NEW_USER_DEFAULT_MODELS",
+        "llm:langgenius/openai/openai:gpt-4o-mini,llm:langgenius/anthropic/anthropic:claude-sonnet-4",
+    )
+
+    config = DifyConfig(_env_file=None)
+
+    with pytest.raises(ValueError, match="duplicate model type: llm"):
+        _ = config.NEW_USER_DEFAULT_MODEL_LIST
+
+
 def test_http_timeout_defaults(monkeypatch: pytest.MonkeyPatch):
     """Test that HTTP timeout defaults are correctly set"""
     # clear system environment variables

api/tests/unit_tests/events/event_handlers/test_queue_default_plugin_install_when_tenant_created.py

@@ -0,0 +1,52 @@
+import logging
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import pytest
+
+from events.event_handlers import queue_default_plugin_install_when_tenant_created as handler_module
+
+
+def test_handle_skips_when_no_default_plugins_are_configured(monkeypatch: pytest.MonkeyPatch) -> None:
+    delay = MagicMock()
+    monkeypatch.setattr(handler_module.dify_config, "NEW_USER_DEFAULT_PLUGIN_IDS", "")
+    monkeypatch.setattr(handler_module.install_default_plugins_task, "delay", delay)
+
+    handler_module.handle(SimpleNamespace(id="tenant-1"))
+
+    delay.assert_not_called()
+
+
+def test_handle_queues_configured_plugins(monkeypatch: pytest.MonkeyPatch) -> None:
+    delay = MagicMock()
+    plugins = [
+        "langgenius/openai",
+        "langgenius/gemini",
+    ]
+    monkeypatch.setattr(handler_module.dify_config, "NEW_USER_DEFAULT_PLUGIN_IDS", ",".join(plugins))
+    monkeypatch.setattr(handler_module.install_default_plugins_task, "delay", delay)
+
+    handler_module.handle(SimpleNamespace(id="tenant-1"))
+
+    delay.assert_called_once_with("tenant-1", plugins)
+
+
+def test_handle_does_not_fail_tenant_creation_when_queue_is_unavailable(
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    monkeypatch.setattr(
+        handler_module.dify_config,
+        "NEW_USER_DEFAULT_PLUGIN_IDS",
+        "langgenius/openai",
+    )
+    monkeypatch.setattr(
+        handler_module.install_default_plugins_task,
+        "delay",
+        MagicMock(side_effect=ConnectionError("broker unavailable")),
+    )
+
+    with caplog.at_level(logging.ERROR, logger=handler_module.logger.name):
+        handler_module.handle(SimpleNamespace(id="tenant-1"))
+
+    assert "Failed to queue default plugin installation for tenant tenant-1" in caplog.text

api/tests/unit_tests/tasks/test_install_default_plugins_task.py

@@ -0,0 +1,149 @@
+import logging
+from types import SimpleNamespace
+from unittest.mock import MagicMock, call
+
+import pytest
+from celery.exceptions import Retry
+
+
+def test_install_default_plugins_task_uses_plugin_queue() -> None:
+    from tasks.install_default_plugins_task import install_default_plugins_task
+
+    assert install_default_plugins_task.queue == "plugin"
+
+
+def test_configure_default_models_task_uses_plugin_queue() -> None:
+    from tasks.install_default_plugins_task import configure_default_models_task
+
+    assert configure_default_models_task.queue == "plugin"
+
+
+def test_install_default_plugins_task_installs_latest_identifiers(monkeypatch: pytest.MonkeyPatch) -> None:
+    import tasks.install_default_plugins_task as task_module
+    from tasks.install_default_plugins_task import install_default_plugins_task
+
+    plugin_ids = ["langgenius/openai", "langgenius/gemini"]
+    plugin_identifiers = ["langgenius/openai:1.0.0@aaa", "langgenius/gemini:0.9.1@bbb"]
+    fetch = MagicMock(
+        return_value=[
+            SimpleNamespace(plugin_id=plugin_ids[1], latest_package_identifier=plugin_identifiers[1]),
+            SimpleNamespace(plugin_id=plugin_ids[0], latest_package_identifier=plugin_identifiers[0]),
+        ]
+    )
+    install = MagicMock()
+    monkeypatch.setattr(task_module.marketplace, "batch_fetch_plugin_manifests", fetch)
+    monkeypatch.setattr(task_module.PluginService, "install_from_marketplace_pkg", install)
+
+    install_default_plugins_task.run("tenant-1", plugin_ids)
+
+    fetch.assert_called_once_with(plugin_ids)
+    install.assert_called_once_with("tenant-1", plugin_identifiers)
+
+
+def test_install_default_plugins_task_skips_missing_plugins(
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    import tasks.install_default_plugins_task as task_module
+    from tasks.install_default_plugins_task import install_default_plugins_task
+
+    fetch = MagicMock(
+        return_value=[
+            SimpleNamespace(plugin_id="langgenius/openai", latest_package_identifier="langgenius/openai:1.0.0@aaa")
+        ]
+    )
+    install = MagicMock()
+    monkeypatch.setattr(task_module.marketplace, "batch_fetch_plugin_manifests", fetch)
+    monkeypatch.setattr(task_module.PluginService, "install_from_marketplace_pkg", install)
+
+    with caplog.at_level(logging.WARNING, logger=task_module.logger.name):
+        install_default_plugins_task.run("tenant-1", ["langgenius/openai", "missing/plugin"])
+
+    install.assert_called_once_with("tenant-1", ["langgenius/openai:1.0.0@aaa"])
+    assert "Default plugins not found in marketplace: missing/plugin" in caplog.text
+
+
+def test_install_default_plugins_task_queues_model_configuration_after_daemon_install(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    import tasks.install_default_plugins_task as task_module
+    from tasks.install_default_plugins_task import install_default_plugins_task
+
+    plugin_id = "langgenius/openai"
+    plugin_identifier = "langgenius/openai:1.0.0@aaa"
+    monkeypatch.setattr(task_module.dify_config, "NEW_USER_DEFAULT_MODELS", "llm:provider:model")
+    monkeypatch.setattr(
+        task_module.marketplace,
+        "batch_fetch_plugin_manifests",
+        MagicMock(return_value=[SimpleNamespace(plugin_id=plugin_id, latest_package_identifier=plugin_identifier)]),
+    )
+    monkeypatch.setattr(
+        task_module.PluginService,
+        "install_from_marketplace_pkg",
+        MagicMock(return_value=SimpleNamespace(all_installed=False, task_id="install-task-1")),
+    )
+    delay = MagicMock()
+    monkeypatch.setattr(task_module.configure_default_models_task, "delay", delay)
+
+    install_default_plugins_task.run("tenant-1", [plugin_id])
+
+    delay.assert_called_once_with("tenant-1", "install-task-1")
+
+
+def test_configure_default_models_task_retries_while_plugins_are_installing(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    import tasks.install_default_plugins_task as task_module
+    from tasks.install_default_plugins_task import configure_default_models_task
+
+    monkeypatch.setattr(task_module.dify_config, "NEW_USER_DEFAULT_MODELS", "llm:provider:model")
+    monkeypatch.setattr(
+        task_module.PluginService,
+        "fetch_install_task",
+        MagicMock(return_value=SimpleNamespace(status=task_module.PluginInstallTaskStatus.Running)),
+    )
+    retry = MagicMock(side_effect=Retry())
+    monkeypatch.setattr(configure_default_models_task, "retry", retry)
+
+    with pytest.raises(Retry):
+        configure_default_models_task.run("tenant-1", "install-task-1")
+
+    retry.assert_called_once_with()
+
+
+def test_configure_default_models_task_sets_each_explicit_model(monkeypatch: pytest.MonkeyPatch) -> None:
+    import tasks.install_default_plugins_task as task_module
+    from tasks.install_default_plugins_task import configure_default_models_task
+
+    monkeypatch.setattr(
+        task_module.dify_config,
+        "NEW_USER_DEFAULT_MODELS",
+        ("llm:langgenius/openai/openai:gpt-4o-mini,text-embedding:langgenius/openai/openai:text-embedding-3-small"),
+    )
+    fetch_install_task = MagicMock(
+        return_value=SimpleNamespace(
+            status=task_module.PluginInstallTaskStatus.Success,
+            plugins=[],
+        )
+    )
+    monkeypatch.setattr(task_module.PluginService, "fetch_install_task", fetch_install_task)
+    model_provider_service = MagicMock()
+    monkeypatch.setattr(task_module, "ModelProviderService", MagicMock(return_value=model_provider_service))
+
+    configure_default_models_task.run("tenant-1", "install-task-1")
+
+    fetch_install_task.assert_called_once_with("tenant-1", "install-task-1")
+    assert model_provider_service.update_default_model_of_model_type.call_args_list == [
+        call(
+            tenant_id="tenant-1",
+            model_type="llm",
+            provider="langgenius/openai/openai",
+            model="gpt-4o-mini",
+        ),
+        call(
+            tenant_id="tenant-1",
+            model_type="text-embedding",
+            provider="langgenius/openai/openai",
+            model="text-embedding-3-small",
+        ),
+    ]

docker/envs/core-services/shared.env.example

@@ -64,6 +64,12 @@ PGDATA=/var/lib/postgresql/data/pgdata
 PLUGIN_MAX_PACKAGE_SIZE=52428800
 PLUGIN_MODEL_SCHEMA_CACHE_TTL=3600
 PLUGIN_MODEL_PROVIDERS_CACHE_TTL=86400
+# Comma-separated marketplace plugin IDs whose latest versions are installed for newly registered users.
+# Example: langgenius/openai,langgenius/gemini
+NEW_USER_DEFAULT_PLUGIN_IDS=
+# Comma-separated model_type:provider:model entries assigned after default plugins finish installing.
+# Example: llm:langgenius/openai/openai:gpt-4o-mini,text-embedding:langgenius/openai/openai:text-embedding-3-small
+NEW_USER_DEFAULT_MODELS=
 ENDPOINT_URL_TEMPLATE=http://localhost/e/{hook_id}
 LOG_LEVEL=INFO
 LOG_OUTPUT_FORMAT=text