dify/api/tests/unit_tests/extensions/test_ext_application_services.py

"""Tests for application-service dependency wiring."""

import json
import logging
from collections.abc import Mapping
from decimal import Decimal
from io import BytesIO
from types import SimpleNamespace
from typing import cast
from unittest.mock import MagicMock, call, create_autospec, patch
from uuid import uuid4

import httpx
import pytest
from flask import Flask
from redis import Redis
from sqlalchemy import Connection, event, select
from sqlalchemy.orm import Session, SessionTransaction, sessionmaker

from core.credit_usage import CreditUsageAppType, CreditUsageCreatedBy
from core.llm_generator.prompts import CONVERSATION_TITLE_PROMPT
from core.model_context import get_credit_usage_metadata, use_credit_usage_metadata
from core.plugin.impl import base as plugin_http
from core.tools.tool_file_manager import ToolFileManager
from enums import DeploymentEdition, WebAppAccessMode
from extensions import ext_application_services
from extensions.application_services.app import AppServices
from extensions.ext_database import db
from extensions.ext_redis import RedisClientWrapper
from extensions.tracing import application_tracing_hooks, create_trace_scope_for_app
from graphon.model_runtime.entities.model_entities import ModelType
from machinery.context import RequestContext
from models.account import Account, Tenant, TenantAccountJoin, TenantAccountRole
from models.enums import AppStatus, ConversationFromSource, CustomizeTokenStrategy
from models.enums import InvokeFrom as ModelInvokeFrom
from models.model import (
    AccountTrialAppRecord,
    App,
    AppMode,
    AppModelConfig,
    Conversation,
    DifySetup,
    InstalledApp,
    Message,
    Site,
    TrialApp,
)
from models.provider import Provider, ProviderCredential, TenantDefaultModel
from repositories.account.repository import SQLAlchemyAccountRepository
from repositories.account_activation_repository import SQLAlchemyAccountActivationRepository
from repositories.account_integration_repository import SQLAlchemyAccountIntegrationRepository
from repositories.app.console_repository import ConsoleAppRepository
from repositories.app.site_command_repository import AppSiteCommandRepository
from repositories.app.tracing_config_repository import SQLAlchemyAppTracingConfigRepository
from repositories.app.workflow_comment_repository import WorkflowCommentRepository
from repositories.app_scoped_end_user_repository import AppScopedEndUserRepo
from repositories.app_statistic_query_repository import AppStatisticQueryRepository
from repositories.human_input_file_upload_repository import SQLAlchemyHumanInputFileUploadRepository
from repositories.installed_app_repository import SQLAlchemyInstalledAppRepository
from repositories.message_file_preview_repository import MessageFilePreviewQueryRepository
from repositories.network_access_group_repository import SQLAlchemyNetworkAccessGroupAppRepository
from repositories.plugin_file_upload_repository import SQLAlchemyPluginFileUploadOwnerRepository
from repositories.sqlalchemy_api_workflow_run_repository import DifyAPISQLAlchemyWorkflowRunRepository
from repositories.upload_file_delivery_repository import UploadFileDeliveryQueryRepository
from repositories.workflow_app_log_query_repository import WorkflowAppLogQueryRepository
from repositories.workflow_run_archive_repository import WorkflowRunArchiveBundleQueryRepository
from repositories.workspace.workspace_repository import WorkspaceRepository
from services import audio_provider_gateway, recommended_app_catalog_gateway
from services.account import forgot_password_service as account_forgot_password_service
from services.account.adapters import (
    BillingAccountActivationEligibility,
    BillingWorkspaceMembershipCache,
    DeploymentWorkspaceInvitePolicy,
    RBACWorkspaceMemberAccessSync,
    RedisInvitationTokenStore,
)
from services.account.email_registration_adapters import (
    AccountLifecycleRegistrationGateway,
    BillingAccountRegistrationPolicyGateway,
    RedisEmailRegistrationSecurityGateway,
    TokenManagerEmailRegistrationTokenGateway,
)
from services.account.forgot_password_adapters import (
    RateLimiterForgotPasswordSendLimiter,
    RedisForgotPasswordSecurityGateway,
    RedisForgotPasswordTokenGateway,
)
from services.account.login_adapters import RedisConsoleAuthSecurityGateway
from services.account.oauth_adapters import (
    AccountActivationOAuthInvitationGateway,
    AccountLifecycleOAuthRegistrationGateway,
    AccountLifecycleOAuthSessionGateway,
    DeploymentOAuthPolicyGateway,
    RedisOAuthAccountClaimLock,
    WorkspaceProvisioningOAuthGateway,
)
from services.account_avatar_file_gateway import SQLAlchemyAccountAvatarFileGateway
from services.api_based_extension_adapters import APIBasedExtensionPingProbe, WorkspaceTokenCipher
from services.api_based_extension_application_service import APIBasedExtensionApplicationService
from services.app.api_key_service import AppApiKeyService
from services.app.console_gateway import EnterpriseConsoleAppAccess
from services.app.creators_platform_gateway import CreatorsPlatformGateway
from services.app.workflow_comment_mention_gateway import CeleryWorkflowCommentMentionNotifier
from services.app.workflow_comment_service import WorkflowCommentService
from services.app_generate_service import AppGenerateService
from services.app_preview_query_service import AppPreviewRef, AppPreviewUnavailableError
from services.app_scoped_end_user_query_service import AppScopedEndUserQueryService
from services.app_scoped_end_user_service import AppScopedEndUserService
from services.app_site_service import AppSiteService
from services.app_tracing_config_service import AppTracingConfigService
from services.audio_types import AudioAppRef, AudioOutput, AudioUpload
from services.billing_portal_service import BillingPortalService
from services.billing_service import BillingService
from services.compliance_download_service import ComplianceDownloadService
from services.data_source.auth.api_key_service import DataSourceApiKeyAuthService
from services.data_source.binding_application_service import DataSourceBindingApplicationService
from services.data_source.credential_gateway import ActorAwareDatasourceCredentialGateway
from services.data_source.notion_import_adapters import PluginNotionSourceGateway
from services.data_source.notion_import_application_service import NotionImportApplicationService
from services.enterprise.enterprise_service import EnterpriseService
from services.errors.enterprise import EnterpriseAPIError, EnterpriseAPINotFoundError, EnterpriseServiceError
from services.file_service import FileService
from services.human_input_file_upload_service import HumanInputFileUploadService
from services.init_validation_service import InvalidInitializationPasswordError
from services.installed_app_access_service import (
    InstalledAppAccessDeniedError,
    InstalledAppAccessService,
    InstalledAppRef,
)
from services.installed_app_generation_adapters import AppGenerateServiceRuntime as InstalledAppGenerateServiceRuntime
from services.installed_app_generation_service import InstalledAppGenerationService
from services.knowledge.api_key_service import DatasetApiKeyService
from services.knowledge.dataset_access import DatasetAccessService
from services.knowledge.datasets.adapters import SQLAlchemyDatasetOperations
from services.knowledge.datasets.application import DatasetApplicationService
from services.knowledge.document_sync import DocumentSyncApplicationService
from services.knowledge.documents.application import DatasetDocumentApplicationService
from services.knowledge.external.adapters import SQLAlchemyExternalKnowledgeOperations
from services.knowledge.external.application import ExternalKnowledgeApplicationService
from services.knowledge.indexing.adapters.estimate import IndexingEstimateAdapter, SQLAlchemyProcessRuleReader
from services.knowledge.indexing.adapters.sources import NotionSourceResolver
from services.knowledge.indexing.estimate import IndexingEstimateApplicationService
from services.knowledge.segments.application import DatasetSegmentApplicationService
from services.knowledge_fs.builtin_tool.knowledge_fs.knowledge_fs import KnowledgeFSProvider
from services.knowledge_fs.feature_service import KnowledgeFSFeatures
from services.message_file_preview_service import MessageFilePreviewService
from services.network_access_group_gateway import (
    BillingNetworkAccessGroupEntitlementGateway,
    NetworkAccessGroupGateway,
)
from services.network_access_group_service import NetworkAccessGroupAccessDeniedError, NetworkAccessGroupService
from services.oauth_device_application_service import OAuthDeviceApplicationService
from services.plugin.plugin_parameter_gateway import DaemonPluginParameterGateway
from services.plugin_file_upload_gateway import ToolFilePluginUploadGateway
from services.plugin_file_upload_service import PluginFileUploadService
from services.retention.workflow_run.archive_download_task_cache import WorkflowRunArchiveDownloadTaskCache
from services.retention.workflow_run.archive_log_service import WorkflowRunArchiveService
from services.tag_application_service import TagApplicationService
from services.tool_file_download_service import ToolFileDownloadService
from services.tracing.adapters import CeleryTraceEmitter
from services.tracing.runtime_bridge import RuntimeApplicationTracingHooks
from services.trigger.endpoint_gateway import PluginTriggerEndpointGateway
from services.trigger.event_gateway import PluginTriggerEventGateway
from services.trigger.oauth_gateway import PluginTriggerOAuthGateway, TriggerOAuthBuilderGateway
from services.trigger.provider_gateway import PluginTriggerProviderQuery
from services.trigger.subscription_gateway import PluginTriggerSubscriptionGateway
from services.upload_file_delivery_service import UploadFileDeliveryService
from services.web_authentication_adapters import (
    AccountWebAuthenticationSecurityGateway,
    PassportWebAppSessionGateway,
    TokenManagerWebAuthenticationGateway,
)
from services.web_authentication_service import WebAuthenticationService
from services.webapp_access_adapters import EnterpriseWebAppAccessPolicyGateway
from services.webapp_access_query_service import WebAppAccessQueryService, WebAppAccessUnavailableError
from services.workflow_app_log_query_service import WorkflowAppLogQueryService
from services.workflow_run_service import WorkflowRunService
from services.workflow_statistic_query_service import WorkflowStatisticQueryService
from tasks.delete_conversation_task import delete_conversation_related_data
from tests.helpers.unit.config_override import apply_config_overrides
from tests.helpers.unit.model_factories import make_app
from tests.helpers.unit.services.conversation_name_fixtures import _MODEL, _PROVIDER, NamingPlugin
from tests.unit_tests.services.test_app_task_service import _AppTaskServiceStopredis as _StopRedis

pytest_plugins = ("tests.helpers.unit.services.conversation_name_fixtures",)


@pytest.fixture(autouse=True)
def _reject_redis_commands(monkeypatch: pytest.MonkeyPatch) -> None:
    """Service composition must register scripts without performing Redis I/O."""

    def execute_command(*args: object, **_kwargs: object) -> None:
        pytest.fail(f"Unexpected Redis command: {args[1:]}")

    monkeypatch.setattr(Redis, "execute_command", execute_command)


def _redis() -> RedisClientWrapper:
    client = RedisClientWrapper()
    client.initialize(Redis())
    return client


@pytest.mark.parametrize(
    ("deployment_edition", "initialization_password", "session_validated", "setup_exists", "expected"),
    [
        pytest.param(DeploymentEdition.CLOUD, "expected", False, False, True, id="cloud"),
        pytest.param(DeploymentEdition.COMMUNITY, "", False, False, True, id="no-password"),
        pytest.param(DeploymentEdition.COMMUNITY, "expected", False, False, False, id="not-validated"),
        pytest.param(DeploymentEdition.ENTERPRISE, "expected", False, False, False, id="enterprise"),
        pytest.param(DeploymentEdition.COMMUNITY, "expected", True, False, True, id="browser-session"),
        pytest.param(DeploymentEdition.COMMUNITY, "expected", False, True, True, id="setup-record"),
    ],
)
def test_build_application_services_configures_init_validation(
    sqlite_session: Session,
    sqlite_session_factory: sessionmaker[Session],
    deployment_edition: DeploymentEdition,
    initialization_password: str,
    session_validated: bool,
    setup_exists: bool,
    expected: bool,
) -> None:
    if setup_exists:
        sqlite_session.add(DifySetup(version="test-version"))
        sqlite_session.commit()

    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=deployment_edition,
        initialization_password=initialization_password,
        redis=_redis(),
    )

    assert services.init_validation.is_validated(session_validated=session_validated) is expected


def test_build_application_services_passes_the_expected_password(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="expected",
        redis=_redis(),
    )

    services.init_validation.validate_password("expected")
    with pytest.raises(InvalidInitializationPasswordError):
        services.init_validation.validate_password("wrong")


def test_init_app_registers_services_for_the_current_app(
    monkeypatch: pytest.MonkeyPatch,
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    app = Flask(__name__)
    monkeypatch.setattr(ext_application_services, "get_session_maker", lambda: sqlite_session_factory)
    apply_config_overrides(
        monkeypatch,
        DEPLOYMENT_EDITION=DeploymentEdition.COMMUNITY,
        INIT_PASSWORD="expected",
    )

    ext_application_services.init_app(app)

    from extensions.application_services.tools import resolve_tool_credentials

    assert ext_application_services.ToolManager._credential_resolver is resolve_tool_credentials

    with app.app_context():
        services = ext_application_services.application_services()
        assert services is app.extensions["application_services"]
        assert services.init_validation.is_validated(session_validated=False) is False
        assert isinstance(services.app_scoped_end_users.commands, AppScopedEndUserService)
        assert isinstance(services.app_scoped_end_users.queries, AppScopedEndUserQueryService)
        repository = services.app_scoped_end_users.queries._app_scoped_end_users
        assert isinstance(repository, AppScopedEndUserRepo)
        assert services.app_scoped_end_users.commands._app_scoped_end_users is repository
        assert repository._session_factory is sqlite_session_factory
        assert isinstance(services.workflow_statistics, WorkflowStatisticQueryService)
        assert isinstance(services.apps, AppServices)
        assert services.apps.console._apps is services.apps.discovery._apps
        dataset_operations = services.knowledge.datasets._operations
        assert isinstance(dataset_operations, SQLAlchemyDatasetOperations)
        assert isinstance(dataset_operations._app_queries, ConsoleAppRepository)
        assert dataset_operations._app_queries._session_factory is sqlite_session_factory
        creators = services.apps.console._creators
        assert isinstance(creators, CreatorsPlatformGateway)
        assert creators._oauth is services.oauth_server


@pytest.mark.parametrize("enabled", [True, False])
def test_init_app_registers_knowledge_fs_tool_only_when_knowledge_fs_is_enabled(
    monkeypatch: pytest.MonkeyPatch,
    sqlite_session_factory: sessionmaker[Session],
    enabled: bool,
) -> None:
    app = Flask(__name__)
    monkeypatch.setattr(ext_application_services, "get_session_maker", lambda: sqlite_session_factory)
    monkeypatch.setattr(
        ext_application_services,
        "get_knowledge_fs_features",
        lambda: KnowledgeFSFeatures(enabled=enabled, upload_enabled=False, agent_enabled=False),
    )
    monkeypatch.setattr(ext_application_services.ToolManager, "_external_hardcoded_provider_classes", [])
    apply_config_overrides(monkeypatch, DEPLOYMENT_EDITION=DeploymentEdition.COMMUNITY, INIT_PASSWORD="")

    ext_application_services.init_app(app)

    registered = ext_application_services.ToolManager._external_hardcoded_provider_classes
    assert (KnowledgeFSProvider in registered) is enabled


@pytest.mark.parametrize("enabled", [True, False])
def test_init_app_binds_each_apps_agent_hooks_only_when_agent_tracing_is_enabled(
    monkeypatch: pytest.MonkeyPatch,
    sqlite_session_factory: sessionmaker[Session],
    enabled: bool,
) -> None:
    first_app, second_app = Flask("first"), Flask("second")
    monkeypatch.setattr(ext_application_services, "get_session_maker", lambda: sqlite_session_factory)
    apply_config_overrides(
        monkeypatch,
        DEPLOYMENT_EDITION=DeploymentEdition.COMMUNITY,
        INIT_PASSWORD="",
        OPS_AGENT_TRACE_ENABLED=enabled,
    )

    ext_application_services.init_app(first_app)
    ext_application_services.init_app(second_app)

    with first_app.app_context():
        first_hooks = application_tracing_hooks()
        first_services = ext_application_services.application_services()
        with second_app.app_context():
            second_hooks = application_tracing_hooks()
        assert application_tracing_hooks() is first_hooks
    if not enabled:
        assert first_hooks is None
        assert second_hooks is None
        return
    assert isinstance(first_hooks, RuntimeApplicationTracingHooks)
    assert isinstance(second_hooks, RuntimeApplicationTracingHooks)
    assert second_hooks is not first_hooks
    assert first_hooks is first_services.tracing.hooks


def test_init_app_binds_trace_scopes_to_the_apps_tracing_policy(
    monkeypatch: pytest.MonkeyPatch,
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    app = Flask(__name__)
    monkeypatch.setattr(ext_application_services, "get_session_maker", lambda: sqlite_session_factory)
    apply_config_overrides(monkeypatch, DEPLOYMENT_EDITION=DeploymentEdition.COMMUNITY, INIT_PASSWORD="")
    with sqlite_session_factory.begin() as session:
        session.add_all(
            [
                make_app(
                    app_id="traced-app",
                    tenant_id="tenant-1",
                    tracing=json.dumps({"enabled": True, "tracing_provider": "langfuse"}),
                ),
                make_app(app_id="untraced-app", tenant_id="tenant-1"),
            ]
        )

    ext_application_services.init_app(app)

    with app.app_context():
        scope = create_trace_scope_for_app(
            tenant_id="tenant-1", app_id="traced-app", user_id="user-1", extras={"external_trace_id": "external-1"}
        )
        assert scope is not None
        assert isinstance(scope.emitter, CeleryTraceEmitter)
        assert (scope.tenant_id, scope.app_id, scope.user_id) == ("tenant-1", "traced-app", "user-1")
        assert scope.correlation.external_trace_id == "external-1"
        assert create_trace_scope_for_app(tenant_id="tenant-1", app_id="untraced-app", user_id=None, extras={}) is None
    with Flask("unbound").app_context():
        assert create_trace_scope_for_app(tenant_id="tenant-1", app_id="traced-app", user_id=None, extras={}) is None


def test_build_application_services_preserves_composed_boundaries(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    redis = _redis()

    with patch.object(redis, "register_script", wraps=redis.register_script) as register_script:
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.COMMUNITY,
            initialization_password="",
            redis=redis,
        )

    assert isinstance(services.app_api_keys, AppApiKeyService)
    assert isinstance(services.dataset_api_keys, DatasetApiKeyService)
    assert services.dataset_api_keys._access is services.knowledge.datasets._dataset_access
    assert isinstance(services.oauth_device, OAuthDeviceApplicationService)
    assert register_script.call_count == 3
    assert services.conversation_queries._files is services.message_queries._files

    assert isinstance(services.installed_apps.generation, InstalledAppGenerationService)
    installed_apps = services.installed_apps.access._installed_apps
    assert isinstance(installed_apps, SQLAlchemyInstalledAppRepository)
    assert services.installed_apps.management._installed_apps is installed_apps
    assert services.installed_apps.generation._usage is installed_apps
    runtime = services.installed_apps.generation._runtime
    assert isinstance(runtime, InstalledAppGenerateServiceRuntime)
    assert runtime._session_factory is sqlite_session_factory

    trial_apps = services.trial_apps.usage
    assert services.trial_apps.access._apps is trial_apps
    assert services.trial_apps.generation._usage is trial_apps
    assert services.recommended_app_queries._trial_apps is trial_apps

    assert services.rbac.resources._apps is services.apps.console._apps
    assert services.rbac.queries._apps is services.rbac.resources._apps
    assert services.rbac.queries._datasets is services.rbac.resources._datasets
    dataset_access = services.knowledge.datasets._dataset_access
    assert isinstance(dataset_access, DatasetAccessService)
    assert services.rbac.resources._datasets is dataset_access._datasets
    assert services.rbac.members._members is services.rbac.resources._members
    assert services.rbac.members._members is services.workspaces.members._workspaces
    assert services.rbac.members._workspace_roles is services.workspaces.members
    app_access = services.apps.console._access
    assert isinstance(app_access, EnterpriseConsoleAppAccess)
    assert app_access._members is services.rbac.members
    dataset_operations = services.knowledge.datasets._operations
    assert isinstance(dataset_operations, SQLAlchemyDatasetOperations)
    assert dataset_operations._members_service is services.rbac.members
    external_operations = services.knowledge.external._operations
    assert isinstance(external_operations, SQLAlchemyExternalKnowledgeOperations)
    assert external_operations._members is services.rbac.members


@pytest.mark.parametrize(
    ("deployment_edition", "setup_completed"),
    [
        pytest.param(DeploymentEdition.CLOUD, True, id="cloud"),
        pytest.param(DeploymentEdition.COMMUNITY, False, id="community"),
        pytest.param(DeploymentEdition.ENTERPRISE, False, id="enterprise"),
    ],
)
def test_build_application_services_configures_setup_policy(
    sqlite_session_factory: sessionmaker[Session],
    deployment_edition: DeploymentEdition,
    setup_completed: bool,
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=deployment_edition,
        initialization_password="",
        redis=_redis(),
    )

    assert services.setup.get_status().completed is setup_completed
    assert services.oauth_server is not None
    assert isinstance(services.oauth_device._accounts, SQLAlchemyAccountRepository)
    assert isinstance(services.oauth_device._workspaces, WorkspaceRepository)


def test_build_application_services_wires_builtin_schema_definitions(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    definitions = services.schema_definitions.list()

    assert definitions
    assert all({"name", "label", "schema"} <= definition.keys() for definition in definitions)


def test_build_application_services_does_not_construct_schema_manager(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    with patch("extensions.ext_application_services.SchemaManager") as schema_manager:
        ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.COMMUNITY,
            initialization_password="",
            redis=_redis(),
        )

    schema_manager.assert_not_called()


def test_build_application_services_wires_tag_boundary(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert isinstance(services.tags, TagApplicationService)


def test_build_application_services_wires_api_based_extension_boundary(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert isinstance(services.api_based_extensions, APIBasedExtensionApplicationService)
    assert isinstance(services.api_based_extensions._secrets, WorkspaceTokenCipher)
    assert isinstance(services.api_based_extensions._probe, APIBasedExtensionPingProbe)


def test_build_application_services_reuses_file_service(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert isinstance(services.files, FileService)
    assert services.files._session_maker is sqlite_session_factory
    assert services.web_app_runtime._file_service is services.files


def test_build_application_services_wires_message_file_previews(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert isinstance(services.message_file_previews, MessageFilePreviewService)
    assert isinstance(services.message_file_previews._files, MessageFilePreviewQueryRepository)
    assert services.message_file_previews._files._session_factory is sqlite_session_factory
    assert services.message_file_previews._storage is ext_application_services.storage


def test_build_application_services_wires_plugin_file_upload_boundary(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert isinstance(services.plugin_file_uploads, PluginFileUploadService)
    assert isinstance(services.plugin_file_uploads._owners, SQLAlchemyPluginFileUploadOwnerRepository)
    assert services.plugin_file_uploads._owners._session_factory is sqlite_session_factory
    assert isinstance(services.plugin_file_uploads._files, ToolFilePluginUploadGateway)


def test_build_application_services_wires_tool_file_downloads(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert isinstance(services.tool_file_downloads, ToolFileDownloadService)
    assert isinstance(services.tool_file_downloads._tool_files, ToolFileManager)


def test_build_application_services_wires_upload_file_delivery(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert isinstance(services.upload_file_delivery, UploadFileDeliveryService)
    assert isinstance(services.upload_file_delivery._files, UploadFileDeliveryQueryRepository)
    assert services.upload_file_delivery._files._session_factory is sqlite_session_factory
    assert services.upload_file_delivery._storage is ext_application_services.storage


def test_build_application_services_wires_workflow_run_archives(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    redis = _redis()

    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=redis,
    )

    workflow_run_archives = services.workflow_run_archives
    assert isinstance(workflow_run_archives, WorkflowRunArchiveService)
    assert isinstance(workflow_run_archives._bundles, WorkflowRunArchiveBundleQueryRepository)
    assert workflow_run_archives._bundles._session_factory is sqlite_session_factory
    assert isinstance(workflow_run_archives._tasks, WorkflowRunArchiveDownloadTaskCache)
    assert workflow_run_archives._tasks._redis is redis
    assert workflow_run_archives._dispatcher is ext_application_services.dispatch_workflow_run_archive_download_task
    assert workflow_run_archives._sign_download_url is ext_application_services.sign_workflow_run_archive_download_url


def test_build_application_services_wires_human_input_file_uploads(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    human_input_file_uploads = services.human_input_file_uploads
    assert isinstance(human_input_file_uploads, HumanInputFileUploadService)
    assert isinstance(human_input_file_uploads._uploads, SQLAlchemyHumanInputFileUploadRepository)
    assert human_input_file_uploads._uploads._session_factory is sqlite_session_factory
    assert human_input_file_uploads._remote_files is services.remote_files
    assert human_input_file_uploads._files is services.files
    assert services.remote_files._files is services.files


def test_build_application_services_wires_app_site_boundary(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert isinstance(services.app_sites, AppSiteService)
    assert isinstance(services.app_sites._sites, AppSiteCommandRepository)
    assert services.app_sites._sites._session_factory is sqlite_session_factory


def test_build_application_services_wires_web_authentication_boundary(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.ENTERPRISE,
        initialization_password="",
        redis=create_autospec(RedisClientWrapper, instance=True),
    )

    assert isinstance(services.web_authentication, WebAuthenticationService)
    assert services.web_authentication._accounts is services.accounts.profile._accounts
    assert isinstance(services.web_authentication._tokens, TokenManagerWebAuthenticationGateway)
    assert (
        services.web_authentication._tokens._access_token_expire_minutes
        == ext_application_services.dify_config.ACCESS_TOKEN_EXPIRE_MINUTES
    )
    assert isinstance(services.web_authentication._security, AccountWebAuthenticationSecurityGateway)
    assert isinstance(services.web_authentication._security._password_security, RedisForgotPasswordSecurityGateway)
    assert isinstance(services.web_authentication._security._login_security, RedisConsoleAuthSecurityGateway)
    assert isinstance(services.web_authentication._app_sessions, PassportWebAppSessionGateway)
    assert services.web_authentication._app_sessions._sessions is services.webapp_access._access
    assert isinstance(services.webapp_access._policy, EnterpriseWebAppAccessPolicyGateway)
    assert services.webapp_access._policy._webapp_auth is EnterpriseService.WebAppAuth


def test_build_application_services_reuses_installed_app_dependencies(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=create_autospec(RedisClientWrapper, instance=True),
    )

    assert isinstance(services.installed_apps.access, InstalledAppAccessService)
    assert isinstance(services.installed_apps.generation, InstalledAppGenerationService)
    assert isinstance(services.installed_apps.access._installed_apps, SQLAlchemyInstalledAppRepository)
    assert services.installed_apps.generation._usage is services.installed_apps.access._installed_apps
    assert isinstance(services.installed_apps.generation._runtime, InstalledAppGenerateServiceRuntime)
    assert services.installed_apps.generation._runtime._session_factory is sqlite_session_factory


def test_build_application_services_wires_app_tracing_config_boundary(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert isinstance(services.app_tracing_configs, AppTracingConfigService)
    assert isinstance(services.app_tracing_configs._configs, SQLAlchemyAppTracingConfigRepository)
    assert services.app_tracing_configs._configs._session_factory is sqlite_session_factory
    assert services.app_tracing_configs._provider is services.tracing.provider_configs


def test_build_application_services_wires_workflow_app_log_boundary(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert isinstance(services.workflow_app_logs, WorkflowAppLogQueryService)
    assert isinstance(services.workflow_app_logs._logs, WorkflowAppLogQueryRepository)
    assert services.workflow_app_logs._logs._session_factory is sqlite_session_factory


def test_build_application_services_wires_workflow_comment_boundary(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert isinstance(services.workflow_comments, WorkflowCommentService)
    assert isinstance(services.workflow_comments._comments, WorkflowCommentRepository)
    assert services.workflow_comments._comments._session_factory is sqlite_session_factory
    assert isinstance(services.workflow_comments._notifier, CeleryWorkflowCommentMentionNotifier)


def test_build_application_services_wires_app_statistic_boundary(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert isinstance(services.app_statistics, AppStatisticQueryRepository)
    assert services.app_statistics._session_factory is sqlite_session_factory


def test_build_application_services_wires_workflow_run_service(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    workflow_runs = services.workflow_runs
    assert isinstance(workflow_runs, WorkflowRunService)
    assert isinstance(workflow_runs._workflow_runs, DifyAPISQLAlchemyWorkflowRunRepository)
    assert workflow_runs._workflow_runs._session_maker is sqlite_session_factory


def test_build_application_services_wires_billing_service(
    sqlite_session: Session,
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    account = Account(name="Billing Owner", email="owner@example.com")
    account.id = "account-1"
    sqlite_session.add(account)
    sqlite_session.commit()

    with (
        patch.object(
            BillingService,
            "get_subscription",
            return_value={"url": "https://billing.example.com/checkout"},
        ) as get_subscription,
        patch.object(
            BillingService,
            "get_invoices",
            return_value={"url": "https://billing.example.com/portal"},
        ) as get_invoices,
    ):
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.COMMUNITY,
            initialization_password="",
            redis=_redis(),
        )

    request_context = RequestContext(
        request_id="request-1",
        trace_id="trace-1",
        account_id="account-1",
        active_workspace_id="workspace-1",
    )
    assert isinstance(services.billing_portal, BillingPortalService)
    assert services.billing_portal.get_subscription(
        request_context,
        plan="professional",
        interval="month",
    ) == {"url": "https://billing.example.com/checkout"}
    assert services.billing_portal.get_invoices(request_context) == {"url": "https://billing.example.com/portal"}
    get_subscription.assert_called_once_with("professional", "month", "owner@example.com", "workspace-1")
    get_invoices.assert_called_once_with("owner@example.com", "workspace-1")


def test_build_application_services_wires_network_access_group_boundary(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.CLOUD,
        initialization_password="",
        redis=_redis(),
    )

    network_access_groups = services.network_access_groups
    assert isinstance(network_access_groups, NetworkAccessGroupService)
    assert isinstance(network_access_groups._control_plane, NetworkAccessGroupGateway)
    assert isinstance(network_access_groups._apps, SQLAlchemyNetworkAccessGroupAppRepository)
    assert network_access_groups._apps._session_factory is sqlite_session_factory
    assert isinstance(network_access_groups._entitlement, BillingNetworkAccessGroupEntitlementGateway)

    # Exercise the composed membership port, not only the class of its owner.
    # A workspace-repository migration must not leave a stale method signature.
    context = RequestContext("request-1", None, "account-1", "workspace-1", "203.0.113.42")
    with pytest.raises(NetworkAccessGroupAccessDeniedError):
        network_access_groups.get_current_ip(context, client_ip_supplier=lambda: "203.0.113.42")
    with sqlite_session_factory.begin() as session:
        account = Account(name="QA owner", email="qa-owner@example.com")
        account.id = context.account_id
        workspace = Tenant(name="QA workspace")
        workspace.id = context.active_workspace_id
        session.add_all(
            [
                account,
                workspace,
                TenantAccountJoin(
                    tenant_id=workspace.id,
                    account_id=account.id,
                    role=TenantAccountRole.OWNER,
                ),
            ]
        )
    assert network_access_groups.get_current_ip(context, client_ip_supplier=lambda: "203.0.113.42") == "203.0.113.42"


def test_build_application_services_wires_compliance_downloads(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    redis = _redis()
    with (
        patch.object(
            BillingService,
            "get_compliance_download_link",
            return_value={"url": "https://billing.example.com/compliance"},
        ) as fetch_link,
        patch("extensions.ext_application_services.RateLimiter") as rate_limiter_type,
    ):
        rate_limiter = rate_limiter_type.return_value
        rate_limiter.is_rate_limited.return_value = False
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.COMMUNITY,
            initialization_password="",
            redis=redis,
        )

    assert isinstance(services.compliance_downloads, ComplianceDownloadService)
    assert services.compliance_downloads.get_link(
        request_context=RequestContext(
            request_id="request-1",
            trace_id="trace-1",
            account_id="account-1",
            active_workspace_id="workspace-1",
        ),
        document_name="SOC2_Type_II",
        ip_address="127.0.0.1",
        device_info="test-agent",
    ) == {"url": "https://billing.example.com/compliance"}
    rate_limiter_type.assert_any_call(
        prefix="compliance_download_rate_limiter",
        max_attempts=4,
        time_window=60,
        redis_client=redis,
    )
    rate_limiter.is_rate_limited.assert_called_once_with("account-1:workspace-1")
    rate_limiter.increment_rate_limit.assert_called_once_with("account-1:workspace-1")
    fetch_link.assert_called_once_with("SOC2_Type_II", "account-1", "workspace-1", "127.0.0.1", "test-agent")


def test_build_application_services_wires_education_rate_limiters(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    redis = _redis()
    with patch("extensions.application_services.account.RateLimiter") as rate_limiter_type:
        ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.COMMUNITY,
            initialization_password="",
            redis=redis,
        )

    rate_limiter_type.assert_any_call(
        prefix="edu_verification_rate_limit",
        max_attempts=10,
        time_window=60,
        redis_client=redis,
    )
    rate_limiter_type.assert_any_call(
        prefix="edu_activation_rate_limit",
        max_attempts=10,
        time_window=60,
        redis_client=redis,
    )


def test_build_application_services_wires_account_profile_repository(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    accounts = services.accounts.profile._accounts
    assert isinstance(accounts, SQLAlchemyAccountRepository)
    assert accounts._session_factory is sqlite_session_factory
    assert services.accounts.password._accounts is accounts
    assert services.accounts.authentication._passwords is services.accounts.password._passwords
    forgot_password = services.accounts.forgot_password
    assert forgot_password._accounts is accounts
    assert forgot_password._passwords is services.accounts.password._passwords
    tokens = cast(RedisForgotPasswordTokenGateway, forgot_password._tokens)
    send_limiter = cast(RateLimiterForgotPasswordSendLimiter, forgot_password._send_limits)
    security = cast(RedisForgotPasswordSecurityGateway, forgot_password._security)
    assert tokens._redis is send_limiter._rate_limiter._redis_client
    assert send_limiter._rate_limiter.prefix == account_forgot_password_service.FORGOT_PASSWORD_SEND_RATE_LIMIT_PREFIX
    assert (
        send_limiter._rate_limiter.max_attempts
        == account_forgot_password_service.FORGOT_PASSWORD_SEND_RATE_LIMIT_MAX_ATTEMPTS
    )
    assert (
        security._verification_failure_limit
        == account_forgot_password_service.FORGOT_PASSWORD_VERIFICATION_FAILURE_LIMIT
    )
    assert security._verification_key_prefix == account_forgot_password_service.FORGOT_PASSWORD_VERIFICATION_KEY_PREFIX
    assert services.accounts.initialization._accounts is accounts
    assert not services.accounts.initialization._invitation_required
    assert services.accounts.change_email._accounts is accounts
    email_registration = services.accounts.email_registration
    assert email_registration._accounts is accounts
    assert isinstance(email_registration._tokens, TokenManagerEmailRegistrationTokenGateway)
    assert isinstance(email_registration._security, RedisEmailRegistrationSecurityGateway)
    assert isinstance(email_registration._account_policy, BillingAccountRegistrationPolicyGateway)
    assert isinstance(email_registration._registration, AccountLifecycleRegistrationGateway)
    assert email_registration._registration._accounts is services.accounts.lifecycle
    assert services.workspaces.invitations._accounts is services.accounts.lifecycle
    assert services.workspaces.invitations._members is services.workspaces.members
    assert services.workspaces.invitations._workspaces is services.workspaces.management._workspaces
    assert services.workspaces.provisioning._members is services.workspaces.members
    assert services.accounts.lifecycle._workspaces is services.workspaces.provisioning
    assert services.workspaces.identity._workspaces is services.workspaces.management._workspaces
    assert services.accounts.education._accounts is accounts
    assert services.accounts.deletion._accounts is accounts
    assert services.accounts.authentication._accounts is accounts
    assert services.accounts.authentication._workspaces is services.workspaces.queries._workspaces
    assert services.step_by_step_tour._accounts is accounts
    assert services.accounts.deletion._memberships is services.workspaces.queries._workspaces
    integrations = services.accounts.integrations._integrations
    assert isinstance(integrations, SQLAlchemyAccountIntegrationRepository)
    assert integrations._session_factory is sqlite_session_factory
    oauth = services.accounts.oauth
    assert oauth._accounts is accounts
    assert oauth._integrations is integrations
    assert oauth._memberships is services.workspaces.queries._workspaces
    assert isinstance(oauth._invitations, AccountActivationOAuthInvitationGateway)
    assert isinstance(oauth._account_claims, RedisOAuthAccountClaimLock)
    assert isinstance(oauth._registration, AccountLifecycleOAuthRegistrationGateway)
    assert isinstance(oauth._workspaces, WorkspaceProvisioningOAuthGateway)
    assert isinstance(oauth._sessions, AccountLifecycleOAuthSessionGateway)
    assert oauth._sessions is not oauth._workspaces
    assert isinstance(oauth._registration_policy, DeploymentOAuthPolicyGateway)
    assert oauth._workspace_policy is oauth._registration_policy
    avatar_files = services.accounts.avatar._files
    assert isinstance(avatar_files, SQLAlchemyAccountAvatarFileGateway)
    assert avatar_files._session_factory is sqlite_session_factory


def test_build_application_services_requires_invitation_for_cloud_initialization(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.CLOUD,
        initialization_password="",
        redis=_redis(),
    )

    assert services.accounts.initialization._invitation_required


@pytest.mark.parametrize(
    ("deployment_edition", "billing_enabled"),
    [
        pytest.param(DeploymentEdition.CLOUD, True, id="cloud"),
        pytest.param(DeploymentEdition.COMMUNITY, False, id="community"),
        pytest.param(DeploymentEdition.ENTERPRISE, False, id="enterprise"),
    ],
)
def test_build_application_services_wires_account_activation(
    sqlite_session_factory: sessionmaker[Session],
    deployment_edition: DeploymentEdition,
    billing_enabled: bool,
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=deployment_edition,
        initialization_password="",
        redis=_redis(),
    )

    activation = services.accounts.activation
    assert isinstance(activation._tokens, RedisInvitationTokenStore)
    assert isinstance(activation._accounts, SQLAlchemyAccountActivationRepository)
    assert activation._accounts._session_factory is sqlite_session_factory
    assert isinstance(activation._workspace_policy, DeploymentWorkspaceInvitePolicy)
    assert isinstance(activation._eligibility, BillingAccountActivationEligibility)
    assert activation._eligibility._enabled is billing_enabled
    assert isinstance(activation._membership_cache, BillingWorkspaceMembershipCache)
    assert activation._membership_cache._enabled is billing_enabled
    assert isinstance(activation._member_access_sync, RBACWorkspaceMemberAccessSync)


def test_build_application_services_wires_data_source_api_key_auth(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert isinstance(services.data_sources.api_key_auth, DataSourceApiKeyAuthService)


def test_build_application_services_groups_dataset_services_and_reuses_repositories(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert isinstance(services.data_sources.bindings, DataSourceBindingApplicationService)
    assert isinstance(services.data_sources.notion_imports, NotionImportApplicationService)
    assert isinstance(services.knowledge.document_sync, DocumentSyncApplicationService)
    assert isinstance(services.knowledge.documents, DatasetDocumentApplicationService)
    assert isinstance(services.knowledge.datasets, DatasetApplicationService)
    assert isinstance(services.knowledge.external, ExternalKnowledgeApplicationService)
    assert services.knowledge.datasets._dataset_access is services.knowledge.document_sync._dataset_access
    assert services.knowledge.external._dataset_access is services.knowledge.document_sync._dataset_access
    assert services.knowledge.documents._dataset_access is services.knowledge.document_sync._dataset_access
    assert isinstance(services.knowledge.indexing_estimates, IndexingEstimateApplicationService)
    assert isinstance(services.knowledge.segments, DatasetSegmentApplicationService)
    assert services.data_sources.bindings._bindings is services.data_sources.oauth["notion"]._bindings
    assert services.data_sources.notion_imports._dataset_access is services.knowledge.document_sync._dataset_access
    assert services.data_sources.notion_imports._dataset_access is services.knowledge.indexing_estimates._dataset_access
    assert services.data_sources.notion_imports._documents is services.knowledge.document_sync._documents
    assert services.data_sources.notion_imports._documents is services.knowledge.indexing_estimates._documents
    notion_source = services.data_sources.notion_imports._source
    estimate_service = services.knowledge.indexing_estimates
    dataset_access = services.data_sources.notion_imports._dataset_access
    assert isinstance(notion_source, PluginNotionSourceGateway)
    assert isinstance(estimate_service._runner, IndexingEstimateAdapter)
    assert isinstance(estimate_service._process_rules, SQLAlchemyProcessRuleReader)
    assert isinstance(dataset_access, DatasetAccessService)
    notion_resolver = estimate_service._notion
    assert isinstance(notion_resolver, NotionSourceResolver)
    actor_credentials = notion_source._credentials
    assert isinstance(actor_credentials, ActorAwareDatasourceCredentialGateway)
    assert actor_credentials is notion_resolver._actor_credentials
    assert services.data_sources.providers._credentials is actor_credentials._credentials
    assert services.knowledge.pipeline_generator._datasource_providers is services.data_sources.providers
    assert notion_resolver._stored_credentials is not notion_resolver._actor_credentials
    assert services.workspaces.member_queries._members is dataset_access._workspace_roles


def test_build_application_services_wires_credential_query(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )
    tenant_id, actor_id = str(uuid4()), str(uuid4())
    with sqlite_session_factory.begin() as session:
        credential = ProviderCredential(
            tenant_id=tenant_id, provider_name="openai", credential_name="Team", encrypted_config="encrypted"
        )
        session.add(credential)
        credential_id = credential.id

    records = services.credential_queries.list_models(workspace_id=tenant_id, provider="openai", actor_id=actor_id)

    assert [(record.id, record.name) for record in records] == [(credential_id, "Team")]
    assert services.triggers.subscriptions._credentials is services.credential_queries
    assert services.triggers.events._subscriptions is services.triggers.subscriptions
    assert services.triggers.binding_queries is services.triggers.subscription_management._subscriptions
    assert services.triggers.events._requests is services.triggers.endpoints._requests
    assert services.workspaces.plugin_parameters._credentials is services.credential_queries
    assert services.workspaces.plugin_parameters._builders is services.triggers.builders
    assert (
        services.workspaces.plugin_parameters._subscriptions is services.triggers.subscription_management._subscriptions
    )
    assert services.triggers.subscriptions._schemas is services.triggers.providers
    assert services.triggers.subscriptions._subscriptions is services.triggers.subscription_management._subscriptions
    gateway = services.triggers.oauth._builders
    assert isinstance(gateway, TriggerOAuthBuilderGateway)
    assert gateway._builders is services.triggers.builders
    assert services.triggers.builders._create_subscription == services.triggers.subscription_management.create
    assert services.triggers.endpoints._subscriptions is services.triggers.subscription_management._subscriptions
    assert (
        services.triggers.subscription_renewal._subscriptions
        is services.triggers.subscription_management._subscriptions
    )
    assert services.triggers.subscription_renewal._providers is services.triggers.subscription_management._providers
    assert services.triggers.subscription_refresh._renewal is services.triggers.subscription_renewal
    assert (
        services.triggers.subscription_refresh._subscriptions
        is services.triggers.subscription_management._subscriptions
    )
    assert services.triggers.subscription_renewal._locks is services.triggers.subscription_management._locks
    assert services.triggers.subscription_renewal._codec is services.triggers.subscription_management._codec
    assert services.triggers.subscription_renewal._clients is services.triggers.oauth_clients
    assert services.triggers.subscription_renewal._oauth is services.triggers.oauth._providers
    provider_query = services.triggers.providers
    events_gateway = services.triggers.events._plugins
    endpoint_gateway = services.triggers.endpoints._provider
    subscription_gateway = services.triggers.subscription_management._providers
    oauth_gateway = services.triggers.oauth._providers
    parameter_gateway = services.workspaces.plugin_parameters._plugins
    assert isinstance(provider_query, PluginTriggerProviderQuery)
    assert isinstance(events_gateway, PluginTriggerEventGateway)
    assert isinstance(endpoint_gateway, PluginTriggerEndpointGateway)
    assert isinstance(subscription_gateway, PluginTriggerSubscriptionGateway)
    assert isinstance(oauth_gateway, PluginTriggerOAuthGateway)
    assert isinstance(parameter_gateway, DaemonPluginParameterGateway)
    provider_loader = provider_query._providers
    assert events_gateway._providers is provider_loader
    assert endpoint_gateway._providers is provider_loader
    assert subscription_gateway._providers is provider_loader
    assert oauth_gateway._providers is provider_loader
    assert parameter_gateway._providers is provider_loader


def test_build_application_services_uses_supplied_redis_for_both_workflow_stop_signals(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    redis = _StopRedis(read_error=AssertionError("Workflow stop must not inspect task ownership"))
    script = MagicMock(side_effect=AssertionError("Workflow stop must not execute Redis scripts"))
    with patch.object(redis, "register_script", return_value=script):
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.COMMUNITY,
            initialization_password="",
            redis=redis,
        )

    services.app_tasks.stop_workflow_task_no_user_check(task_id="workflow-task")

    assert redis.reads == []
    assert redis.operations == ["legacy_flag", "graph_command"]
    assert redis.values["generate_task_stopped:workflow-task"] == b"1"
    assert redis.expirations["generate_task_stopped:workflow-task"] == 600
    assert [json.loads(command) for command in redis.commands["workflow:workflow-task:commands"]] == [
        {"command_type": "abort", "reason": "User requested stop"}
    ]
    assert redis.expirations["workflow:workflow-task:commands"] == 3600


def test_build_application_services_wires_trial_app_usage(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )
    app_id = str(uuid4())
    account_id = str(uuid4())

    services.trial_apps.usage.record(app_id=app_id, account_id=account_id)

    with sqlite_session_factory() as session:
        record = session.scalar(
            select(AccountTrialAppRecord).where(
                AccountTrialAppRecord.app_id == app_id,
                AccountTrialAppRecord.account_id == account_id,
            )
        )
    assert record is not None
    assert record.count == 1


@pytest.fixture
def published_webapp_id(sqlite_session: Session) -> str:
    """Keep public admission real so wiring tests reach the Enterprise adapter."""
    app_id = str(uuid4())
    config = AppModelConfig(app_id=app_id)
    config.id = str(uuid4())
    app = App(
        id=app_id,
        tenant_id=str(uuid4()),
        name="Published adapter fixture",
        mode=AppMode.CHAT,
        status=AppStatus.NORMAL,
        enable_site=True,
        enable_api=True,
        app_model_config_id=config.id,
    )
    site = Site(
        app_id=app_id,
        code="published-adapter-fixture",
        title="Published adapter fixture",
        default_language="en-US",
        customize_token_strategy=CustomizeTokenStrategy.UUID,
        status=AppStatus.NORMAL,
    )
    sqlite_session.add_all([app, config, site])
    sqlite_session.commit()
    return app_id


@pytest.fixture
def installed_app_ref(sqlite_session_factory: sessionmaker[Session]) -> InstalledAppRef:
    with sqlite_session_factory.begin() as session:
        app = App(
            tenant_id=str(uuid4()),
            name="Installed app",
            mode=AppMode.COMPLETION,
            enable_site=True,
            enable_api=True,
        )
        session.add(app)
        session.flush()
        installed_app = InstalledApp(
            tenant_id=str(uuid4()),
            app_id=app.id,
            app_owner_tenant_id=app.tenant_id,
            is_pinned=False,
        )
        session.add(installed_app)
        session.flush()
        result = InstalledAppRef(
            id=installed_app.id,
            app_id=app.id,
            tenant_id=installed_app.tenant_id,
            app_owner_tenant_id=app.tenant_id,
            app_mode=app.mode.value,
        )
    return result


def test_build_application_services_reuses_installed_app_generation_dependencies(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    assert services.installed_apps.access._installed_apps is services.installed_apps.generation._usage


@pytest.mark.parametrize(
    ("deployment_edition", "permission_result"),
    [
        pytest.param(DeploymentEdition.COMMUNITY, False, id="community-skips-permission"),
        pytest.param(DeploymentEdition.CLOUD, False, id="cloud-skips-permission"),
        pytest.param(DeploymentEdition.ENTERPRISE, True, id="enterprise-allowed"),
        pytest.param(DeploymentEdition.ENTERPRISE, False, id="enterprise-denied"),
    ],
)
def test_build_application_services_wires_installed_app_admission(
    sqlite_session_factory: sessionmaker[Session],
    installed_app_ref: InstalledAppRef,
    deployment_edition: DeploymentEdition,
    permission_result: bool,
) -> None:
    account_id = str(uuid4())
    with patch(
        "services.enterprise.enterprise_service.EnterpriseRequest.send_request",
        return_value={"result": permission_result},
    ) as enterprise_request:
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=deployment_edition,
            initialization_password="",
            redis=_redis(),
        )
        if deployment_edition == DeploymentEdition.ENTERPRISE and not permission_result:
            with pytest.raises(InstalledAppAccessDeniedError):
                services.installed_apps.access.get_access(
                    installed_app_id=installed_app_ref.id,
                    tenant_id=installed_app_ref.tenant_id,
                    account_id=account_id,
                )
        else:
            assert (
                services.installed_apps.access.get_access(
                    installed_app_id=installed_app_ref.id,
                    tenant_id=installed_app_ref.tenant_id,
                    account_id=account_id,
                )
                == installed_app_ref
            )

    if deployment_edition == DeploymentEdition.ENTERPRISE:
        enterprise_request.assert_called_once_with(
            "GET", "/webapp/permission", params={"userId": account_id, "appId": installed_app_ref.app_id}
        )
    else:
        enterprise_request.assert_not_called()


@pytest.mark.parametrize(
    "enterprise_error",
    [
        pytest.param(EnterpriseAPINotFoundError(), id="not-found"),
        pytest.param(EnterpriseAPIError("permission unavailable"), id="api-error"),
        pytest.param(httpx.ConnectError("connection failed"), id="transport"),
        pytest.param(json.JSONDecodeError("invalid", "", 0), id="invalid-json"),
        pytest.param(UnicodeDecodeError("utf-8", b"\xff", 0, 1, "invalid"), id="invalid-encoding"),
    ],
)
def test_installed_app_admission_normalizes_known_enterprise_errors(
    sqlite_session_factory: sessionmaker[Session],
    installed_app_ref: InstalledAppRef,
    enterprise_error: Exception,
) -> None:
    account_id = str(uuid4())
    with patch(
        "services.enterprise.enterprise_service.EnterpriseRequest.send_request",
        side_effect=enterprise_error,
    ) as enterprise_request:
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.ENTERPRISE,
            initialization_password="",
            redis=_redis(),
        )
        with pytest.raises(WebAppAccessUnavailableError) as raised:
            services.installed_apps.access.get_access(
                installed_app_id=installed_app_ref.id,
                tenant_id=installed_app_ref.tenant_id,
                account_id=account_id,
            )

    assert raised.value.__cause__ is enterprise_error
    enterprise_request.assert_called_once_with(
        "GET", "/webapp/permission", params={"userId": account_id, "appId": installed_app_ref.app_id}
    )


def test_trial_generation_uses_configured_access_runtime_and_usage(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )
    app_id, tenant_id, account_id = str(uuid4()), str(uuid4()), str(uuid4())
    with sqlite_session_factory.begin() as session:
        session.add(
            App(
                id=app_id,
                tenant_id=tenant_id,
                name="Trial",
                mode=AppMode.COMPLETION,
                enable_site=True,
                enable_api=False,
            )
        )
        account = Account(name="Account", email="trial@example.com")
        account.id = account_id
        session.add(account)
        session.add(TrialApp(app_id=app_id, tenant_id=tenant_id))

    admitted = services.trial_apps.access.get_access(app_id=app_id, account_id=account_id)
    with patch.object(AppGenerateService, "generate", return_value={"answer": "hello"}):
        response = services.trial_apps.generation.generate_completion(
            trial_app=admitted, account_id=account_id, args={"inputs": {}}
        )

    assert response == {"answer": "hello"}
    with sqlite_session_factory() as session:
        record = session.scalar(
            select(AccountTrialAppRecord).where(
                AccountTrialAppRecord.app_id == app_id, AccountTrialAppRecord.account_id == account_id
            )
        )
        assert record is not None
        assert record.count == 1


def test_app_audio_uses_the_configured_database_and_app_owner(
    sqlite_session_factory: sessionmaker[Session], monkeypatch: pytest.MonkeyPatch
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )
    app_id, tenant_id, account_id = str(uuid4()), str(uuid4()), str(uuid4())
    with sqlite_session_factory.begin() as session:
        config = AppModelConfig(app_id=app_id, speech_to_text='{"enabled":true}')
        session.add(config)
        session.flush()
        session.add(
            App(
                id=app_id,
                tenant_id=tenant_id,
                name="Trial",
                mode=AppMode.CHAT,
                app_model_config_id=config.id,
                enable_site=True,
                enable_api=False,
            )
        )
        session.add(TrialApp(app_id=app_id, tenant_id=tenant_id))

    def transcribe(*, app: AudioAppRef, content: bytes, end_user: str | None) -> str:
        assert app == AudioAppRef(app_id=app_id, tenant_id=tenant_id, app_mode="chat")
        assert content == b"audio"
        assert end_user is None
        return "transcript"

    def synthesize(*, app: AudioAppRef, text: str, voice: str | None, end_user: str | None) -> AudioOutput:
        assert app == AudioAppRef(app_id=app_id, tenant_id=tenant_id, app_mode="chat")
        assert (text, voice, end_user) == ("read", "voice", None)
        return AudioOutput(data=b"audio", mime_type="audio/mpeg")

    monkeypatch.setattr(audio_provider_gateway, "speech_to_text", transcribe)
    monkeypatch.setattr(audio_provider_gateway, "text_to_speech", synthesize)
    admitted = services.trial_apps.access.get_access(app_id=app_id, account_id=account_id)
    assert services.app_audio.transcript_asr(
        app=AudioAppRef(admitted.app_id, admitted.tenant_id, admitted.app_mode),
        audio=AudioUpload(stream=BytesIO(b"audio"), mime_type="audio/mp3"),
    ) == {"text": "transcript"}
    assert services.app_audio.transcript_tts(
        app=AudioAppRef(admitted.app_id, admitted.tenant_id, admitted.app_mode),
        account_id=account_id,
        text=" read ",
        voice="voice",
        message_id=None,
    ) == AudioOutput(data=b"audio", mime_type="audio/mpeg")


def test_app_previews_use_the_configured_catalog_and_app_owner(
    sqlite_session_factory: sessionmaker[Session], monkeypatch: pytest.MonkeyPatch
) -> None:
    apply_config_overrides(monkeypatch, HOSTED_FETCH_APP_TEMPLATES_MODE="builtin")
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )
    app_id, tenant_id, other_id = str(uuid4()), str(uuid4()), str(uuid4())
    with sqlite_session_factory.begin() as session:
        session.add_all(
            [
                App(
                    id=app_id,
                    tenant_id=tenant_id,
                    name="Preview",
                    mode=AppMode.CHAT,
                    enable_site=False,
                    enable_api=False,
                ),
                App(
                    id=other_id,
                    tenant_id=tenant_id,
                    name="Private",
                    mode=AppMode.CHAT,
                    enable_site=False,
                    enable_api=False,
                ),
            ]
        )

    # Catalog-only previews must work without a Trial registration or account.
    payload = json.dumps({"app_details": {app_id: {"id": app_id}}})
    with patch.object(recommended_app_catalog_gateway.Path, "read_text", return_value=payload):
        assert services.app_previews.get_access(app_id=app_id) == AppPreviewRef(app_id=app_id, tenant_id=tenant_id)
        with pytest.raises(AppPreviewUnavailableError, match=other_id):
            services.app_previews.get_access(app_id=other_id)


def test_app_preview_details_use_the_configured_database_without_request_globals(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )
    app_id, owner_id, viewer_workspace_id = str(uuid4()), str(uuid4()), str(uuid4())
    account = Account(name="Preview viewer", email="preview@example.com")
    with sqlite_session_factory.begin() as session:
        session.add_all(
            [
                account,
                App(
                    id=app_id,
                    tenant_id=owner_id,
                    name="Preview",
                    mode=AppMode.CHAT,
                    enable_site=True,
                    enable_api=False,
                ),
                Site(
                    app_id=app_id,
                    title="Preview site",
                    default_language="en-US",
                    customize_token_strategy=CustomizeTokenStrategy.UUID,
                ),
            ]
        )

    detail = services.app_preview_details.get_detail(
        app=AppPreviewRef(app_id=app_id, tenant_id=owner_id),
        account_id=account.id,
        active_workspace_id=viewer_workspace_id,
    )

    assert detail.id == app_id
    assert detail.name == "Preview"
    assert detail.site.title == "Preview site"
    assert detail.model_config is None


def test_build_application_services_adapts_enterprise_webapp_access_mode(
    sqlite_session_factory: sessionmaker[Session],
    published_webapp_id: str,
) -> None:
    with (
        patch("extensions.ext_application_services.SystemFeatureService.is_webapp_auth_enabled", return_value=True),
        patch(
            "extensions.ext_application_services.EnterpriseService.WebAppAuth.get_app_access_mode_by_id",
            return_value=SimpleNamespace(access_mode="private_all"),
        ) as get_access_mode,
    ):
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.COMMUNITY,
            initialization_password="",
            redis=_redis(),
        )
        result = services.webapp_access.get_access_mode(app_id=published_webapp_id, app_code=None)

    assert result is WebAppAccessMode.PRIVATE_ALL
    get_access_mode.assert_called_once_with(published_webapp_id)


def _query_webapp_access(
    service: WebAppAccessQueryService, query_kind: str, published_webapp_id: str
) -> WebAppAccessMode | bool | Mapping[str, WebAppAccessMode] | Mapping[str, bool]:
    if query_kind == "single-mode":
        return service.get_access_mode(app_id=published_webapp_id, app_code=None)
    if query_kind == "single-permission":
        return service.is_user_allowed(user_id="viewer", app_id="app-1")
    if query_kind == "batch-modes":
        return service.batch_get_access_modes(app_ids=("app-1",))
    assert query_kind == "batch-permissions"
    return service.batch_get_user_permissions(user_id="viewer", app_ids=("app-1",))


@pytest.mark.parametrize("query_kind", ["single-mode", "single-permission", "batch-modes", "batch-permissions"])
@pytest.mark.parametrize(
    "enterprise_error",
    [
        pytest.param(httpx.ReadTimeout("Enterprise timed out"), id="timeout"),
        pytest.param(httpx.ConnectError("connection failed"), id="connection"),
        pytest.param(EnterpriseServiceError("upstream failure"), id="upstream"),
        pytest.param(json.JSONDecodeError("invalid", "", 0), id="invalid-json"),
        pytest.param(UnicodeDecodeError("utf-8", b"\xff", 0, 1, "invalid"), id="invalid-encoding"),
    ],
)
def test_webapp_access_queries_map_known_enterprise_errors_to_unavailable(
    sqlite_session_factory: sessionmaker[Session],
    published_webapp_id: str,
    query_kind: str,
    enterprise_error: Exception,
) -> None:
    with patch(
        "services.enterprise.enterprise_service.EnterpriseRequest.send_request", side_effect=enterprise_error
    ) as enterprise_request:
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.ENTERPRISE,
            initialization_password="",
            redis=_redis(),
        )
        with pytest.raises(WebAppAccessUnavailableError) as raised:
            _query_webapp_access(services.webapp_access, query_kind, published_webapp_id)

    assert type(raised.value) is WebAppAccessUnavailableError
    assert raised.value.__cause__ is enterprise_error
    assert enterprise_request.call_count == 1


@pytest.mark.parametrize("access_mode", ["invalid", 123])
def test_single_webapp_mode_maps_invalid_enum_or_field_value_to_unavailable(
    sqlite_session_factory: sessionmaker[Session], published_webapp_id: str, access_mode: str | int
) -> None:
    with patch(
        "services.enterprise.enterprise_service.EnterpriseRequest.send_request",
        return_value={"accessMode": access_mode},
    ):
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.ENTERPRISE,
            initialization_password="",
            redis=_redis(),
        )
        with pytest.raises(WebAppAccessUnavailableError) as raised:
            services.webapp_access.get_access_mode(app_id=published_webapp_id, app_code=None)

    assert type(raised.value) is WebAppAccessUnavailableError
    assert isinstance(raised.value.__cause__, ValueError)


@pytest.mark.parametrize("query_kind", ["single-mode", "single-permission", "batch-modes", "batch-permissions"])
@pytest.mark.parametrize("failure", [TypeError("adapter bug"), ValueError("unexpected programming error")])
def test_webapp_access_queries_do_not_hide_unknown_programming_errors(
    sqlite_session_factory: sessionmaker[Session], published_webapp_id: str, query_kind: str, failure: Exception
) -> None:
    with patch("services.enterprise.enterprise_service.EnterpriseRequest.send_request", side_effect=failure):
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.ENTERPRISE,
            initialization_password="",
            redis=_redis(),
        )
        with pytest.raises(type(failure)) as raised:
            _query_webapp_access(services.webapp_access, query_kind, published_webapp_id)

    assert raised.value is failure


def test_build_application_services_wires_webapp_permission(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    with (
        patch(
            "extensions.ext_application_services.SystemFeatureService.is_webapp_auth_enabled", return_value=True
        ) as enabled,
        patch(
            "extensions.ext_application_services.EnterpriseService.WebAppAuth.get_app_access_mode_by_id",
            return_value=SimpleNamespace(access_mode="private"),
        ) as get_access_mode,
        patch(
            "extensions.ext_application_services.EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp",
            return_value=False,
        ) as is_user_allowed,
    ):
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.COMMUNITY,
            initialization_password="",
            redis=_redis(),
        )
        requires_permission = services.webapp_access.requires_permission_check("app-1")
        allowed = services.webapp_access.is_user_allowed(user_id="user-1", app_id="app-1")

    assert requires_permission is True
    assert allowed is False
    enabled.assert_has_calls(
        [
            call(deployment_edition=DeploymentEdition.COMMUNITY),
            call(deployment_edition=DeploymentEdition.COMMUNITY),
        ]
    )
    get_access_mode.assert_called_once_with("app-1")
    is_user_allowed.assert_called_once_with("user-1", "app-1")


def test_webapp_permission_adapter_maps_connection_failure() -> None:
    failure = httpx.ConnectError("connection failed")
    adapter = EnterpriseWebAppAccessPolicyGateway(webapp_auth=EnterpriseService.WebAppAuth)
    with (
        patch(
            "extensions.ext_application_services.EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp",
            side_effect=failure,
        ),
        pytest.raises(WebAppAccessUnavailableError) as raised,
    ):
        adapter.is_user_allowed(user_id="user-1", app_id="app-1")

    assert raised.value.__cause__ is failure


def test_build_application_services_wires_dynamic_recommended_catalog(
    sqlite_session_factory: sessionmaker[Session],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    apply_config_overrides(monkeypatch, HOSTED_FETCH_APP_TEMPLATES_MODE="builtin")
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )

    builtin_payload = json.dumps(
        {
            "recommended_apps": {
                "en-US": {
                    "recommended_apps": [{"app": None, "app_id": "app-1", "categories": []}],
                    "categories": [],
                }
            }
        }
    )
    with patch.object(recommended_app_catalog_gateway.Path, "read_text", return_value=builtin_payload):
        result = services.recommended_app_queries.list_recommended(
            language="en-US",
        )
    assert result.recommended_apps

    apply_config_overrides(monkeypatch, HOSTED_FETCH_APP_TEMPLATES_MODE="invalid")
    with pytest.raises(ValueError, match="invalid fetch recommended apps mode: invalid"):
        services.recommended_app_queries.list_recommended(
            language="en-US",
        )


@pytest.mark.parametrize(
    ("deployment_edition", "permission_result"),
    [
        pytest.param(DeploymentEdition.COMMUNITY, False, id="community-skips-enterprise"),
        pytest.param(DeploymentEdition.CLOUD, False, id="cloud-skips-enterprise"),
        pytest.param(DeploymentEdition.ENTERPRISE, True, id="enterprise-allowed"),
        pytest.param(DeploymentEdition.ENTERPRISE, False, id="enterprise-denied"),
    ],
)
def test_installed_app_management_composition_reads_real_installations_and_current_workspace_role(
    sqlite_session_factory: sessionmaker[Session],
    installed_app_ref: InstalledAppRef,
    deployment_edition: DeploymentEdition,
    permission_result: bool,
) -> None:
    account_id = str(uuid4())
    with sqlite_session_factory.begin() as session:
        account = Account(name="Viewer", email="management@example.com")
        account.id = account_id
        tenant = Tenant(name="Viewer workspace")
        tenant.id = installed_app_ref.tenant_id
        session.add_all([account, tenant])
        membership = TenantAccountJoin(
            tenant_id=installed_app_ref.tenant_id, account_id=account_id, role=TenantAccountRole.OWNER
        )
        configuration = AppModelConfig(app_id=installed_app_ref.app_id)
        session.add_all([membership, configuration])
        session.flush()
        app = session.get(App, installed_app_ref.app_id)
        assert app is not None
        app.app_model_config_id = configuration.id
        membership_id = membership.id

    active_connections = 0
    engine = sqlite_session_factory.kw["bind"]

    @event.listens_for(engine, "checkout")
    def connection_checked_out(_connection: object, _record: object, _proxy: object) -> None:
        nonlocal active_connections
        active_connections += 1

    @event.listens_for(engine, "checkin")
    def connection_checked_in(_connection: object, _record: object) -> None:
        nonlocal active_connections
        active_connections -= 1

    def enterprise_response(method: str, path: str, *, json: dict[str, object]) -> dict[str, object]:
        assert deployment_edition == DeploymentEdition.ENTERPRISE
        # The candidate read must release its DB connection before either network call.
        assert active_connections == 0
        assert method == "POST"
        if path == "/webapp/access-mode/batch/id":
            assert json == {"appIds": [installed_app_ref.app_id]}
            return {"accessModes": {installed_app_ref.app_id: "private"}}
        assert path == "/webapp/permission/batch"
        assert json == {"userId": account_id, "appIds": [installed_app_ref.app_id]}
        return {"permissions": {installed_app_ref.app_id: permission_result}}

    with patch(
        "services.enterprise.enterprise_service.EnterpriseRequest.send_request", side_effect=enterprise_response
    ) as enterprise_request:
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=deployment_edition,
            initialization_password="",
            redis=_redis(),
        )
        if deployment_edition != DeploymentEdition.ENTERPRISE:
            assert services.webapp_access.batch_get_access_modes(app_ids=(installed_app_ref.app_id,)) == {
                installed_app_ref.app_id: WebAppAccessMode.PUBLIC
            }
            assert services.webapp_access.batch_get_user_permissions(
                user_id=account_id, app_ids=(installed_app_ref.app_id,)
            ) == {installed_app_ref.app_id: True}
            assert services.installed_apps.access.get_visible_app_ids(
                user_id=account_id, app_ids=(installed_app_ref.app_id,)
            ) == frozenset({installed_app_ref.app_id})
        page = services.installed_apps.management.get_visible_page(
            tenant_id=installed_app_ref.tenant_id,
            user_id=account_id,
            cursor=None,
            limit=1,
            app_id=None,
            name=None,
        )
        assert page.editable is True
        assert page.has_more is False
        assert page.next_cursor is None
        expected_ids: list[str] = (
            [installed_app_ref.id] if deployment_edition != DeploymentEdition.ENTERPRISE or permission_result else []
        )
        assert [installation.id for installation in page.data] == expected_ids

        with sqlite_session_factory.begin() as session:
            stored_membership = session.get(TenantAccountJoin, membership_id)
            assert stored_membership is not None
            stored_membership.role = TenantAccountRole.NORMAL
        services.installed_apps.management.set_pinned(installed_app=installed_app_ref, is_pinned=True)
        detail = services.installed_apps.management.get_detail(installed_app=installed_app_ref, account_id=account_id)
        assert detail.editable is False
        assert detail.installation.is_pinned is True
        assert detail.installation.id == installed_app_ref.id
        assert detail.installation.app.id == installed_app_ref.app_id
        assert active_connections == 0

    if deployment_edition == DeploymentEdition.ENTERPRISE:
        assert enterprise_request.call_args_list == [
            call("POST", "/webapp/access-mode/batch/id", json={"appIds": [installed_app_ref.app_id]}),
            call(
                "POST",
                "/webapp/permission/batch",
                json={"userId": account_id, "appIds": [installed_app_ref.app_id]},
            ),
        ]
    else:
        enterprise_request.assert_not_called()


@pytest.mark.parametrize("naming_fails", [False, True])
def test_installed_app_conversations_wire_real_persistence_naming_and_cleanup(
    sqlite_session_factory: sessionmaker[Session],
    installed_app_ref: InstalledAppRef,
    naming_app: Flask,
    naming_plugin: NamingPlugin,
    monkeypatch: pytest.MonkeyPatch,
    naming_fails: bool,
) -> None:
    naming_plugin.answer = '{"Your Output": "Generated name"}'
    naming_plugin.fail_invoke = naming_fails
    account_id = str(uuid4())
    with sqlite_session_factory.begin() as session:
        app = session.get(App, installed_app_ref.app_id)
        assert app is not None
        app.mode = AppMode.CHAT
        owner_tenant_id = app.tenant_id
        session.add_all(
            [
                Provider(tenant_id=owner_tenant_id, provider_name=_PROVIDER, is_valid=True),
                TenantDefaultModel(
                    tenant_id=owner_tenant_id, provider_name=_PROVIDER, model_name=_MODEL, model_type=ModelType.LLM
                ),
            ]
        )
        conversation = Conversation(
            app_id=app.id,
            mode=AppMode.CHAT,
            name="Original",
            _inputs={"enabled": False},
            from_source=ConversationFromSource.CONSOLE,
            from_account_id=account_id,
            from_end_user_id=None,
            invoke_from=ModelInvokeFrom.EXPLORE,
        )
        session.add(conversation)
        session.flush()
        conversation_id = conversation.id
        session.add(
            Message(
                app_id=app.id,
                conversation_id=conversation.id,
                _inputs={},
                query="First question",
                message={},
                message_unit_price=Decimal(0),
                answer="Answer",
                answer_unit_price=Decimal(0),
                currency="USD",
                from_source=ConversationFromSource.CONSOLE,
                from_account_id=account_id,
            )
        )

    active_connections = 0

    @event.listens_for(sqlite_session_factory.kw["bind"], "checkout")
    def connection_checked_out(_connection: object, _record: object, _proxy: object) -> None:
        nonlocal active_connections
        active_connections += 1

    @event.listens_for(sqlite_session_factory.kw["bind"], "checkin")
    def connection_checked_in(_connection: object, _record: object) -> None:
        nonlocal active_connections
        active_connections -= 1

    naming_sessions: list[Session] = []

    def observe_naming_session(session: Session, _transaction: SessionTransaction, _connection: Connection) -> None:
        naming_sessions.append(session)

    def observe_naming_request(request: httpx.Request) -> None:
        if request.url.path.endswith("/llm/invoke"):
            assert active_connections == 0
            assert naming_sessions
            assert all(not session.in_transaction() for session in naming_sessions)
            assert get_credit_usage_metadata() == {
                "app_type": CreditUsageAppType.CHATBOT,
                "created_by": CreditUsageCreatedBy.CONVERSATION_NAME,
                "request_id": "naming-request",
            }
            assert f"/{owner_tenant_id}/" in request.url.path
            assert json.loads(request.content)["data"]["prompt_messages"] == [
                {"role": "user", "content": CONVERSATION_TITLE_PROMPT + "First question\n", "name": None}
            ]

    queued: list[str] = []

    def enqueue_cleanup(c_id: str) -> None:
        assert active_connections == 0
        assert all(not session.in_transaction() for session in naming_sessions)
        with sqlite_session_factory() as session:
            stored = session.get(Conversation, c_id)
            assert stored is not None
            assert stored.is_deleted
        queued.append(c_id)

    monkeypatch.setattr(delete_conversation_related_data, "delay", enqueue_cleanup)
    services = ext_application_services.build_application_services(
        database_client=sqlite_session_factory,
        deployment_edition=DeploymentEdition.COMMUNITY,
        initialization_password="",
        redis=_redis(),
    )
    admitted_app = services.installed_apps.access.get_access(
        installed_app_id=installed_app_ref.id,
        tenant_id=installed_app_ref.tenant_id,
        account_id=account_id,
    )
    assert admitted_app.app_mode == "chat"
    previous_metadata = get_credit_usage_metadata()
    event.listen(db.session.session_factory, "after_begin", observe_naming_session)
    plugin_http._httpx_client.event_hooks["request"].append(observe_naming_request)
    try:
        with naming_app.app_context(), use_credit_usage_metadata({"request_id": "naming-request"}):
            outer_session = db.session()
            renamed = services.installed_apps.conversations.rename(
                installed_app=admitted_app,
                account_id=account_id,
                conversation_id=conversation_id,
                name=None,
                auto_generate=True,
            )
            assert get_credit_usage_metadata() == {"request_id": "naming-request"}
            assert renamed.name == ("Original" if naming_fails else "Generated name")
            assert renamed.inputs == {"enabled": False}
            assert naming_sessions
            assert all(session is not outer_session and not session.in_transaction() for session in naming_sessions)
            assert db.session() is outer_session
            services.installed_apps.conversations.delete(
                installed_app=admitted_app, account_id=account_id, conversation_id=conversation_id
            )
    finally:
        plugin_http._httpx_client.event_hooks["request"].remove(observe_naming_request)
        event.remove(db.session.session_factory, "after_begin", observe_naming_session)
    assert len([path for path, _payload in naming_plugin.requests if path.endswith("/llm/invoke")]) == 1
    assert get_credit_usage_metadata() == previous_metadata
    assert queued == [conversation_id]
    assert active_connections == 0


def test_installed_app_visibility_batches_settings_before_permissions_and_preserves_truthiness_filter(
    sqlite_session_factory: sessionmaker[Session],
) -> None:
    app_ids = (
        "allowed",
        "truthy-permission",
        "missing-setting",
        "sso",
        "denied",
        "zero",
        "empty",
        "null",
        "missing-permission",
    )
    permission_candidates = ["allowed", "truthy-permission", "denied", "zero", "empty", "null", "missing-permission"]
    with patch(
        "services.enterprise.enterprise_service.EnterpriseRequest.send_request",
        side_effect=[
            {
                "accessModes": {
                    "allowed": "private",
                    "truthy-permission": "private_all",
                    "sso": "sso_verified",
                    "denied": "public",
                    "zero": "private_all",
                    "empty": "public",
                    "null": "private",
                    "missing-permission": "private",
                }
            },
            {
                "permissions": {
                    "allowed": True,
                    "truthy-permission": 1,
                    "missing-setting": True,
                    "sso": True,
                    "denied": False,
                    "zero": 0,
                    "empty": "",
                    "null": None,
                }
            },
        ],
    ) as enterprise_request:
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.ENTERPRISE,
            initialization_password="",
            redis=_redis(),
        )
        visible = services.installed_apps.access.get_visible_app_ids(user_id="viewer", app_ids=app_ids)

    assert visible == frozenset({"allowed", "truthy-permission"})
    assert enterprise_request.call_args_list == [
        call("POST", "/webapp/access-mode/batch/id", json={"appIds": list(app_ids)}),
        call("POST", "/webapp/permission/batch", json={"userId": "viewer", "appIds": permission_candidates}),
    ]


@pytest.mark.parametrize("include_valid_apps", [True, False], ids=["mixed-modes", "all-invalid"])
def test_installed_app_visibility_skips_and_logs_each_invalid_access_mode(
    sqlite_session_factory: sessionmaker[Session], caplog: pytest.LogCaptureFixture, include_valid_apps: bool
) -> None:
    modes = {"invalid-empty": "", "invalid-unknown": "future-mode"}
    valid_ids: list[str] = ["valid-before", "valid-after"] if include_valid_apps else []
    if include_valid_apps:
        modes = {"valid-before": "private", **modes, "valid-after": "public"}
    app_ids = tuple(modes)
    responses: list[object] = [{"accessModes": modes}]
    if include_valid_apps:
        # Even an over-inclusive permission response must not restore invalid apps.
        responses.append({"permissions": dict.fromkeys(app_ids, True)})
    with patch(
        "services.enterprise.enterprise_service.EnterpriseRequest.send_request", side_effect=responses
    ) as enterprise_request:
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.ENTERPRISE,
            initialization_password="",
            redis=_redis(),
        )
        visible = services.installed_apps.access.get_visible_app_ids(user_id="viewer", app_ids=app_ids)

    assert visible == frozenset(valid_ids)
    expected_calls = [call("POST", "/webapp/access-mode/batch/id", json={"appIds": list(app_ids)})]
    if include_valid_apps:
        expected_calls.append(call("POST", "/webapp/permission/batch", json={"userId": "viewer", "appIds": valid_ids}))
    assert enterprise_request.call_args_list == expected_calls
    warnings = [
        message
        for logger_name, level, message in caplog.record_tuples
        if logger_name == ext_application_services.__name__ and level == logging.WARNING
    ]
    assert len(warnings) == 2
    assert any("invalid-empty" in message and repr("") in message for message in warnings)
    assert any("invalid-unknown" in message and repr("future-mode") in message for message in warnings)


@pytest.mark.parametrize("app_ids", [(), ("sso", "missing")])
def test_installed_app_visibility_skips_unnecessary_enterprise_requests(
    sqlite_session_factory: sessionmaker[Session], app_ids: tuple[str, ...]
) -> None:
    with patch(
        "services.enterprise.enterprise_service.EnterpriseRequest.send_request",
        return_value={"accessModes": {"sso": "sso_verified"}},
    ) as enterprise_request:
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.ENTERPRISE,
            initialization_password="",
            redis=_redis(),
        )
        visible = services.installed_apps.access.get_visible_app_ids(user_id="viewer", app_ids=app_ids)

    assert visible == frozenset()
    if app_ids:
        enterprise_request.assert_called_once_with(
            "POST", "/webapp/access-mode/batch/id", json={"appIds": list(app_ids)}
        )
    else:
        enterprise_request.assert_not_called()


@pytest.mark.parametrize("failure_stage", ["settings", "permissions"])
def test_installed_app_visibility_propagates_access_unavailable_with_original_cause(
    sqlite_session_factory: sessionmaker[Session],
    failure_stage: str,
) -> None:
    enterprise_error = EnterpriseAPIError("batch unavailable")
    responses: list[object] = []
    if failure_stage == "permissions":
        responses.append({"accessModes": {"app-1": "private"}})
    responses.append(enterprise_error)
    with patch(
        "services.enterprise.enterprise_service.EnterpriseRequest.send_request", side_effect=responses
    ) as enterprise_request:
        services = ext_application_services.build_application_services(
            database_client=sqlite_session_factory,
            deployment_edition=DeploymentEdition.ENTERPRISE,
            initialization_password="",
            redis=_redis(),
        )
        with pytest.raises(WebAppAccessUnavailableError) as caught:
            services.installed_apps.access.get_visible_app_ids(user_id="viewer", app_ids=("app-1",))

    assert type(caught.value) is WebAppAccessUnavailableError
    assert caught.value.__cause__ is enterprise_error
    assert enterprise_request.call_count == (2 if failure_stage == "permissions" else 1)