chore(deps): bump httplib2, setuptools, wandb for CVE fixes (#38976)

(cherry picked from commit 651111fb0d)

api/commands/__init__.py

@@ -22,7 +22,7 @@ from .plugin import (
     setup_system_trigger_oauth_client,
     transform_datasource_credentials,
 )
-from .rbac import migrate_member_roles_to_rbac
+from .rbac import migrate_dataset_permissions_to_rbac, migrate_member_roles_to_rbac
 from .retention import (
     archive_workflow_runs,
     archive_workflow_runs_plan,
@@ -76,6 +76,7 @@ __all__ = [
     "legacy_model_types",
     "migrate_annotation_vector_database",
     "migrate_data_for_plugin",
+    "migrate_dataset_permissions_to_rbac",
     "migrate_knowledge_vector_database",
     "migrate_member_roles_to_rbac",
     "migrate_oss",

api/commands/data_migrate.py

@@ -7,6 +7,7 @@ from typing import cast
 
 import click
 
+from commands.rbac import migrate_dataset_permissions_to_rbac
 from extensions.ext_database import db
 from graphon.model_runtime.entities.model_entities import ModelType
 from services.legacy_model_type_migration import (
@@ -177,3 +178,4 @@ def legacy_model_types(
 
 
 data_migrate.add_command(legacy_model_types)
+data_migrate.add_command(migrate_dataset_permissions_to_rbac)

api/commands/rbac.py

@@ -1,11 +1,55 @@
 from __future__ import annotations
 
+import json
+from collections.abc import Iterator
+from concurrent.futures import ThreadPoolExecutor, as_completed
+
 import click
 from sqlalchemy import select
 
+from configs import dify_config
 from core.db.session_factory import session_factory
-from models import TenantAccountJoin, TenantAccountRole
-from services.enterprise.rbac_service import ListOption, RBACService
+from core.rbac import RBACResourceWhitelistScope
+from models import Dataset, DatasetPermission, DatasetPermissionEnum, TenantAccountJoin, TenantAccountRole
+from services.enterprise.rbac_service import ListOption, RBACService, ReplaceMemberBindings, ReplaceUserAccessPolicies
+
+_RBAC_DEFAULT_ACCESS_POLICY_ID = "default"
+
+_LEGACY_ROLE_TO_BUILTIN_TAG = {
+    TenantAccountRole.OWNER.value: "owner",
+    TenantAccountRole.ADMIN.value: "admin",
+    TenantAccountRole.EDITOR.value: "editor",
+    TenantAccountRole.NORMAL.value: "normal",
+    TenantAccountRole.DATASET_OPERATOR.value: "dataset_operator",
+}
+
+
+def _resolve_builtin_role_ids(tenant_id: str, operator_account_id: str) -> dict[str, str]:
+    """Resolve every legacy workspace role to the current tenant's builtin RBAC role id.
+
+    The migration replays the old `TenantAccountJoin.role` values onto the
+    RBAC member-role binding API. Builtin RBAC roles are tenant-scoped and
+    identified by runtime ids, so the command must look them up per tenant.
+    """
+    roles = RBACService.Roles.list(
+        tenant_id=tenant_id,
+        account_id=operator_account_id,
+        options=ListOption(page_number=1, results_per_page=100),
+    ).data
+    role_id_by_tag = {
+        role.role_tag: role.id
+        for role in roles
+        if role.is_builtin and role.category == "global_system_default" and role.role_tag
+    }
+    resolved: dict[str, str] = {}
+    for legacy_role, expected_builtin_tag in _LEGACY_ROLE_TO_BUILTIN_TAG.items():
+        role_id = role_id_by_tag.get(expected_builtin_tag)
+        if expected_builtin_tag == "dataset_operator" and not dify_config.DATASET_OPERATOR_ENABLED:
+            continue
+        if not role_id:
+            raise ValueError(f"Builtin RBAC role not found for tenant={tenant_id}, legacy_role={legacy_role}")
+        resolved[legacy_role] = role_id
+    return resolved
 
 
 def _resolve_builtin_role_id(tenant_id: str, operator_account_id: str, legacy_role: str) -> str:
@@ -15,26 +59,86 @@ def _resolve_builtin_role_id(tenant_id: str, operator_account_id: str, legacy_ro
     RBAC member-role binding API. Builtin RBAC roles are tenant-scoped and
     identified by runtime ids, so the command must look them up per tenant.
     """
-    expected_builtin_tag = {
-        TenantAccountRole.OWNER.value: "owner",
-        TenantAccountRole.ADMIN.value: "admin",
-        TenantAccountRole.EDITOR.value: "editor",
-        TenantAccountRole.NORMAL.value: "normal",
-        TenantAccountRole.DATASET_OPERATOR.value: "dataset_operator",
-    }.get(legacy_role)
-    if not expected_builtin_tag:
+    if legacy_role not in _LEGACY_ROLE_TO_BUILTIN_TAG:
         raise ValueError(f"Unsupported legacy workspace role: {legacy_role}")
 
-    roles = RBACService.Roles.list(
+    return _resolve_builtin_role_ids(tenant_id, operator_account_id)[legacy_role]
+
+
+def _iter_tenant_member_batches(
+    tenant_id: str | None,
+    *,
+    db_batch_size: int,
+    api_batch_size: int,
+) -> Iterator[tuple[str, str, list[tuple[str, str]]]]:
+    """Yield legacy member roles in tenant-scoped API-sized batches.
+
+    Rows are projected to primitive values and streamed from the database, so
+    the command never materializes every TenantAccountJoin ORM object. The
+    iterator only keeps one tenant's API-sized batches in memory while it
+    finds that tenant's owner account.
+    """
+    with session_factory.create_session() as session:
+        stmt = (
+            select(TenantAccountJoin.tenant_id, TenantAccountJoin.account_id, TenantAccountJoin.role)
+            .order_by(TenantAccountJoin.tenant_id.asc(), TenantAccountJoin.id.asc())
+            .execution_options(yield_per=db_batch_size)
+        )
+        if tenant_id:
+            stmt = stmt.where(TenantAccountJoin.tenant_id == tenant_id)
+
+        current_tenant_id: str | None = None
+        owner_account_id: str | None = None
+        batches: list[list[tuple[str, str]]] = []
+        batch: list[tuple[str, str]] = []
+
+        def flush_current_tenant() -> Iterator[tuple[str, str, list[tuple[str, str]]]]:
+            if current_tenant_id is None:
+                return
+            if batch:
+                batches.append(batch.copy())
+            if not owner_account_id:
+                raise ValueError(f"Workspace owner not found for tenant={current_tenant_id}")
+            for item in batches:
+                yield current_tenant_id, owner_account_id, item
+
+        for row in session.execute(stmt):
+            workspace_id = str(row.tenant_id)
+            if current_tenant_id is not None and workspace_id != current_tenant_id:
+                yield from flush_current_tenant()
+                owner_account_id = None
+                batches = []
+                batch = []
+            current_tenant_id = workspace_id
+            account_id = str(row.account_id)
+            role = str(row.role)
+            if role == TenantAccountRole.OWNER.value:
+                owner_account_id = account_id
+            batch.append((account_id, role))
+            if len(batch) >= api_batch_size:
+                batches.append(batch)
+                batch = []
+
+        yield from flush_current_tenant()
+
+
+def _member_already_has_role(current_roles_by_account_id: dict[str, set[str]], account_id: str, role_id: str) -> bool:
+    return current_roles_by_account_id.get(account_id) == {role_id}
+
+
+def _replace_member_role(
+    tenant_id: str,
+    operator_account_id: str,
+    member_account_id: str,
+    role_id: str,
+) -> str:
+    RBACService.MemberRoles.replace(
         tenant_id=tenant_id,
         account_id=operator_account_id,
-        options=ListOption(page_number=1, results_per_page=100),
-    ).data
-    for role in roles:
-        if role.is_builtin and role.category == "global_system_default" and role.role_tag == expected_builtin_tag:
-            return str(role.id)
-
-    raise ValueError(f"Builtin RBAC role not found for tenant={tenant_id}, legacy_role={legacy_role}")
+        member_account_id=member_account_id,
+        role_ids=[role_id],
+    )
+    return member_account_id
 
 
 @click.command(
@@ -42,7 +146,16 @@ def _resolve_builtin_role_id(tenant_id: str, operator_account_id: str, legacy_ro
 )
 @click.option("--tenant-id", help="Only migrate a single workspace.")
 @click.option("--dry-run", is_flag=True, default=False, help="Preview the migration without writing RBAC bindings.")
-def migrate_member_roles_to_rbac(tenant_id: str | None, dry_run: bool) -> None:
+@click.option("--db-batch-size", default=5000, show_default=True, help="Rows fetched per database batch.")
+@click.option("--api-batch-size", default=200, show_default=True, help="Members checked per RBAC batch_get call.")
+@click.option("--workers", default=1, show_default=True, help="Concurrent member role replace calls per tenant batch.")
+def migrate_member_roles_to_rbac(
+    tenant_id: str | None,
+    dry_run: bool,
+    db_batch_size: int,
+    api_batch_size: int,
+    workers: int,
+) -> None:
     """Backfill RBAC member-role bindings from legacy `TenantAccountJoin.role` data.
 
     This is an offline migration command for workspaces that already have
@@ -50,63 +163,322 @@ def migrate_member_roles_to_rbac(tenant_id: str | None, dry_run: bool) -> None:
     member-role binding store.
     """
     click.echo(click.style("Starting RBAC member-role migration.", fg="green"))
+    if workers < 1:
+        raise click.BadParameter("workers must be >= 1", param_hint="--workers")
 
-    with session_factory.create_session() as session:
-        stmt = select(TenantAccountJoin).order_by(TenantAccountJoin.tenant_id.asc(), TenantAccountJoin.id.asc())
-        if tenant_id:
-            stmt = stmt.where(TenantAccountJoin.tenant_id == tenant_id)
+    tenant_count = 0
+    scanned_count = 0
+    skipped_count = 0
+    migrated_count = 0
+    current_tenant_id: str | None = None
+    role_ids_by_legacy_role: dict[str, str] = {}
 
-        joins = list(session.scalars(stmt).all())
+    for workspace_id, owner_account_id, batch in _iter_tenant_member_batches(
+        tenant_id,
+        db_batch_size=db_batch_size,
+        api_batch_size=api_batch_size,
+    ):
+        scanned_count += len(batch)
+        if workspace_id != current_tenant_id:
+            tenant_count += 1
+            current_tenant_id = workspace_id
+            role_ids_by_legacy_role = _resolve_builtin_role_ids(workspace_id, owner_account_id)
+            click.echo(f"tenant={workspace_id}")
 
-    if not joins:
+        current_roles_by_account_id: dict[str, set[str]] = {}
+        if not dry_run:
+            current_roles = RBACService.MemberRoles.batch_get(
+                tenant_id=workspace_id,
+                account_id=owner_account_id,
+                member_account_ids=[account_id for account_id, _ in batch],
+            )
+            current_roles_by_account_id = {item.account_id: {role.id for role in item.roles} for item in current_roles}
+
+        replace_jobs: list[tuple[str, str]] = []
+        for member_account_id, legacy_role in batch:
+            resolved_role_id = role_ids_by_legacy_role.get(legacy_role)
+            if not resolved_role_id:
+                raise ValueError(f"Unsupported legacy workspace role: {legacy_role}")
+
+            if dry_run:
+                click.echo(
+                    f"tenant={workspace_id} member={member_account_id} "
+                    f"legacy_role={legacy_role} -> rbac_role_id={resolved_role_id}"
+                )
+                continue
+
+            if _member_already_has_role(current_roles_by_account_id, member_account_id, resolved_role_id):
+                skipped_count += 1
+                continue
+
+            replace_jobs.append((member_account_id, resolved_role_id))
+
+        if replace_jobs:
+            if workers == 1:
+                for member_account_id, resolved_role_id in replace_jobs:
+                    _replace_member_role(workspace_id, owner_account_id, member_account_id, resolved_role_id)
+                    migrated_count += 1
+            else:
+                with ThreadPoolExecutor(max_workers=workers) as executor:
+                    futures = [
+                        executor.submit(
+                            _replace_member_role,
+                            workspace_id,
+                            owner_account_id,
+                            member_account_id,
+                            resolved_role_id,
+                        )
+                        for member_account_id, resolved_role_id in replace_jobs
+                    ]
+                    for future in as_completed(futures):
+                        future.result()
+                        migrated_count += 1
+
+        if scanned_count % 10000 == 0:
+            click.echo(
+                f"progress scanned={scanned_count} migrated={migrated_count} skipped={skipped_count}",
+                err=True,
+            )
+
+    if scanned_count == 0:
         click.echo(click.style("No workspace members found for migration.", fg="yellow"))
         return
 
-    owner_account_by_tenant: dict[str, str] = {}
-    resolved_role_ids: dict[tuple[str, str], str] = {}
-    migrated_count = 0
-
-    for join in joins:
-        workspace_id = str(join.tenant_id)
-        member_account_id = str(join.account_id)
-        legacy_role = str(join.role)
-
-        if workspace_id not in owner_account_by_tenant:
-            owner_join = next(
-                (
-                    item
-                    for item in joins
-                    if str(item.tenant_id) == workspace_id and str(item.role) == TenantAccountRole.OWNER.value
-                ),
-                None,
-            )
-            if not owner_join:
-                raise ValueError(f"Workspace owner not found for tenant={workspace_id}")
-            owner_account_by_tenant[workspace_id] = str(owner_join.account_id)
-
-        operator_account_id = owner_account_by_tenant[workspace_id]
-        cache_key = (workspace_id, legacy_role)
-        if cache_key not in resolved_role_ids:
-            resolved_role_ids[cache_key] = _resolve_builtin_role_id(workspace_id, operator_account_id, legacy_role)
-
-        resolved_role_id = resolved_role_ids[cache_key]
+    if dry_run:
         click.echo(
-            f"tenant={workspace_id} member={member_account_id} "
-            f"legacy_role={legacy_role} -> rbac_role_id={resolved_role_id}"
+            click.style(
+                f"Dry run completed. Scanned {scanned_count} members across {tenant_count} tenants. "
+                "No RBAC bindings were written.",
+                fg="yellow",
+            )
+        )
+    else:
+        click.echo(
+            click.style(
+                f"RBAC member-role migration completed. Scanned {scanned_count} members across {tenant_count} tenants, "
+                f"migrated {migrated_count}, skipped {skipped_count} already up-to-date.",
+                fg="green",
+            )
         )
 
-        if dry_run:
-            continue
 
-        RBACService.MemberRoles.replace(
-            tenant_id=workspace_id,
-            account_id=operator_account_id,
-            member_account_id=member_account_id,
-            role_ids=[resolved_role_id],
-        )
-        migrated_count += 1
+def _dataset_permission_enum(permission: DatasetPermissionEnum | str | None) -> DatasetPermissionEnum:
+    if permission is None:
+        return DatasetPermissionEnum.ONLY_ME
+    try:
+        return DatasetPermissionEnum(permission)
+    except ValueError as exc:
+        raise ValueError(f"Unsupported legacy dataset permission: {permission}") from exc
+
+
+def _rbac_dataset_scope_for_legacy_permission(permission: DatasetPermissionEnum) -> RBACResourceWhitelistScope:
+    if permission is DatasetPermissionEnum.ALL_TEAM:
+        return RBACResourceWhitelistScope.ALL
+    if permission in {DatasetPermissionEnum.ONLY_ME, DatasetPermissionEnum.PARTIAL_TEAM}:
+        return RBACResourceWhitelistScope.SPECIFIC
+    raise ValueError(f"Unsupported legacy dataset permission: {permission}")
+
+
+def _emit_dataset_permission_migration_event(payload: dict[str, object]) -> None:
+    click.echo(json.dumps(payload, sort_keys=True))
+
+
+@click.command(
+    "rbac-migrate-dataset-permissions",
+    help=(
+        "Migrate legacy dataset permission scopes and partial members into RBAC dataset access bindings. "
+        "Side effect: replacing each dataset whitelist clears existing per-user policy bindings; "
+        "the command then recreates legacy partial-member default bindings."
+    ),
+)
+@click.option("--tenant-id", help="Only migrate datasets in a single workspace.")
+@click.option("--dataset-id", help="Only migrate a single dataset.")
+@click.option("--batch-size", default=500, show_default=True, type=click.IntRange(min=1))
+@click.option(
+    "--dry-run/--apply",
+    default=True,
+    show_default=True,
+    help="Preview the migration without writing RBAC bindings. Use --apply to write changes.",
+)
+def migrate_dataset_permissions_to_rbac(
+    tenant_id: str | None,
+    dataset_id: str | None,
+    batch_size: int,
+    dry_run: bool,
+) -> None:
+    """Backfill RBAC dataset access config from legacy `Dataset.permission`.
+
+    Legacy mapping:
+    - all_team_members -> RBAC dataset whitelist scope "all"
+    - partial_members  -> RBAC dataset whitelist scope "specific" plus each partial member gets the
+      virtual default policy
+    - only_me          -> RBAC dataset whitelist scope "specific" with no member policy bindings
+
+    The command replaces each dataset's RBAC whitelist scope first. RBAC clears
+    existing per-user policy bindings during that replace, then this command
+    recreates the legacy partial-member default bindings. Re-running it is
+    therefore idempotent for a dataset's current legacy configuration.
+    """
+    click.echo(click.style("Starting RBAC dataset permission migration.", fg="green"))
+
+    scanned_count = 0
+    scope_migrated_count = 0
+    user_policy_migrated_count = 0
+    partial_dataset_count = 0
+
+    last_dataset_id: str | None = None
+    while True:
+        with session_factory.create_session() as session:
+            stmt = (
+                select(Dataset.id, Dataset.tenant_id, Dataset.permission, Dataset.created_by)
+                .order_by(Dataset.id.asc())
+                .limit(batch_size)
+            )
+            if tenant_id:
+                stmt = stmt.where(Dataset.tenant_id == tenant_id)
+            if dataset_id:
+                stmt = stmt.where(Dataset.id == dataset_id)
+            if last_dataset_id:
+                stmt = stmt.where(Dataset.id > last_dataset_id)
+
+            dataset_rows = list(session.execute(stmt).all())
+            if not dataset_rows:
+                break
+
+            dataset_ids = [str(row.id) for row in dataset_rows]
+            partial_members_by_dataset_id: dict[str, list[str]] = {item: [] for item in dataset_ids}
+            permission_rows = session.execute(
+                select(DatasetPermission.dataset_id, DatasetPermission.account_id).where(
+                    DatasetPermission.dataset_id.in_(dataset_ids)
+                )
+            ).all()
+            for row in permission_rows:
+                partial_members_by_dataset_id[str(row.dataset_id)].append(str(row.account_id))
+
+        for dataset in dataset_rows:
+            workspace_id = str(dataset.tenant_id)
+            current_dataset_id = str(dataset.id)
+            operator_account_id = str(dataset.created_by)
+            permission_value = _dataset_permission_enum(dataset.permission)
+            scope = _rbac_dataset_scope_for_legacy_permission(permission_value)
+            partial_member_ids = sorted(set(partial_members_by_dataset_id[current_dataset_id]))
+            should_bind_partial_members = permission_value is DatasetPermissionEnum.PARTIAL_TEAM
+
+            click.echo(
+                f"tenant={workspace_id} dataset={current_dataset_id} "
+                f"operator={operator_account_id} "
+                f"legacy_permission={permission_value} -> rbac_scope={scope} "
+                f"partial_members={len(partial_member_ids) if should_bind_partial_members else 0}"
+            )
+
+            scanned_count += 1
+            replace_whitelist_payload = ReplaceMemberBindings(scope=scope)
+            if dry_run:
+                _emit_dataset_permission_migration_event(
+                    {
+                        "event": "dataset_permission_migration_proposed_change",
+                        "action": "replace_whitelist",
+                        "dry_run": True,
+                        "tenant_id": workspace_id,
+                        "dataset_id": current_dataset_id,
+                        "operator_account_id": operator_account_id,
+                        "before": {
+                            "legacy_dataset_permission": permission_value.value,
+                            "legacy_partial_member_ids": partial_member_ids if should_bind_partial_members else [],
+                        },
+                        "after": {
+                            "rbac_whitelist_scope": scope.value,
+                        },
+                        "call": {
+                            "method": "RBACService.DatasetAccess.replace_whitelist",
+                            "kwargs": {
+                                "tenant_id": workspace_id,
+                                "account_id": operator_account_id,
+                                "dataset_id": current_dataset_id,
+                                "payload": replace_whitelist_payload.model_dump(mode="json"),
+                            },
+                        },
+                    }
+                )
+            if not dry_run:
+                RBACService.DatasetAccess.replace_whitelist(
+                    tenant_id=workspace_id,
+                    account_id=operator_account_id,
+                    dataset_id=current_dataset_id,
+                    payload=replace_whitelist_payload,
+                )
+                scope_migrated_count += 1
+
+            if should_bind_partial_members:
+                partial_dataset_count += 1
+                for member_account_id in partial_member_ids:
+                    replace_user_access_policies_payload = ReplaceUserAccessPolicies(
+                        access_policy_ids=[_RBAC_DEFAULT_ACCESS_POLICY_ID],
+                    )
+                    if dry_run:
+                        _emit_dataset_permission_migration_event(
+                            {
+                                "event": "dataset_permission_migration_proposed_change",
+                                "action": "replace_user_access_policies",
+                                "dry_run": True,
+                                "tenant_id": workspace_id,
+                                "dataset_id": current_dataset_id,
+                                "operator_account_id": operator_account_id,
+                                "target_account_id": member_account_id,
+                                "before": {
+                                    "legacy_dataset_permission": permission_value.value,
+                                    "legacy_partial_member_id": member_account_id,
+                                },
+                                "after": {
+                                    "rbac_user_access_policy_ids": [_RBAC_DEFAULT_ACCESS_POLICY_ID],
+                                },
+                                "call": {
+                                    "method": "RBACService.DatasetAccess.replace_user_access_policies",
+                                    "kwargs": {
+                                        "tenant_id": workspace_id,
+                                        "account_id": operator_account_id,
+                                        "dataset_id": current_dataset_id,
+                                        "target_account_id": member_account_id,
+                                        "payload": replace_user_access_policies_payload.model_dump(
+                                            mode="json", exclude_unset=True
+                                        ),
+                                    },
+                                },
+                            }
+                        )
+                        continue
+                    RBACService.DatasetAccess.replace_user_access_policies(
+                        tenant_id=workspace_id,
+                        account_id=operator_account_id,
+                        dataset_id=current_dataset_id,
+                        target_account_id=member_account_id,
+                        payload=replace_user_access_policies_payload,
+                    )
+                    user_policy_migrated_count += 1
+
+        last_dataset_id = dataset_ids[-1]
+
+        if dataset_id:
+            break
+
+    if scanned_count == 0:
+        click.echo(click.style("No datasets found for migration.", fg="yellow"))
+        return
 
     if dry_run:
-        click.echo(click.style("Dry run completed. No RBAC bindings were written.", fg="yellow"))
+        click.echo(
+            click.style(
+                f"Dry run completed. Scanned {scanned_count} datasets; "
+                f"{partial_dataset_count} partial-member datasets would be migrated.",
+                fg="yellow",
+            )
+        )
     else:
-        click.echo(click.style(f"RBAC member-role migration completed. Migrated {migrated_count} members.", fg="green"))
+        click.echo(
+            click.style(
+                "RBAC dataset permission migration completed. "
+                f"Scanned {scanned_count} datasets, migrated {scope_migrated_count} scopes, "
+                f"wrote {user_policy_migrated_count} user default-policy bindings.",
+                fg="green",
+            )
+        )

api/configs/enterprise/__init__.py

@@ -34,6 +34,12 @@ class EnterpriseFeatureConfig(BaseSettings):
         default=False,
     )
 
+    ENTERPRISE_RBAC_REQUEST_TIMEOUT: int = Field(
+        ge=1,
+        description="Maximum timeout in seconds for inner RBAC requests.",
+        default=30,
+    )
+
 
 class EnterpriseTelemetryConfig(BaseSettings):
     """

api/controllers/console/app/app.py

@@ -43,6 +43,7 @@ from controllers.console.wraps import (
 from core.ops.ops_trace_manager import OpsTraceManager
 from core.rag.entities import PreProcessingRule, Rule, Segmentation
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
+from core.rbac import RBACResourceWhitelistScope
 from core.trigger.constants import TRIGGER_NODE_TYPES
 from extensions.ext_database import db
 from fields.base import ResponseModel
@@ -69,6 +70,7 @@ from services.entities.knowledge_entities.knowledge_entities import (
     WeightVectorSetting,
 )
 from services.feature_service import FeatureService
+from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task
 
 ALLOW_CREATE_APP_MODES = ["chat", "agent-chat", "advanced-chat", "workflow", "completion"]
 
@@ -666,6 +668,14 @@ class AppListApi(Resource):
 
         app_service = AppService()
         app = app_service.create_app(current_tenant_id, params, current_user)
+        if dify_config.RBAC_ENABLED:
+            enterprise_rbac_service.RBACService.AppAccess.replace_whitelist(
+                tenant_id=str(current_tenant_id),
+                account_id=current_user.id,
+                app_id=str(app.id),
+                payload=enterprise_rbac_service.ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
+            )
+            initialize_created_app_rbac_access_task.delay(current_tenant_id, current_user.id, app_id=app.id)
         permission_keys_map = enterprise_rbac_service.RBACService.AppPermissions.batch_get(
             str(current_tenant_id),
             current_user.id,

api/controllers/console/app/ops_trace.py

@@ -13,6 +13,7 @@ from controllers.console.wraps import (
     RBACPermission,
     RBACResourceScope,
     account_initialization_required,
+    edit_permission_required,
     rbac_permission_required,
     setup_required,
 )
@@ -95,9 +96,12 @@ class TraceAppConfigApi(Resource):
         console_ns.models[TraceAppConfigResponse.__name__],
     )
     @console_ns.response(400, "Invalid request parameters or configuration already exists")
+    @console_ns.response(403, "Insufficient permissions")
     @setup_required
     @login_required
     @account_initialization_required
+    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_TRACING_CONFIG)
     @get_app_model
     def post(self, app_model: App):
         """Create a new trace app configuration"""
@@ -125,9 +129,12 @@ class TraceAppConfigApi(Resource):
         console_ns.models[TraceAppConfigResponse.__name__],
     )
     @console_ns.response(400, "Invalid request parameters or configuration not found")
+    @console_ns.response(403, "Insufficient permissions")
     @setup_required
     @login_required
     @account_initialization_required
+    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_TRACING_CONFIG)
     @get_app_model
     def patch(self, app_model: App):
         """Update an existing trace app configuration"""
@@ -149,9 +156,12 @@ class TraceAppConfigApi(Resource):
     @console_ns.doc(params=query_params_from_model(TraceProviderQuery))
     @console_ns.response(204, "Tracing configuration deleted successfully")
     @console_ns.response(400, "Invalid request parameters or configuration not found")
+    @console_ns.response(403, "Insufficient permissions")
     @setup_required
     @login_required
     @account_initialization_required
+    @edit_permission_required
+    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_TRACING_CONFIG)
     @get_app_model
     def delete(self, app_model: App):
         """Delete an existing trace app configuration"""

api/controllers/console/app/workflow_comment.py

@@ -8,11 +8,8 @@ from controllers.common.schema import register_response_schema_models, register_
 from controllers.console import console_ns
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import (
-    RBACPermission,
-    RBACResourceScope,
     account_initialization_required,
     edit_permission_required,
-    rbac_permission_required,
     setup_required,
     with_current_tenant_id,
     with_current_user,
@@ -223,7 +220,6 @@ class WorkflowCommentListApi(Resource):
     @setup_required
     @account_initialization_required
     @with_current_tenant_id
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_VIEW_LAYOUT)
     @get_app_model()
     def get(self, current_tenant_id: str, app_model: App):
         """Get all comments for a workflow."""
@@ -240,7 +236,6 @@ class WorkflowCommentListApi(Resource):
     @setup_required
     @account_initialization_required
     @edit_permission_required
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
     @with_current_user
     @with_current_tenant_id
     @get_app_model()
@@ -273,7 +268,6 @@ class WorkflowCommentDetailApi(Resource):
     @setup_required
     @account_initialization_required
     @with_current_tenant_id
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_VIEW_LAYOUT)
     @get_app_model()
     def get(self, current_tenant_id: str, app_model: App, comment_id: str):
         """Get a specific workflow comment."""
@@ -292,7 +286,6 @@ class WorkflowCommentDetailApi(Resource):
     @setup_required
     @account_initialization_required
     @edit_permission_required
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
     @with_current_user
     @with_current_tenant_id
     @get_app_model()
@@ -321,7 +314,6 @@ class WorkflowCommentDetailApi(Resource):
     @setup_required
     @account_initialization_required
     @edit_permission_required
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
     @with_current_user
     @with_current_tenant_id
     @get_app_model()
@@ -349,7 +341,6 @@ class WorkflowCommentResolveApi(Resource):
     @setup_required
     @account_initialization_required
     @edit_permission_required
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
     @with_current_user
     @with_current_tenant_id
     @get_app_model()
@@ -378,7 +369,6 @@ class WorkflowCommentReplyApi(Resource):
     @setup_required
     @account_initialization_required
     @edit_permission_required
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
     @with_current_user
     @with_current_tenant_id
     @get_app_model()
@@ -414,7 +404,6 @@ class WorkflowCommentReplyDetailApi(Resource):
     @setup_required
     @account_initialization_required
     @edit_permission_required
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
     @with_current_user
     @with_current_tenant_id
     @get_app_model()
@@ -447,7 +436,6 @@ class WorkflowCommentReplyDetailApi(Resource):
     @setup_required
     @account_initialization_required
     @edit_permission_required
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_EDIT)
     @with_current_user
     @with_current_tenant_id
     @get_app_model()
@@ -483,7 +471,6 @@ class WorkflowCommentMentionUsersApi(Resource):
     @setup_required
     @account_initialization_required
     @with_current_user
-    @rbac_permission_required(RBACResourceScope.APP, RBACPermission.APP_VIEW_LAYOUT)
     @get_app_model()
     def get(self, current_user: Account, app_model: App):
         """Get all users in current tenant for mentions."""

api/controllers/console/app/wraps.py

@@ -19,7 +19,7 @@ from controllers.console.app.error import AppNotFoundError
 from core.db.session_factory import session_factory
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant
-from models import App, AppMode
+from models import App, AppMode, TrialApp
 
 
 def _load_app_model(session: Session, app_id: str) -> App | None:
@@ -41,7 +41,10 @@ def _load_app_model_from_scoped_session(app_id: str) -> App | None:
 
 
 def _load_app_model_with_trial(app_id: str) -> App | None:
-    app_model = db.session.scalar(select(App).where(App.id == app_id, App.status == "normal").limit(1))
+    """Load a normal app through its trial registration without applying current-tenant scope."""
+    app_model = db.session.scalar(
+        select(App).join(TrialApp, TrialApp.app_id == App.id).where(App.id == app_id, App.status == "normal").limit(1)
+    )
     return app_model
 
 

api/controllers/console/auth/email_register.py

@@ -23,9 +23,9 @@ from libs.password import valid_password
 from models import Account
 from services.account_service import AccountService
 from services.billing_service import BillingService
-from services.errors.account import AccountRegisterError
+from services.errors.account import AccountRegisterError, SeatsLimitExceededError
 
-from ..error import AccountInFreezeError, EmailSendIpLimitError
+from ..error import AccountInFreezeError, EmailSendIpLimitError, SeatsLimitExceeded
 from ..wraps import email_password_login_enabled, email_register_enabled, setup_required
 
 
@@ -208,5 +208,7 @@ class EmailRegisterResetApi(Resource):
                 timezone=timezone,
                 session=db.session,
             )
+        except SeatsLimitExceededError:
+            raise SeatsLimitExceeded()
         except AccountRegisterError:
             raise AccountInFreezeError()

api/controllers/console/auth/login.py

@@ -25,6 +25,7 @@ from controllers.console.error import (
     AccountNotFound,
     EmailSendIpLimitError,
     NotAllowedCreateWorkspace,
+    SeatsLimitExceeded,
     WorkspacesLimitExceeded,
 )
 from controllers.console.wraps import (
@@ -51,7 +52,7 @@ from models.account import Account
 from services.account_service import AccountService, InvitationDetailDict, RegisterService, TenantService
 from services.billing_service import BillingService
 from services.entities.auth_entities import LoginFailureReason, LoginPayloadBase
-from services.errors.account import AccountRegisterError
+from services.errors.account import AccountRegisterError, SeatsLimitExceededError
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
 from services.feature_service import FeatureService
 
@@ -317,6 +318,8 @@ class EmailCodeLoginApi(Resource):
                 )
             except WorkSpaceNotAllowedCreateError:
                 raise NotAllowedCreateWorkspace()
+            except SeatsLimitExceededError:
+                raise SeatsLimitExceeded()
             except AccountRegisterError:
                 _log_console_login_failure(email=user_email, reason=LoginFailureReason.ACCOUNT_IN_FREEZE)
                 raise AccountInFreezeError()

api/controllers/console/auth/oauth.py

@@ -25,7 +25,7 @@ from libs.token import (
 from models import Account, AccountStatus
 from services.account_service import AccountService, RegisterService, TenantService
 from services.billing_service import BillingService
-from services.errors.account import AccountNotFoundError, AccountRegisterError
+from services.errors.account import AccountNotFoundError, AccountRegisterError, SeatsLimitExceededError
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkSpaceNotFoundError
 from services.feature_service import FeatureService
 
@@ -38,6 +38,7 @@ class OAuthLoginQuery(BaseModel):
     invite_token: str | None = Field(default=None, description="Optional invitation token")
     timezone: str | None = Field(default=None, description="Preferred timezone")
     language: str | None = Field(default=None, description="Preferred interface language")
+    redirect_url: str | None = Field(default=None, description="Relative page to resume after login")
 
 
 class OAuthCallbackQuery(BaseModel):
@@ -87,6 +88,36 @@ def _validated_language(value: str | None) -> str | None:
     return None
 
 
+def _url_origin(url: str) -> tuple[str, str, int] | None:
+    parsed_url = urllib.parse.urlsplit(url)
+    if parsed_url.scheme not in {"http", "https"} or parsed_url.hostname is None:
+        return None
+
+    try:
+        port = parsed_url.port
+    except ValueError:
+        return None
+
+    if port is None:
+        port = 443 if parsed_url.scheme == "https" else 80
+    return parsed_url.scheme, parsed_url.hostname, port
+
+
+def _get_redirect_target(redirect_url: str | None) -> str:
+    if not redirect_url:
+        return dify_config.CONSOLE_WEB_URL
+
+    parsed_url = urllib.parse.urlsplit(redirect_url)
+    normalized_path = redirect_url.lstrip().replace("\\", "/")
+    if not parsed_url.scheme and not parsed_url.netloc and not normalized_path.startswith("//"):
+        return redirect_url
+
+    redirect_origin = _url_origin(redirect_url)
+    if redirect_origin is not None and redirect_origin == _url_origin(dify_config.CONSOLE_WEB_URL):
+        return redirect_url
+    return dify_config.CONSOLE_WEB_URL
+
+
 def _preferred_interface_language(language: str | None = None) -> str:
     if language:
         return language
@@ -109,6 +140,7 @@ class OAuthLogin(Resource):
         invite_token = request.args.get("invite_token") or None
         timezone = _validated_timezone(request.args.get("timezone") or None)
         language = _validated_language(request.args.get("language") or None)
+        redirect_url = request.args.get("redirect_url") or None
         OAUTH_PROVIDERS = get_oauth_providers()
         with current_app.app_context():
             oauth_provider = OAUTH_PROVIDERS.get(provider)
@@ -119,6 +151,7 @@ class OAuthLogin(Resource):
             invite_token=invite_token,
             timezone=timezone,
             language=language,
+            redirect_url=redirect_url,
         )
         return redirect(auth_url)
 
@@ -144,6 +177,7 @@ class OAuthCallback(Resource):
         invite_token = oauth_state.get("invite_token")
         timezone = _validated_timezone(oauth_state.get("timezone"))
         language = _validated_language(oauth_state.get("language"))
+        redirect_url = oauth_state.get("redirect_url")
 
         if not code:
             return {"error": "Authorization code is required"}, 400
@@ -182,6 +216,8 @@ class OAuthCallback(Resource):
                 f"{dify_config.CONSOLE_WEB_URL}/signin"
                 "?message=Workspace not found, please contact system admin to invite you to join in a workspace."
             )
+        except SeatsLimitExceededError:
+            return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message=Licensed seats limit exceeded.")
         except AccountRegisterError as e:
             return redirect(f"{dify_config.CONSOLE_WEB_URL}/signin?message={e.description}")
 
@@ -210,9 +246,9 @@ class OAuthCallback(Resource):
             ip_address=extract_remote_ip(request),
         )
 
-        base_url = dify_config.CONSOLE_WEB_URL
-        query_char = "&" if "?" in base_url else "?"
-        target_url = f"{base_url}{query_char}oauth_new_user={str(oauth_new_user).lower()}"
+        target_url = _get_redirect_target(redirect_url)
+        query_char = "&" if "?" in target_url else "?"
+        target_url = f"{target_url}{query_char}oauth_new_user={str(oauth_new_user).lower()}"
         response = redirect(target_url)
 
         set_access_token_to_cookie(request, response, token_pair.access_token)

api/controllers/console/datasets/datasets.py

@@ -50,6 +50,8 @@ from models.provider_ids import ModelProviderID
 from services.api_token_service import ApiTokenCache
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
 from services.enterprise import rbac_service as enterprise_rbac_service
+from services.enterprise.rbac_service import RBACResourceWhitelistScope, ReplaceMemberBindings
+from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task
 
 register_response_schema_models(console_ns, ApiBaseUrlResponse, SimpleResultResponse, UsageCheckResponse)
 
@@ -565,6 +567,16 @@ class DatasetListApi(Resource):
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()
 
+        if dify_config.RBAC_ENABLED:
+            if permission == DatasetPermissionEnum.ALL_TEAM:
+                enterprise_rbac_service.RBACService.DatasetAccess.replace_whitelist(
+                    current_tenant_id,
+                    current_user.id,
+                    dataset.id,
+                    ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
+                )
+                initialize_created_app_rbac_access_task.delay(current_tenant_id, current_user.id, dataset_id=dataset.id)
+
         permission_keys_map = enterprise_rbac_service.RBACService.DatasetPermissions.batch_get(
             str(current_tenant_id),
             current_user.id,

api/controllers/console/error.py

@@ -58,6 +58,12 @@ class WorkspacesLimitExceeded(BaseHTTPException):
     code = 400
 
 
+class SeatsLimitExceeded(BaseHTTPException):
+    error_code = "limit_exceeded"
+    description = "Unable to create account because the licensed seats limit was exceeded"
+    code = 400
+
+
 class AccountBannedError(BaseHTTPException):
     error_code = "account_banned"
     description = "Account is banned."

api/controllers/console/workspace/members.py

@@ -36,7 +36,7 @@ from libs.login import current_account_with_tenant, login_required
 from models.account import Account, TenantAccountJoin, TenantAccountRole
 from services.account_service import AccountService, RegisterService, TenantService
 from services.enterprise import rbac_service as enterprise_rbac_service
-from services.errors.account import AccountAlreadyInTenantError
+from services.errors.account import AccountAlreadyInTenantError, SeatsLimitExceededError
 from services.feature_service import FeatureService
 
 
@@ -291,6 +291,14 @@ class MemberInviteEmailApi(Resource):
                             "message": "Account already in workspace.",
                         }
                     )
+                except SeatsLimitExceededError:
+                    invitation_results.append(
+                        MemberInviteFailedResponse(
+                            status="failed",
+                            email=invitee_email,
+                            message="Licensed seats limit exceeded.",
+                        )
+                    )
                 except Exception as e:
                     invitation_results.append({"status": "failed", "email": invitee_email, "message": str(e)})
 

api/controllers/console/workspace/rbac.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-from enum import StrEnum
 from typing import Any
 
 from flask import request
@@ -14,9 +13,11 @@ from controllers.common.schema import register_response_schema_models
 from controllers.console import console_ns
 from controllers.console.wraps import RBACPermission, RBACResourceScope, rbac_permission_required
 from core.db.session_factory import session_factory
+from core.rbac import RBACResourceWhitelistScope
 from libs.login import current_account_with_tenant, login_required
 from models import Account
 from services.enterprise import rbac_service as svc
+from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task
 
 
 class _RBACRoleList(svc.Paginated[svc.RBACRole]):
@@ -511,14 +512,8 @@ class RBACAccessPolicyBindingUnlockApi(Resource):
 # ---------------------------------------------------------------------------
 
 
-class _AccessScope(StrEnum):
-    ALL = "all"
-    SPECIFIC = "specific"
-    ONLY_ME = "only_me"
-
-
 class _ResourceAccessScopeRequest(BaseModel):
-    scope: _AccessScope
+    scope: RBACResourceWhitelistScope
 
 
 class _ReplaceBindingsRequest(BaseModel):
@@ -584,14 +579,15 @@ class RBACAppWhitelistApi(Resource):
     def put(self, app_id):
         tenant_id, account_id = _current_ids()
         request = _payload(_ResourceAccessScopeRequest)
-        return _dump(
-            svc.RBACService.AppAccess.replace_whitelist(
-                tenant_id,
-                account_id,
-                str(app_id),
-                svc.ReplaceMemberBindings(scope=request.scope.value),
-            )
+        result = svc.RBACService.AppAccess.replace_whitelist(
+            tenant_id,
+            account_id,
+            str(app_id),
+            svc.ReplaceMemberBindings(scope=request.scope.value),
         )
+        if dify_config.RBAC_ENABLED and request.scope is RBACResourceWhitelistScope.ALL:
+            initialize_created_app_rbac_access_task.delay(tenant_id, account_id, str(app_id))
+        return _dump(result)
 
 
 @console_ns.route("/workspaces/current/rbac/apps/<uuid:app_id>/user-access-policies")
@@ -616,8 +612,8 @@ class RBACAppUserAccessPolicyAssignmentApi(Resource):
             svc.RBACService.AppAccess.replace_user_access_policies(
                 tenant_id,
                 account_id,
-                str(app_id),
-                str(target_account_id),
+                app_id,
+                target_account_id,
                 payload,
             )
         )

api/controllers/openapi/workspaces.py

@@ -48,6 +48,7 @@ from services.errors.account import (
     MemberNotInTenantError,
     NoPermissionError,
     RoleAlreadyAssignedError,
+    SeatsLimitExceededError,
 )
 from services.feature_service import FeatureService
 
@@ -190,6 +191,8 @@ class WorkspaceMembersApi(Resource):
             raise BadRequest(str(exc))
         except NoPermissionError as exc:
             raise BadRequest(str(exc))
+        except SeatsLimitExceededError:
+            raise BadRequest("licensed seats limit exceeded")
         except AccountRegisterError as exc:
             raise BadRequest(str(exc))
 

api/controllers/service_api/dataset/dataset.py

@@ -43,6 +43,7 @@ from models.dataset import DatasetPermissionEnum
 from models.enums import TagType
 from models.provider_ids import ModelProviderID
 from services.dataset_service import DatasetPermissionService, DatasetService, DocumentService
+from services.enterprise.rbac_service import RBACResourceWhitelistScope, RBACService, ReplaceMemberBindings
 from services.entities.knowledge_entities.knowledge_entities import (
     ExternalRetrievalModel,
     KnowledgeProvider,
@@ -58,6 +59,7 @@ from services.tag_service import (
 from services.tag_service import (
     UpdateTagPayload as UpdateTagServicePayload,
 )
+from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task
 
 register_enum_models(service_api_ns, DatasetPermissionEnum)
 
@@ -523,6 +525,15 @@ class DatasetListApi(DatasetApiResource):
         except services.errors.dataset.DatasetNameDuplicateError:
             raise DatasetNameDuplicateError()
 
+        if payload.permission == DatasetPermissionEnum.ALL_TEAM and dify_config.RBAC_ENABLED:
+            RBACService.DatasetAccess.replace_whitelist(
+                tenant_id,
+                current_user.id,
+                dataset.id,
+                ReplaceMemberBindings(scope=RBACResourceWhitelistScope.ALL),
+            )
+            initialize_created_app_rbac_access_task.delay(tenant_id, current_user.id, dataset_id=dataset.id)
+
         return _dump_service_dataset_detail(dataset), 200
 
 

api/controllers/service_api/wraps.py

@@ -12,6 +12,7 @@ from flask_restx import Resource
 from flask_restx.utils import merge
 from pydantic import BaseModel
 from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
 from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
 
 from configs import dify_config
@@ -269,8 +270,8 @@ def cloud_edition_billing_rate_limit_check[**P, R](
                             subscription_plan=knowledge_rate_limit.subscription_plan,
                             operation="knowledge",
                         )
-                        db.session.add(rate_limit_log)
-                        db.session.commit()
+                        with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
+                            session.add(rate_limit_log)
                         raise Forbidden(
                             "Sorry, you have reached the knowledge base request rate limit of your subscription."
                         )

api/core/callback_handler/workflow_tool_callback_handler.py

@@ -1,6 +1,7 @@
 from collections.abc import Generator, Iterable, Mapping
 from typing import Any
 
+from configs import dify_config
 from core.callback_handler.agent_tool_callback_handler import DifyAgentCallbackHandler, print_text
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.tools.entities.tool_entities import ToolInvokeMessage
@@ -19,8 +20,9 @@ class DifyWorkflowCallbackHandler(DifyAgentCallbackHandler):
         trace_manager: TraceQueueManager | None = None,
     ) -> Generator[ToolInvokeMessage, None, None]:
         for tool_output in tool_outputs:
-            print_text("\n[on_tool_execution]\n", color=self.color)
-            print_text("Tool: " + tool_name + "\n", color=self.color)
-            print_text("Outputs: " + tool_output.model_dump_json()[:1000] + "\n", color=self.color)
-            print_text("\n")
+            if dify_config.DEBUG:
+                print_text("\n[on_tool_execution]\n", color=self.color)
+                print_text("Tool: " + tool_name + "\n", color=self.color)
+                print_text("Outputs: " + tool_output.model_dump_json()[:1000] + "\n", color=self.color)
+                print_text("\n")
             yield tool_output

api/core/helper/code_executor/code_executor.py

@@ -13,7 +13,7 @@ from core.helper.code_executor.jinja2.jinja2_transformer import Jinja2TemplateTr
 from core.helper.code_executor.python3.python3_transformer import Python3TemplateTransformer
 from core.helper.code_executor.template_transformer import TemplateTransformer
 from core.helper.http_client_pooling import get_pooled_http_client
-from graphon.nodes.code.entities import CodeLanguage
+from graphon.nodes.code.entities import CodeLanguage as CodeLanguage  # noqa: PLC0414
 
 logger = logging.getLogger(__name__)
 code_execution_endpoint_url = URL(str(dify_config.CODE_EXECUTION_ENDPOINT))
@@ -133,7 +133,9 @@ class CodeExecutor:
         return response_code.data.stdout or ""
 
     @classmethod
-    def execute_workflow_code_template(cls, language: CodeLanguage, code: str, inputs: Mapping[str, Any]):
+    def execute_workflow_code_template(
+        cls, language: CodeLanguage, code: str, inputs: Mapping[str, Any]
+    ) -> dict[str, Any]:
         """
         Execute code
         :param language: code language

api/core/helper/code_executor/jinja2/jinja2_transformer.py

@@ -11,7 +11,7 @@ class Jinja2TemplateTransformer(TemplateTransformer):
 
     @classmethod
     @override
-    def transform_response(cls, response: str):
+    def transform_response(cls, response: str) -> dict[str, Any]:
         """
         Transform response to dict
         :param response: response

api/core/helper/code_executor/template_transformer.py

@@ -36,14 +36,14 @@ class TemplateTransformer(ABC):
         return runner_script, preload_script
 
     @classmethod
-    def extract_result_str_from_response(cls, response: str):
+    def extract_result_str_from_response(cls, response: str) -> str:
         result = re.search(rf"{cls._result_tag}(.*){cls._result_tag}", response, re.DOTALL)
         if not result:
             raise ValueError(f"Failed to parse result: no result tag found in response. Response: {response[:200]}...")
         return result.group(1)
 
     @classmethod
-    def transform_response(cls, response: str) -> Mapping[str, Any]:
+    def transform_response(cls, response: str) -> dict[str, Any]:
         """
         Transform response to dict
         :param response: response
@@ -71,7 +71,7 @@ class TemplateTransformer(ABC):
         return result
 
     @classmethod
-    def _post_process_result(cls, result: dict[Any, Any]) -> dict[Any, Any]:
+    def _post_process_result(cls, result: dict[str, Any]) -> dict[str, Any]:
         """
         Post-process the result to convert scientific notation strings back to numbers
         """
@@ -89,7 +89,7 @@ class TemplateTransformer(ABC):
                     return [convert_scientific_notation(v) for v in value]
             return value
 
-        return convert_scientific_notation(result)
+        return {key: convert_scientific_notation(value) for key, value in result.items()}
 
     @classmethod
     @abstractmethod

api/core/helper/creators.py

@@ -24,7 +24,7 @@ def upload_dsl(dsl_file_bytes: bytes, filename: str = "template.yaml") -> str:
     response.raise_for_status()
     data = response.json()
     claim_code = data.get("data", {}).get("claim_code")
-    if not claim_code:
+    if not isinstance(claim_code, str) or not claim_code:
         raise ValueError("Creators Platform did not return a valid claim_code")
     return claim_code
 

api/core/helper/credential_utils.py

@@ -10,18 +10,21 @@ def is_credential_exists(credential_id: str, credential_type: "PluginCredentialT
     """
     Check if the credential still exists in the database.
 
+    Uses the configured SQLAlchemy session factory instead of Flask-SQLAlchemy's
+    ``db.engine`` because workflow graph node construction may run without an
+    active Flask application context.
+
     :param credential_id: The credential ID to check
     :param credential_type: The type of credential (MODEL or TOOL)
     :return: True if credential exists, False otherwise
     """
     from sqlalchemy import select
-    from sqlalchemy.orm import Session
 
-    from extensions.ext_database import db
+    from core.db import session_factory
     from models.provider import ProviderCredential, ProviderModelCredential
     from models.tools import BuiltinToolProvider
 
-    with Session(db.engine) as session:
+    with session_factory.create_session() as session:
         if credential_type == PluginCredentialType.MODEL:
             # Check both pre-defined and custom model credentials using a single UNION query
             stmt = (
@@ -42,7 +45,7 @@ def is_credential_exists(credential_id: str, credential_type: "PluginCredentialT
 
 def runtime_check_credential_policy_compliance(
     credential_id: str, provider: str, credential_type: "PluginCredentialType", check_existence: bool = True
-):
+) -> None:
     if dify_config.ENTERPRISE_DISABLE_RUNTIME_CREDENTIAL_CHECK:
         return
     check_credential_policy_compliance(

api/core/helper/download.py

@@ -1,4 +1,7 @@
-def download_with_size_limit(url, max_download_size: int, **kwargs):
+from typing import Any
+
+
+def download_with_size_limit(url: str, max_download_size: int, **kwargs: Any) -> bytes:
     from core.file import remote_fetcher
 
     response = remote_fetcher.make_request("GET", url, follow_redirects=True, **kwargs)

api/core/helper/encrypter.py

@@ -1,5 +1,7 @@
 import base64
 
+from Crypto.PublicKey import RSA
+
 from libs import rsa
 
 
@@ -11,13 +13,13 @@ def obfuscated_token(token: str) -> str:
     return token[:6] + "*" * 12 + token[-2:]
 
 
-def full_mask_token(token_length=20):
+def full_mask_token(token_length: int = 20) -> str:
     return "*" * token_length
 
 
-def encrypt_token(tenant_id: str, token: str):
-    from extensions.ext_database import db
+def encrypt_token(tenant_id: str, token: str) -> str:
     from models.account import Tenant
+    from models.engine import db
 
     if not (tenant := db.session.get(Tenant, tenant_id)):
         raise ValueError(f"Tenant with id {tenant_id} not found")
@@ -30,15 +32,15 @@ def decrypt_token(tenant_id: str, token: str) -> str:
     return rsa.decrypt(base64.b64decode(token), tenant_id)
 
 
-def batch_decrypt_token(tenant_id: str, tokens: list[str]):
+def batch_decrypt_token(tenant_id: str, tokens: list[str]) -> list[str]:
     rsa_key, cipher_rsa = rsa.get_decrypt_decoding(tenant_id)
 
     return [rsa.decrypt_token_with_decoding(base64.b64decode(token), rsa_key, cipher_rsa) for token in tokens]
 
 
-def get_decrypt_decoding(tenant_id: str):
+def get_decrypt_decoding(tenant_id: str) -> tuple[RSA.RsaKey, object]:
     return rsa.get_decrypt_decoding(tenant_id)
 
 
-def decrypt_token_with_decoding(token: str, rsa_key, cipher_rsa):
+def decrypt_token_with_decoding(token: str, rsa_key: RSA.RsaKey, cipher_rsa: object) -> str:
     return rsa.decrypt_token_with_decoding(base64.b64decode(token), rsa_key, cipher_rsa)

api/core/helper/marketplace.py

@@ -1,5 +1,6 @@
 import logging
 from collections.abc import Sequence
+from typing import Any
 
 import httpx
 from yarl import URL
@@ -19,7 +20,7 @@ def get_plugin_pkg_url(plugin_unique_identifier: str) -> str:
     return str((marketplace_api_url / "api/v1/plugins/download").with_query(unique_identifier=plugin_unique_identifier))
 
 
-def download_plugin_pkg(plugin_unique_identifier: str):
+def download_plugin_pkg(plugin_unique_identifier: str) -> bytes:
     return download_with_size_limit(get_plugin_pkg_url(plugin_unique_identifier), dify_config.PLUGIN_MAX_PACKAGE_SIZE)
 
 
@@ -39,7 +40,7 @@ def batch_fetch_plugin_manifests(plugin_ids: list[str]) -> Sequence[MarketplaceP
     return [MarketplacePluginDeclaration.model_validate(plugin) for plugin in response.json()["data"]["plugins"]]
 
 
-def batch_fetch_plugin_by_ids(plugin_ids: list[str]) -> list[dict]:
+def batch_fetch_plugin_by_ids(plugin_ids: list[str]) -> list[dict[str, Any]]:
     if not plugin_ids:
         return []
 
@@ -53,10 +54,19 @@ def batch_fetch_plugin_by_ids(plugin_ids: list[str]) -> list[dict]:
     response.raise_for_status()
 
     data = response.json()
-    return data.get("data", {}).get("plugins", [])
+    plugins = data.get("data", {}).get("plugins", [])
+    if not isinstance(plugins, list):
+        raise ValueError("Marketplace did not return a valid plugins list")
+
+    result: list[dict[str, Any]] = []
+    for plugin in plugins:
+        if not isinstance(plugin, dict) or not all(isinstance(key, str) for key in plugin):
+            raise ValueError("Marketplace did not return a valid plugins list")
+        result.append(plugin)
+    return result
 
 
-def record_install_plugin_event(plugin_unique_identifier: str):
+def record_install_plugin_event(plugin_unique_identifier: str) -> None:
     url = str(marketplace_api_url / "api/v1/stats/plugins/install_count")
     response = httpx.post(url, json={"unique_identifier": plugin_unique_identifier}, timeout=MARKETPLACE_TIMEOUT)
     response.raise_for_status()

api/core/helper/model_provider_cache.py

@@ -34,7 +34,7 @@ class ProviderCredentialsCache:
         else:
             return None
 
-    def set(self, credentials: dict[str, Any]):
+    def set(self, credentials: dict[str, Any]) -> None:
         """
         Cache model provider credentials.
 
@@ -43,7 +43,7 @@ class ProviderCredentialsCache:
         """
         redis_client.setex(self.cache_key, 86400, json.dumps(credentials))
 
-    def delete(self):
+    def delete(self) -> None:
         """
         Delete cached model provider credentials.
 

api/core/helper/module_import_helper.py

@@ -20,17 +20,18 @@ def import_module_from_source[T: (str, bytes)](
                 raise Exception(f"Failed to load module {module_name} from {py_file_path!r}")
         else:
             # Refer to: https://docs.python.org/3/library/importlib.html#importing-a-source-file-directly
-            # FIXME: mypy does not support the type of spec.loader
-            spec = importlib.util.spec_from_file_location(module_name, py_file_path)  # type: ignore[assignment]
-            if not spec or not spec.loader:
+            new_spec = importlib.util.spec_from_file_location(module_name, py_file_path)
+            if not new_spec or not new_spec.loader:
                 raise Exception(f"Failed to load module {module_name} from {py_file_path!r}")
             if use_lazy_loader:
                 # Refer to: https://docs.python.org/3/library/importlib.html#implementing-lazy-imports
-                spec.loader = importlib.util.LazyLoader(spec.loader)
+                new_spec.loader = importlib.util.LazyLoader(new_spec.loader)
+            spec = new_spec
         module = importlib.util.module_from_spec(spec)
         if not existed_spec:
             sys.modules[module_name] = module
-        spec.loader.exec_module(module)
+        if spec.loader is not None:
+            spec.loader.exec_module(module)
         return module
     except Exception as e:
         logger.exception("Failed to load module %s from script file '%s'", module_name, repr(py_file_path))

api/core/helper/provider_cache.py

@@ -9,11 +9,11 @@ from extensions.ext_redis import redis_client
 class ProviderCredentialsCache(ABC):
     """Base class for provider credentials cache"""
 
-    def __init__(self, **kwargs):
+    def __init__(self, **kwargs: Any) -> None:
         self.cache_key = self._generate_cache_key(**kwargs)
 
     @abstractmethod
-    def _generate_cache_key(self, **kwargs) -> str:
+    def _generate_cache_key(self, **kwargs: Any) -> str:
         """Generate cache key based on subclass implementation"""
         pass
 
@@ -28,11 +28,11 @@ class ProviderCredentialsCache(ABC):
                 return None
         return None
 
-    def set(self, config: dict[str, Any]):
+    def set(self, config: dict[str, Any]) -> None:
         """Cache provider credentials"""
         redis_client.setex(self.cache_key, 86400, json.dumps(config))
 
-    def delete(self):
+    def delete(self) -> None:
         """Delete cached provider credentials"""
         redis_client.delete(self.cache_key)
 
@@ -48,7 +48,7 @@ class SingletonProviderCredentialsCache(ProviderCredentialsCache):
         )
 
     @override
-    def _generate_cache_key(self, **kwargs) -> str:
+    def _generate_cache_key(self, **kwargs: Any) -> str:
         tenant_id = kwargs["tenant_id"]
         provider_type = kwargs["provider_type"]
         identity_name = kwargs["provider_identity"]
@@ -63,7 +63,7 @@ class ToolProviderCredentialsCache(ProviderCredentialsCache):
         super().__init__(tenant_id=tenant_id, provider=provider, credential_id=credential_id)
 
     @override
-    def _generate_cache_key(self, **kwargs) -> str:
+    def _generate_cache_key(self, **kwargs: Any) -> str:
         tenant_id = kwargs["tenant_id"]
         provider = kwargs["provider"]
         credential_id = kwargs["credential_id"]
@@ -77,10 +77,10 @@ class NoOpProviderCredentialCache:
         """Get cached provider credentials"""
         return None
 
-    def set(self, config: dict[str, Any]):
+    def set(self, config: dict[str, Any]) -> None:
         """Cache provider credentials"""
         pass
 
-    def delete(self):
+    def delete(self) -> None:
         """Delete cached provider credentials"""
         pass

api/core/helper/provider_encryption.py

@@ -125,5 +125,7 @@ class ProviderConfigEncrypter:
         return data
 
 
-def create_provider_encrypter(tenant_id: str, config: list[BasicProviderConfig], cache: ProviderConfigCache):
+def create_provider_encrypter(
+    tenant_id: str, config: list[BasicProviderConfig], cache: ProviderConfigCache
+) -> tuple[ProviderConfigEncrypter, ProviderConfigCache]:
     return ProviderConfigEncrypter(tenant_id=tenant_id, config=config, provider_config_cache=cache), cache

api/core/helper/tool_parameter_cache.py

@@ -37,11 +37,11 @@ class ToolParameterCache:
         else:
             return None
 
-    def set(self, parameters: dict[str, Any]):
+    def set(self, parameters: dict[str, Any]) -> None:
         """Cache model provider credentials."""
         redis_client.setex(self.cache_key, 86400, json.dumps(parameters))
 
-    def delete(self):
+    def delete(self) -> None:
         """
         Delete cached model provider credentials.
 

api/core/helper/trace_id_helper.py

@@ -61,7 +61,7 @@ def get_external_trace_id(request: Any) -> str | None:
     return None
 
 
-def extract_external_trace_id_from_args(args: Mapping[str, Any]):
+def extract_external_trace_id_from_args(args: Mapping[str, Any]) -> dict[str, Any]:
     """
     Extract 'external_trace_id' from args.
 

api/core/plugin/backwards_invocation/app.py

@@ -76,7 +76,11 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
         if not user_id:
             user = EndUserService.get_or_create_end_user(app)
         else:
-            user = cls._get_user(user_id, app)
+            try:
+                user = cls._get_user(user_id, app)
+            except ValueError:
+                # Plugins such as WeCom Bot pass external sender IDs rather than EndUser UUIDs.
+                user = EndUserService.get_or_create_end_user(app, user_id=user_id)
 
         conversation_id = conversation_id or ""
 
@@ -226,6 +230,13 @@ class PluginAppBackwardsInvocation(BaseBackwardsInvocation):
                 EndUser.app_id == app.id,
             )
             user = session.scalar(stmt)
+            if not user:
+                stmt = select(EndUser).where(
+                    EndUser.session_id == user_id,
+                    EndUser.tenant_id == app.tenant_id,
+                    EndUser.app_id == app.id,
+                )
+                user = session.scalar(stmt)
             if not user:
                 stmt = select(Account).where(
                     Account.id == user_id,

api/core/plugin/entities/plugin_daemon.py

@@ -228,7 +228,7 @@ class CredentialType(enum.StrEnum):
     OAUTH2 = "oauth2"
     UNAUTHORIZED = "unauthorized"
 
-    def get_name(self):
+    def get_name(self) -> str:
         if self == CredentialType.API_KEY:
             return "API KEY"
         elif self == CredentialType.OAUTH2:

api/core/plugin/plugin_service.py

@@ -4,6 +4,8 @@ This module owns plugin daemon management calls that are shared by API services
 and core runtimes. Plugin model provider discovery is cached here, alongside
 plugin install, uninstall, and upgrade invalidation, so all cache mutations for
 plugin-owned provider metadata stay tenant-scoped and in one place.
+Provider cache payloads may be stored as prefixed zstd bytes; readers also
+accept legacy plain JSON payloads for rolling upgrades and existing Redis keys.
 
 The console plugin list also normalizes endpoint setup counters against live
 endpoint records. Some plugin daemon builds return stale ``endpoints_*``
@@ -14,12 +16,15 @@ metadata.
 
 import logging
 import time
-from collections.abc import Mapping, Sequence
+from collections.abc import Iterator, Mapping, Sequence
+from contextlib import contextmanager
 from mimetypes import guess_type
-from typing import ClassVar
+from typing import Literal, Protocol
 
+import zstandard
 from pydantic import BaseModel, TypeAdapter, ValidationError
 from redis import RedisError
+from redis.exceptions import LockError
 from sqlalchemy import delete, select, update
 from sqlalchemy.orm import Session
 from yarl import URL
@@ -67,14 +72,18 @@ logger = logging.getLogger(__name__)
 _provider_entities_adapter: TypeAdapter[list[ProviderEntity]] = TypeAdapter(list[ProviderEntity])
 
 
-class PluginService:
-    _plugin_model_providers_memory_cache: ClassVar[dict[str, tuple[int, float, tuple[ProviderEntity, ...]]]] = {}
+class _RedisLock(Protocol):
+    def acquire(self, *, blocking: bool = True, blocking_timeout: float | None = None) -> bool: ...
 
+    def release(self) -> None: ...
+
+
+class PluginService:
     class LatestPluginCache(BaseModel):
         plugin_id: str
         version: str
         unique_identifier: str
-        status: str
+        status: Literal["active", "deleted"]
         deprecated_reason: str
         alternative_plugin_id: str
 
@@ -82,6 +91,12 @@ class PluginService:
     REDIS_TTL = 60 * 5  # 5 minutes
     PLUGIN_MODEL_PROVIDERS_REDIS_KEY_PREFIX = "plugin_model_providers:tenant_id:"
     PLUGIN_MODEL_PROVIDERS_GENERATION_REDIS_KEY_PREFIX = "plugin_model_providers_generation:tenant_id:"
+    PLUGIN_MODEL_PROVIDERS_LOCK_REDIS_KEY_PREFIX = "plugin_model_providers_refresh_lock:tenant_id:"
+    PLUGIN_MODEL_PROVIDERS_LOCK_TTL = 30
+    PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT = 2.0
+    PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL = 0.05
+    PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX = b"\x00dify-plugin-model-providers-zstd-v1:"
+    PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_MIN_BYTES = 64 * 1024
     PLUGIN_INSTALL_TASK_TERMINAL_STATUSES = (PluginInstallTaskStatus.Success, PluginInstallTaskStatus.Failed)
     # Mirror the detail-panel endpoint query size so list reconciliation and
     # the visible endpoint drawer exercise the same daemon pagination path.
@@ -98,6 +113,10 @@ class PluginService:
     def _get_plugin_model_providers_generation_cache_key(cls, tenant_id: str) -> str:
         return f"{cls.PLUGIN_MODEL_PROVIDERS_GENERATION_REDIS_KEY_PREFIX}{tenant_id}"
 
+    @classmethod
+    def _get_plugin_model_providers_lock_key(cls, tenant_id: str, generation: int) -> str:
+        return f"{cls.PLUGIN_MODEL_PROVIDERS_LOCK_REDIS_KEY_PREFIX}{tenant_id}:generation:{generation}"
+
     @staticmethod
     def _get_provider_short_name_alias(provider: PluginModelProviderEntity) -> str:
         """
@@ -129,8 +148,25 @@ class PluginService:
         return declaration
 
     @classmethod
-    def _copy_provider_entities(cls, providers: Sequence[ProviderEntity]) -> tuple[ProviderEntity, ...]:
-        return tuple(provider.model_copy(deep=True) for provider in providers)
+    def _encode_plugin_model_providers_cache_payload(cls, payload: bytes) -> bytes:
+        if len(payload) < cls.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_MIN_BYTES:
+            return payload
+
+        return cls.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX + zstandard.compress(payload, level=1)
+
+    @classmethod
+    def _decode_plugin_model_providers_cache_payload(cls, payload: bytes | bytearray | str) -> bytes | bytearray | str:
+        if isinstance(payload, str):
+            return payload
+
+        prefix = cls.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX
+        if not payload.startswith(prefix):
+            return payload
+
+        try:
+            return zstandard.decompress(payload[len(prefix) :])
+        except zstandard.ZstdError as exc:
+            raise ValueError("Invalid compressed plugin model providers cache payload.") from exc
 
     @classmethod
     def _load_plugin_model_providers_generation(cls, tenant_id: str) -> int | None:
@@ -163,76 +199,35 @@ class PluginService:
             return None
 
     @classmethod
-    def _load_in_memory_plugin_model_providers(
-        cls, memory_cache_key: str, generation: int
-    ) -> tuple[ProviderEntity, ...] | None:
-        cached_entry = cls._plugin_model_providers_memory_cache.get(memory_cache_key)
-        if cached_entry is None:
-            return None
+    def _load_cached_plugin_model_providers_for_generation(
+        cls, tenant_id: str, generation: int | None
+    ) -> tuple[tuple[ProviderEntity, ...] | None, bool]:
+        if generation is None:
+            return None, False
 
-        cached_generation, expires_at, providers = cached_entry
-        if cached_generation != generation or time.monotonic() >= expires_at:
-            cls._plugin_model_providers_memory_cache.pop(memory_cache_key, None)
-            return None
-
-        return cls._copy_provider_entities(providers)
-
-    @classmethod
-    def _store_in_memory_plugin_model_providers(
-        cls, memory_cache_key: str, generation: int, providers: Sequence[ProviderEntity]
-    ) -> None:
-        ttl = dify_config.PLUGIN_MODEL_PROVIDERS_CACHE_TTL
-        if ttl <= 0:
-            cls._plugin_model_providers_memory_cache.pop(memory_cache_key, None)
-            return
-
-        cls._plugin_model_providers_memory_cache[memory_cache_key] = (
-            generation,
-            time.monotonic() + ttl,
-            cls._copy_provider_entities(providers),
-        )
-
-    @classmethod
-    def _load_cached_plugin_model_providers(
-        cls, tenant_id: str, *, client: PluginModelClient | None = None
-    ) -> tuple[ProviderEntity, ...] | None:
-        generation = cls._load_plugin_model_providers_generation(tenant_id)
-        if generation is not None:
-            in_memory_cached_providers = cls._load_in_memory_plugin_model_providers(tenant_id, generation)
-            if in_memory_cached_providers is not None:
-                return in_memory_cached_providers
-
-        cache_keys = []
-        if generation is not None:
-            cache_keys.append(cls._get_plugin_model_providers_cache_key(tenant_id, generation))
-            if generation == 0:
-                cache_keys.append(cls._get_plugin_model_providers_cache_key(tenant_id))
-
-        if not cache_keys:
-            return None
+        cache_keys = [cls._get_plugin_model_providers_cache_key(tenant_id, generation)]
 
         try:
             cached_provider_entries = redis_client.mget(cache_keys)
-        except (RedisError, RuntimeError):
+        except (LockError, RedisError, RuntimeError):
             logger.warning("Failed to read cached plugin model providers for tenant %s.", tenant_id, exc_info=True)
-            return None
+            return None, False
 
         if len(cached_provider_entries) != len(cache_keys):
             logger.warning(
                 "Unexpected cached plugin model providers response size for tenant %s.",
                 tenant_id,
             )
-            return None
+            return None, False
 
         for cache_key, cached_providers in zip(cache_keys, cached_provider_entries):
             if not cached_providers:
                 continue
 
             try:
-                providers = tuple(_provider_entities_adapter.validate_json(cached_providers))
-                if generation is not None:
-                    cls._store_in_memory_plugin_model_providers(tenant_id, generation, providers)
-                return providers
+                payload = cls._decode_plugin_model_providers_cache_payload(cached_providers)
+                providers = tuple(_provider_entities_adapter.validate_json(payload))
+                return providers, True
             except (TypeError, ValueError, ValidationError):
                 logger.warning(
                     "Invalid cached plugin model providers for tenant %s; deleting cache key %s.",
@@ -249,7 +244,7 @@ class PluginService:
                         exc_info=True,
                     )
 
-        return None
+        return None, True
 
     @classmethod
     def _store_cached_plugin_model_providers(
@@ -257,15 +252,94 @@ class PluginService:
     ) -> None:
         cache_key = cls._get_plugin_model_providers_cache_key(tenant_id, generation)
         try:
-            payload = _provider_entities_adapter.dump_json(list(providers)).decode("utf-8")
+            payload = cls._encode_plugin_model_providers_cache_payload(
+                _provider_entities_adapter.dump_json(list(providers))
+            )
             redis_client.setex(cache_key, dify_config.PLUGIN_MODEL_PROVIDERS_CACHE_TTL, payload)
         except (RedisError, RuntimeError):
             logger.warning("Failed to cache plugin model providers for tenant %s.", tenant_id, exc_info=True)
 
+    @classmethod
+    @contextmanager
+    def _plugin_model_providers_refresh_lock(
+        cls, tenant_id: str, generation: int, *, wait_timeout: float
+    ) -> Iterator[bool]:
+        lock_key = cls._get_plugin_model_providers_lock_key(tenant_id, generation)
+        try:
+            refresh_lock: _RedisLock = redis_client.lock(
+                lock_key,
+                timeout=cls.PLUGIN_MODEL_PROVIDERS_LOCK_TTL,
+                sleep=cls.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL,
+            )
+        except (RedisError, RuntimeError):
+            logger.warning(
+                "Failed to create plugin model providers refresh lock for tenant %s.",
+                tenant_id,
+                exc_info=True,
+            )
+            yield False
+            return
+
+        try:
+            lock_acquired = refresh_lock.acquire(blocking=True, blocking_timeout=wait_timeout)
+        except LockError:
+            logger.warning(
+                "Provider refresh lock timed out; direct daemon fallback. tenant_id=%s generation=%s",
+                tenant_id,
+                generation,
+                exc_info=True,
+            )
+            yield False
+            return
+        except (RedisError, RuntimeError):
+            # Redis failures should not block provider discovery; callers fetch directly from the daemon.
+            logger.warning(
+                "Failed to acquire plugin model providers refresh lock for tenant %s.",
+                tenant_id,
+                exc_info=True,
+            )
+            yield False
+            return
+
+        if not lock_acquired:
+            logger.warning(
+                "Provider refresh lock timed out; direct daemon fallback. tenant_id=%s generation=%s",
+                tenant_id,
+                generation,
+            )
+            yield False
+            return
+
+        try:
+            yield True
+        finally:
+            try:
+                refresh_lock.release()
+            except (LockError, RedisError, RuntimeError):
+                # Release failures must not hide the daemon result or the original exception.
+                logger.warning(
+                    "Failed to release plugin model providers refresh lock for tenant %s generation %s.",
+                    tenant_id,
+                    generation,
+                    exc_info=True,
+                )
+
+    @classmethod
+    def _fetch_and_cache_plugin_model_providers(
+        cls, tenant_id: str, client: PluginModelClient | None, *, refresh_generation: int | None
+    ) -> tuple[ProviderEntity, ...]:
+        model_client = client or PluginModelClient()
+        providers = tuple(
+            cls._to_provider_entity(provider) for provider in model_client.fetch_model_providers(tenant_id)
+        )
+        generation = cls._load_plugin_model_providers_generation(tenant_id)
+        if generation is not None and generation == refresh_generation:
+            cls._store_cached_plugin_model_providers(tenant_id, generation, providers)
+        return providers
+
     @classmethod
     def invalidate_plugin_model_providers_cache(cls, tenant_id: str) -> None:
-        """Invalidate tenant-scoped provider metadata across Redis and worker-local mirrors."""
-        cls._plugin_model_providers_memory_cache.pop(tenant_id, None)
+        """Invalidate tenant-scoped provider metadata stored in Redis."""
         cache_key = cls._get_plugin_model_providers_cache_key(tenant_id)
         generation_key = cls._get_plugin_model_providers_generation_cache_key(tenant_id)
         try:
@@ -287,21 +361,68 @@ class PluginService:
         are intentionally owned by this service so tenant isolation and cache
         expiry are handled in one place.
         """
-        cached_providers = cls._load_cached_plugin_model_providers(tenant_id, client=client)
-        if cached_providers is not None:
-            return cached_providers
+        deadline = time.monotonic() + cls.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT
 
-        model_client = client or PluginModelClient()
-        providers = tuple(
-            cls._to_provider_entity(provider) for provider in model_client.fetch_model_providers(tenant_id)
-        )
-        if not providers:
-            return providers
-        generation = cls._load_plugin_model_providers_generation(tenant_id)
-        if generation is not None:
-            cls._store_in_memory_plugin_model_providers(tenant_id, generation, providers)
-            cls._store_cached_plugin_model_providers(tenant_id, generation, providers)
-        return providers
+        while True:
+            generation = cls._load_plugin_model_providers_generation(tenant_id)
+            cached_providers, cache_available = cls._load_cached_plugin_model_providers_for_generation(
+                tenant_id, generation
+            )
+            if cached_providers is not None:
+                return cached_providers
+
+            if generation is None or not cache_available:
+                return cls._fetch_and_cache_plugin_model_providers(
+                    tenant_id,
+                    client,
+                    refresh_generation=generation,
+                )
+
+            wait_timeout = deadline - time.monotonic()
+            if wait_timeout < 0:
+                logger.warning(
+                    "Provider refresh lock timed out; direct daemon fallback. tenant_id=%s generation=%s",
+                    tenant_id,
+                    generation,
+                )
+                return cls._fetch_and_cache_plugin_model_providers(
+                    tenant_id,
+                    client,
+                    refresh_generation=generation,
+                )
+
+            with cls._plugin_model_providers_refresh_lock(
+                tenant_id,
+                generation,
+                wait_timeout=wait_timeout,
+            ) as lock_acquired:
+                if not lock_acquired:
+                    return cls._fetch_and_cache_plugin_model_providers(
+                        tenant_id,
+                        client,
+                        refresh_generation=generation,
+                    )
+
+                latest_generation = cls._load_plugin_model_providers_generation(tenant_id)
+                cached_providers, cache_available = cls._load_cached_plugin_model_providers_for_generation(
+                    tenant_id, latest_generation
+                )
+                if cached_providers is not None:
+                    return cached_providers
+                if latest_generation is None or not cache_available:
+                    return cls._fetch_and_cache_plugin_model_providers(
+                        tenant_id,
+                        client,
+                        refresh_generation=latest_generation,
+                    )
+                if latest_generation != generation:
+                    continue
+
+                return cls._fetch_and_cache_plugin_model_providers(
+                    tenant_id,
+                    client,
+                    refresh_generation=generation,
+                )
 
     @staticmethod
     def fetch_latest_plugin_version(plugin_ids: Sequence[str]) -> Mapping[str, LatestPluginCache | None]:

api/core/rag/retrieval/dataset_retrieval.py

@@ -1030,6 +1030,10 @@ class DatasetRetrieval:
     ):
         """
         Persist dataset query audit rows for retrieval requests.
+
+        Query audit logging is a side effect of retrieval. Keep it in an
+        independent transaction so failures or commits here do not affect the
+        request/workflow transaction that called the retriever.
         """
         if not query and not attachment_ids:
             return
@@ -1041,6 +1045,9 @@ class DatasetRetrieval:
                 app_id,
             )
             return
+        created_by_role = self._resolve_creator_user_role(user_from)
+        if created_by_role is None:
+            return
         dataset_queries = []
         for dataset_id in dataset_ids:
             contents = []
@@ -1055,13 +1062,16 @@ class DatasetRetrieval:
                     content=json.dumps(contents),
                     source=DatasetQuerySource.APP,
                     source_app_id=app_id,
-                    created_by_role=CreatorUserRole(user_from),
+                    created_by_role=created_by_role,
                     created_by=created_by,
                 )
                 dataset_queries.append(dataset_query)
-            if dataset_queries:
-                db.session.add_all(dataset_queries)
-        db.session.commit()
+
+        if not dataset_queries:
+            return
+
+        with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
+            session.add_all(dataset_queries)
 
     def _retriever(
         self,

api/core/rbac/__init__.py

@@ -1,3 +1,3 @@
-from core.rbac.entities import RBACPermission, RBACResourceScope
+from core.rbac.entities import RBACPermission, RBACResourceScope, RBACResourceWhitelistScope
 
-__all__ = ["RBACPermission", "RBACResourceScope"]
+__all__ = ["RBACPermission", "RBACResourceScope", "RBACResourceWhitelistScope"]

api/core/rbac/entities.py

@@ -13,6 +13,14 @@ class RBACResourceScope(StrEnum):
     WORKSPACE = "workspace"
 
 
+class RBACResourceWhitelistScope(StrEnum):
+    """Whitelist scopes accepted by RBAC app and dataset access config APIs."""
+
+    ALL = "all"
+    SPECIFIC = "specific"
+    ONLY_ME = "only_me"
+
+
 class RBACPermission(StrEnum):
     """Permission points (RBAC scenes) checked by ``rbac_permission_required``.
 

api/core/tools/custom_tool/tool.py

@@ -102,16 +102,17 @@ class ApiTool(Tool):
             elif not isinstance(credentials["api_key_value"], str):
                 raise ToolProviderCredentialValidationError("api_key_value must be a string")
 
+            api_key_value = credentials["api_key_value"]
             if "api_key_header_prefix" in credentials:
                 api_key_header_prefix = credentials["api_key_header_prefix"]
-                if api_key_header_prefix == "basic" and credentials["api_key_value"]:
-                    credentials["api_key_value"] = f"Basic {credentials['api_key_value']}"
-                elif api_key_header_prefix == "bearer" and credentials["api_key_value"]:
-                    credentials["api_key_value"] = f"Bearer {credentials['api_key_value']}"
+                if api_key_header_prefix == "basic" and api_key_value:
+                    api_key_value = f"Basic {api_key_value}"
+                elif api_key_header_prefix == "bearer" and api_key_value:
+                    api_key_value = f"Bearer {api_key_value}"
                 elif api_key_header_prefix == "custom":
                     pass
 
-            headers[api_key_header] = credentials["api_key_value"]
+            headers[api_key_header] = api_key_value
 
         elif credentials["auth_type"] == "api_key_query":
             # For query parameter authentication, we don't add anything to headers

api/core/tools/tool_engine.py

@@ -7,6 +7,7 @@ from datetime import UTC, datetime
 from mimetypes import guess_type
 from typing import Any, Union, cast
 
+from sqlalchemy.orm import sessionmaker
 from yarl import URL
 
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -338,47 +339,49 @@ class ToolEngine:
         user_id: str,
     ) -> list[str]:
         """
-        Create message file
+        Create message files produced by a tool call.
+
+        Tool file persistence is a side effect of agent execution. Use an
+        independent transaction so this helper never commits or closes the
+        caller's request-scoped session.
 
         :return: message file ids
         """
         result = []
 
-        for message in tool_messages:
-            if "image" in message.mimetype:
-                file_type = FileType.IMAGE
-            elif "video" in message.mimetype:
-                file_type = FileType.VIDEO
-            elif "audio" in message.mimetype:
-                file_type = FileType.AUDIO
-            elif "text" in message.mimetype or "pdf" in message.mimetype:
-                file_type = FileType.DOCUMENT
-            else:
-                file_type = FileType.CUSTOM
+        with sessionmaker(bind=db.engine, expire_on_commit=False).begin() as session:
+            for message in tool_messages:
+                # extract tool file id from url
+                tool_file_id = message.url.split("/")[-1].split(".")[0]
+                message_file = MessageFile(
+                    message_id=agent_message.id,
+                    type=ToolEngine._resolve_tool_file_type(message),
+                    transfer_method=FileTransferMethod.TOOL_FILE,
+                    belongs_to=MessageFileBelongsTo.ASSISTANT,
+                    url=message.url,
+                    upload_file_id=tool_file_id,
+                    created_by_role=(
+                        CreatorUserRole.ACCOUNT
+                        if invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}
+                        else CreatorUserRole.END_USER
+                    ),
+                    created_by=user_id,
+                )
 
-            # extract tool file id from url
-            tool_file_id = message.url.split("/")[-1].split(".")[0]
-            message_file = MessageFile(
-                message_id=agent_message.id,
-                type=file_type,
-                transfer_method=FileTransferMethod.TOOL_FILE,
-                belongs_to=MessageFileBelongsTo.ASSISTANT,
-                url=message.url,
-                upload_file_id=tool_file_id,
-                created_by_role=(
-                    CreatorUserRole.ACCOUNT
-                    if invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}
-                    else CreatorUserRole.END_USER
-                ),
-                created_by=user_id,
-            )
-
-            db.session.add(message_file)
-            db.session.commit()
-            db.session.refresh(message_file)
-
-            result.append(message_file.id)
-
-        db.session.close()
+                session.add(message_file)
+                result.append(message_file.id)
 
         return result
+
+    @staticmethod
+    def _resolve_tool_file_type(message: ToolInvokeMessageBinary) -> FileType:
+        if "image" in message.mimetype:
+            return FileType.IMAGE
+        elif "video" in message.mimetype:
+            return FileType.VIDEO
+        elif "audio" in message.mimetype:
+            return FileType.AUDIO
+        elif "text" in message.mimetype or "pdf" in message.mimetype:
+            return FileType.DOCUMENT
+        else:
+            return FileType.CUSTOM

api/core/tools/utils/parser.py

@@ -5,10 +5,10 @@ from json import loads as json_loads
 from json.decoder import JSONDecodeError
 from typing import Any, TypedDict
 
-import httpx
 from flask import has_request_context, request
 from yaml import YAMLError, safe_load
 
+from core.helper import ssrf_proxy
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_bundle import ApiToolBundle
 from core.tools.entities.tool_entities import ApiProviderSchemaType, ToolParameter
@@ -376,7 +376,7 @@ class ApiBasedToolSchemaParser:
             raise ToolNotSupportedError("Only openapi is supported now.")
 
         # get openapi yaml
-        response = httpx.get(
+        response = ssrf_proxy.get(
             api_url, headers={"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "}, timeout=5
         )
 

api/docker/entrypoint.sh

@@ -35,10 +35,10 @@ if [[ "${MODE}" == "worker" ]]; then
   if [[ -z "${CELERY_QUEUES}" ]]; then
     if [[ "${EDITION}" == "CLOUD" ]]; then
       # Cloud edition: separate queues for dataset and trigger tasks
-      DEFAULT_QUEUES="api_token,dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow_professional,workflow_team,workflow_sandbox,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_publisher,trigger_refresh_executor,retention,workflow_based_app_execution"
+      DEFAULT_QUEUES="api_token,dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,app_rbac,plugin,workflow_storage,conversation,workflow_professional,workflow_team,workflow_sandbox,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_publisher,trigger_refresh_executor,retention,workflow_based_app_execution"
     else
       # Community edition (SELF_HOSTED): dataset, pipeline and workflow have separate queues
-      DEFAULT_QUEUES="api_token,dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_publisher,trigger_refresh_executor,retention,workflow_based_app_execution"
+      DEFAULT_QUEUES="api_token,dataset,dataset_summary,priority_dataset,priority_pipeline,pipeline,mail,ops_trace,app_deletion,app_rbac,plugin,workflow_storage,conversation,workflow,schedule_poller,schedule_executor,triggered_workflow_dispatcher,trigger_refresh_publisher,trigger_refresh_executor,retention,workflow_based_app_execution"
     fi
   else
     DEFAULT_QUEUES="${CELERY_QUEUES}"

api/extensions/ext_celery.py

@@ -153,6 +153,7 @@ def init_app(app: DifyApp) -> Celery:
         "tasks.trigger_processing_tasks",  # async trigger processing
         "tasks.generate_summary_index_task",  # summary index generation
         "tasks.regenerate_summary_index_task",  # summary index regeneration
+        "tasks.initialize_created_app_rbac_access_task",  # app access initialization
         "tasks.app_generate.resume_agent_app_task",  # ENG-635: Agent v2 chat ask_human resume
     ]
     day = dify_config.CELERY_BEAT_SCHEDULER_TIME

api/extensions/ext_commands.py

@@ -27,6 +27,7 @@ def init_app(app: DifyApp):
         install_plugins,
         install_rag_pipeline_plugins,
         migrate_data_for_plugin,
+        migrate_dataset_permissions_to_rbac,
         migrate_member_roles_to_rbac,
         migrate_oss,
         migration_data_wizard,
@@ -56,6 +57,7 @@ def init_app(app: DifyApp):
         upgrade_db,
         fix_app_site_missing,
         migrate_data_for_plugin,
+        migrate_dataset_permissions_to_rbac,
         migrate_member_roles_to_rbac,
         backfill_plugin_auto_upgrade,
         extract_plugins,

api/libs/oauth.py

@@ -34,6 +34,7 @@ class OAuthState(TypedDict, total=False):
     invite_token: str
     timezone: str
     language: str
+    redirect_url: str
 
 
 class GitHubEmailRecord(TypedDict, total=False):
@@ -72,6 +73,7 @@ def encode_oauth_state(
     invite_token: str | None = None,
     timezone: str | None = None,
     language: str | None = None,
+    redirect_url: str | None = None,
 ) -> str | None:
     state: OAuthState = {}
     if invite_token:
@@ -80,6 +82,8 @@ def encode_oauth_state(
         state["timezone"] = timezone
     if language:
         state["language"] = language
+    if redirect_url:
+        state["redirect_url"] = redirect_url
     if not state:
         return None
 
@@ -122,6 +126,7 @@ class OAuth:
         invite_token: str | None = None,
         timezone: str | None = None,
         language: str | None = None,
+        redirect_url: str | None = None,
     ) -> str:
         raise NotImplementedError()
 
@@ -151,13 +156,19 @@ class GitHubOAuth(OAuth):
         invite_token: str | None = None,
         timezone: str | None = None,
         language: str | None = None,
+        redirect_url: str | None = None,
     ) -> str:
         params = {
             "client_id": self.client_id,
             "redirect_uri": self.redirect_uri,
             "scope": "user:email",  # Request only basic user information
         }
-        state = encode_oauth_state(invite_token=invite_token, timezone=timezone, language=language)
+        state = encode_oauth_state(
+            invite_token=invite_token,
+            timezone=timezone,
+            language=language,
+            redirect_url=redirect_url,
+        )
         if state:
             params["state"] = state
         return f"{self._AUTH_URL}?{urllib.parse.urlencode(params)}"
@@ -248,6 +259,7 @@ class GoogleOAuth(OAuth):
         invite_token: str | None = None,
         timezone: str | None = None,
         language: str | None = None,
+        redirect_url: str | None = None,
     ) -> str:
         params = {
             "client_id": self.client_id,
@@ -255,7 +267,12 @@ class GoogleOAuth(OAuth):
             "redirect_uri": self.redirect_uri,
             "scope": "openid email",
         }
-        state = encode_oauth_state(invite_token=invite_token, timezone=timezone, language=language)
+        state = encode_oauth_state(
+            invite_token=invite_token,
+            timezone=timezone,
+            language=language,
+            redirect_url=redirect_url,
+        )
         if state:
             params["state"] = state
         return f"{self._AUTH_URL}?{urllib.parse.urlencode(params)}"

api/models/enums.py

@@ -214,6 +214,18 @@ class EndUserType(StrEnum):
     SERVICE_API = "service-api"
     TRIGGER = "trigger"
 
+    @classmethod
+    @override
+    def _missing_(cls, value):
+        # Legacy rows persisted the service-api type with an underscore before it
+        # was normalized to the hyphenated value. The
+        # `4f7b2c8d9a10_normalize_legacy_end_user_type` migration rewrites those
+        # rows, but tolerate the old value here as well so an unmigrated end user
+        # keeps loading instead of failing enum validation on every request.
+        if value == "service_api":
+            return cls.SERVICE_API
+        return super()._missing_(value)
+
 
 class DocumentDocType(StrEnum):
     """Document doc_type classification"""

api/openapi/markdown/console-openapi.md

@@ -2868,6 +2868,7 @@ Delete an existing tracing configuration for an application
 | ---- | ----------- |
 | 204 | Tracing configuration deleted successfully |
 | 400 | Invalid request parameters or configuration not found |
+| 403 | Insufficient permissions |
 
 ### [GET] /apps/{app_id}/trace-config
 Get tracing configuration for an application
@@ -2909,6 +2910,7 @@ Update an existing tracing configuration for an application
 | ---- | ----------- | ------ |
 | 200 | Tracing configuration updated successfully | **application/json**: [TraceAppConfigResponse](#traceappconfigresponse)<br> |
 | 400 | Invalid request parameters or configuration not found |  |
+| 403 | Insufficient permissions |  |
 
 ### [POST] /apps/{app_id}/trace-config
 **Create a new trace app configuration**
@@ -2933,6 +2935,7 @@ Create a new tracing configuration for an application
 | ---- | ----------- | ------ |
 | 201 | Tracing configuration created successfully | **application/json**: [TraceAppConfigResponse](#traceappconfigresponse)<br> |
 | 400 | Invalid request parameters or configuration already exists |  |
+| 403 | Insufficient permissions |  |
 
 ### [POST] /apps/{app_id}/trigger-enable
 **Update app trigger (enable/disable)**
@@ -7000,6 +7003,7 @@ Initiate OAuth login process
 | provider | path | OAuth provider name (github/google) | Yes | string |
 | invite_token | query | Optional invitation token | No | string |
 | language | query | Preferred interface language | No | string |
+| redirect_url | query | Relative page to resume after login | No | string |
 | timezone | query | Preferred timezone | No | string |
 
 #### Responses
@@ -16859,6 +16863,7 @@ Enum class for large language model mode.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | expired_at | string |  | Yes |
+| seats | [LicenseLimitationModel](#licenselimitationmodel) |  | Yes |
 | status | [LicenseStatus](#licensestatus) |  | Yes |
 | workspaces | [LicenseLimitationModel](#licenselimitationmodel) |  | Yes |
 
@@ -17529,6 +17534,7 @@ Coarse node-level status used by Inspector to pick a banner.
 | ---- | ---- | ----------- | -------- |
 | invite_token | string | Optional invitation token | No |
 | language | string | Preferred interface language | No |
+| redirect_url | string | Relative page to resume after login | No |
 | timezone | string | Preferred timezone | No |
 
 #### OAuthProviderAccountResponse
@@ -18795,6 +18801,13 @@ Model class for provider quota configuration.
 | ---- | ---- | ----------- | -------- |
 | url | string | URL to fetch | Yes |
 
+#### ReplaceUserAccessPolicies
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| access_policy_ids | [ string ] |  | No |
+| account_ids | [ string ] |  | No |
+
 #### ReplaceUserAccessPoliciesResponse
 
 | Name | Type | Description | Required |

api/openapi/markdown/web-openapi.md

@@ -1337,6 +1337,7 @@ Parsed multipart form fields for HITL uploads.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | expired_at | string |  | Yes |
+| seats | [LicenseLimitationModel](#licenselimitationmodel) |  | Yes |
 | status | [LicenseStatus](#licensestatus) |  | Yes |
 | workspaces | [LicenseLimitationModel](#licenselimitationmodel) |  | Yes |
 

api/providers/vdb/vdb-myscale/src/dify_vdb_myscale/myscale_vector.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import re
 import uuid
 from enum import StrEnum
 from typing import Any, override
@@ -17,6 +18,8 @@ from models.dataset import Dataset
 
 logger = logging.getLogger(__name__)
 
+METADATA_KEY_PATTERN = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+
 
 class MyScaleConfig(BaseModel):
     host: str
@@ -102,7 +105,10 @@ class MyScaleVector(BaseVector):
 
     @override
     def text_exists(self, id: str) -> bool:
-        results = self._client.query(f"SELECT id FROM {self._config.database}.{self._collection_name} WHERE id='{id}'")
+        results = self._client.query(
+            f"SELECT id FROM {self._config.database}.{self._collection_name} WHERE id={{id:String}}",
+            parameters={"id": id},
+        )
         return results.row_count > 0
 
     @override
@@ -110,20 +116,26 @@ class MyScaleVector(BaseVector):
         if not ids:
             return
         self._client.command(
-            f"DELETE FROM {self._config.database}.{self._collection_name} WHERE id IN {str(tuple(ids))}"
+            f"DELETE FROM {self._config.database}.{self._collection_name} WHERE id IN {{ids:Array(String)}}",
+            parameters={"ids": ids},
         )
 
     @override
     def get_ids_by_metadata_field(self, key: str, value: str):
+        self._validate_metadata_key(key)
         rows = self._client.query(
-            f"SELECT DISTINCT id FROM {self._config.database}.{self._collection_name} WHERE metadata.{key}='{value}'"
+            f"SELECT DISTINCT id FROM {self._config.database}.{self._collection_name} "
+            f"WHERE metadata.{key}={{value:String}}",
+            parameters={"value": value},
         ).result_rows
         return [row[0] for row in rows]
 
     @override
     def delete_by_metadata_field(self, key: str, value: str):
+        self._validate_metadata_key(key)
         self._client.command(
-            f"DELETE FROM {self._config.database}.{self._collection_name} WHERE metadata.{key}='{value}'"
+            f"DELETE FROM {self._config.database}.{self._collection_name} WHERE metadata.{key}={{value:String}}",
+            parameters={"value": value},
         )
 
     @override
@@ -132,22 +144,29 @@ class MyScaleVector(BaseVector):
 
     @override
     def search_by_full_text(self, query: str, **kwargs: Any) -> list[Document]:
-        return self._search(f"TextSearch('enable_nlq=false')(text, '{query}')", SortOrder.DESC, **kwargs)
+        return self._search(
+            "TextSearch('enable_nlq=false')(text, {query:String})",
+            SortOrder.DESC,
+            parameters={"query": query},
+            **kwargs,
+        )
 
-    def _search(self, dist: str, order: SortOrder, **kwargs: Any) -> list[Document]:
+    def _search(
+        self, dist: str, order: SortOrder, parameters: dict[str, Any] | None = None, **kwargs: Any
+    ) -> list[Document]:
         top_k = kwargs.get("top_k", 4)
         if not isinstance(top_k, int) or top_k <= 0:
             raise ValueError("top_k must be a positive integer")
         score_threshold = float(kwargs.get("score_threshold") or 0.0)
-        where_str = (
-            f"WHERE dist < {1 - score_threshold}"
-            if self._metric.upper() == "COSINE" and order == SortOrder.ASC and score_threshold > 0.0
-            else ""
-        )
+        where_conditions = []
+        query_parameters = dict(parameters or {})
+        if self._metric.upper() == "COSINE" and order == SortOrder.ASC and score_threshold > 0.0:
+            where_conditions.append(f"dist < {1 - score_threshold}")
         document_ids_filter = kwargs.get("document_ids_filter")
         if document_ids_filter:
-            document_ids = ", ".join(f"'{id}'" for id in document_ids_filter)
-            where_str = f"{where_str} AND metadata['document_id'] in ({document_ids})"
+            where_conditions.append("metadata['document_id'] IN {document_ids_filter:Array(String)}")
+            query_parameters["document_ids_filter"] = document_ids_filter
+        where_str = f"WHERE {' AND '.join(where_conditions)}" if where_conditions else ""
         sql = f"""
             SELECT text, vector, metadata, {dist} as dist FROM {self._config.database}.{self._collection_name}
             {where_str} ORDER BY dist {order.value} LIMIT {top_k}
@@ -159,12 +178,17 @@ class MyScaleVector(BaseVector):
                     vector=r["vector"],
                     metadata=r["metadata"],
                 )
-                for r in self._client.query(sql).named_results()
+                for r in self._client.query(sql, parameters=query_parameters).named_results()
             ]
         except Exception:
             logger.exception("Vector search operation failed")
             return []
 
+    @staticmethod
+    def _validate_metadata_key(key: str) -> None:
+        if not METADATA_KEY_PATTERN.match(key):
+            raise ValueError("metadata key must be a valid identifier")
+
     @override
     def delete(self):
         self._client.command(f"DROP TABLE IF EXISTS {self._config.database}.{self._collection_name}")

api/providers/vdb/vdb-myscale/tests/unit_tests/test_myscale_vector.py

@@ -181,14 +181,41 @@ def test_text_exists_and_metadata_operations(myscale_module):
     vector = myscale_module.MyScaleVector("collection_1", _config(myscale_module))
     vector._client.query.return_value = SimpleNamespace(row_count=1, result_rows=[("id-1",), ("id-2",)])
 
-    assert vector.text_exists("id-1") is True
-    assert vector.get_ids_by_metadata_field("document_id", "doc-1") == ["id-1", "id-2"]
+    assert vector.text_exists("id-1' OR '1'='1") is True
+    text_exists_call = vector._client.query.call_args
+    assert "id={id:String}" in text_exists_call.args[0]
+    assert "id-1' OR '1'='1" not in text_exists_call.args[0]
+    assert text_exists_call.kwargs["parameters"] == {"id": "id-1' OR '1'='1"}
+
+    assert vector.get_ids_by_metadata_field("document_id", "doc-1' OR '1'='1") == ["id-1", "id-2"]
+    metadata_query_call = vector._client.query.call_args
+    assert "metadata.document_id={value:String}" in metadata_query_call.args[0]
+    assert "doc-1' OR '1'='1" not in metadata_query_call.args[0]
+    assert metadata_query_call.kwargs["parameters"] == {"value": "doc-1' OR '1'='1"}
 
     vector.delete_by_ids(["id-1", "id-2"])
-    vector.delete_by_metadata_field("document_id", "doc-1")
+    delete_ids_call = vector._client.command.call_args
+    assert "id IN {ids:Array(String)}" in delete_ids_call.args[0]
+    assert delete_ids_call.kwargs["parameters"] == {"ids": ["id-1", "id-2"]}
+
+    vector.delete_by_metadata_field("document_id", "doc-1' OR '1'='1")
+    delete_metadata_call = vector._client.command.call_args
+    assert "metadata.document_id={value:String}" in delete_metadata_call.args[0]
+    assert "doc-1' OR '1'='1" not in delete_metadata_call.args[0]
+    assert delete_metadata_call.kwargs["parameters"] == {"value": "doc-1' OR '1'='1"}
     assert vector._client.command.call_count >= 2
 
 
+def test_metadata_operations_reject_invalid_key(myscale_module):
+    vector = myscale_module.MyScaleVector("collection_1", _config(myscale_module))
+
+    with pytest.raises(ValueError, match="metadata key must be a valid identifier"):
+        vector.get_ids_by_metadata_field("document_id) OR 1=1 --", "doc-1")
+
+    with pytest.raises(ValueError, match="metadata key must be a valid identifier"):
+        vector.delete_by_metadata_field("document_id) OR 1=1 --", "doc-1")
+
+
 def test_search_delegation_methods(myscale_module):
     vector = myscale_module.MyScaleVector("collection_1", _config(myscale_module))
     vector._search = MagicMock(return_value=["result"])
@@ -199,6 +226,28 @@ def test_search_delegation_methods(myscale_module):
     assert result_vector == ["result"]
     assert result_text == ["result"]
     assert vector._search.call_count == 2
+    vector._search.assert_any_call(
+        "TextSearch('enable_nlq=false')(text, {query:String})",
+        myscale_module.SortOrder.DESC,
+        parameters={"query": "hello"},
+        top_k=2,
+    )
+
+
+def test_search_by_full_text_uses_query_parameters(myscale_module):
+    vector = myscale_module.MyScaleVector("collection_1", _config(myscale_module))
+    vector._client.query.return_value = SimpleNamespace(
+        named_results=lambda: [{"text": "doc", "vector": [0.1], "metadata": {"doc_id": "1"}}]
+    )
+    payload = "x') AS dist FROM dify.collection_1 UNION ALL SELECT secret FROM users --"
+
+    docs = vector.search_by_full_text(payload, top_k=2)
+
+    assert len(docs) == 1
+    sql = vector._client.query.call_args.args[0]
+    assert payload not in sql
+    assert "TextSearch('enable_nlq=false')(text, {query:String})" in sql
+    assert vector._client.query.call_args.kwargs["parameters"] == {"query": payload}
 
 
 def test_search_with_document_filter_and_exception(myscale_module):
@@ -215,7 +264,8 @@ def test_search_with_document_filter_and_exception(myscale_module):
     )
     assert len(docs) == 1
     sql = vector._client.query.call_args.args[0]
-    assert "metadata['document_id'] in ('doc-1', 'doc-2')" in sql
+    assert "WHERE metadata['document_id'] IN {document_ids_filter:Array(String)}" in sql
+    assert vector._client.query.call_args.kwargs["parameters"] == {"document_ids_filter": ["doc-1", "doc-2"]}
 
     vector._client.query.side_effect = RuntimeError("boom")
     assert vector._search("distance(vector, [0.1])", myscale_module.SortOrder.ASC, top_k=1) == []

api/providers/vdb/vdb-qdrant/src/dify_vdb_qdrant/qdrant_vector.py

@@ -33,7 +33,6 @@ from models.dataset import Dataset, DatasetCollectionBinding
 
 if TYPE_CHECKING:
     from qdrant_client.conversions import common_types
-    from qdrant_client.http import models as rest
 
     type DictFilter = dict[str, str | int | bool | dict | list]
     type MetadataFilter = DictFilter | common_types.Filter

api/providers/vdb/vdb-tidb-on-qdrant/src/dify_vdb_tidb_on_qdrant/tidb_on_qdrant_vector.py

@@ -41,7 +41,6 @@ from models.enums import TidbAuthBindingStatus
 if TYPE_CHECKING:
     from qdrant_client import grpc  # noqa
     from qdrant_client.conversions import common_types
-    from qdrant_client.http import models as rest
 
     type DictFilter = dict[str, str | int | bool | dict | list]
     type MetadataFilter = DictFilter | common_types.Filter

api/pyproject.toml

@@ -42,11 +42,12 @@ dependencies = [
     "opentelemetry-propagator-b3>=1.41.1,<2.0.0",
     "readabilipy==0.3.0",
     "resend>=2.27.0,<3.0.0",
+    "zstandard==0.25.0",
     # Emerging: newer and fast-moving, use compatible pins
     "fastopenapi[flask]==0.7.0",
     "graphon==0.5.3",
     "httpx-sse==0.4.3",
-    "json-repair==0.59.4",
+    "json-repair==0.60.1",
 ]
 # Before adding new dependency, consider place it in
 # alphabet order (a-z) and suitable group.
@@ -107,6 +108,7 @@ override-dependencies = [
     "litellm>=1.83.10,<2.0.0",
     "pyarrow>=23.0.1,<24.0.0",
     "cryptography>=49.0.0,<50.0.0",
+    "setuptools>=80.10.2,<81",
 ]
 
 [dependency-groups]

api/services/account_service.py

@@ -10,6 +10,7 @@ import json
 import logging
 import secrets
 import uuid
+from collections.abc import Iterator
 from datetime import UTC, datetime, timedelta
 from hashlib import sha256
 from typing import Any, NotRequired, TypedDict, cast
@@ -64,6 +65,7 @@ from services.errors.account import (
     MemberNotInTenantError,
     NoPermissionError,
     RoleAlreadyAssignedError,
+    SeatsLimitExceededError,
     TenantNotFoundError,
 )
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
@@ -433,6 +435,13 @@ class AccountService:
 
             raise AccountNotFound()
 
+        # A licensed seat is one Account row, deployment-wide; joining an existing
+        # account into another workspace does not pass through here and costs no seat.
+        # is_authenticated=True: server-side enforcement needs the full license payload,
+        # which the enterprise fill withholds from unauthenticated (browser-facing) calls.
+        if not FeatureService.get_system_features(is_authenticated=True).license.seats.is_available():
+            raise SeatsLimitExceededError("licensed seats limit exceeded")
+
         if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(email):
             raise AccountRegisterError(
                 description=(
@@ -1576,6 +1585,25 @@ class TenantService:
 
         return updated_accounts
 
+    @staticmethod
+    def iter_member_account_id_batches(tenant_id: str, batch_size: int, *, session: Session) -> Iterator[list[str]]:
+        """Yield workspace member account ids in bounded, ordered batches."""
+        offset = 0
+        while True:
+            stmt = (
+                select(TenantAccountJoin.account_id)
+                .where(TenantAccountJoin.tenant_id == tenant_id)
+                .order_by(TenantAccountJoin.id)
+                .offset(offset)
+                .limit(batch_size)
+            )
+            account_ids = list(session.scalars(stmt).all())
+            if not account_ids:
+                return
+
+            yield account_ids
+            offset += batch_size
+
     @staticmethod
     def get_dataset_operator_members(tenant: Tenant, *, session: scoped_session | Session) -> list[Account]:
         """Get dataset admin members"""
@@ -1981,6 +2009,10 @@ class RegisterService:
             session.rollback()
             logger.exception("Register failed")
             raise AccountRegisterError("Workspace is not allowed to create.")
+        except SeatsLimitExceededError:
+            session.rollback()
+            logger.exception("Register failed")
+            raise
         except AccountRegisterError as are:
             session.rollback()
             logger.exception("Register failed")

api/services/billing_service.py

@@ -50,9 +50,26 @@ class QuotaReleaseResult(TypedDict):
     released: int
 
 
+class QuotaBalanceResult(TypedDict):
+    available: int
+    reserved: int
+    quota: int
+    usage: int
+
+
+class QuotaConsumeCappedResult(TypedDict):
+    deducted: int
+    available: int
+    reserved: int
+    quota: int
+    usage: int
+
+
 _quota_reserve_adapter = TypeAdapter(QuotaReserveResult)
 _quota_commit_adapter = TypeAdapter(QuotaCommitResult)
 _quota_release_adapter = TypeAdapter(QuotaReleaseResult)
+_quota_balance_adapter = TypeAdapter(QuotaBalanceResult)
+_quota_consume_capped_adapter = TypeAdapter(QuotaConsumeCappedResult)
 
 
 class _TenantFeatureQuota(TypedDict):
@@ -176,6 +193,7 @@ class DismissNotificationDict(TypedDict):
 
 class BillingService:
     base_url = os.environ.get("BILLING_API_URL", "BILLING_API_URL")
+    quota_base_url = os.environ.get("BILLING_QUOTA_API_URL") or base_url
     secret_key = os.environ.get("BILLING_API_SECRET_KEY", "BILLING_API_SECRET_KEY")
 
     compliance_download_rate_limiter = RateLimiter("compliance_download_rate_limiter", 4, 60)
@@ -215,12 +233,18 @@ class BillingService:
     def get_quota_info(cls, tenant_id: str) -> TenantFeatureQuotaInfo:
         params = {"tenant_id": tenant_id}
         return _tenant_feature_quota_info_adapter.validate_python(
-            cls._send_request("GET", "/quota/info", params=params)
+            cls._send_quota_request("GET", "/quota/info", params=params)
         )
 
     @classmethod
     def quota_reserve(
-        cls, tenant_id: str, feature_key: str, request_id: str, amount: int = 1, meta: dict | None = None
+        cls,
+        tenant_id: str,
+        feature_key: str,
+        request_id: str,
+        amount: int = 1,
+        meta: dict | None = None,
+        bucket: str = "",
     ) -> QuotaReserveResult:
         """Reserve quota before task execution."""
         payload: dict = {
@@ -229,13 +253,21 @@ class BillingService:
             "request_id": request_id,
             "amount": amount,
         }
+        if bucket:
+            payload["bucket"] = bucket
         if meta:
             payload["meta"] = meta
-        return _quota_reserve_adapter.validate_python(cls._send_request("POST", "/quota/reserve", json=payload))
+        return _quota_reserve_adapter.validate_python(cls._send_quota_request("POST", "/quota/reserve", json=payload))
 
     @classmethod
     def quota_commit(
-        cls, tenant_id: str, feature_key: str, reservation_id: str, actual_amount: int, meta: dict | None = None
+        cls,
+        tenant_id: str,
+        feature_key: str,
+        reservation_id: str,
+        actual_amount: int,
+        meta: dict | None = None,
+        bucket: str = "",
     ) -> QuotaCommitResult:
         """Commit a reservation with actual consumption."""
         payload: dict = {
@@ -244,23 +276,57 @@ class BillingService:
             "reservation_id": reservation_id,
             "actual_amount": actual_amount,
         }
+        if bucket:
+            payload["bucket"] = bucket
         if meta:
             payload["meta"] = meta
-        return _quota_commit_adapter.validate_python(cls._send_request("POST", "/quota/commit", json=payload))
+        return _quota_commit_adapter.validate_python(cls._send_quota_request("POST", "/quota/commit", json=payload))
 
     @classmethod
-    def quota_release(cls, tenant_id: str, feature_key: str, reservation_id: str) -> QuotaReleaseResult:
+    def quota_release(
+        cls, tenant_id: str, feature_key: str, reservation_id: str, bucket: str = ""
+    ) -> QuotaReleaseResult:
         """Release a reservation (cancel, return frozen quota)."""
-        return _quota_release_adapter.validate_python(
-            cls._send_request(
-                "POST",
-                "/quota/release",
-                json={
-                    "tenant_id": tenant_id,
-                    "feature_key": feature_key,
-                    "reservation_id": reservation_id,
-                },
-            )
+        payload = {
+            "tenant_id": tenant_id,
+            "feature_key": feature_key,
+            "reservation_id": reservation_id,
+        }
+        if bucket:
+            payload["bucket"] = bucket
+        return _quota_release_adapter.validate_python(cls._send_quota_request("POST", "/quota/release", json=payload))
+
+    @classmethod
+    def quota_get_balance(cls, tenant_id: str, feature_key: str, bucket: str = "") -> QuotaBalanceResult:
+        """Get quota balance for a feature bucket."""
+        params = {"tenant_id": tenant_id, "feature_key": feature_key}
+        if bucket:
+            params["bucket"] = bucket
+        return _quota_balance_adapter.validate_python(cls._send_quota_request("GET", "/quota/balance", params=params))
+
+    @classmethod
+    def quota_consume_capped(
+        cls,
+        tenant_id: str,
+        feature_key: str,
+        request_id: str,
+        amount: int,
+        meta: dict | None = None,
+        bucket: str = "",
+    ) -> QuotaConsumeCappedResult:
+        """Consume up to the available quota and return the actual deducted amount."""
+        payload: dict = {
+            "tenant_id": tenant_id,
+            "feature_key": feature_key,
+            "request_id": request_id,
+            "amount": amount,
+        }
+        if bucket:
+            payload["bucket"] = bucket
+        if meta:
+            payload["meta"] = meta
+        return _quota_consume_capped_adapter.validate_python(
+            cls._send_quota_request("POST", "/quota/consume-capped", json=payload)
         )
 
     @classmethod
@@ -334,6 +400,12 @@ class BillingService:
         params = {"tenant_id": tenant_id, "feature_key": feature_key}
         return cls._send_request("GET", "/billing/tenant_feature_plan/usage", params=params)
 
+    @classmethod
+    def _send_quota_request(
+        cls, method: Literal["GET", "POST", "DELETE", "PUT"], endpoint: str, json=None, params=None
+    ):
+        return cls._send_request(method, endpoint, json=json, params=params, base_url=cls.quota_base_url)
+
     @classmethod
     @retry(
         wait=wait_fixed(2),
@@ -341,10 +413,17 @@ class BillingService:
         retry=retry_if_exception_type(httpx.RequestError),
         reraise=True,
     )
-    def _send_request(cls, method: Literal["GET", "POST", "DELETE", "PUT"], endpoint: str, json=None, params=None):
+    def _send_request(
+        cls,
+        method: Literal["GET", "POST", "DELETE", "PUT"],
+        endpoint: str,
+        json=None,
+        params=None,
+        base_url: str | None = None,
+    ):
         headers = {"Content-Type": "application/json", "Billing-Api-Secret-Key": cls.secret_key}
 
-        url = f"{cls.base_url}{endpoint}"
+        url = f"{base_url or cls.base_url}{endpoint}"
         response = _http_client.request(method, url, json=json, params=params, headers=headers, follow_redirects=True)
         if method == "GET" and response.status_code != httpx.codes.OK:
             raise ValueError("Unable to retrieve billing information. Please try again later or contact support.")

api/services/credit_pool_service.py

@@ -7,6 +7,8 @@ from piling up database transactions while preserving cross-tenant concurrency.
 
 import logging
 from collections.abc import Callable
+from dataclasses import dataclass
+from uuid import uuid4
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -21,11 +23,37 @@ from models.enums import ProviderQuotaType
 
 logger = logging.getLogger(__name__)
 
+FEATURE_KEY_CREDIT_POOL = "credit_pool"
 CREDIT_POOL_TENANT_LOCK_TIMEOUT_SECONDS = 10
 CREDIT_POOL_TENANT_LOCK_BLOCKING_TIMEOUT_SECONDS = 5
 
 
+@dataclass(frozen=True)
+class CreditPoolBalance:
+    tenant_id: str
+    pool_type: str
+    quota_limit: int
+    quota_used: int
+
+    @property
+    def remaining_credits(self) -> int:
+        if self.quota_limit == -1:
+            return -1
+        return max(0, self.quota_limit - self.quota_used)
+
+    def has_sufficient_credits(self, required_credits: int) -> bool:
+        return self.quota_limit == -1 or self.remaining_credits >= required_credits
+
+
 class CreditPoolService:
+    @staticmethod
+    def _normalize_pool_type(pool_type: str | ProviderQuotaType) -> str:
+        return pool_type.value if isinstance(pool_type, ProviderQuotaType) else str(pool_type)
+
+    @staticmethod
+    def _use_billing_quota() -> bool:
+        return bool(dify_config.BILLING_ENABLED)
+
     @staticmethod
     def _get_tenant_lock_key(tenant_id: str) -> str:
         return f"credit_pool:tenant:{tenant_id}:deduct_lock"
@@ -79,14 +107,32 @@ class CreditPoolService:
         return credit_pool
 
     @classmethod
-    def get_pool(cls, tenant_id: str, pool_type: str = "trial") -> TenantCreditPool | None:
+    def get_pool(
+        cls, tenant_id: str, pool_type: str | ProviderQuotaType = "trial"
+    ) -> TenantCreditPool | CreditPoolBalance | None:
         """get tenant credit pool"""
+        normalized_pool_type = cls._normalize_pool_type(pool_type)
+        if cls._use_billing_quota():
+            from services.billing_service import BillingService
+
+            balance = BillingService.quota_get_balance(
+                tenant_id=tenant_id,
+                feature_key=FEATURE_KEY_CREDIT_POOL,
+                bucket=normalized_pool_type,
+            )
+            return CreditPoolBalance(
+                tenant_id=tenant_id,
+                pool_type=normalized_pool_type,
+                quota_limit=balance["quota"],
+                quota_used=balance["usage"],
+            )
+
         with session_factory.get_session_maker().begin() as session:
             return session.scalar(
                 select(TenantCreditPool)
                 .where(
                     TenantCreditPool.tenant_id == tenant_id,
-                    TenantCreditPool.pool_type == pool_type,
+                    TenantCreditPool.pool_type == normalized_pool_type,
                 )
                 .limit(1)
             )
@@ -102,7 +148,7 @@ class CreditPoolService:
         pool = cls.get_pool(tenant_id, pool_type)
         if not pool:
             return False
-        return pool.remaining_credits >= credits_required
+        return pool.has_sufficient_credits(credits_required)
 
     @classmethod
     def check_and_deduct_credits(
@@ -114,10 +160,54 @@ class CreditPoolService:
         """Deduct exactly the requested credits or raise without mutating the pool."""
         if credits_required <= 0:
             return 0
+        normalized_pool_type = cls._normalize_pool_type(pool_type)
+
+        if cls._use_billing_quota():
+            from services.billing_service import BillingService
+
+            request_id = str(uuid4())
+            result = BillingService.quota_reserve(
+                tenant_id=tenant_id,
+                feature_key=FEATURE_KEY_CREDIT_POOL,
+                bucket=normalized_pool_type,
+                request_id=request_id,
+                amount=credits_required,
+                meta={"source": "credit_pool.check_and_deduct"},
+            )
+            reservation_id = result.get("reservation_id", "")
+            if not reservation_id:
+                raise QuotaExceededError("Insufficient credits remaining")
+            try:
+                BillingService.quota_commit(
+                    tenant_id=tenant_id,
+                    feature_key=FEATURE_KEY_CREDIT_POOL,
+                    bucket=normalized_pool_type,
+                    reservation_id=reservation_id,
+                    actual_amount=credits_required,
+                    meta={"source": "credit_pool.check_and_deduct"},
+                )
+            except Exception:
+                try:
+                    BillingService.quota_release(
+                        tenant_id=tenant_id,
+                        feature_key=FEATURE_KEY_CREDIT_POOL,
+                        bucket=normalized_pool_type,
+                        reservation_id=reservation_id,
+                    )
+                except Exception:
+                    logger.warning(
+                        "Failed to release reserved credit pool quota, tenant_id=%s, pool_type=%s, reservation_id=%s",
+                        tenant_id,
+                        normalized_pool_type,
+                        reservation_id,
+                        exc_info=True,
+                    )
+                raise
+            return credits_required
 
         def deduct() -> int:
             with session_factory.get_session_maker().begin() as session:
-                pool = cls._get_locked_pool(session=session, tenant_id=tenant_id, pool_type=pool_type)
+                pool = cls._get_locked_pool(session=session, tenant_id=tenant_id, pool_type=normalized_pool_type)
                 if not pool:
                     raise QuotaExceededError("Credit pool not found")
 
@@ -148,12 +238,26 @@ class CreditPoolService:
         """Deduct up to the available balance and return the actual deducted credits."""
         if credits_required <= 0:
             return 0
+        normalized_pool_type = cls._normalize_pool_type(pool_type)
+
+        if cls._use_billing_quota():
+            from services.billing_service import BillingService
+
+            result = BillingService.quota_consume_capped(
+                tenant_id=tenant_id,
+                feature_key=FEATURE_KEY_CREDIT_POOL,
+                bucket=normalized_pool_type,
+                request_id=str(uuid4()),
+                amount=credits_required,
+                meta={"source": "credit_pool.deduct_capped"},
+            )
+            return result["deducted"]
 
         def deduct() -> int:
             with session_factory.get_session_maker().begin() as session:
-                pool = cls._get_locked_pool(session=session, tenant_id=tenant_id, pool_type=pool_type)
+                pool = cls._get_locked_pool(session=session, tenant_id=tenant_id, pool_type=normalized_pool_type)
                 if not pool:
-                    logger.warning("Credit pool not found, tenant_id=%s, pool_type=%s", tenant_id, pool_type)
+                    logger.warning("Credit pool not found, tenant_id=%s, pool_type=%s", tenant_id, normalized_pool_type)
                     return 0
 
                 deducted_credits = min(credits_required, pool.remaining_credits)

api/services/enterprise/rbac_service.py

@@ -12,6 +12,7 @@ from sqlalchemy.exc import SQLAlchemyError
 
 from configs import dify_config
 from core.db.session_factory import session_factory
+from core.rbac import RBACResourceWhitelistScope
 from models import TenantAccountJoin, TenantAccountRole
 from services.enterprise.base import EnterpriseRequest
 
@@ -254,6 +255,7 @@ class ResourceUserAccessPoliciesResponse(_RBACModel):
 
 class ReplaceUserAccessPolicies(_RBACModel):
     access_policy_ids: list[str] = Field(default_factory=list)
+    account_ids: list[str] = Field(default_factory=list)
 
     @field_validator("access_policy_ids", mode="before")
     @classmethod
@@ -364,7 +366,6 @@ _LEGACY_WORKSPACE_ADMIN_KEYS: list[str] = [
 ]
 
 _LEGACY_WORKSPACE_EDITOR_KEYS: list[str] = [
-    "workspace.member.manage",
     "api_extension.manage",
     "plugin.install",
     "credential.use",
@@ -435,6 +436,7 @@ _LEGACY_APP_EDITOR_KEYS: list[str] = [
     "app.acl.delete",
     "app.acl.release_and_version",
     "app.acl.monitor",
+    "app.acl.log_and_annotation",
     "app.acl.access_config",
 ]
 
@@ -649,17 +651,18 @@ class ReplaceRoleBindings(_RBACModel):
 
 
 class ReplaceMemberBindings(_RBACModel):
-    scope: str = "specific"
+    scope: RBACResourceWhitelistScope = RBACResourceWhitelistScope.SPECIFIC
 
     @field_validator("scope")
     @classmethod
-    def _normalize_scope(cls, value: Any) -> str:
+    def _normalize_scope(cls, value: Any) -> RBACResourceWhitelistScope:
         scope = str(value or "").strip().lower()
-        if scope in {"", "specific"}:
-            return "specific"
-        if scope in {"all", "only_me"}:
-            return scope
-        raise ValueError(f"invalid scope: {value}")
+        if scope == "":
+            return RBACResourceWhitelistScope.SPECIFIC
+        try:
+            return RBACResourceWhitelistScope(scope)
+        except ValueError as exc:
+            raise ValueError(f"invalid scope: {value}") from exc
 
 
 class DeleteMemberBindings(_RBACModel):
@@ -743,6 +746,7 @@ def _inner_call(
         account_id=account_id,
         json=json,
         params=params,
+        timeout=dify_config.ENTERPRISE_RBAC_REQUEST_TIMEOUT,
     )
 
 
@@ -1122,16 +1126,17 @@ class RBACService:
             tenant_id: str,
             account_id: str | None,
             app_id: str,
-            target_account_id: str,
+            target_account_id: str | None,
             payload: ReplaceUserAccessPolicies,
         ) -> ReplaceUserAccessPoliciesResponse:
+            request_data = payload.model_dump(mode="json")
             data = _inner_call(
                 "PUT",
                 f"{_INNER_PREFIX}/apps/user-access-policies",
                 tenant_id=tenant_id,
                 account_id=account_id,
                 params={"app_id": app_id, "account_id": target_account_id},
-                json=payload.model_dump(mode="json"),
+                json=request_data,
             )
             return ReplaceUserAccessPoliciesResponse.model_validate(data or {})
 
@@ -1291,7 +1296,7 @@ class RBACService:
             tenant_id: str,
             account_id: str | None,
             dataset_id: str,
-            target_account_id: str,
+            target_account_id: str | None,
             payload: ReplaceUserAccessPolicies,
         ) -> ReplaceUserAccessPoliciesResponse:
             data = _inner_call(
@@ -1300,7 +1305,7 @@ class RBACService:
                 tenant_id=tenant_id,
                 account_id=account_id,
                 params={"dataset_id": dataset_id, "account_id": target_account_id},
-                json=payload.model_dump(mode="json"),
+                json=payload.model_dump(mode="json", exclude_unset=True),
             )
             return ReplaceUserAccessPoliciesResponse.model_validate(data or {})
 

api/services/errors/account.py

@@ -37,6 +37,10 @@ class AccountAlreadyInTenantError(BaseServiceError):
     pass
 
 
+class SeatsLimitExceededError(BaseServiceError):
+    pass
+
+
 class InvalidActionError(BaseServiceError):
     pass
 

api/services/feature_service.py

@@ -79,6 +79,7 @@ class LicenseModel(FeatureResponseModel):
     status: LicenseStatus = LicenseStatus.NONE
     expired_at: str = ""
     workspaces: LicenseLimitationModel = LicenseLimitationModel(enabled=False, size=0, limit=0)
+    seats: LicenseLimitationModel = LicenseLimitationModel(enabled=False, size=0, limit=0)
 
 
 class BrandingModel(FeatureResponseModel):
@@ -457,6 +458,11 @@ class FeatureService:
                     features.license.workspaces.limit = workspaces_info.get("limit", 0)
                     features.license.workspaces.size = workspaces_info.get("used", 0)
 
+                if seats_info := license_info.get("licensedSeats"):
+                    features.license.seats.enabled = seats_info.get("enabled", False)
+                    features.license.seats.limit = seats_info.get("limit", 0)
+                    features.license.seats.size = seats_info.get("used", 0)
+
         if "PluginInstallationPermission" in enterprise_info:
             plugin_installation_info = enterprise_info["PluginInstallationPermission"]
             features.plugin_installation_permission.plugin_installation_scope = plugin_installation_info[

api/services/tools/api_tools_manage_service.py

@@ -2,11 +2,11 @@ import json
 import logging
 from typing import Any, TypedDict, cast
 
-from httpx import get
 from sqlalchemy import select
 from sqlalchemy.orm import sessionmaker
 
 from core.entities.provider_entities import ProviderConfig
+from core.helper import ssrf_proxy
 from core.tools.__base.tool_runtime import ToolRuntime
 from core.tools.custom_tool.provider import ApiToolProviderController
 from core.tools.entities.api_entities import ToolApiEntity, ToolProviderApiEntity
@@ -212,7 +212,7 @@ class ApiToolManageService:
         }
 
         try:
-            response = get(url, headers=headers, timeout=10)
+            response = ssrf_proxy.get(url, headers=headers, timeout=10)
             if response.status_code != 200:
                 raise ValueError(f"Got status code {response.status_code}")
             schema = response.text

api/tasks/initialize_created_app_rbac_access_task.py

@@ -0,0 +1,67 @@
+"""Initialize default RBAC access for existing workspace members after app creation."""
+
+import logging
+
+from celery import shared_task
+
+from configs import dify_config
+from extensions.ext_database import db
+from services.account_service import TenantService
+from services.enterprise import rbac_service as enterprise_rbac_service
+
+logger = logging.getLogger(__name__)
+
+APP_RBAC_ACCOUNT_POLICY_BATCH_SIZE = 500
+APP_RBAC_DEFAULT_ACCESS_POLICY_ID = "default"
+APP_RBAC_QUEUE = "app_rbac"
+
+
+@shared_task(queue=APP_RBAC_QUEUE, bind=True, max_retries=3, default_retry_delay=60)
+def initialize_created_app_rbac_access_task(
+    self, tenant_id: str, account_id: str, app_id: str | None = None, dataset_id: str | None = None
+) -> None:
+    """Grant the default app policy to current workspace members.
+
+    App scope is persisted synchronously before this task is queued. Replacing
+    member policies is idempotent, so retrying the whole synchronization is safe
+    when the enterprise RBAC service is temporarily unavailable.
+    """
+    if not dify_config.RBAC_ENABLED:
+        return
+
+    try:
+        for account_ids in TenantService.iter_member_account_id_batches(
+            tenant_id,
+            APP_RBAC_ACCOUNT_POLICY_BATCH_SIZE,
+            session=db.session(),
+        ):
+            if app_id is not None:
+                enterprise_rbac_service.RBACService.AppAccess.replace_user_access_policies(
+                    tenant_id=tenant_id,
+                    account_id=account_id,
+                    app_id=app_id,
+                    target_account_id=None,
+                    payload=enterprise_rbac_service.ReplaceUserAccessPolicies(
+                        access_policy_ids=[APP_RBAC_DEFAULT_ACCESS_POLICY_ID],
+                        account_ids=account_ids,
+                    ),
+                )
+            elif dataset_id is not None:
+                enterprise_rbac_service.RBACService.DatasetAccess.replace_user_access_policies(
+                    tenant_id=tenant_id,
+                    account_id=account_id,
+                    dataset_id=dataset_id,
+                    target_account_id=None,
+                    payload=enterprise_rbac_service.ReplaceUserAccessPolicies(
+                        access_policy_ids=[APP_RBAC_DEFAULT_ACCESS_POLICY_ID],
+                        account_ids=account_ids,
+                    ),
+                )
+    except Exception as exc:
+        logger.exception(
+            "Failed to initialize app RBAC access; retrying: tenant_id=%s app_id=%s attempt=%s",
+            tenant_id,
+            app_id,
+            self.request.retries + 1,
+        )
+        raise self.retry(exc=exc)

api/tasks/remove_app_and_related_data_task.py

@@ -701,16 +701,29 @@ def _delete_records(query_sql: str, params: dict[str, Any], delete_func: Callabl
             if not rows:
                 break
 
+            success_count = 0
             for i in rows:
                 record_id = str(i.id)
                 try:
                     delete_func(session, record_id)
                     logger.info(click.style(f"Deleted {name} {record_id}", fg="green"))
+                    session.commit()
+                    success_count += 1
                 except Exception:
                     logger.exception("Error occurred while deleting %s %s", name, record_id)
                     # continue with next record even if one deletion fails
                     session.rollback()
-                    break
-                session.commit()
+                    continue
 
             rs.close()
+
+            # If we couldn't delete ANY records in this batch, we must break out of the while loop
+            # to prevent an infinite loop where we keep fetching the same failing records.
+            if success_count == 0:
+                logger.warning(
+                    click.style(
+                        f"Failed to delete any {name} in the current batch. Stopping to prevent infinite loop.",
+                        fg="yellow",
+                    )
+                )
+                break

api/tests/test_containers_integration_tests/controllers/console/auth/test_oauth.py

@@ -105,6 +105,7 @@ class TestOAuthLogin:
             invite_token=expected_token,
             timezone=None,
             language=None,
+            redirect_url=None,
         )
         mock_redirect.assert_called_once_with("https://github.com/login/oauth/authorize?...")
 
@@ -127,6 +128,7 @@ class TestOAuthLogin:
             invite_token=None,
             timezone="Asia/Shanghai",
             language=None,
+            redirect_url=None,
         )
         mock_redirect.assert_called_once_with("https://github.com/login/oauth/authorize?...")
 
@@ -149,6 +151,7 @@ class TestOAuthLogin:
             invite_token=None,
             timezone=None,
             language="zh-Hans",
+            redirect_url=None,
         )
         mock_redirect.assert_called_once_with("https://github.com/login/oauth/authorize?...")
 

api/tests/test_containers_integration_tests/services/test_account_service.py

@@ -17,6 +17,7 @@ from services.errors.account import (
     AccountPasswordError,
     AccountRegisterError,
     CurrentPasswordIncorrectError,
+    SeatsLimitExceededError,
     TenantNotFoundError,
 )
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
@@ -477,6 +478,32 @@ class TestAccountService:
                 session=db_session_with_containers,
             )
 
+    def test_create_account_seats_limit_exceeded(
+        self, db_session_with_containers: Session, mock_external_service_dependencies
+    ):
+        """
+        Test account creation when the licensed seats limit is exceeded.
+        """
+        fake = Faker()
+        email = fake.email()
+        name = fake.name()
+        password = generate_valid_password(fake)
+        # Setup mocks
+        mock_external_service_dependencies["feature_service"].get_system_features.return_value.is_allow_register = True
+        mock_external_service_dependencies[
+            "feature_service"
+        ].get_system_features.return_value.license.seats.is_available.return_value = False
+        mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
+
+        with pytest.raises(SeatsLimitExceededError):
+            AccountService.create_account(
+                email=email,
+                name=name,
+                interface_language="en-US",
+                password=password,
+                session=db_session_with_containers,
+            )
+
     def test_link_account_integrate_new_provider(
         self, db_session_with_containers: Session, mock_external_service_dependencies
     ):

api/tests/unit_tests/commands/test_data_migration_wizard.py

@@ -377,8 +377,8 @@ def test_import_options_prompts_explain_secrets_reuse_and_conflicts(monkeypatch)
     assert prompt_calls[0][0] == "Import ID strategy. Enter one of: preserve-id, generate-new-id"
     assert prompt_calls[0][1]["default"] == "preserve-id"
     assert prompt_calls[0][1]["show_default"] is True
-    assert prompt_calls[0][1]["type"].choices == ID_STRATEGY_CHOICES
+    assert list(prompt_calls[0][1]["type"].choices) == ID_STRATEGY_CHOICES
     assert prompt_calls[1][0] == "Import conflict strategy. Enter one of: fail, skip, update"
     assert prompt_calls[1][1]["default"] == "update"
     assert prompt_calls[1][1]["show_default"] is True
-    assert prompt_calls[1][1]["type"].choices == CONFLICT_STRATEGY_CHOICES
+    assert list(prompt_calls[1][1]["type"].choices) == CONFLICT_STRATEGY_CHOICES

api/tests/unit_tests/commands/test_legacy_model_type_migration.py

@@ -336,6 +336,174 @@ def _insert_load_balancing_model_config(
         )
 
 
+def test_data_migrate_group_registers_dataset_permission_rbac_migration(command_module) -> None:
+    command = command_module.data_migrate.commands["rbac-migrate-dataset-permissions"]
+
+    assert command is command_module.migrate_dataset_permissions_to_rbac
+    assert "operator_account_id" not in {param.name for param in command.params}
+
+
+def test_dataset_permission_rbac_migration_help_mentions_binding_clear_side_effect(command_module) -> None:
+    result = CliRunner().invoke(
+        command_module.data_migrate,
+        ["rbac-migrate-dataset-permissions", "--help"],
+    )
+
+    assert result.exit_code == 0
+    normalized_output = " ".join(result.output.split())
+    assert "clears existing per-user policy bindings" in normalized_output
+    assert "recreates legacy partial-member default bindings" in normalized_output
+
+
+def test_dataset_permission_rbac_migration_maps_legacy_permissions_to_enum_scopes() -> None:
+    rbac_module = importlib.import_module("commands.rbac")
+
+    assert (
+        rbac_module._rbac_dataset_scope_for_legacy_permission(rbac_module.DatasetPermissionEnum.ALL_TEAM)
+        is rbac_module.RBACResourceWhitelistScope.ALL
+    )
+    assert (
+        rbac_module._rbac_dataset_scope_for_legacy_permission(rbac_module.DatasetPermissionEnum.PARTIAL_TEAM)
+        is rbac_module.RBACResourceWhitelistScope.SPECIFIC
+    )
+    assert rbac_module._dataset_permission_enum("partial_members") is rbac_module.DatasetPermissionEnum.PARTIAL_TEAM
+    assert rbac_module._dataset_permission_enum(None) is rbac_module.DatasetPermissionEnum.ONLY_ME
+
+
+def test_dataset_permission_rbac_migration_uses_dataset_creator_as_operator(
+    command_module,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    rbac_module = importlib.import_module("commands.rbac")
+    dataset_row = SimpleNamespace(
+        id="dataset-1",
+        tenant_id="tenant-1",
+        permission="only_me",
+        created_by="creator-account-1",
+    )
+    execute_results = [[dataset_row], [], []]
+    calls: list[dict[str, object]] = []
+    session_closed = False
+
+    class FakeExecuteResult:
+        def __init__(self, rows: list[object]) -> None:
+            self._rows = rows
+
+        def all(self) -> list[object]:
+            return self._rows
+
+    class FakeSession:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, traceback) -> None:
+            nonlocal session_closed
+            session_closed = True
+            pass
+
+        def execute(self, stmt):
+            return FakeExecuteResult(execute_results.pop(0))
+
+    class FakeSessionFactory:
+        @staticmethod
+        def create_session() -> FakeSession:
+            return FakeSession()
+
+    def fake_replace_whitelist(**kwargs):
+        assert session_closed is True
+        calls.append(kwargs)
+
+    monkeypatch.setattr(rbac_module, "session_factory", FakeSessionFactory)
+    monkeypatch.setattr(rbac_module.RBACService.DatasetAccess, "replace_whitelist", fake_replace_whitelist)
+
+    command_module.migrate_dataset_permissions_to_rbac.callback(
+        tenant_id=None,
+        dataset_id=None,
+        batch_size=500,
+        dry_run=False,
+    )
+
+    assert calls[0]["tenant_id"] == "tenant-1"
+    assert calls[0]["account_id"] == "creator-account-1"
+    assert calls[0]["dataset_id"] == "dataset-1"
+    assert calls[0]["payload"].scope is rbac_module.RBACResourceWhitelistScope.SPECIFIC
+
+
+def test_dataset_permission_rbac_migration_dry_run_outputs_structured_proposed_changes(
+    command_module,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    rbac_module = importlib.import_module("commands.rbac")
+    dataset_row = SimpleNamespace(
+        id="dataset-1",
+        tenant_id="tenant-1",
+        permission="partial_members",
+        created_by="creator-account-1",
+    )
+    permission_row = SimpleNamespace(dataset_id="dataset-1", account_id="member-account-1")
+    execute_results = [[dataset_row], [permission_row], []]
+
+    class FakeExecuteResult:
+        def __init__(self, rows: list[object]) -> None:
+            self._rows = rows
+
+        def all(self) -> list[object]:
+            return self._rows
+
+    class FakeSession:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, traceback) -> None:
+            pass
+
+        def execute(self, stmt):
+            return FakeExecuteResult(execute_results.pop(0))
+
+    class FakeSessionFactory:
+        @staticmethod
+        def create_session() -> FakeSession:
+            return FakeSession()
+
+    monkeypatch.setattr(rbac_module, "session_factory", FakeSessionFactory)
+    monkeypatch.setattr(
+        rbac_module.RBACService.DatasetAccess,
+        "replace_whitelist",
+        lambda **kwargs: pytest.fail("dry-run must not replace whitelist"),
+    )
+    monkeypatch.setattr(
+        rbac_module.RBACService.DatasetAccess,
+        "replace_user_access_policies",
+        lambda **kwargs: pytest.fail("dry-run must not replace user access policies"),
+    )
+
+    result = CliRunner().invoke(
+        command_module.data_migrate,
+        ["rbac-migrate-dataset-permissions", "--dry-run"],
+    )
+
+    assert result.exit_code == 0
+    events = [json.loads(line) for line in result.output.splitlines() if line.startswith("{")]
+    assert [event["action"] for event in events] == ["replace_whitelist", "replace_user_access_policies"]
+    assert events[0]["before"] == {
+        "legacy_dataset_permission": "partial_members",
+        "legacy_partial_member_ids": ["member-account-1"],
+    }
+    assert events[0]["after"] == {"rbac_whitelist_scope": "specific"}
+    assert events[0]["call"] == {
+        "method": "RBACService.DatasetAccess.replace_whitelist",
+        "kwargs": {
+            "tenant_id": "tenant-1",
+            "account_id": "creator-account-1",
+            "dataset_id": "dataset-1",
+            "payload": {"scope": "specific"},
+        },
+    }
+    assert events[1]["target_account_id"] == "member-account-1"
+    assert events[1]["after"] == {"rbac_user_access_policy_ids": ["default"]}
+    assert events[1]["call"]["kwargs"]["payload"] == {"access_policy_ids": ["default"]}
+
+
 def test_data_migrate_command_defaults_output_to_stdout_stream(
     command_module,
     monkeypatch: pytest.MonkeyPatch,

api/tests/unit_tests/controllers/console/app/test_app_response_models.py

@@ -543,6 +543,7 @@ def test_app_create_api_attaches_permission_keys(app, app_module):
 
     with app.test_request_context("/apps", method="POST", json={}):
         with pytest.MonkeyPatch.context() as monkeypatch:
+            monkeypatch.setattr(app_module.dify_config, "RBAC_ENABLED", True)
             app_module.console_ns.payload = {
                 "name": "Created App",
                 "description": "Summary",
@@ -558,11 +559,25 @@ def test_app_create_api_attaches_permission_keys(app, app_module):
                 "batch_get",
                 lambda tenant_id, account_id, app_ids: {"app-new": ["app.acl.view_layout", "app.acl.edit"]},
             )
+            initialize_rbac_task = MagicMock()
+            monkeypatch.setattr(
+                app_module,
+                "initialize_created_app_rbac_access_task",
+                initialize_rbac_task,
+            )
+            replace_whitelist = MagicMock()
+            monkeypatch.setattr(
+                app_module.enterprise_rbac_service.RBACService.AppAccess,
+                "replace_whitelist",
+                replace_whitelist,
+            )
 
             resp, status = method(app_module.AppListApi(), "tenant-1", SimpleNamespace(id="acct-1"))
 
     assert status == 201
     assert resp["permission_keys"] == ["app.acl.view_layout", "app.acl.edit"]
+    assert replace_whitelist.call_args.kwargs["payload"].scope is app_module.RBACResourceWhitelistScope.ALL
+    initialize_rbac_task.delay.assert_called_once_with("tenant-1", "acct-1", app_id="app-new")
 
 
 def test_app_list_api_attaches_permission_keys(app, app_module):

api/tests/unit_tests/controllers/console/app/test_ops_trace_api.py

@@ -0,0 +1,154 @@
+from __future__ import annotations
+
+from contextlib import nullcontext
+from types import SimpleNamespace
+from unittest.mock import MagicMock, PropertyMock, patch
+
+import pytest
+from flask import Flask
+from werkzeug.exceptions import Forbidden
+
+from controllers.common import wraps as common_wraps
+from controllers.console import console_ns
+from controllers.console import wraps as console_wraps
+from controllers.console.app import ops_trace as ops_trace_module
+from controllers.console.app import wraps as app_wraps
+from libs import login as login_lib
+from models.account import Account, AccountStatus, TenantAccountRole
+
+
+def _make_account(role: TenantAccountRole) -> Account:
+    account = Account(name="tester", email="tester@example.com")
+    account.id = "account-123"  # type: ignore[assignment]
+    account.status = AccountStatus.ACTIVE
+    account.role = role
+    account._current_tenant = SimpleNamespace(id="tenant-123")  # type: ignore[assignment]
+    account._get_current_object = lambda: account  # type: ignore[attr-defined]
+    return account
+
+
+def _make_app() -> SimpleNamespace:
+    return SimpleNamespace(id="app-123", tenant_id="tenant-123", status="normal", mode="chat")
+
+
+def _patch_console_guards(
+    monkeypatch: pytest.MonkeyPatch,
+    account: Account,
+    app_model: SimpleNamespace,
+    *,
+    rbac_enabled: bool = False,
+) -> None:
+    monkeypatch.setattr(login_lib.dify_config, "LOGIN_DISABLED", True)
+    monkeypatch.setattr(login_lib.dify_config, "RBAC_ENABLED", rbac_enabled)
+    monkeypatch.setattr(console_wraps.dify_config, "EDITION", "CLOUD")
+    monkeypatch.setattr(login_lib, "current_user", account)
+    monkeypatch.setattr(login_lib, "current_account_with_tenant", lambda: (account, account.current_tenant_id))
+    monkeypatch.setattr(console_wraps, "current_account_with_tenant", lambda: (account, account.current_tenant_id))
+    monkeypatch.setattr(common_wraps, "current_account_with_tenant", lambda: (account, account.current_tenant_id))
+    monkeypatch.setattr(app_wraps, "_load_app_model_from_scoped_session", lambda _app_id: app_model)
+
+
+def _patch_payload(payload: dict[str, object] | None):
+    if payload is None:
+        return nullcontext()
+    return patch.object(type(console_ns), "payload", new_callable=PropertyMock, return_value=payload)
+
+
+@pytest.mark.parametrize(
+    ("method_name", "path", "payload", "service_method_name", "service_result"),
+    [
+        (
+            "post",
+            "/console/api/apps/app-123/trace-config",
+            {"tracing_provider": "mlflow", "tracing_config": {"endpoint": "https://trace.example.com"}},
+            "create_tracing_app_config",
+            {"id": "trace-config-1"},
+        ),
+        (
+            "patch",
+            "/console/api/apps/app-123/trace-config",
+            {"tracing_provider": "mlflow", "tracing_config": {"endpoint": "https://trace.example.com"}},
+            "update_tracing_app_config",
+            True,
+        ),
+        (
+            "delete",
+            "/console/api/apps/app-123/trace-config?tracing_provider=mlflow",
+            None,
+            "delete_tracing_app_config",
+            True,
+        ),
+    ],
+)
+def test_trace_config_mutations_require_edit_permission(
+    app: Flask,
+    monkeypatch: pytest.MonkeyPatch,
+    method_name: str,
+    path: str,
+    payload: dict[str, object] | None,
+    service_method_name: str,
+    service_result: object,
+) -> None:
+    app.config.setdefault("RESTX_MASK_HEADER", "X-Fields")
+    account = _make_account(TenantAccountRole.NORMAL)
+    _patch_console_guards(monkeypatch, account, _make_app())
+    service_mock = MagicMock(return_value=service_result)
+    monkeypatch.setattr(ops_trace_module.OpsService, service_method_name, service_mock)
+
+    with app.test_request_context(path, method=method_name.upper(), json=payload):
+        with _patch_payload(payload):
+            with pytest.raises(Forbidden):
+                getattr(ops_trace_module.TraceAppConfigApi(), method_name)(app_id="app-123")
+
+    service_mock.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("method_name", "path", "payload", "service_method_name", "service_result"),
+    [
+        (
+            "post",
+            "/console/api/apps/app-123/trace-config",
+            {"tracing_provider": "mlflow", "tracing_config": {"endpoint": "https://trace.example.com"}},
+            "create_tracing_app_config",
+            {"id": "trace-config-1"},
+        ),
+        (
+            "patch",
+            "/console/api/apps/app-123/trace-config",
+            {"tracing_provider": "mlflow", "tracing_config": {"endpoint": "https://trace.example.com"}},
+            "update_tracing_app_config",
+            True,
+        ),
+        (
+            "delete",
+            "/console/api/apps/app-123/trace-config?tracing_provider=mlflow",
+            None,
+            "delete_tracing_app_config",
+            True,
+        ),
+    ],
+)
+def test_trace_config_mutations_require_rbac_permission(
+    app: Flask,
+    monkeypatch: pytest.MonkeyPatch,
+    method_name: str,
+    path: str,
+    payload: dict[str, object] | None,
+    service_method_name: str,
+    service_result: object,
+) -> None:
+    app.config.setdefault("RESTX_MASK_HEADER", "X-Fields")
+    account = _make_account(TenantAccountRole.NORMAL)
+    _patch_console_guards(monkeypatch, account, _make_app(), rbac_enabled=True)
+    monkeypatch.setattr(common_wraps.db, "session", SimpleNamespace(scalar=lambda _stmt: "other-account"))
+    monkeypatch.setattr(common_wraps.RBACService.CheckAccess, "check", MagicMock(return_value=False))
+    service_mock = MagicMock(return_value=service_result)
+    monkeypatch.setattr(ops_trace_module.OpsService, service_method_name, service_mock)
+
+    with app.test_request_context(path, method=method_name.upper(), json=payload):
+        with _patch_payload(payload):
+            with pytest.raises(Forbidden):
+                getattr(ops_trace_module.TraceAppConfigApi(), method_name)(app_id="app-123")
+
+    service_mock.assert_not_called()

api/tests/unit_tests/controllers/console/app/test_wraps.py

@@ -3,10 +3,11 @@ from __future__ import annotations
 from types import SimpleNamespace
 
 import pytest
+from sqlalchemy import Select
 
 from controllers.console.app import wraps as wraps_module
 from controllers.console.app.error import AppNotFoundError
-from models.model import AppMode
+from models.model import App, AppMode, TrialApp
 
 
 class FakeSession:
@@ -117,6 +118,25 @@ def test_get_app_model_rejects_wrong_mode(monkeypatch: pytest.MonkeyPatch) -> No
         handler(app_id="app-1")
 
 
+def test_get_app_model_with_trial_requires_trial_app_registration(monkeypatch: pytest.MonkeyPatch) -> None:
+    app_model = SimpleNamespace(id="app-1", mode=AppMode.CHAT.value, status="normal", tenant_id="t1")
+
+    def scalar(statement: Select[tuple[App]]) -> object | None:
+        has_trial_app_join = any(
+            from_clause.is_derived_from(TrialApp.__table__) for from_clause in statement.get_final_froms()
+        )
+        return None if has_trial_app_join else app_model
+
+    monkeypatch.setattr(wraps_module.db, "session", SimpleNamespace(scalar=scalar))
+
+    @wraps_module.get_app_model_with_trial
+    def handler(app_model):
+        return app_model.id
+
+    with pytest.raises(AppNotFoundError):
+        handler(app_id="app-1")
+
+
 def test_get_app_model_requires_app_id() -> None:
     @wraps_module.get_app_model
     def handler(app_model):

api/tests/unit_tests/controllers/console/auth/test_login_logout.py

@@ -26,10 +26,11 @@ from controllers.console.auth.login import EmailCodeLoginApi, LoginApi, LogoutAp
 from controllers.console.error import (
     AccountBannedError,
     AccountInFreezeError,
+    SeatsLimitExceeded,
     WorkspacesLimitExceeded,
 )
 from services.entities.auth_entities import LoginFailureReason
-from services.errors.account import AccountLoginError, AccountPasswordError
+from services.errors.account import AccountLoginError, AccountPasswordError, SeatsLimitExceededError
 
 
 def encode_password(password: str) -> str:
@@ -487,6 +488,45 @@ class TestLoginApi:
         assert warn_records[0].args[0] == "user@example.com"
         assert warn_records[0].args[1] == LoginFailureReason.ACCOUNT_BANNED
 
+    @patch("controllers.console.wraps.db")
+    @patch("controllers.console.auth.login.db")
+    @patch("controllers.console.auth.login.AccountService.create_account_and_tenant")
+    @patch("controllers.console.auth.login.AccountService.get_email_code_login_data")
+    @patch("controllers.console.auth.login.AccountService.revoke_email_code_login_token")
+    @patch("controllers.console.auth.login._get_account_with_case_fallback")
+    def test_email_code_login_fails_when_seats_limit_exceeded(
+        self,
+        mock_get_account: MagicMock,
+        mock_revoke_token: MagicMock,
+        mock_get_token_data: MagicMock,
+        mock_create_account: MagicMock,
+        mock_login_db: MagicMock,
+        mock_db: MagicMock,
+        app: Flask,
+    ):
+        """
+        Test email-code login failure when creating the account would exceed the licensed seats.
+
+        Verifies that:
+        - the new-account path is taken when no account exists for the email
+        - the service-layer SeatsLimitExceededError is translated to the SeatsLimitExceeded HTTP error
+        """
+        # Arrange: valid token, no existing account -> account-creation path
+        mock_get_token_data.return_value = {"email": "User@Example.com", "code": "123456"}
+        mock_get_account.return_value = None
+        mock_create_account.side_effect = SeatsLimitExceededError("licensed seats limit exceeded")
+
+        # Act & Assert
+        with app.test_request_context(
+            "/email-code-login/validity",
+            method="POST",
+            json={"email": "User@Example.com", "code": encode_code("123456"), "token": "token-123"},
+        ):
+            with pytest.raises(SeatsLimitExceeded):
+                EmailCodeLoginApi().post()
+
+        mock_create_account.assert_called_once()
+
 
 class TestLogoutApi:
     """Test cases for the LogoutApi endpoint."""

api/tests/unit_tests/controllers/console/auth/test_oauth_redirect.py

@@ -0,0 +1,93 @@
+import urllib.parse
+from unittest.mock import MagicMock, patch
+
+import pytest
+from flask import Flask
+
+from controllers.console.auth.oauth import OAuthCallback, OAuthLogin
+from libs.oauth import OAuthUserInfo, encode_oauth_state
+from models.account import AccountStatus
+
+REDIRECT_URL = "/apps?category=workflow"
+CONSOLE_WEB_URL = "https://console.example.com"
+
+
+@pytest.fixture
+def app() -> Flask:
+    app = Flask(__name__)
+    app.config["TESTING"] = True
+    return app
+
+
+def test_oauth_login_passes_relative_redirect_url_through(app: Flask) -> None:
+    oauth_provider = MagicMock()
+    oauth_provider.get_authorization_url.return_value = "https://accounts.google.com/o/oauth2/v2/auth?state=..."
+    query = urllib.parse.urlencode({"redirect_url": REDIRECT_URL})
+
+    with (
+        patch("controllers.console.auth.oauth.get_oauth_providers", return_value={"google": oauth_provider}),
+        app.test_request_context(f"/oauth/login/google?{query}"),
+    ):
+        response = OAuthLogin().get("google")
+
+    oauth_provider.get_authorization_url.assert_called_once_with(
+        invite_token=None,
+        timezone=None,
+        language=None,
+        redirect_url=REDIRECT_URL,
+    )
+    assert response.status_code == 302
+    assert response.headers["Location"] == "https://accounts.google.com/o/oauth2/v2/auth?state=..."
+
+
+@pytest.mark.parametrize(
+    ("redirect_url", "expected_target_url"),
+    [
+        (REDIRECT_URL, REDIRECT_URL),
+        (f"{CONSOLE_WEB_URL}{REDIRECT_URL}", f"{CONSOLE_WEB_URL}{REDIRECT_URL}"),
+        ("https://console.example.com.malicious.example/apps", CONSOLE_WEB_URL),
+        ("//malicious.example.com/apps", CONSOLE_WEB_URL),
+        ("///malicious.example.com/apps", CONSOLE_WEB_URL),
+        (r"\\malicious.example.com/apps", CONSOLE_WEB_URL),
+    ],
+)
+@pytest.mark.parametrize("oauth_new_user", [False, True])
+def test_oauth_callback_validates_redirect_url_and_appends_new_user_flag(
+    app: Flask,
+    redirect_url: str,
+    expected_target_url: str,
+    oauth_new_user: bool,
+) -> None:
+    oauth_provider = MagicMock()
+    oauth_provider.get_access_token.return_value = "google-access-token"
+    oauth_provider.get_user_info.return_value = OAuthUserInfo(
+        id="google-user-123",
+        name="Test User",
+        email="test@example.com",
+    )
+    account = MagicMock()
+    account.status = AccountStatus.ACTIVE
+    token_pair = MagicMock()
+    token_pair.access_token = "dify-access-token"
+    token_pair.refresh_token = "dify-refresh-token"
+    token_pair.csrf_token = "dify-csrf-token"
+    state = encode_oauth_state(redirect_url=redirect_url)
+
+    with (
+        patch("controllers.console.auth.oauth.get_oauth_providers", return_value={"google": oauth_provider}),
+        patch("controllers.console.auth.oauth.dify_config.CONSOLE_WEB_URL", CONSOLE_WEB_URL),
+        patch("controllers.console.auth.oauth._generate_account", return_value=(account, oauth_new_user)),
+        patch("controllers.console.auth.oauth.TenantService.create_owner_tenant_if_not_exist"),
+        patch("controllers.console.auth.oauth.AccountService.login", return_value=token_pair),
+        patch("controllers.console.auth.oauth.set_access_token_to_cookie"),
+        patch("controllers.console.auth.oauth.set_refresh_token_to_cookie"),
+        patch("controllers.console.auth.oauth.set_csrf_token_to_cookie"),
+        app.test_request_context(f"/oauth/authorize/google?code=test-code&state={state}"),
+    ):
+        response = OAuthCallback().get("google")
+
+    assert response.status_code == 302
+    query_char = "&" if "?" in expected_target_url else "?"
+    assert response.headers["Location"] == (
+        f"{expected_target_url}{query_char}oauth_new_user={str(oauth_new_user).lower()}"
+    )

api/tests/unit_tests/controllers/console/auth/test_oauth_timezone.py

@@ -33,6 +33,7 @@ def test_oauth_login_passes_language_and_timezone_to_authorization_url(
         invite_token=None,
         timezone="Asia/Shanghai",
         language="zh-Hans",
+        redirect_url=None,
     )
     mock_redirect.assert_called_once_with("https://github.com/login/oauth/authorize?state=...")
 

api/tests/unit_tests/controllers/console/workspace/test_rbac.py

@@ -136,7 +136,7 @@ class TestPydanticModels:
 
     def test_resource_access_scope_defaults_empty_account_ids(self):
         parsed = rbac_mod._ResourceAccessScopeRequest.model_validate({"scope": "specific"})
-        assert parsed.scope is rbac_mod._AccessScope.SPECIFIC
+        assert parsed.scope is rbac_mod.RBACResourceWhitelistScope.SPECIFIC
 
     def test_resource_access_scope_coerce_null_account_ids(self):
         rbac_mod._ResourceAccessScopeRequest.model_validate({"scope": "all"})
@@ -262,6 +262,25 @@ class TestPaginationMapping:
 
 
 class TestResourceAccessScopeBindings:
+    def test_app_whitelist_all_schedules_member_policy_sync(self, app):
+        with (
+            app.test_request_context(
+                "/workspaces/current/rbac/apps/app-1/whitelist",
+                method="PUT",
+                json={"scope": "all"},
+            ),
+            patch("controllers.console.workspace.rbac._current_ids", return_value=("tenant-1", "acct-actor")),
+            patch("controllers.console.workspace.rbac.dify_config.RBAC_ENABLED", True),
+            patch(
+                "controllers.console.workspace.rbac.svc.RBACService.AppAccess.replace_whitelist",
+                return_value=rbac_mod.svc.ResourceWhitelist(),
+            ),
+            patch("controllers.console.workspace.rbac.initialize_created_app_rbac_access_task") as mock_sync_task,
+        ):
+            inspect.unwrap(rbac_mod.RBACAppWhitelistApi.put)(rbac_mod.RBACAppWhitelistApi(), "app-1")
+
+        mock_sync_task.delay.assert_called_once_with("tenant-1", "acct-actor", "app-1")
+
     def test_app_user_access_policy_assignment_forwards_ids(self, app):
         with (
             app.test_request_context(

api/tests/unit_tests/controllers/service_api/test_wraps.py

@@ -3,7 +3,7 @@ Unit tests for Service API wraps (authentication decorators)
 """
 
 import uuid
-from unittest.mock import Mock, patch
+from unittest.mock import MagicMock, Mock, patch
 
 import pytest
 from flask import Flask
@@ -469,7 +469,10 @@ class TestCloudEditionBillingRateLimitCheck:
     @patch("controllers.service_api.wraps.validate_and_get_api_token")
     @patch("controllers.service_api.wraps.FeatureService.get_knowledge_rate_limit")
     @patch("controllers.service_api.wraps.db")
-    def test_rejects_over_rate_limit(self, mock_db, mock_get_rate_limit, mock_validate_token, app: Flask):
+    @patch("controllers.service_api.wraps.sessionmaker")
+    def test_rejects_over_rate_limit(
+        self, mock_sessionmaker, mock_db, mock_get_rate_limit, mock_validate_token, app: Flask
+    ):
         """Test that Forbidden is raised when over rate limit."""
         # Arrange
         mock_validate_token.return_value = Mock(tenant_id="tenant123")
@@ -479,6 +482,10 @@ class TestCloudEditionBillingRateLimitCheck:
         mock_rate_limit.limit = 10
         mock_rate_limit.subscription_plan = "pro"
         mock_get_rate_limit.return_value = mock_rate_limit
+        rate_limit_log_session = MagicMock()
+        session_factory = MagicMock()
+        session_factory.begin.return_value.__enter__.return_value = rate_limit_log_session
+        mock_sessionmaker.return_value = session_factory
 
         with patch("controllers.service_api.wraps.redis_client") as mock_redis:
             mock_redis.zcard.return_value = 15  # Over limit
@@ -492,6 +499,9 @@ class TestCloudEditionBillingRateLimitCheck:
                 with pytest.raises(Forbidden) as exc_info:
                     knowledge_request()
                 assert "rate limit" in str(exc_info.value)
+            mock_sessionmaker.assert_called_once_with(bind=mock_db.engine, expire_on_commit=False)
+            rate_limit_log_session.add.assert_called_once()
+            mock_db.session.commit.assert_not_called()
 
 
 class TestValidateDatasetToken:

api/tests/unit_tests/core/callback_handler/test_workflow_tool_callback_handler.py

@@ -32,8 +32,16 @@ def mock_print_text(mocker: MockerFixture):
     return mocker.patch("core.callback_handler.workflow_tool_callback_handler.print_text")
 
 
+@pytest.fixture
+def enable_debug(mocker: MockerFixture):
+    """Force DEBUG on so the handler emits its verbose stdout traces."""
+    mocker.patch("core.callback_handler.workflow_tool_callback_handler.dify_config.DEBUG", True)
+
+
 class TestDifyWorkflowCallbackHandler:
-    def test_on_tool_execution_single_output_success(self, handler: DifyWorkflowCallbackHandler, mock_print_text):
+    def test_on_tool_execution_single_output_success(
+        self, handler: DifyWorkflowCallbackHandler, mock_print_text, enable_debug
+    ):
         # Arrange
         tool_name = "test_tool"
         tool_inputs = {"a": 1}
@@ -63,7 +71,9 @@ class TestDifyWorkflowCallbackHandler:
             ]
         )
 
-    def test_on_tool_execution_multiple_outputs(self, handler: DifyWorkflowCallbackHandler, mock_print_text):
+    def test_on_tool_execution_multiple_outputs(
+        self, handler: DifyWorkflowCallbackHandler, mock_print_text, enable_debug
+    ):
         # Arrange
         tool_name = "multi_tool"
         outputs = [
@@ -101,6 +111,29 @@ class TestDifyWorkflowCallbackHandler:
         assert results == []
         mock_print_text.assert_not_called()
 
+    def test_on_tool_execution_skips_print_when_debug_disabled(
+        self, handler: DifyWorkflowCallbackHandler, mock_print_text, mocker: MockerFixture
+    ):
+        """When DEBUG is off, outputs are still yielded but nothing is printed
+        and model_dump_json() is never invoked."""
+        # Arrange
+        mocker.patch("core.callback_handler.workflow_tool_callback_handler.dify_config.DEBUG", False)
+        message = MagicMock()
+
+        # Act
+        results = list(
+            handler.on_tool_execution(
+                tool_name="quiet_tool",
+                tool_inputs={},
+                tool_outputs=[message],
+            )
+        )
+
+        # Assert
+        assert results == [message]
+        mock_print_text.assert_not_called()
+        message.model_dump_json.assert_not_called()
+
     @pytest.mark.parametrize(
         ("invalid_outputs", "expected_exception"),
         [
@@ -110,7 +143,7 @@ class TestDifyWorkflowCallbackHandler:
         ],
     )
     def test_on_tool_execution_invalid_outputs_type(
-        self, handler: DifyWorkflowCallbackHandler, invalid_outputs, expected_exception
+        self, handler: DifyWorkflowCallbackHandler, invalid_outputs, expected_exception, enable_debug
     ):
         # Arrange
         tool_name = "invalid_tool"
@@ -125,7 +158,9 @@ class TestDifyWorkflowCallbackHandler:
                 )
             )
 
-    def test_on_tool_execution_long_json_truncation(self, handler: DifyWorkflowCallbackHandler, mock_print_text):
+    def test_on_tool_execution_long_json_truncation(
+        self, handler: DifyWorkflowCallbackHandler, mock_print_text, enable_debug
+    ):
         # Arrange
         tool_name = "long_json_tool"
         long_json = "x" * 1500
@@ -147,7 +182,9 @@ class TestDifyWorkflowCallbackHandler:
             color="blue",
         )
 
-    def test_on_tool_execution_model_dump_json_exception(self, handler: DifyWorkflowCallbackHandler, mock_print_text):
+    def test_on_tool_execution_model_dump_json_exception(
+        self, handler: DifyWorkflowCallbackHandler, mock_print_text, enable_debug
+    ):
         # Arrange
         tool_name = "exception_tool"
         bad_message = MagicMock()
@@ -167,7 +204,7 @@ class TestDifyWorkflowCallbackHandler:
         assert mock_print_text.call_count >= 2
 
     def test_on_tool_execution_none_message_id_and_trace_manager(
-        self, handler: DifyWorkflowCallbackHandler, mock_print_text
+        self, handler: DifyWorkflowCallbackHandler, mock_print_text, enable_debug
     ):
         # Arrange
         tool_name = "optional_params_tool"

api/tests/unit_tests/core/helper/code_executor/test_template_transformer.py

@@ -1,6 +1,5 @@
 import json
 from base64 import b64decode
-from collections.abc import Mapping
 from typing import Any
 
 import pytest
@@ -44,7 +43,7 @@ def test_serialize_inputs_encodes_payload() -> None:
 def test_transform_response_parses_json_result_and_converts_scientific_notation() -> None:
     response = '<<RESULT>>{"value": "1e+3", "nested": {"x": "2E-2"}, "arr": ["3e+1"]}<<RESULT>>'
 
-    result: Mapping[str, Any] = _DummyTransformer.transform_response(response)
+    result: dict[str, Any] = _DummyTransformer.transform_response(response)
 
     assert result == {"value": 1000.0, "nested": {"x": 0.02}, "arr": [30.0]}
 

api/tests/unit_tests/core/helper/test_creators.py

@@ -46,6 +46,18 @@ class TestUploadDSL:
         with pytest.raises(ValueError, match="claim_code"):
             upload_dsl(b"app: demo")
 
+    @patch("core.helper.creators.httpx.post")
+    def test_raises_on_non_string_claim_code(self, mock_post):
+        mock_response = MagicMock(spec=httpx.Response)
+        mock_response.json.return_value = {"data": {"claim_code": 123}}
+        mock_response.raise_for_status = MagicMock()
+        mock_post.return_value = mock_response
+
+        from core.helper.creators import upload_dsl
+
+        with pytest.raises(ValueError, match="claim_code"):
+            upload_dsl(b"app: demo")
+
     @patch("core.helper.creators.httpx.post")
     def test_raises_on_http_error(self, mock_post):
         mock_response = MagicMock(spec=httpx.Response)

api/tests/unit_tests/core/helper/test_credential_utils.py

@@ -114,10 +114,11 @@ def test_is_credential_exists_by_type(
     scalar_result: str | None,
     expected: bool,
 ) -> None:
-    mocker.patch("extensions.ext_database.db", new=SimpleNamespace(engine=object()))
-    session_cls = mocker.patch("sqlalchemy.orm.Session")
-    session = session_cls.return_value.__enter__.return_value
+    session = mocker.MagicMock()
     session.scalar.return_value = scalar_result
+    session_context = mocker.MagicMock()
+    session_context.__enter__.return_value = session
+    mocker.patch("core.db.session_factory.create_session", return_value=session_context)
 
     result = is_credential_exists("cred-1", credential_type)
 
@@ -128,11 +129,33 @@ def test_is_credential_exists_by_type(
 def test_is_credential_exists_returns_false_for_unknown_type(
     mocker: MockerFixture,
 ) -> None:
-    mocker.patch("extensions.ext_database.db", new=SimpleNamespace(engine=object()))
-    session_cls = mocker.patch("sqlalchemy.orm.Session")
-    session = session_cls.return_value.__enter__.return_value
+    session = mocker.MagicMock()
+    session_context = mocker.MagicMock()
+    session_context.__enter__.return_value = session
+    mocker.patch("core.db.session_factory.create_session", return_value=session_context)
 
     result = is_credential_exists("cred-1", cast(PluginCredentialType, "unknown"))
 
     assert result is False
     session.scalar.assert_not_called()
+
+
+def test_is_credential_exists_uses_configured_session_factory_without_flask_app_context(
+    mocker: MockerFixture,
+) -> None:
+    class RaisingDB:
+        @property
+        def engine(self):
+            raise RuntimeError("Working outside of application context.")
+
+    session = mocker.MagicMock()
+    session.scalar.return_value = "model-credential"
+    session_context = mocker.MagicMock()
+    session_context.__enter__.return_value = session
+    create_session = mocker.patch("core.db.session_factory.create_session", return_value=session_context)
+    mocker.patch("extensions.ext_database.db", new=RaisingDB())
+
+    result = is_credential_exists("cred-1", PluginCredentialType.MODEL)
+
+    assert result is True
+    create_session.assert_called_once_with()

api/tests/unit_tests/core/helper/test_marketplace.py

@@ -1,6 +1,7 @@
 from types import SimpleNamespace
 from unittest.mock import MagicMock
 
+import pytest
 from pytest_mock import MockerFixture
 
 from core.helper.marketplace import (
@@ -53,6 +54,16 @@ def test_batch_fetch_plugin_by_ids_returns_plugins_from_response(mocker: MockerF
     response.raise_for_status.assert_called_once()
 
 
+def test_batch_fetch_plugin_by_ids_rejects_invalid_plugins_response(mocker: MockerFixture) -> None:
+    response = MagicMock()
+    response.json.return_value = {"data": {"plugins": ["p1"]}}
+    response.raise_for_status.return_value = None
+    mocker.patch("core.helper.marketplace.httpx.post", return_value=response)
+
+    with pytest.raises(ValueError, match="plugins list"):
+        batch_fetch_plugin_by_ids(["p1"])
+
+
 def test_batch_fetch_plugin_manifests_returns_empty_for_empty_input(mocker: MockerFixture) -> None:
     post_mock = mocker.patch("core.helper.marketplace.httpx.post")
 

api/tests/unit_tests/core/plugin/test_backwards_invocation_app.py

@@ -346,14 +346,32 @@ class TestPluginAppBackwardsInvocation:
         assert "end_users.app_id" in compiled
         assert stmt.compile().params == {"id_1": "uid", "tenant_id_1": "tenant-1", "app_id_1": "app-1"}
 
+    def test_get_user_returns_end_user_by_session_id(self, mocker: MockerFixture):
+        session = self.patch_create_session(mocker, side_effect=[None, MagicMock(id="session-user")])
+        app = SimpleNamespace(id="app-1", tenant_id="tenant-1")
+
+        user = PluginAppBackwardsInvocation._get_user("wecom-sender-1", app)
+
+        assert user.id == "session-user"
+        stmt = session.scalar.call_args_list[1].args[0]
+        compiled = str(stmt.compile(dialect=postgresql.dialect()))
+        assert "end_users.session_id" in compiled
+        assert "end_users.tenant_id" in compiled
+        assert "end_users.app_id" in compiled
+        assert stmt.compile().params == {
+            "session_id_1": "wecom-sender-1",
+            "tenant_id_1": "tenant-1",
+            "app_id_1": "app-1",
+        }
+
     def test_get_user_falls_back_to_account_user(self, mocker: MockerFixture):
-        session = self.patch_create_session(mocker, side_effect=[None, MagicMock(id="account-user")])
+        session = self.patch_create_session(mocker, side_effect=[None, None, MagicMock(id="account-user")])
         app = SimpleNamespace(id="app-1", tenant_id="tenant-1")
 
         user = PluginAppBackwardsInvocation._get_user("uid", app)
 
         assert user.id == "account-user"
-        stmt = session.scalar.call_args_list[1].args[0]
+        stmt = session.scalar.call_args_list[2].args[0]
         compiled = str(stmt.compile(dialect=postgresql.dialect()))
         assert "accounts.id" in compiled
         assert "tenant_account_joins.account_id" in compiled
@@ -361,12 +379,41 @@ class TestPluginAppBackwardsInvocation:
         assert stmt.compile().params == {"id_1": "uid", "tenant_id_1": "tenant-1"}
 
     def test_get_user_raises_when_user_not_found(self, mocker: MockerFixture):
-        self.patch_create_session(mocker, side_effect=[None, None])
+        self.patch_create_session(mocker, side_effect=[None, None, None])
         app = SimpleNamespace(id="app-1", tenant_id="tenant-1")
 
         with pytest.raises(ValueError, match="user not found"):
             PluginAppBackwardsInvocation._get_user("uid", app)
 
+    def test_invoke_app_creates_end_user_for_unknown_external_user_id(self, mocker: MockerFixture):
+        app = MagicMock(mode=AppMode.WORKFLOW)
+        end_user = MagicMock()
+        workflow = MagicMock()
+        mocker.patch.object(PluginAppBackwardsInvocation, "_get_app", return_value=app)
+        mocker.patch.object(PluginAppBackwardsInvocation, "_get_workflow", return_value=workflow)
+        mocker.patch.object(PluginAppBackwardsInvocation, "_get_user", side_effect=ValueError("user not found"))
+        get_or_create = mocker.patch(
+            "core.plugin.backwards_invocation.app.EndUserService.get_or_create_end_user",
+            return_value=end_user,
+        )
+        route = mocker.patch.object(PluginAppBackwardsInvocation, "invoke_workflow_app", return_value={"ok": True})
+
+        result = PluginAppBackwardsInvocation.invoke_app(
+            MagicMock(),
+            app_id="app",
+            user_id="wecom-sender-1",
+            tenant_id="tenant",
+            conversation_id="",
+            query=None,
+            stream=True,
+            inputs={},
+            files=[],
+        )
+
+        assert result == {"ok": True}
+        get_or_create.assert_called_once_with(app, user_id="wecom-sender-1")
+        assert route.call_args.args[2] is end_user
+
     def test_get_app_returns_app(self, mocker: MockerFixture):
         app_obj = MagicMock(id="app")
         self.patch_create_session(mocker, return_value=app_obj)

api/tests/unit_tests/core/plugin/test_model_runtime_adapter.py

@@ -3,7 +3,7 @@
 import datetime
 import uuid
 from types import SimpleNamespace
-from unittest.mock import Mock, patch, sentinel
+from unittest.mock import MagicMock, Mock, patch, sentinel
 
 import pytest
 
@@ -44,12 +44,34 @@ class _FakeRedis:
     def delete(self, key: str) -> None:
         self._values.pop(key, None)
 
+    def lock(
+        self,
+        key: str,
+        *,
+        timeout: int,
+        sleep: float,
+    ) -> "_FakeRedisLock":
+        return _FakeRedisLock(self, key)
 
-@pytest.fixture(autouse=True)
-def clear_plugin_model_provider_memory_cache() -> None:
-    PluginService._plugin_model_providers_memory_cache.clear()
-    yield
-    PluginService._plugin_model_providers_memory_cache.clear()
+
+class _FakeRedisLock:
+    def __init__(self, redis: _FakeRedis, key: str) -> None:
+        self._redis = redis
+        self._key = key
+        self._acquired = False
+
+    def acquire(self, *, blocking: bool = True, blocking_timeout: float | None = None) -> bool:
+        if self._key in self._redis._values:
+            return False
+
+        self._redis._values[self._key] = "locked"
+        self._acquired = True
+        return True
+
+    def release(self) -> None:
+        if self._acquired:
+            self._redis.delete(self._key)
+            self._acquired = False
 
 
 def _build_model_schema() -> AIModelEntity:
@@ -413,12 +435,13 @@ class TestPluginModelRuntime:
             "redis_client",
             SimpleNamespace(
                 get=Mock(return_value=None),
-                mget=Mock(return_value=[None, None]),
+                mget=Mock(return_value=[None]),
                 delete=Mock(),
                 setex=Mock(),
+                lock=Mock(return_value=MagicMock()),
             ),
         )
-        monkeypatch.setattr(plugin_service_module.dify_config, "PLUGIN_MODEL_PROVIDERS_CACHE_TTL", 300)
+        monkeypatch.setattr(plugin_service_module.dify_config, "PLUGIN_MODEL_PROVIDERS_CACHE_TTL", 0)
         runtime = PluginModelRuntime(tenant_id="tenant", user_id="user", client=client, plugin_service=PluginService)
 
         runtime.fetch_model_providers()

api/tests/unit_tests/core/rag/retrieval/test_dataset_retrieval.py

@@ -3888,7 +3888,17 @@ class TestDatasetRetrievalAdditionalHelpers:
         trace_manager.add_trace_task.assert_not_called()
 
     def test_on_query(self, retrieval: DatasetRetrieval) -> None:
-        with patch("core.rag.retrieval.dataset_retrieval.db.session") as mock_session:
+        db_mock = Mock()
+        audit_session = MagicMock()
+        session_factory = MagicMock()
+        session_factory.begin.return_value.__enter__.return_value = audit_session
+
+        with (
+            patch("core.rag.retrieval.dataset_retrieval.db", db_mock),
+            patch(
+                "core.rag.retrieval.dataset_retrieval.sessionmaker", return_value=session_factory
+            ) as sessionmaker_mock,
+        ):
             retrieval._on_query(
                 query=None,
                 attachment_ids=None,
@@ -3897,7 +3907,7 @@ class TestDatasetRetrievalAdditionalHelpers:
                 user_from="account",
                 user_id="u1",
             )
-            mock_session.add_all.assert_not_called()
+            audit_session.add_all.assert_not_called()
 
             retrieval._on_query(
                 query="python",
@@ -3907,11 +3917,22 @@ class TestDatasetRetrievalAdditionalHelpers:
                 user_from="account",
                 user_id="u1",
             )
-            mock_session.add_all.assert_called()
-            mock_session.commit.assert_called()
+            sessionmaker_mock.assert_called_once_with(bind=db_mock.engine, expire_on_commit=False)
+            audit_session.add_all.assert_called_once()
+            added_queries = audit_session.add_all.call_args.args[0]
+            assert len(added_queries) == 2
+            db_mock.session.commit.assert_not_called()
 
     def test_on_query_normalizes_workflow_end_user_role(self, retrieval: DatasetRetrieval) -> None:
-        with patch("core.rag.retrieval.dataset_retrieval.db.session") as mock_session:
+        db_mock = Mock()
+        audit_session = MagicMock()
+        session_factory = MagicMock()
+        session_factory.begin.return_value.__enter__.return_value = audit_session
+
+        with (
+            patch("core.rag.retrieval.dataset_retrieval.db", db_mock),
+            patch("core.rag.retrieval.dataset_retrieval.sessionmaker", return_value=session_factory),
+        ):
             retrieval._on_query(
                 query="python",
                 attachment_ids=None,
@@ -3921,12 +3942,11 @@ class TestDatasetRetrievalAdditionalHelpers:
                 user_id="u1",
             )
 
-            mock_session.add_all.assert_called_once()
-            added_queries = mock_session.add_all.call_args.args[0]
+            audit_session.add_all.assert_called_once()
+            added_queries = audit_session.add_all.call_args.args[0]
 
             assert len(added_queries) == 1
             assert added_queries[0].created_by_role == CreatorUserRole.END_USER
-            mock_session.commit.assert_called_once()
 
     def test_handle_invoke_result(self, retrieval: DatasetRetrieval) -> None:
         usage = LLMUsage.empty_usage()

api/tests/unit_tests/core/tools/test_custom_tool.py

@@ -84,6 +84,10 @@ def test_assembling_request_auth_header_assembly():
     assert headers["Authorization"] == "Bearer abc"
 
     tool.runtime.credentials = {"auth_type": "api_key_header", "api_key_header_prefix": "basic", "api_key_value": "abc"}
+    headers = tool.assembling_request(parameters={})
+    assert headers["Authorization"] == "Basic abc"
+    assert tool.runtime.credentials["api_key_value"] == "abc"
+
     headers = tool.assembling_request(parameters={})
     assert headers["Authorization"] == "Basic abc"
 

api/tests/unit_tests/core/tools/test_tool_engine.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from collections.abc import Generator
 from types import SimpleNamespace
 from typing import Any
-from unittest.mock import Mock, patch
+from unittest.mock import MagicMock, Mock, patch
 
 import pytest
 
@@ -131,18 +131,25 @@ def test_create_message_files_and_invoke_generator():
         created.append(obj)
         return obj
 
-    with patch("core.tools.tool_engine.MessageFile", side_effect=_message_file_factory):
-        with patch("core.tools.tool_engine.db") as mock_db:
-            ids = ToolEngine._create_message_files(
-                tool_messages=binaries,
-                agent_message=SimpleNamespace(id="msg-1"),
-                invoke_from=InvokeFrom.DEBUGGER,
-                user_id="user-1",
-            )
+    file_session = MagicMock()
+    session_factory = MagicMock()
+    session_factory.begin.return_value.__enter__.return_value = file_session
+    with (
+        patch("core.tools.tool_engine.MessageFile", side_effect=_message_file_factory),
+        patch("core.tools.tool_engine.db") as mock_db,
+        patch("core.tools.tool_engine.sessionmaker", return_value=session_factory) as mock_sessionmaker,
+    ):
+        ids = ToolEngine._create_message_files(
+            tool_messages=binaries,
+            agent_message=SimpleNamespace(id="msg-1"),
+            invoke_from=InvokeFrom.DEBUGGER,
+            user_id="user-1",
+        )
 
     assert ids == ["mf-1", "mf-2"]
-    assert mock_db.session.add.call_count == 2
-    mock_db.session.close.assert_called_once()
+    mock_sessionmaker.assert_called_once_with(bind=mock_db.engine, expire_on_commit=False)
+    assert file_session.add.call_count == 2
+    mock_db.session.close.assert_not_called()
 
     tool = _build_tool()
     invoked = list(ToolEngine._invoke(tool, {"a": 1}, user_id="u"))

api/tests/unit_tests/core/tools/utils/test_parser.py

@@ -327,7 +327,7 @@ def test_parse_openai_plugin_json_branches(app):
 def test_parse_openai_plugin_json_http_branches(app):
     with app.test_request_context():
         response = type("Resp", (), {"status_code": 500, "text": "", "close": Mock()})()
-        with patch("core.tools.utils.parser.httpx.get", return_value=response):
+        with patch("core.tools.utils.parser.ssrf_proxy.get", return_value=response):
             with pytest.raises(ToolProviderNotFoundError, match="cannot get openapi yaml"):
                 ApiBasedToolSchemaParser.parse_openai_plugin_json_to_tool_bundle(
                     '{"api": {"url": "https://x", "type": "openapi"}}'
@@ -335,7 +335,7 @@ def test_parse_openai_plugin_json_http_branches(app):
         response.close.assert_called_once()
 
         success_response = type("Resp", (), {"status_code": 200, "text": "openapi: 3.0.0", "close": Mock()})()
-        with patch("core.tools.utils.parser.httpx.get", return_value=success_response):
+        with patch("core.tools.utils.parser.ssrf_proxy.get", return_value=success_response):
             with patch(
                 "core.tools.utils.parser.ApiBasedToolSchemaParser.parse_openapi_yaml_to_tool_bundle",
                 return_value=["bundle"],

api/tests/unit_tests/libs/test_oauth_base.py

@@ -19,13 +19,19 @@ def test_oauth_base_methods_raise_not_implemented():
         oauth._transform_user_info({})
 
 
-def test_oauth_state_round_trips_invite_token_timezone_and_language():
-    state = encode_oauth_state(invite_token="invite-123", timezone="Asia/Shanghai", language="zh-Hans")
+def test_oauth_state_round_trips_login_context():
+    state = encode_oauth_state(
+        invite_token="invite-123",
+        timezone="Asia/Shanghai",
+        language="zh-Hans",
+        redirect_url="/apps?category=workflow",
+    )
 
     assert decode_oauth_state(state) == {
         "invite_token": "invite-123",
         "timezone": "Asia/Shanghai",
         "language": "zh-Hans",
+        "redirect_url": "/apps?category=workflow",
     }
 
 

api/tests/unit_tests/libs/test_oauth_clients.py

@@ -70,6 +70,14 @@ class TestGitHubOAuth(BaseOAuthTest):
         else:
             assert "state" not in params
 
+    def test_should_preserve_redirect_url_in_state(self, oauth):
+        redirect_url = "/apps?category=workflow"
+
+        url = oauth.get_authorization_url(redirect_url=redirect_url)
+        _, params = self.parse_auth_url(url)
+
+        assert decode_oauth_state(params["state"][0]) == {"redirect_url": redirect_url}
+
     @pytest.mark.parametrize(
         ("response_data", "expected_token", "should_raise"),
         [
@@ -252,6 +260,14 @@ class TestGoogleOAuth(BaseOAuthTest):
         else:
             assert "state" not in params
 
+    def test_should_preserve_redirect_url_in_state(self, oauth):
+        redirect_url = "/apps?category=workflow"
+
+        url = oauth.get_authorization_url(redirect_url=redirect_url)
+        _, params = self.parse_auth_url(url)
+
+        assert decode_oauth_state(params["state"][0]) == {"redirect_url": redirect_url}
+
     @pytest.mark.parametrize(
         ("response_data", "expected_token", "should_raise"),
         [

api/tests/unit_tests/models/test_end_user_type.py

@@ -2,6 +2,9 @@ import ast
 import inspect
 from pathlib import Path
 
+import pytest
+from sqlalchemy.engine.default import DefaultDialect
+
 from models.enums import EndUserType
 from models.model import EndUser
 from models.types import EnumText
@@ -24,6 +27,24 @@ def test_end_user_type_is_plain_persisted_value_enum():
     assert not hasattr(EndUserType, "from_invoke_from")
 
 
+def test_end_user_type_accepts_legacy_service_api_value():
+    # Legacy rows stored the service-api type with an underscore before it was
+    # normalized to the hyphenated value; loading them must not raise. See #38201.
+    assert EndUserType("service_api") is EndUserType.SERVICE_API
+    assert EndUserType("service-api") is EndUserType.SERVICE_API
+
+
+def test_end_user_type_still_rejects_unknown_values():
+    with pytest.raises(ValueError):
+        EndUserType("not-a-real-end-user-type")
+
+
+def test_enum_text_deserializes_legacy_service_api_value():
+    # Exercises the actual column load path that raised for unmigrated rows.
+    column_type = EnumText(EndUserType)
+    assert column_type.process_result_value("service_api", DefaultDialect()) is EndUserType.SERVICE_API
+
+
 def test_end_user_service_creation_methods_accept_end_user_type():
     assert inspect.signature(EndUserService.get_or_create_end_user_by_type).parameters["type"].annotation is EndUserType
     assert inspect.signature(EndUserService.create_end_user_batch).parameters["type"].annotation is EndUserType

api/tests/unit_tests/services/enterprise/test_rbac_service.py

@@ -633,6 +633,8 @@ class TestMyPermissions:
             assert "dataset.acl.preview" in out.workspace.permission_keys
             assert "app.acl.preview" in out.app.default_permission_keys
             assert "dataset.acl.preview" in out.dataset.default_permission_keys
+        if role == "editor":
+            assert "app.acl.log_and_annotation" in out.app.default_permission_keys
 
     @pytest.mark.parametrize(
         ("role", "expected_snippet_keys"),

api/tests/unit_tests/services/plugin/test_plugin_service.py

@@ -4,6 +4,7 @@ from types import SimpleNamespace
 from unittest.mock import MagicMock, Mock, call, patch
 
 import pytest
+import zstandard
 from pydantic import TypeAdapter
 from redis import RedisError
 
@@ -14,15 +15,6 @@ from graphon.model_runtime.entities.provider_entities import ConfigurateMethod,
 MODULE = "core.plugin.plugin_service"
 
 
-@pytest.fixture(autouse=True)
-def clear_plugin_model_provider_memory_cache() -> None:
-    from core.plugin.plugin_service import PluginService
-
-    PluginService._plugin_model_providers_memory_cache.clear()
-    yield
-    PluginService._plugin_model_providers_memory_cache.clear()
-
-
 class _FakeSession:
     def __init__(self) -> None:
         self.execute = Mock()
@@ -137,17 +129,71 @@ class TestFetchLatestPluginVersion:
 
 
 class TestPluginModelProviderCache:
-    def test_fetch_plugin_model_providers_returns_cached_provider_without_calling_daemon(self) -> None:
-        """A valid tenant cache entry is reused across runtime calls without plugin daemon access."""
+    def test_store_cached_plugin_model_providers_compresses_large_payload(self) -> None:
+        """Large provider metadata payloads are compressed before being stored in Redis."""
+        large_provider = _build_provider_entity()
+        large_provider.label = I18nObject(en_US="OpenAI " * 10000)
+        raw_payload = TypeAdapter(list[ProviderEntity]).dump_json([large_provider])
+        cache_key = _provider_cache_key("tenant-1", 0)
+
+        with (
+            patch(f"{MODULE}.redis_client") as redis_client,
+            patch(f"{MODULE}.dify_config") as mock_config,
+        ):
+            mock_config.PLUGIN_MODEL_PROVIDERS_CACHE_TTL = 86400
+
+            from core.plugin.plugin_service import PluginService
+
+            PluginService._store_cached_plugin_model_providers("tenant-1", 0, [large_provider])
+
+        redis_client.setex.assert_called_once()
+        stored_key, ttl, stored_payload = redis_client.setex.call_args.args
+        assert stored_key == cache_key
+        assert ttl == 86400
+        assert isinstance(stored_payload, bytes)
+        prefix = PluginService.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX
+        assert stored_payload.startswith(prefix)
+        assert len(stored_payload) < len(raw_payload)
+        assert zstandard.decompress(stored_payload[len(prefix) :]) == raw_payload
+
+    def test_fetch_plugin_model_providers_reads_compressed_cached_provider_without_calling_daemon(self) -> None:
+        """Compressed tenant cache entries are decoded before provider schema validation."""
         cached_provider = _build_provider_entity()
-        cached_payload = TypeAdapter(list[ProviderEntity]).dump_json([cached_provider]).decode("utf-8")
+        cached_provider.label = I18nObject(en_US="OpenAI " * 10000)
+        cached_payload = TypeAdapter(list[ProviderEntity]).dump_json([cached_provider])
         generation_key = _provider_generation_key("tenant-1")
         cache_key = _provider_cache_key("tenant-1", 0)
-        legacy_cache_key = _provider_cache_key("tenant-1")
+
+        from core.plugin.plugin_service import PluginService
+
+        compressed_payload = PluginService.PLUGIN_MODEL_PROVIDERS_CACHE_COMPRESSION_PREFIX + zstandard.compress(
+            cached_payload, level=1
+        )
 
         with patch(f"{MODULE}.redis_client") as redis_client:
             redis_client.get.return_value = None
-            redis_client.mget.return_value = [cached_payload, None]
+            redis_client.mget.return_value = [compressed_payload]
+            client = Mock()
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
+        assert result[0].label.en_us == "OpenAI " * 10000
+        client.fetch_model_providers.assert_not_called()
+        redis_client.setex.assert_not_called()
+        redis_client.get.assert_called_once_with(generation_key)
+        redis_client.mget.assert_called_once_with([cache_key])
+
+    def test_fetch_plugin_model_providers_returns_cached_provider_without_calling_daemon(self) -> None:
+        """A valid tenant cache entry is reused across runtime calls without plugin daemon access."""
+        cached_provider = _build_provider_entity()
+        cached_payload = TypeAdapter(list[ProviderEntity]).dump_json([cached_provider])
+        generation_key = _provider_generation_key("tenant-1")
+        cache_key = _provider_cache_key("tenant-1", 0)
+
+        with patch(f"{MODULE}.redis_client") as redis_client:
+            redis_client.get.return_value = None
+            redis_client.mget.return_value = [cached_payload]
 
             from core.plugin.plugin_service import PluginService
 
@@ -158,19 +204,18 @@ class TestPluginModelProviderCache:
         client.fetch_model_providers.assert_not_called()
         redis_client.setex.assert_not_called()
         redis_client.get.assert_called_once_with(generation_key)
-        redis_client.mget.assert_called_once_with([cache_key, legacy_cache_key])
+        redis_client.mget.assert_called_once_with([cache_key])
 
     def test_fetch_plugin_model_providers_deletes_invalid_cache_and_refetches(self) -> None:
         """Invalid generation-scoped cache payloads are removed before falling back to the daemon."""
         generation_key = _provider_generation_key("tenant-1")
         cache_key = _provider_cache_key("tenant-1", 0)
-        legacy_cache_key = _provider_cache_key("tenant-1")
         with (
             patch(f"{MODULE}.redis_client") as redis_client,
             patch(f"{MODULE}.dify_config") as mock_config,
         ):
-            redis_client.get.side_effect = [None, None]
-            redis_client.mget.return_value = ["not-json", None]
+            redis_client.get.side_effect = [None, None, None]
+            redis_client.mget.side_effect = [["not-json"], [None]]
             mock_config.PLUGIN_MODEL_PROVIDERS_CACHE_TTL = 86400
             client = Mock()
             client.fetch_model_providers.return_value = [_build_plugin_model_provider()]
@@ -184,8 +229,11 @@ class TestPluginModelProviderCache:
         assert redis_client.setex.call_args.args[0] == cache_key
         assert redis_client.setex.call_args.args[1] == 86400
         assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
-        redis_client.get.assert_has_calls([call(generation_key), call(generation_key)])
-        redis_client.mget.assert_called_once_with([cache_key, legacy_cache_key])
+        redis_client.get.assert_has_calls([call(generation_key), call(generation_key), call(generation_key)])
+        assert redis_client.mget.call_args_list == [
+            call([cache_key]),
+            call([cache_key]),
+        ]
 
     def test_fetch_plugin_model_providers_refetches_when_cache_read_fails(self) -> None:
         """Redis read failures do not block provider discovery for the tenant."""
@@ -204,7 +252,6 @@ class TestPluginModelProviderCache:
     def test_fetch_plugin_model_providers_refetches_when_cached_payload_batch_read_fails(self) -> None:
         """Redis mget failures do not block provider discovery for the tenant."""
         cache_key = _provider_cache_key("tenant-1", 0)
-        legacy_cache_key = _provider_cache_key("tenant-1")
         with patch(f"{MODULE}.redis_client") as redis_client:
             redis_client.get.return_value = None
             redis_client.mget.side_effect = RedisError("redis unavailable")
@@ -216,14 +263,14 @@ class TestPluginModelProviderCache:
             result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
 
         client.fetch_model_providers.assert_called_once_with("tenant-1")
-        redis_client.mget.assert_called_once_with([cache_key, legacy_cache_key])
+        redis_client.mget.assert_called_once_with([cache_key])
         assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
 
     def test_fetch_plugin_model_providers_returns_fresh_result_when_cache_write_fails(self) -> None:
         """Redis write failures are non-fatal after fresh provider data has been fetched."""
         with patch(f"{MODULE}.redis_client") as redis_client:
             redis_client.get.return_value = None
-            redis_client.mget.return_value = [None, None]
+            redis_client.mget.return_value = [None]
             redis_client.setex.side_effect = RedisError("redis unavailable")
             client = Mock()
             client.fetch_model_providers.return_value = [_build_plugin_model_provider()]
@@ -235,6 +282,352 @@ class TestPluginModelProviderCache:
         client.fetch_model_providers.assert_called_once_with("tenant-1")
         assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
 
+    def test_fetch_plugin_model_providers_waits_for_concurrent_refresh_cache_fill(self) -> None:
+        """A cache miss waits for the active tenant refresh instead of stampeding the daemon."""
+        cached_provider = _build_provider_entity()
+        cached_payload = TypeAdapter(list[ProviderEntity]).dump_json([cached_provider])
+        cache_key = _provider_cache_key("tenant-1", 0)
+
+        with (
+            patch(f"{MODULE}.redis_client") as redis_client,
+            patch(f"{MODULE}.time.monotonic", return_value=100.0),
+        ):
+            redis_client.get.return_value = None
+            redis_client.mget.side_effect = [[None], [cached_payload]]
+            client = Mock()
+            client.fetch_model_providers.return_value = [_build_plugin_model_provider(provider="anthropic")]
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        redis_client.lock.assert_called_once_with(
+            PluginService._get_plugin_model_providers_lock_key("tenant-1", 0),
+            timeout=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_TTL,
+            sleep=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL,
+        )
+        redis_client.lock.return_value.acquire.assert_called_once_with(
+            blocking=True,
+            blocking_timeout=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT,
+        )
+        assert redis_client.mget.call_args_list == [
+            call([cache_key]),
+            call([cache_key]),
+        ]
+        redis_client.lock.return_value.release.assert_called_once()
+        client.fetch_model_providers.assert_not_called()
+        assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
+
+    def test_fetch_plugin_model_providers_falls_back_when_refresh_lock_wait_times_out(self) -> None:
+        """A request should stop waiting and fetch directly instead of surfacing lock contention."""
+        cache_key = _provider_cache_key("tenant-1", 0)
+        with (
+            patch(f"{MODULE}.redis_client") as redis_client,
+            patch(f"{MODULE}.PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT", 0),
+            patch(f"{MODULE}.time.monotonic", return_value=100.0),
+        ):
+            redis_client.get.return_value = None
+            redis_client.mget.return_value = [None]
+            redis_client.lock.return_value.acquire.return_value = False
+            client = Mock()
+            client.fetch_model_providers.return_value = [_build_plugin_model_provider()]
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        redis_client.lock.assert_called_once_with(
+            PluginService._get_plugin_model_providers_lock_key("tenant-1", 0),
+            timeout=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_TTL,
+            sleep=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL,
+        )
+        redis_client.lock.return_value.acquire.assert_called_once_with(blocking=True, blocking_timeout=0)
+        redis_client.lock.return_value.release.assert_not_called()
+        client.fetch_model_providers.assert_called_once_with("tenant-1")
+        redis_client.setex.assert_called_once()
+        assert redis_client.setex.call_args.args[0] == cache_key
+        assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
+
+    def test_fetch_plugin_model_providers_restarts_lock_path_after_generation_changes(self) -> None:
+        """Waiters re-read provider generation before trying to become the next refresh owner."""
+        generation_key = _provider_generation_key("tenant-1")
+        stale_cache_key = _provider_cache_key("tenant-1", 0)
+        new_cache_key = _provider_cache_key("tenant-1", 1)
+        with (
+            patch(f"{MODULE}.redis_client") as redis_client,
+            patch(f"{MODULE}.time.monotonic", side_effect=[100.0, 100.0, 100.5, 101.0]),
+        ):
+            redis_client.get.side_effect = [None, b"1", b"1", b"1", b"1"]
+            redis_client.mget.side_effect = [[None], [None], [None], [None]]
+            client = Mock()
+            client.fetch_model_providers.return_value = [_build_plugin_model_provider(provider="anthropic")]
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        assert redis_client.get.call_args_list == [
+            call(generation_key),
+            call(generation_key),
+            call(generation_key),
+            call(generation_key),
+            call(generation_key),
+        ]
+        assert redis_client.mget.call_args_list == [
+            call([stale_cache_key]),
+            call([new_cache_key]),
+            call([new_cache_key]),
+            call([new_cache_key]),
+        ]
+        assert redis_client.lock.call_args_list == [
+            call(
+                PluginService._get_plugin_model_providers_lock_key("tenant-1", 0),
+                timeout=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_TTL,
+                sleep=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL,
+            ),
+            call(
+                PluginService._get_plugin_model_providers_lock_key("tenant-1", 1),
+                timeout=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_TTL,
+                sleep=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL,
+            ),
+        ]
+        assert redis_client.lock.return_value.acquire.call_args_list == [
+            call(blocking=True, blocking_timeout=2.0),
+            call(blocking=True, blocking_timeout=1.5),
+        ]
+        assert redis_client.lock.return_value.release.call_count == 2
+        client.fetch_model_providers.assert_called_once_with("tenant-1")
+        redis_client.setex.assert_called_once()
+        assert redis_client.setex.call_args.args[0] == new_cache_key
+        assert [provider.provider for provider in result] == ["langgenius/anthropic/anthropic"]
+
+    def test_fetch_plugin_model_providers_falls_back_when_generation_retries_exhaust_wait_budget(self) -> None:
+        """Generation retry loops share one request-local lock wait deadline before direct fetch fallback."""
+        generation_key = _provider_generation_key("tenant-1")
+        stale_cache_key = _provider_cache_key("tenant-1", 0)
+        new_cache_key = _provider_cache_key("tenant-1", 1)
+
+        with (
+            patch(f"{MODULE}.redis_client") as redis_client,
+            patch(f"{MODULE}.time.monotonic", side_effect=[100.0, 100.0, 102.1]),
+        ):
+            redis_client.get.side_effect = [None, b"1", b"1", b"1"]
+            redis_client.mget.side_effect = [[None], [None], [None]]
+            client = Mock()
+            client.fetch_model_providers.return_value = [_build_plugin_model_provider(provider="anthropic")]
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        assert redis_client.get.call_args_list == [
+            call(generation_key),
+            call(generation_key),
+            call(generation_key),
+            call(generation_key),
+        ]
+        assert redis_client.mget.call_args_list == [
+            call([stale_cache_key]),
+            call([new_cache_key]),
+            call([new_cache_key]),
+        ]
+        redis_client.lock.assert_called_once_with(
+            PluginService._get_plugin_model_providers_lock_key("tenant-1", 0),
+            timeout=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_TTL,
+            sleep=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL,
+        )
+        redis_client.lock.return_value.acquire.assert_called_once_with(blocking=True, blocking_timeout=2.0)
+        redis_client.lock.return_value.release.assert_called_once()
+        client.fetch_model_providers.assert_called_once_with("tenant-1")
+        redis_client.setex.assert_called_once()
+        assert redis_client.setex.call_args.args[0] == new_cache_key
+        assert [provider.provider for provider in result] == ["langgenius/anthropic/anthropic"]
+
+    def test_fetch_plugin_model_providers_releases_owned_refresh_lock_after_store(self) -> None:
+        """The refresh owner releases only its token after storing provider metadata."""
+        cache_key = _provider_cache_key("tenant-1", 0)
+
+        with (
+            patch(f"{MODULE}.redis_client") as redis_client,
+            patch(f"{MODULE}.time.monotonic", return_value=100.0),
+        ):
+            redis_client.get.return_value = None
+            redis_client.mget.return_value = [None]
+            client = Mock()
+            client.fetch_model_providers.return_value = [_build_plugin_model_provider()]
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        redis_client.lock.assert_called_once_with(
+            PluginService._get_plugin_model_providers_lock_key("tenant-1", 0),
+            timeout=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_TTL,
+            sleep=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_INTERVAL,
+        )
+        redis_client.lock.return_value.acquire.assert_called_once_with(
+            blocking=True,
+            blocking_timeout=PluginService.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT,
+        )
+        assert redis_client.mget.call_args_list == [
+            call([cache_key]),
+            call([cache_key]),
+        ]
+        redis_client.lock.return_value.release.assert_called_once()
+        redis_client.eval.assert_not_called()
+        client.fetch_model_providers.assert_called_once_with("tenant-1")
+        assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
+
+    def test_fetch_plugin_model_providers_returns_fresh_result_when_refresh_lock_release_fails(self) -> None:
+        """Release failures are logged, not allowed to hide a successful daemon refresh."""
+        cache_key = _provider_cache_key("tenant-1", 0)
+
+        with (
+            patch(f"{MODULE}.redis_client") as redis_client,
+            patch(f"{MODULE}.time.monotonic", return_value=100.0),
+        ):
+            redis_client.get.return_value = None
+            redis_client.mget.return_value = [None]
+            redis_client.lock.return_value.release.side_effect = RedisError("release failed")
+            client = Mock()
+            client.fetch_model_providers.return_value = [_build_plugin_model_provider()]
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        assert redis_client.mget.call_args_list == [
+            call([cache_key]),
+            call([cache_key]),
+        ]
+        client.fetch_model_providers.assert_called_once_with("tenant-1")
+        redis_client.setex.assert_called_once()
+        redis_client.lock.return_value.release.assert_called_once()
+        assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
+
+    def test_fetch_plugin_model_providers_releases_owned_refresh_lock_when_fetch_fails(self) -> None:
+        """Release failures must not hide the daemon failure that happened while owning the lock."""
+        cache_key = _provider_cache_key("tenant-1", 0)
+
+        with (
+            patch(f"{MODULE}.redis_client") as redis_client,
+            patch(f"{MODULE}.time.monotonic", return_value=100.0),
+        ):
+            redis_client.get.return_value = None
+            redis_client.mget.return_value = [None]
+            redis_client.lock.return_value.release.side_effect = RedisError("release failed")
+            client = Mock()
+            client.fetch_model_providers.side_effect = RuntimeError("daemon failed")
+
+            from core.plugin.plugin_service import PluginService
+
+            with pytest.raises(RuntimeError, match="daemon failed"):
+                PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        assert redis_client.mget.call_args_list == [
+            call([cache_key]),
+            call([cache_key]),
+        ]
+        client.fetch_model_providers.assert_called_once_with("tenant-1")
+        redis_client.setex.assert_not_called()
+        redis_client.lock.return_value.release.assert_called_once()
+
+    def test_fetch_plugin_model_providers_falls_back_when_refresh_lock_acquire_fails(self) -> None:
+        """Redis acquire failures degrade to a direct daemon fetch instead of hiding provider data."""
+        with (
+            patch(f"{MODULE}.redis_client") as redis_client,
+            patch(f"{MODULE}.time.monotonic", return_value=100.0),
+        ):
+            redis_client.get.return_value = None
+            redis_client.mget.return_value = [None]
+            redis_client.lock.return_value.acquire.side_effect = RedisError("redis unavailable")
+            client = Mock()
+            client.fetch_model_providers.return_value = [_build_plugin_model_provider()]
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        redis_client.lock.return_value.acquire.assert_called_once()
+        redis_client.lock.return_value.release.assert_not_called()
+        client.fetch_model_providers.assert_called_once_with("tenant-1")
+        assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
+
+    def test_fetch_plugin_model_providers_skips_wait_when_refresh_lock_fails(self) -> None:
+        """Lock API failures should fall back directly instead of adding timeout latency."""
+        with (
+            patch(f"{MODULE}.redis_client") as redis_client,
+            patch(f"{MODULE}.time.sleep") as sleep,
+        ):
+            redis_client.get.return_value = None
+            redis_client.mget.return_value = [None]
+            redis_client.lock.side_effect = RedisError("redis unavailable")
+            redis_client.set.side_effect = AssertionError("raw redis set must not be used for refresh locks")
+            client = Mock()
+            client.fetch_model_providers.return_value = [_build_plugin_model_provider()]
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        sleep.assert_not_called()
+        redis_client.lock.assert_called_once()
+        client.fetch_model_providers.assert_called_once_with("tenant-1")
+        assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
+
+    def test_fetch_plugin_model_providers_caches_empty_provider_list(self) -> None:
+        """An empty provider list is still a valid refresh result for single-flight waiters."""
+        cache_key = _provider_cache_key("tenant-1", 0)
+        with patch(f"{MODULE}.redis_client") as redis_client:
+            redis_client.get.return_value = None
+            redis_client.mget.return_value = [None]
+            client = Mock()
+            client.fetch_model_providers.return_value = []
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        assert result == ()
+        redis_client.setex.assert_called_once()
+        assert redis_client.setex.call_args.args[0] == cache_key
+        redis_client.lock.return_value.release.assert_called_once()
+
+    def test_fetch_plugin_model_providers_skips_cache_write_when_generation_changes_during_refresh(self) -> None:
+        """A refresh that started before invalidation must not populate the newer generation cache."""
+        with patch(f"{MODULE}.redis_client") as redis_client:
+            redis_client.get.side_effect = [None, None, "1"]
+            redis_client.mget.return_value = [None]
+            client = Mock()
+            client.fetch_model_providers.return_value = []
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        assert result == ()
+        client.fetch_model_providers.assert_called_once_with("tenant-1")
+        redis_client.setex.assert_not_called()
+        redis_client.lock.return_value.release.assert_called_once()
+
+    def test_fetch_plugin_model_providers_reuses_cached_empty_provider_list(self) -> None:
+        """A cached empty list should prevent repeated daemon fetches for tenants without plugin models."""
+        empty_payload = TypeAdapter(list[ProviderEntity]).dump_json([])
+        cache_key = _provider_cache_key("tenant-1", 0)
+
+        with patch(f"{MODULE}.redis_client") as redis_client:
+            redis_client.get.return_value = None
+            redis_client.mget.return_value = [empty_payload]
+            client = Mock()
+
+            from core.plugin.plugin_service import PluginService
+
+            result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
+
+        assert result == ()
+        redis_client.mget.assert_called_once_with([cache_key])
+        client.fetch_model_providers.assert_not_called()
+
     def test_fetch_plugin_model_providers_creates_default_client_on_cache_miss(self) -> None:
         """The service owns plugin daemon access when no runtime-provided client is injected."""
         with (
@@ -242,7 +635,7 @@ class TestPluginModelProviderCache:
             patch(f"{MODULE}.PluginModelClient") as client_cls,
         ):
             redis_client.get.return_value = None
-            redis_client.mget.return_value = [None, None]
+            redis_client.mget.return_value = [None]
             client = client_cls.return_value
             client.fetch_model_providers.return_value = [_build_plugin_model_provider()]
 
@@ -254,35 +647,6 @@ class TestPluginModelProviderCache:
         client.fetch_model_providers.assert_called_once_with("tenant-1")
         assert [provider.provider for provider in result] == ["langgenius/openai/openai"]
 
-    def test_fetch_plugin_model_providers_reuses_process_local_cache(self) -> None:
-        generation_key = _provider_generation_key("tenant-1")
-        with (
-            patch(f"{MODULE}.redis_client") as redis_client,
-            patch(f"{MODULE}.PluginModelClient") as client_cls,
-        ):
-            redis_client.get.side_effect = [None, None, None]
-            redis_client.mget.return_value = [None, None]
-            client = client_cls.return_value
-            client.fetch_model_providers.return_value = [_build_plugin_model_provider()]
-
-            from core.plugin.plugin_service import PluginService
-
-            first_result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1")
-            redis_client.get.reset_mock()
-            redis_client.mget.reset_mock()
-            redis_client.setex.reset_mock()
-            client.fetch_model_providers.reset_mock()
-
-            second_result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1")
-
-        redis_client.get.assert_called_once_with(generation_key)
-        redis_client.mget.assert_not_called()
-        redis_client.setex.assert_not_called()
-        client.fetch_model_providers.assert_not_called()
-        assert [provider.provider for provider in second_result] == ["langgenius/openai/openai"]
-        assert second_result[0] == first_result[0]
-        assert second_result[0] is not first_result[0]
-
     def test_invalidate_plugin_model_providers_cache_uses_redis_pipeline(self) -> None:
         with patch(f"{MODULE}.redis_client") as redis_client:
             pipe = redis_client.pipeline.return_value
@@ -310,41 +674,29 @@ class TestPluginModelProviderCache:
         pipe.incr.assert_called_once_with(_provider_generation_key("tenant-1"))
         pipe.execute.assert_called_once_with()
 
-    def test_invalidate_plugin_model_providers_cache_clears_process_local_cache(self) -> None:
-        with patch(f"{MODULE}.redis_client") as redis_client:
-            pipe = redis_client.pipeline.return_value
-
-            from core.plugin.plugin_service import PluginService
-
-            PluginService._store_in_memory_plugin_model_providers("tenant-1", 0, [_build_provider_entity()])
-            PluginService.invalidate_plugin_model_providers_cache("tenant-1")
-
-        assert PluginService._plugin_model_providers_memory_cache == {}
-        redis_client.pipeline.assert_called_once_with(transaction=False)
-        pipe.delete.assert_called_once_with(_provider_cache_key("tenant-1"))
-        pipe.incr.assert_called_once_with(_provider_generation_key("tenant-1"))
-        pipe.execute.assert_called_once_with()
-
-    def test_fetch_plugin_model_providers_ignores_stale_process_local_cache_after_generation_bump(self) -> None:
+    def test_fetch_plugin_model_providers_uses_new_generation_cache_after_generation_bump(self) -> None:
         generation_key = _provider_generation_key("tenant-1")
         new_cache_key = _provider_cache_key("tenant-1", 1)
         with patch(f"{MODULE}.redis_client") as redis_client:
-            redis_client.get.side_effect = [b"1", b"1"]
+            redis_client.get.side_effect = [b"1", b"1", b"1"]
             redis_client.mget.return_value = [None]
             client = Mock()
             client.fetch_model_providers.return_value = [_build_plugin_model_provider(provider="anthropic")]
 
             from core.plugin.plugin_service import PluginService
 
-            PluginService._store_in_memory_plugin_model_providers("tenant-1", 0, [_build_provider_entity()])
             result = PluginService.fetch_plugin_model_providers(tenant_id="tenant-1", client=client)
 
         client.fetch_model_providers.assert_called_once_with("tenant-1")
-        redis_client.get.assert_has_calls([call(generation_key), call(generation_key)])
-        redis_client.mget.assert_called_once_with([new_cache_key])
+        redis_client.get.assert_has_calls([call(generation_key), call(generation_key), call(generation_key)])
+        assert redis_client.mget.call_args_list == [
+            call([new_cache_key]),
+            call([new_cache_key]),
+        ]
         redis_client.setex.assert_called_once()
         assert redis_client.setex.call_args.args[0] == new_cache_key
-        assert PluginService._plugin_model_providers_memory_cache["tenant-1"][0] == 1
+        redis_client.lock.return_value.acquire.assert_called_once()
+        redis_client.lock.return_value.release.assert_called_once()
         assert [provider.provider for provider in result] == ["langgenius/anthropic/anthropic"]
 
 

api/tests/unit_tests/services/test_account_service.py

@@ -671,6 +671,31 @@ class TestTenantService:
         with pytest.raises(exception_type):
             callable_func(*args, **kwargs)
 
+    def test_iter_member_account_id_batches_uses_offset_limit(self):
+        class FakeScalarResult:
+            def __init__(self, items: list[str]) -> None:
+                self.items = items
+
+            def all(self) -> list[str]:
+                return self.items
+
+        offsets: list[int] = []
+
+        def scalars(stmt):
+            offsets.append(stmt._offset_clause.value)
+            if len(offsets) == 1:
+                return FakeScalarResult(["acct-1", "acct-2"])
+            if len(offsets) == 2:
+                return FakeScalarResult(["acct-3"])
+            return FakeScalarResult([])
+
+        batches = list(
+            TenantService.iter_member_account_id_batches("tenant-1", 2, session=SimpleNamespace(scalars=scalars))
+        )
+
+        assert batches == [["acct-1", "acct-2"], ["acct-3"]]
+        assert offsets == [0, 2, 4]
+
     # ==================== get_account_role_in_tenant Tests ====================
     # Backs the auth pipeline's `load_workspace_role`: None => non-member
     # (pipeline maps to 404), otherwise the caller's role (out-of-set role => 403).

api/tests/unit_tests/services/test_billing_service.py

@@ -73,6 +73,23 @@ class TestBillingServiceSendRequest:
         assert call_args[1]["headers"]["Billing-Api-Secret-Key"] == "test-secret-key"
         assert call_args[1]["headers"]["Content-Type"] == "application/json"
 
+    def test_send_request_with_base_url_override(self, mock_httpx_request, mock_billing_config):
+        """Quota APIs can use the new billing service without changing legacy billing calls."""
+        # Arrange
+        expected_response = {"result": "success"}
+        mock_response = MagicMock()
+        mock_response.status_code = httpx.codes.OK
+        mock_response.json.return_value = expected_response
+        mock_httpx_request.return_value = mock_response
+
+        # Act
+        result = BillingService._send_request("GET", "/quota/balance", base_url="https://quota.example.com")
+
+        # Assert
+        assert result == expected_response
+        call_args = mock_httpx_request.call_args
+        assert call_args[0][1] == "https://quota.example.com/quota/balance"
+
     @pytest.mark.parametrize(
         "status_code", [httpx.codes.NOT_FOUND, httpx.codes.INTERNAL_SERVER_ERROR, httpx.codes.BAD_REQUEST]
     )
@@ -393,6 +410,20 @@ class TestBillingServiceSubscriptionInfo:
             params={"tenant_id": tenant_id},
         )
 
+    def test_quota_get_balance_uses_quota_request(self):
+        tenant_id = "tenant-123"
+        with patch.object(BillingService, "_send_quota_request") as mock_send_quota_request:
+            mock_send_quota_request.return_value = {"quota": "200", "usage": "6", "available": "194", "reserved": "0"}
+
+            result = BillingService.quota_get_balance(tenant_id, "credit_pool", bucket="trial")
+
+        assert result == {"quota": 200, "usage": 6, "available": 194, "reserved": 0}
+        mock_send_quota_request.assert_called_once_with(
+            "GET",
+            "/quota/balance",
+            params={"tenant_id": tenant_id, "feature_key": "credit_pool", "bucket": "trial"},
+        )
+
     def test_get_knowledge_rate_limit_with_defaults(self, mock_send_request):
         """Test knowledge rate limit retrieval with default values."""
         # Arrange
@@ -518,19 +549,20 @@ class TestBillingServiceUsageCalculation:
         assert result == expected_response
         mock_send_request.assert_called_once_with("GET", "/tenant-feature-usage/info", params={"tenant_id": tenant_id})
 
-    def test_get_quota_info(self, mock_send_request):
+    def test_get_quota_info(self):
         """Test retrieval of quota info from new endpoint."""
         # Arrange
         tenant_id = "tenant-123"
         expected_response = {"trigger_event": {"limit": 100, "usage": 30}, "api_rate_limit": {"limit": -1, "usage": 0}}
-        mock_send_request.return_value = expected_response
+        with patch.object(BillingService, "_send_quota_request") as mock_send_quota_request:
+            mock_send_quota_request.return_value = expected_response
 
-        # Act
-        result = BillingService.get_quota_info(tenant_id)
+            # Act
+            result = BillingService.get_quota_info(tenant_id)
 
         # Assert
         assert result == expected_response
-        mock_send_request.assert_called_once_with("GET", "/quota/info", params={"tenant_id": tenant_id})
+        mock_send_quota_request.assert_called_once_with("GET", "/quota/info", params={"tenant_id": tenant_id})
 
     def test_update_tenant_feature_plan_usage_positive_delta(self, mock_send_request):
         """Test updating tenant feature usage with positive delta (adding credits)."""
@@ -614,7 +646,7 @@ class TestBillingServiceQuotaOperations:
 
     @pytest.fixture
     def mock_send_request(self):
-        with patch.object(BillingService, "_send_request") as mock:
+        with patch.object(BillingService, "_send_quota_request") as mock:
             yield mock
 
     def test_quota_reserve_success(self, mock_send_request):
@@ -652,6 +684,16 @@ class TestBillingServiceQuotaOperations:
         call_json = mock_send_request.call_args[1]["json"]
         assert call_json["meta"] == {"source": "webhook"}
 
+    def test_quota_reserve_with_bucket(self, mock_send_request):
+        mock_send_request.return_value = {"reservation_id": "rid-2", "available": 98, "reserved": 1}
+
+        BillingService.quota_reserve(
+            tenant_id="t1", feature_key="credit_pool", request_id="req-2", amount=1, bucket="trial"
+        )
+
+        call_json = mock_send_request.call_args[1]["json"]
+        assert call_json["bucket"] == "trial"
+
     def test_quota_commit_success(self, mock_send_request):
         expected = {"available": 98, "reserved": 0, "refunded": 0}
         mock_send_request.return_value = expected
@@ -696,6 +738,20 @@ class TestBillingServiceQuotaOperations:
         call_json = mock_send_request.call_args[1]["json"]
         assert call_json["meta"] == {"reason": "partial"}
 
+    def test_quota_commit_with_bucket(self, mock_send_request):
+        mock_send_request.return_value = {"available": 97, "reserved": 0, "refunded": 0}
+
+        BillingService.quota_commit(
+            tenant_id="t1",
+            feature_key="credit_pool",
+            reservation_id="rid-1",
+            actual_amount=1,
+            bucket="paid",
+        )
+
+        call_json = mock_send_request.call_args[1]["json"]
+        assert call_json["bucket"] == "paid"
+
     def test_quota_release_success(self, mock_send_request):
         expected = {"available": 100, "reserved": 0, "released": 1}
         mock_send_request.return_value = expected
@@ -720,6 +776,64 @@ class TestBillingServiceQuotaOperations:
         assert result["released"] == 1
         assert isinstance(result["released"], int)
 
+    def test_quota_release_with_bucket(self, mock_send_request):
+        mock_send_request.return_value = {"available": 100, "reserved": 0, "released": 1}
+
+        BillingService.quota_release(tenant_id="t1", feature_key="credit_pool", reservation_id="rid-1", bucket="trial")
+
+        call_json = mock_send_request.call_args[1]["json"]
+        assert call_json["bucket"] == "trial"
+
+    def test_quota_consume_capped_success(self, mock_send_request):
+        mock_send_request.return_value = {
+            "deducted": "2",
+            "available": "8",
+            "reserved": "0",
+            "quota": "10",
+            "usage": "2",
+        }
+
+        result = BillingService.quota_consume_capped(
+            tenant_id="t1",
+            feature_key="credit_pool",
+            request_id="req-1",
+            amount=5,
+            bucket="paid",
+            meta={"source": "test"},
+        )
+
+        assert result == {"deducted": 2, "available": 8, "reserved": 0, "quota": 10, "usage": 2}
+        mock_send_request.assert_called_once_with(
+            "POST",
+            "/quota/consume-capped",
+            json={
+                "tenant_id": "t1",
+                "feature_key": "credit_pool",
+                "request_id": "req-1",
+                "amount": 5,
+                "bucket": "paid",
+                "meta": {"source": "test"},
+            },
+        )
+
+    def test_send_quota_request_uses_quota_base_url(self):
+        with (
+            patch.object(BillingService, "quota_base_url", "https://quota.example.com/v1"),
+            patch.object(BillingService, "_send_request") as mock_send_request,
+        ):
+            mock_send_request.return_value = {"ok": True}
+
+            result = BillingService._send_quota_request("GET", "/quota/info", params={"tenant_id": "t1"})
+
+        assert result == {"ok": True}
+        mock_send_request.assert_called_once_with(
+            "GET",
+            "/quota/info",
+            json=None,
+            params={"tenant_id": "t1"},
+            base_url="https://quota.example.com/v1",
+        )
+
     def test_get_quota_info_coerces_string_to_int(self, mock_send_request):
         """Test that TypeAdapter coerces string values to int for get_quota_info."""
         mock_send_request.return_value = {

api/tests/unit_tests/services/test_credit_pool_service.py

@@ -1,7 +1,7 @@
 from collections.abc import Generator
 from contextlib import contextmanager
 from types import SimpleNamespace
-from unittest.mock import MagicMock, patch
+from unittest.mock import ANY, MagicMock, patch
 from uuid import uuid4
 
 import pytest
@@ -15,6 +15,8 @@ from models.enums import ProviderQuotaType
 from services.credit_pool_service import (
     CREDIT_POOL_TENANT_LOCK_BLOCKING_TIMEOUT_SECONDS,
     CREDIT_POOL_TENANT_LOCK_TIMEOUT_SECONDS,
+    FEATURE_KEY_CREDIT_POOL,
+    CreditPoolBalance,
     CreditPoolService,
 )
 
@@ -64,6 +66,12 @@ def _make_redis_lock() -> MagicMock:
     return lock
 
 
+@pytest.fixture(autouse=True)
+def _disable_billing_quota_by_default() -> Generator[None, None, None]:
+    with patch("services.credit_pool_service.dify_config.BILLING_ENABLED", False):
+        yield
+
+
 def test_get_pool_uses_configured_session_factory_without_flask_app_context() -> None:
     engine, tenant_id, _ = _create_engine_with_pool(quota_limit=10, quota_used=2)
 
@@ -75,6 +83,13 @@ def test_get_pool_uses_configured_session_factory_without_flask_app_context() ->
     assert pool.quota_used == 2
 
 
+def test_credit_pool_balance_unlimited_remaining_and_sufficiency() -> None:
+    pool = CreditPoolBalance(tenant_id="tenant-1", pool_type="paid", quota_limit=-1, quota_used=999)
+
+    assert pool.remaining_credits == -1
+    assert pool.has_sufficient_credits(10_000)
+
+
 def test_check_and_deduct_credits_deducts_exact_amount_when_sufficient() -> None:
     engine, tenant_id, pool_id = _create_engine_with_pool(quota_limit=10, quota_used=2)
 
@@ -224,7 +239,7 @@ def test_check_and_deduct_credits_uses_tenant_redis_lock_before_db_deduction() -
     )
     redis_lock.acquire.assert_called_once_with(blocking=True)
     redis_lock.release.assert_called_once_with()
-    get_locked_pool.assert_called_once_with(session=session, tenant_id=tenant_id, pool_type=ProviderQuotaType.TRIAL)
+    get_locked_pool.assert_called_once_with(session=session, tenant_id=tenant_id, pool_type="trial")
 
 
 def test_deduct_credits_capped_uses_tenant_redis_lock_before_db_deduction() -> None:
@@ -254,7 +269,152 @@ def test_deduct_credits_capped_uses_tenant_redis_lock_before_db_deduction() -> N
     )
     redis_lock.acquire.assert_called_once_with(blocking=True)
     redis_lock.release.assert_called_once_with()
-    get_locked_pool.assert_called_once_with(session=session, tenant_id=tenant_id, pool_type=ProviderQuotaType.PAID)
+    get_locked_pool.assert_called_once_with(session=session, tenant_id=tenant_id, pool_type="paid")
+
+
+def test_get_pool_uses_billing_quota_balance_when_enabled() -> None:
+    tenant_id = "tenant-1"
+    with (
+        patch("services.credit_pool_service.dify_config.BILLING_ENABLED", True),
+        patch("services.billing_service.BillingService.quota_get_balance") as quota_get_balance,
+    ):
+        quota_get_balance.return_value = {"quota": 1000, "usage": 250, "available": 750, "reserved": 0}
+
+        pool = CreditPoolService.get_pool(tenant_id=tenant_id, pool_type=ProviderQuotaType.PAID)
+
+    assert isinstance(pool, CreditPoolBalance)
+    assert pool.quota_limit == 1000
+    assert pool.quota_used == 250
+    assert pool.remaining_credits == 750
+    quota_get_balance.assert_called_once_with(
+        tenant_id=tenant_id,
+        feature_key=FEATURE_KEY_CREDIT_POOL,
+        bucket="paid",
+    )
+
+
+def test_check_and_deduct_credits_uses_billing_reserve_and_commit_when_enabled() -> None:
+    tenant_id = "tenant-1"
+    with (
+        patch("services.credit_pool_service.dify_config.BILLING_ENABLED", True),
+        patch("services.billing_service.BillingService.quota_reserve") as quota_reserve,
+        patch("services.billing_service.BillingService.quota_commit") as quota_commit,
+        patch("services.billing_service.BillingService.quota_release") as quota_release,
+    ):
+        quota_reserve.return_value = {"reservation_id": "reservation-1", "available": 7, "reserved": 3}
+
+        result = CreditPoolService.check_and_deduct_credits(
+            tenant_id=tenant_id,
+            credits_required=3,
+            pool_type=ProviderQuotaType.TRIAL,
+        )
+
+    assert result == 3
+    quota_reserve.assert_called_once_with(
+        tenant_id=tenant_id,
+        feature_key=FEATURE_KEY_CREDIT_POOL,
+        bucket="trial",
+        request_id=ANY,
+        amount=3,
+        meta={"source": "credit_pool.check_and_deduct"},
+    )
+    quota_commit.assert_called_once_with(
+        tenant_id=tenant_id,
+        feature_key=FEATURE_KEY_CREDIT_POOL,
+        bucket="trial",
+        reservation_id="reservation-1",
+        actual_amount=3,
+        meta={"source": "credit_pool.check_and_deduct"},
+    )
+    quota_release.assert_not_called()
+
+
+def test_check_and_deduct_credits_raises_when_billing_reserve_is_insufficient() -> None:
+    with (
+        patch("services.credit_pool_service.dify_config.BILLING_ENABLED", True),
+        patch("services.billing_service.BillingService.quota_reserve") as quota_reserve,
+    ):
+        quota_reserve.return_value = {"reservation_id": "", "available": 1, "reserved": 0}
+
+        with pytest.raises(QuotaExceededError, match="Insufficient credits remaining"):
+            CreditPoolService.check_and_deduct_credits(tenant_id="tenant-1", credits_required=3)
+
+
+def test_check_and_deduct_credits_releases_billing_reservation_when_commit_fails() -> None:
+    with (
+        patch("services.credit_pool_service.dify_config.BILLING_ENABLED", True),
+        patch("services.billing_service.BillingService.quota_reserve") as quota_reserve,
+        patch("services.billing_service.BillingService.quota_commit", side_effect=RuntimeError("commit failed")),
+        patch("services.billing_service.BillingService.quota_release") as quota_release,
+    ):
+        quota_reserve.return_value = {"reservation_id": "reservation-1", "available": 7, "reserved": 3}
+
+        with pytest.raises(RuntimeError, match="commit failed"):
+            CreditPoolService.check_and_deduct_credits(tenant_id="tenant-1", credits_required=3)
+
+    quota_release.assert_called_once_with(
+        tenant_id="tenant-1",
+        feature_key=FEATURE_KEY_CREDIT_POOL,
+        bucket="trial",
+        reservation_id="reservation-1",
+    )
+
+
+def test_check_and_deduct_credits_logs_when_billing_release_fails() -> None:
+    with (
+        patch("services.credit_pool_service.dify_config.BILLING_ENABLED", True),
+        patch("services.billing_service.BillingService.quota_reserve") as quota_reserve,
+        patch("services.billing_service.BillingService.quota_commit", side_effect=RuntimeError("commit failed")),
+        patch(
+            "services.billing_service.BillingService.quota_release", side_effect=RuntimeError("release failed")
+        ) as quota_release,
+        patch("services.credit_pool_service.logger.warning") as logger_warning,
+    ):
+        quota_reserve.return_value = {"reservation_id": "reservation-1", "available": 7, "reserved": 3}
+
+        with pytest.raises(RuntimeError, match="commit failed"):
+            CreditPoolService.check_and_deduct_credits(tenant_id="tenant-1", credits_required=3)
+
+    quota_release.assert_called_once_with(
+        tenant_id="tenant-1",
+        feature_key=FEATURE_KEY_CREDIT_POOL,
+        bucket="trial",
+        reservation_id="reservation-1",
+    )
+    logger_warning.assert_called_once()
+    assert logger_warning.call_args.args[3] == "reservation-1"
+    assert logger_warning.call_args.kwargs["exc_info"] is True
+
+
+def test_deduct_credits_capped_uses_billing_consume_capped_when_enabled() -> None:
+    tenant_id = "tenant-1"
+    with (
+        patch("services.credit_pool_service.dify_config.BILLING_ENABLED", True),
+        patch("services.billing_service.BillingService.quota_consume_capped") as quota_consume_capped,
+    ):
+        quota_consume_capped.return_value = {
+            "deducted": 2,
+            "available": 0,
+            "reserved": 0,
+            "quota": 10,
+            "usage": 10,
+        }
+
+        result = CreditPoolService.deduct_credits_capped(
+            tenant_id=tenant_id,
+            credits_required=5,
+            pool_type=ProviderQuotaType.PAID,
+        )
+
+    assert result == 2
+    quota_consume_capped.assert_called_once_with(
+        tenant_id=tenant_id,
+        feature_key=FEATURE_KEY_CREDIT_POOL,
+        bucket="paid",
+        request_id=ANY,
+        amount=5,
+        meta={"source": "credit_pool.deduct_capped"},
+    )
 
 
 @pytest.mark.parametrize(

api/tests/unit_tests/services/test_feature_service_licensed_seats.py

@@ -0,0 +1,38 @@
+import pytest
+
+from services import feature_service as feature_service_module
+from services.feature_service import FeatureService, SystemFeatureModel
+
+_ENTERPRISE_INFO = {"License": {"licensedSeats": {"enabled": True, "limit": 3, "used": 1}}}
+
+
+def test_fulfill_params_from_enterprise_parses_licensed_seats(monkeypatch: pytest.MonkeyPatch):
+    """Authenticated fill copies the licensed-seat quota out of the enterprise payload."""
+    monkeypatch.setattr(
+        feature_service_module.EnterpriseService,
+        "get_info",
+        staticmethod(lambda: _ENTERPRISE_INFO),
+    )
+
+    features = SystemFeatureModel()
+    FeatureService._fulfill_params_from_enterprise(features, is_authenticated=True)
+
+    assert features.license.seats.enabled is True
+    assert features.license.seats.limit == 3
+    assert features.license.seats.size == 1
+
+
+def test_fulfill_params_from_enterprise_withholds_seats_when_unauthenticated(monkeypatch: pytest.MonkeyPatch):
+    """Seat counts are auth-gated: unauthenticated callers keep the zeroed default."""
+    monkeypatch.setattr(
+        feature_service_module.EnterpriseService,
+        "get_info",
+        staticmethod(lambda: _ENTERPRISE_INFO),
+    )
+
+    features = SystemFeatureModel()
+    FeatureService._fulfill_params_from_enterprise(features, is_authenticated=False)
+
+    assert features.license.seats.enabled is False
+    assert features.license.seats.limit == 0
+    assert features.license.seats.size == 0

api/tests/unit_tests/services/tools/test_api_tools_manage_service.py

@@ -0,0 +1,31 @@
+from unittest.mock import Mock
+
+from services.tools.api_tools_manage_service import ApiToolManageService
+
+
+def test_get_api_tool_provider_remote_schema_uses_ssrf_proxy_get(monkeypatch) -> None:
+    schema = """
+    {
+        "openapi": "3.0.0",
+        "info": {"title": "Demo API", "version": "1.0.0"},
+        "servers": [{"url": "https://api.example.com"}],
+        "paths": {}
+    }
+    """
+    url = "https://example.com/openapi.json"
+    expected_headers = {
+        "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko)"
+        " Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0",
+        "Accept": "*/*",
+    }
+    mock_get = Mock(return_value=Mock(status_code=200, text=schema))
+    mock_parser = Mock()
+
+    monkeypatch.setattr("services.tools.api_tools_manage_service.ssrf_proxy.get", mock_get)
+    monkeypatch.setattr(ApiToolManageService, "parser_api_schema", mock_parser)
+
+    result = ApiToolManageService.get_api_tool_provider_remote_schema("user-1", "tenant-1", url)
+
+    assert result == {"schema": schema}
+    mock_get.assert_called_once_with(url, headers=expected_headers, timeout=10)
+    mock_parser.assert_called_once_with(schema)

api/tests/unit_tests/tasks/test_initialize_created_app_rbac_access_task.py

@@ -0,0 +1,69 @@
+from unittest.mock import MagicMock
+
+import pytest
+
+APP_RBAC_QUEUE = "app_rbac"
+
+
+def test_initialize_created_app_rbac_access_task_uses_rbac_queue():
+    from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task
+
+    assert initialize_created_app_rbac_access_task.queue == APP_RBAC_QUEUE
+
+
+def test_initialize_created_app_rbac_access_task_batches_workspace_members(monkeypatch):
+    import tasks.initialize_created_app_rbac_access_task as task_module
+    from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task
+
+    monkeypatch.setattr(task_module.dify_config, "RBAC_ENABLED", True)
+    monkeypatch.setattr(
+        task_module.TenantService,
+        "iter_member_account_id_batches",
+        lambda tenant_id, batch_size, session: iter([["acct-1", "acct-2"], ["acct-3"]]),
+    )
+    replace_whitelist = MagicMock()
+    replace_user_access_policies = MagicMock()
+    monkeypatch.setattr(
+        task_module.enterprise_rbac_service.RBACService.AppAccess,
+        "replace_whitelist",
+        replace_whitelist,
+    )
+    monkeypatch.setattr(
+        task_module.enterprise_rbac_service.RBACService.AppAccess,
+        "replace_user_access_policies",
+        replace_user_access_policies,
+    )
+
+    initialize_created_app_rbac_access_task.run("tenant-1", "actor-1", "app-1")
+
+    replace_whitelist.assert_not_called()
+    assert replace_user_access_policies.call_count == 2
+    assert replace_user_access_policies.call_args_list[0].kwargs["payload"].account_ids == ["acct-1", "acct-2"]
+    assert replace_user_access_policies.call_args_list[1].kwargs["payload"].account_ids == ["acct-3"]
+    for call in replace_user_access_policies.call_args_list:
+        assert call.kwargs["payload"].access_policy_ids == [task_module.APP_RBAC_DEFAULT_ACCESS_POLICY_ID]
+
+
+def test_initialize_created_app_rbac_access_task_retries_on_failure(monkeypatch):
+    import tasks.initialize_created_app_rbac_access_task as task_module
+    from tasks.initialize_created_app_rbac_access_task import initialize_created_app_rbac_access_task
+
+    monkeypatch.setattr(task_module.dify_config, "RBAC_ENABLED", True)
+    monkeypatch.setattr(
+        task_module.TenantService,
+        "iter_member_account_id_batches",
+        lambda tenant_id, batch_size, session: iter([["acct-1"]]),
+    )
+    monkeypatch.setattr(
+        task_module.enterprise_rbac_service.RBACService.AppAccess,
+        "replace_user_access_policies",
+        MagicMock(side_effect=ConnectionError("RBAC unavailable")),
+    )
+    retry = MagicMock(return_value=RuntimeError("retry requested"))
+    monkeypatch.setattr(initialize_created_app_rbac_access_task, "retry", retry)
+
+    with pytest.raises(RuntimeError, match="retry requested"):
+        initialize_created_app_rbac_access_task.run("tenant-1", "actor-1", "app-1")
+
+    retry.assert_called_once()
+    assert isinstance(retry.call_args.kwargs["exc"], ConnectionError)

api/uv.lock

@@ -54,6 +54,7 @@ overrides = [
     { name = "cryptography", specifier = ">=49.0.0,<50.0.0" },
     { name = "litellm", specifier = ">=1.83.10,<2.0.0" },
     { name = "pyarrow", specifier = ">=23.0.1,<24.0.0" },
+    { name = "setuptools", specifier = ">=80.10.2,<81" },
 ]
 
 [[package]]
@@ -883,14 +884,14 @@ wheels = [
 
 [[package]]
 name = "click"
-version = "8.1.8"
+version = "8.4.2"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "colorama", marker = "sys_platform == 'win32'" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/b9/2e/0090cbf739cee7d23781ad4b89a9894a41538e4fcf4c31dcdd705b78eb8b/click-8.1.8.tar.gz", hash = "sha256:ed53c9d8990d83c2a27deae68e4ee337473f6330c040a31d4225c9574d16096a", size = 226593, upload-time = "2024-12-21T18:38:44.339Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/76/d4/81420972a676e8ffea40450d8c8c92943e7218a78fe9b64359836cc9876b/click-8.4.2.tar.gz", hash = "sha256:9a6cea6e60b17ebe0a44c5cc636d94f09bd66142c1cd7d8b4cd731c4917a15f6", size = 338000, upload-time = "2026-06-24T17:45:15.148Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/7e/d4/7ebdbd03970677812aac39c869717059dbb71a4cfc033ca6e5221787892c/click-8.1.8-py3-none-any.whl", hash = "sha256:63c132bbbed01578a06712a2d1f497bb62d9c1c0d329b7903a866228027263b2", size = 98188, upload-time = "2024-12-21T18:38:41.666Z" },
+    { url = "https://files.pythonhosted.org/packages/fb/e2/79c688af8b210d232694e31e59da9f6ec747bae31c3f5946e4e9b98860d5/click-8.4.2-py3-none-any.whl", hash = "sha256:e6f9f66136c816745b9d65817da91d61d957fb16e02e4dcd0552553c5a197b76", size = 119243, upload-time = "2026-06-24T17:45:13.73Z" },
 ]
 
 [[package]]
@@ -1374,6 +1375,7 @@ dependencies = [
     { name = "resend" },
     { name = "sendgrid" },
     { name = "sseclient-py" },
+    { name = "zstandard" },
 ]
 
 [package.dev-dependencies]
@@ -1641,7 +1643,7 @@ requires-dist = [
     { name = "gunicorn", specifier = ">=26.0.0,<27.0.0" },
     { name = "httpx", extras = ["socks"], specifier = "==0.28.1" },
     { name = "httpx-sse", specifier = "==0.4.3" },
-    { name = "json-repair", specifier = "==0.59.4" },
+    { name = "json-repair", specifier = "==0.60.1" },
     { name = "opentelemetry-distro", specifier = "==0.62b1" },
     { name = "opentelemetry-instrumentation-celery", specifier = "==0.62b1" },
     { name = "opentelemetry-instrumentation-flask", specifier = "==0.62b1" },
@@ -1657,6 +1659,7 @@ requires-dist = [
     { name = "resend", specifier = ">=2.27.0,<3.0.0" },
     { name = "sendgrid", specifier = ">=6.12.5,<7.0.0" },
     { name = "sseclient-py", specifier = ">=1.8.0,<2.0.0" },
+    { name = "zstandard", specifier = "==0.25.0" },
 ]
 
 [package.metadata.requires-dev]
@@ -2459,11 +2462,11 @@ wheels = [
 
 [[package]]
 name = "fickling"
-version = "0.1.10"
+version = "0.1.12"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/9f/06/1818b8f52267599e54041349c553d5894e17ec8a539a246eb3f9eaf05629/fickling-0.1.10.tar.gz", hash = "sha256:8c8b76abd29936f1a5932e4087b8c8becb2d7ab1cf08549e63519ebcb2f71644", size = 338062, upload-time = "2026-03-13T16:34:29.287Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/d7/20/d3c2bdb9235b777763a4afc7cc3673afcd162a707ec0988ae7141a540802/fickling-0.1.12.tar.gz", hash = "sha256:83f6ccc948e21edb9ebd92795069536b47f481ce6add62598eac608b31576821", size = 357026, upload-time = "2026-06-26T23:55:57.829Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/05/86/620960dff970da5311f05e25fc045dac8495557d51030e5a0827084b18fd/fickling-0.1.10-py3-none-any.whl", hash = "sha256:962c35c38ece1b3632fc119c0f4cb1eebc02dc6d65bfd93a1803afd42ca91d25", size = 52853, upload-time = "2026-03-13T16:34:27.821Z" },
+    { url = "https://files.pythonhosted.org/packages/ba/c0/c65003b71abc4ffddbae99ae86952a21ff859d81b0f1331f79239fe5a58e/fickling-0.1.12-py3-none-any.whl", hash = "sha256:6232b72857e6ee9d729922811b681132d49dc39abc896581390dad1c0eada814", size = 58960, upload-time = "2026-06-26T23:55:56.401Z" },
 ]
 
 [[package]]
@@ -3279,14 +3282,14 @@ wheels = [
 
 [[package]]
 name = "httplib2"
-version = "0.31.0"
+version = "0.32.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "pyparsing" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/52/77/6653db69c1f7ecfe5e3f9726fdadc981794656fcd7d98c4209fecfea9993/httplib2-0.31.0.tar.gz", hash = "sha256:ac7ab497c50975147d4f7b1ade44becc7df2f8954d42b38b3d69c515f531135c", size = 250759, upload-time = "2025-09-11T12:16:03.403Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/84/f5/ccf58de92d61e3ad921119668f54ed36ca1d0cf5dcc5c1657dfb164fd78b/httplib2-0.32.0.tar.gz", hash = "sha256:48a0ef30a42db65d8f3399045e1d09ab0ba66e3b9efc360d07f80ea55d286025", size = 254283, upload-time = "2026-06-26T10:13:56.265Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/8c/a2/0d269db0f6163be503775dc8b6a6fa15820cc9fdc866f6ba608d86b721f2/httplib2-0.31.0-py3-none-any.whl", hash = "sha256:b9cd78abea9b4e43a7714c6e0f8b6b8561a6fc1e95d5dbd367f5bf0ef35f5d24", size = 91148, upload-time = "2025-09-11T12:16:01.803Z" },
+    { url = "https://files.pythonhosted.org/packages/33/a0/550eec327e5f5c7b732531c489f5307efec41f047b0d703bd4ca1e5ad2db/httplib2-0.32.0-py3-none-any.whl", hash = "sha256:dc6705cacdf3fb0a2aba7629fa33c90fd93e30035db0c157325826be177e4816", size = 93148, upload-time = "2026-06-26T10:13:54.985Z" },
 ]
 
 [[package]]
@@ -3554,11 +3557,11 @@ wheels = [
 
 [[package]]
 name = "json-repair"
-version = "0.59.4"
+version = "0.60.1"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/32/41/4ae9c6e711647a41b4e0c04bce815113ce9c0286eff6dc6fb86979b2fb9f/json_repair-0.59.4.tar.gz", hash = "sha256:559ca1828f6f566530663cd96d64bee29f8282b9d2ff0e661e05fa87b4171ab3", size = 47624, upload-time = "2026-04-15T06:48:40.776Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/5e/a6/d69888cb4ffde30e80db1e6c32caaadd2f984a80067d5ea72c2cb3f61c3f/json_repair-0.60.1.tar.gz", hash = "sha256:841661cdd2df507c9a4e189097f38ca6bc372e06d4b4e36d72e590f68176c290", size = 49451, upload-time = "2026-06-03T17:28:44.451Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/74/c4/ec3068436d2275731539b7a43fbc947f502bc3fe149856a5d00368c7b087/json_repair-0.59.4-py3-none-any.whl", hash = "sha256:46052e646bc0b0c39db672ebbf732f774f3c1a5bde81a54f0b0e19d3af4f45cd", size = 46697, upload-time = "2026-04-15T06:48:39.61Z" },
+    { url = "https://files.pythonhosted.org/packages/32/1f/2a2b5eea8ef5762a86ad3f8fddddaaba2c0d76dd44e644b9158900868bec/json_repair-0.60.1-py3-none-any.whl", hash = "sha256:ba6ff974f2a8bef2f7768144a7f03f870a816443f03da27a49cdd0ec31a78049", size = 48045, upload-time = "2026-06-03T17:28:43.038Z" },
 ]
 
 [[package]]
@@ -4790,21 +4793,19 @@ wheels = [
 
 [[package]]
 name = "pillow"
-version = "12.2.0"
+version = "12.3.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/8c/21/c2bcdd5906101a30244eaffc1b6e6ce71a31bd0742a01eb89e660ebfac2d/pillow-12.2.0.tar.gz", hash = "sha256:a830b1a40919539d07806aa58e1b114df53ddd43213d9c8b75847eee6c0182b5", size = 46987819, upload-time = "2026-04-01T14:46:17.687Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/1c/3d/bb7fca845737cf9d7dbde16ed1843984665ff2e0a518f5db43e77ec540b9/pillow-12.3.0.tar.gz", hash = "sha256:3b8182a766685eaa002637e28b4ec8d6b18819a0c71f579bf0dbaa5830297cce", size = 47025035, upload-time = "2026-07-01T11:56:38.965Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/58/be/7482c8a5ebebbc6470b3eb791812fff7d5e0216c2be3827b30b8bb6603ed/pillow-12.2.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:2d192a155bbcec180f8564f693e6fd9bccff5a7af9b32e2e4bf8c9c69dbad6b5", size = 5308279, upload-time = "2026-04-01T14:43:13.246Z" },
-    { url = "https://files.pythonhosted.org/packages/d8/95/0a351b9289c2b5cbde0bacd4a83ebc44023e835490a727b2a3bd60ddc0f4/pillow-12.2.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:f3f40b3c5a968281fd507d519e444c35f0ff171237f4fdde090dd60699458421", size = 4695490, upload-time = "2026-04-01T14:43:15.584Z" },
-    { url = "https://files.pythonhosted.org/packages/de/af/4e8e6869cbed569d43c416fad3dc4ecb944cb5d9492defaed89ddd6fe871/pillow-12.2.0-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:03e7e372d5240cc23e9f07deca4d775c0817bffc641b01e9c3af208dbd300987", size = 6284462, upload-time = "2026-04-01T14:43:18.268Z" },
-    { url = "https://files.pythonhosted.org/packages/e9/9e/c05e19657fd57841e476be1ab46c4d501bffbadbafdc31a6d665f8b737b6/pillow-12.2.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:b86024e52a1b269467a802258c25521e6d742349d760728092e1bc2d135b4d76", size = 8094744, upload-time = "2026-04-01T14:43:20.716Z" },
-    { url = "https://files.pythonhosted.org/packages/2b/54/1789c455ed10176066b6e7e6da1b01e50e36f94ba584dc68d9eebfe9156d/pillow-12.2.0-cp312-cp312-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:7371b48c4fa448d20d2714c9a1f775a81155050d383333e0a6c15b1123dda005", size = 6398371, upload-time = "2026-04-01T14:43:23.443Z" },
-    { url = "https://files.pythonhosted.org/packages/43/e3/fdc657359e919462369869f1c9f0e973f353f9a9ee295a39b1fea8ee1a77/pillow-12.2.0-cp312-cp312-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:62f5409336adb0663b7caa0da5c7d9e7bdbaae9ce761d34669420c2a801b2780", size = 7087215, upload-time = "2026-04-01T14:43:26.758Z" },
-    { url = "https://files.pythonhosted.org/packages/8b/f8/2f6825e441d5b1959d2ca5adec984210f1ec086435b0ed5f52c19b3b8a6e/pillow-12.2.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:01afa7cf67f74f09523699b4e88c73fb55c13346d212a59a2db1f86b0a63e8c5", size = 6509783, upload-time = "2026-04-01T14:43:29.56Z" },
-    { url = "https://files.pythonhosted.org/packages/67/f9/029a27095ad20f854f9dba026b3ea6428548316e057e6fc3545409e86651/pillow-12.2.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:fc3d34d4a8fbec3e88a79b92e5465e0f9b842b628675850d860b8bd300b159f5", size = 7212112, upload-time = "2026-04-01T14:43:32.091Z" },
-    { url = "https://files.pythonhosted.org/packages/be/42/025cfe05d1be22dbfdb4f264fe9de1ccda83f66e4fc3aac94748e784af04/pillow-12.2.0-cp312-cp312-win32.whl", hash = "sha256:58f62cc0f00fd29e64b29f4fd923ffdb3859c9f9e6105bfc37ba1d08994e8940", size = 6378489, upload-time = "2026-04-01T14:43:34.601Z" },
-    { url = "https://files.pythonhosted.org/packages/5d/7b/25a221d2c761c6a8ae21bfa3874988ff2583e19cf8a27bf2fee358df7942/pillow-12.2.0-cp312-cp312-win_amd64.whl", hash = "sha256:7f84204dee22a783350679a0333981df803dac21a0190d706a50475e361c93f5", size = 7084129, upload-time = "2026-04-01T14:43:37.213Z" },
-    { url = "https://files.pythonhosted.org/packages/10/e1/542a474affab20fd4a0f1836cb234e8493519da6b76899e30bcc5d990b8b/pillow-12.2.0-cp312-cp312-win_arm64.whl", hash = "sha256:af73337013e0b3b46f175e79492d96845b16126ddf79c438d7ea7ff27783a414", size = 2463612, upload-time = "2026-04-01T14:43:39.421Z" },
+    { url = "https://files.pythonhosted.org/packages/37/bf/fb3ebff8ddcb76aac5a01389251bbbb9519922a9b520d8247c1ca864a25d/pillow-12.3.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:ba09209fbe443b4acccebe845d8a138b89a8f4fbaeedd44953490b5315d5e965", size = 5345969, upload-time = "2026-07-01T11:54:06.397Z" },
+    { url = "https://files.pythonhosted.org/packages/d8/66/9a386a92561f402389a4fc70c18838bf6d35eb5eb5c6850b4b2dc64f5048/pillow-12.3.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:ffd0c5368496f41b0944be820fcb7a838aa6e623d250b01acf2643939c3f99d7", size = 4780323, upload-time = "2026-07-01T11:54:09.351Z" },
+    { url = "https://files.pythonhosted.org/packages/25/27/ac8f99618ffd3dde21db0f4d4b1d2ab00c0880595bfd17df103f7f39fd0c/pillow-12.3.0-cp312-cp312-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:d9c7f76c0673154f044e9d78c8655fb4213f6ca31a836df48b40fe5d187717b9", size = 6266838, upload-time = "2026-07-01T11:54:11.71Z" },
+    { url = "https://files.pythonhosted.org/packages/84/21/a35af28dcc61f37ed850a2d64c65c701321dfbf25085e469d5559360cbbf/pillow-12.3.0-cp312-cp312-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:78cb2c6865a35ab8ff8b75fd122f6033b92a62c82801110e48ddd6c936a45d91", size = 6940830, upload-time = "2026-07-01T11:54:13.732Z" },
+    { url = "https://files.pythonhosted.org/packages/eb/51/8b08617af3ad95e33ce6d7dd2c99ed6c8298f7fb131636303956be022e25/pillow-12.3.0-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:e491916b378fba47242221bb9ead245211b70d504f495d105d17b14a24b4907c", size = 6344383, upload-time = "2026-07-01T11:54:15.756Z" },
+    { url = "https://files.pythonhosted.org/packages/1d/72/cf78ac9780bb93c28328f408973845a309d4d145041665f734572ced1b52/pillow-12.3.0-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:0dd2064cbc55aaec028ef5fbb60fa47bb6c3e7918e07ff17935284b227a9d2df", size = 7052934, upload-time = "2026-07-01T11:54:17.721Z" },
+    { url = "https://files.pythonhosted.org/packages/20/20/25e0f4dc178a6bc0696793720055519a0de89e7661dae886992decbd2f81/pillow-12.3.0-cp312-cp312-win32.whl", hash = "sha256:dbce0b29841537a2fa4a214c2bbf14de3587c9680caa9b4e217568472490b28f", size = 6472684, upload-time = "2026-07-01T11:54:19.839Z" },
+    { url = "https://files.pythonhosted.org/packages/45/89/da2f7971a317f83d807fdd4065c0af40208e59e692cc43d315a71a0e96d1/pillow-12.3.0-cp312-cp312-win_amd64.whl", hash = "sha256:a2b55dd6b2a4c4b7d87ffa56bdb33fdc5fdb9a462173861a7bc097f17d91cb09", size = 7227137, upload-time = "2026-07-01T11:54:22.025Z" },
+    { url = "https://files.pythonhosted.org/packages/de/47/4845a0a6c0dbf1db8456bd9fc791f13c5ced7ced20606d08a0aacfd25b49/pillow-12.3.0-cp312-cp312-win_arm64.whl", hash = "sha256:331b624368d4f1d069149002f25f44bc61c8919ce8ddb3c45bdad8f6e2d89510", size = 2568267, upload-time = "2026-07-01T11:54:24.051Z" },
 ]
 
 [[package]]
@@ -5527,23 +5528,24 @@ wheels = [
 
 [[package]]
 name = "python-calamine"
-version = "0.5.4"
+version = "0.8.2"
 source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/10/36/473a55063a6af3739cdf4b879869d3067e677f3dfb52ec8fadac23941a33/python_calamine-0.8.2.tar.gz", hash = "sha256:b2000c085722afd01d973af3d58325d26cfd798b3665bb0924e3a658351ebfad", size = 227187, upload-time = "2026-07-13T13:58:26.909Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/ff/60/b1ace7a0fd636581b3bb27f1011cb7b2fe4d507b58401c4d328cfcb5c849/python_calamine-0.5.4-cp312-cp312-macosx_10_12_x86_64.whl", hash = "sha256:4d711f91283d28f19feb111ed666764de69e6d2a0201df8f84e81a238f68d193", size = 850087, upload-time = "2025-10-21T07:11:17.002Z" },
-    { url = "https://files.pythonhosted.org/packages/7f/32/32ca71ce50f9b7c7d6e7ec5fcc579a97ddd8b8ce314fe143ba2a19441dc7/python_calamine-0.5.4-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:ed67afd3adedb5bcfb428cf1f2d7dfd936dea9fe979ab631194495ab092973ba", size = 825659, upload-time = "2025-10-21T07:11:18.248Z" },
-    { url = "https://files.pythonhosted.org/packages/63/c5/27ba71a9da2a09be9ff2f0dac522769956c8c89d6516565b21c9c78bfae6/python_calamine-0.5.4-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:13662895dac487315ccce25ea272a1ea7e7ac05d899cde4e33d59d6c43274c54", size = 897332, upload-time = "2025-10-21T07:11:19.89Z" },
-    { url = "https://files.pythonhosted.org/packages/5a/e7/c4be6ff8e8899ace98cacc9604a2dd1abc4901839b733addfb6ef32c22ba/python_calamine-0.5.4-cp312-cp312-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:23e354755583cfaa824ddcbe8b099c5c7ac19bf5179320426e7a88eea2f14bc5", size = 886885, upload-time = "2025-10-21T07:11:21.912Z" },
-    { url = "https://files.pythonhosted.org/packages/38/24/80258fb041435021efa10d0b528df6842e442585e48cbf130e73fed2529b/python_calamine-0.5.4-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:4e1bc3f22107dcbdeb32d4d3c5c1e8831d3c85d4b004a8606dd779721b29843d", size = 1043907, upload-time = "2025-10-21T07:11:23.3Z" },
-    { url = "https://files.pythonhosted.org/packages/f2/20/157340787d03ef6113a967fd8f84218e867ba4c2f7fc58cc645d8665a61a/python_calamine-0.5.4-cp312-cp312-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:182b314117e47dbd952adaa2b19c515555083a48d6f9146f46faaabd9dab2f81", size = 942376, upload-time = "2025-10-21T07:11:24.866Z" },
-    { url = "https://files.pythonhosted.org/packages/98/f5/aec030f567ee14c60b6fc9028a78767687f484071cb080f7cfa328d6496e/python_calamine-0.5.4-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:b8f882e092ab23f72ea07e2e48f5f2efb1885c1836fb949f22fd4540ae11742e", size = 906455, upload-time = "2025-10-21T07:11:26.203Z" },
-    { url = "https://files.pythonhosted.org/packages/29/58/4affc0d1389f837439ad45f400f3792e48030b75868ec757e88cb35d7626/python_calamine-0.5.4-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:62a9b4b7b9bd99d03373e58884dfb60d5a1c292c8e04e11f8b7420b77a46813e", size = 948132, upload-time = "2025-10-21T07:11:27.507Z" },
-    { url = "https://files.pythonhosted.org/packages/b4/2e/70ed04f39e682a9116730f56b7fbb54453244ccc1c3dae0662d4819f1c1d/python_calamine-0.5.4-cp312-cp312-musllinux_1_1_aarch64.whl", hash = "sha256:98bb011d33c0e2d183ff30ab3d96792c3493f56f67a7aa2fcadad9a03539e79b", size = 1077436, upload-time = "2025-10-21T07:11:28.801Z" },
-    { url = "https://files.pythonhosted.org/packages/cb/ce/806f8ce06b5bb9db33007f85045c304cda410970e7aa07d08f6eaee67913/python_calamine-0.5.4-cp312-cp312-musllinux_1_1_armv7l.whl", hash = "sha256:6b218a95489ff2f1cc1de0bba2a16fcc82981254bbb23f31d41d29191282b9ad", size = 1150570, upload-time = "2025-10-21T07:11:30.237Z" },
-    { url = "https://files.pythonhosted.org/packages/18/da/61f13c8d107783128c1063cf52ca9cacdc064c58d58d3cf49c1728ce8296/python_calamine-0.5.4-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:e8296a4872dbe834205d25d26dd6cfcb33ee9da721668d81b21adc25a07c07e4", size = 1080286, upload-time = "2025-10-21T07:11:31.564Z" },
-    { url = "https://files.pythonhosted.org/packages/99/85/c5612a63292eb7d0648b17c5ff32ad5d6c6f3e1d78825f01af5c765f4d3f/python_calamine-0.5.4-cp312-cp312-win32.whl", hash = "sha256:cebb9c88983ae676c60c8c02aa29a9fe13563f240579e66de5c71b969ace5fd9", size = 676617, upload-time = "2025-10-21T07:11:32.833Z" },
-    { url = "https://files.pythonhosted.org/packages/bb/18/5a037942de8a8df0c805224b2fba06df6d25c1be3c9484ba9db1ca4f3ee6/python_calamine-0.5.4-cp312-cp312-win_amd64.whl", hash = "sha256:15abd7aff98fde36d7df91ac051e86e66e5d5326a7fa98d54697afe95a613501", size = 721464, upload-time = "2025-10-21T07:11:34.383Z" },
-    { url = "https://files.pythonhosted.org/packages/d1/8b/89ca17b44bcd8be5d0e8378d87b880ae17a837573553bd2147cceca7e759/python_calamine-0.5.4-cp312-cp312-win_arm64.whl", hash = "sha256:1cef0d0fc936974020a24acf1509ed2a285b30a4e1adf346c057112072e84251", size = 687268, upload-time = "2025-10-21T07:11:36.324Z" },
+    { url = "https://files.pythonhosted.org/packages/13/18/adfb951c33dedf37c3c9f5804af04eb8a18ba12dc0d3901c98563462f565/python_calamine-0.8.2-cp312-cp312-macosx_10_12_x86_64.whl", hash = "sha256:0efea8e1fc617d43351abe7c9f8ce7585b3a546d08e5c26d87d1629369a4b6e5", size = 864109, upload-time = "2026-07-13T13:56:31.591Z" },
+    { url = "https://files.pythonhosted.org/packages/c4/5e/d9b2cee430feb7eeee2d84b19e61266cda0e2880d126441a7f19b31772c7/python_calamine-0.8.2-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:a99eb1a74f9a93af4c5a5586e9fa72de1879721c47fd21aa9e042deb6832e30a", size = 846205, upload-time = "2026-07-13T13:56:33.282Z" },
+    { url = "https://files.pythonhosted.org/packages/de/a5/4cbd1a01279372a20319629ddc832806914827bd1edc8801ee7956c275fe/python_calamine-0.8.2-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:96daed873be1f70d22795fbf197d3d833655aad6a149c15154651b7773d3a1ad", size = 923414, upload-time = "2026-07-13T13:56:34.952Z" },
+    { url = "https://files.pythonhosted.org/packages/45/1b/9d646373c241082ce4488a0a297f8834a2513e17c6ce098506f91bb0f4c4/python_calamine-0.8.2-cp312-cp312-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:17314a59fe01bb1bb7010240a578bf85e49b47af955622ae5f06f286d6d6954f", size = 918317, upload-time = "2026-07-13T13:56:36.467Z" },
+    { url = "https://files.pythonhosted.org/packages/34/0d/73f4bcba43fc7a89ec48cbe1459d0436d6205a85ce720af54682bd14116e/python_calamine-0.8.2-cp312-cp312-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:1396287f66a54ec2e68c6448f0c782173dbb3e8c2aca4089195d8c9a52b46c72", size = 1087802, upload-time = "2026-07-13T13:56:38.113Z" },
+    { url = "https://files.pythonhosted.org/packages/47/54/b71763efd1c868d096fbad6f080de39ac83d832b456bc0ff285e74237723/python_calamine-0.8.2-cp312-cp312-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:da1b26061783d2bc1d7aabc83cce63706cdacb6b03c6c6a37d3ea33ad34b7b4b", size = 986451, upload-time = "2026-07-13T13:56:39.743Z" },
+    { url = "https://files.pythonhosted.org/packages/27/61/deb1b368933ddedeedbd7bef8aa71185eaff66dc058b44ca9f4a419925d5/python_calamine-0.8.2-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:5a148bf41efabb39ee99eac2552223c3605d29beaed6bd7a14a2990e07022e3b", size = 920176, upload-time = "2026-07-13T13:56:41.308Z" },
+    { url = "https://files.pythonhosted.org/packages/c9/df/7645e4d3ea983cac1bfe2643661fd03b92db2736cd97ba1bda367d86a61e/python_calamine-0.8.2-cp312-cp312-manylinux_2_5_i686.manylinux1_i686.whl", hash = "sha256:21afb508d71f83c12c1c01fef21d9285f91820d052998e44706571bebd835cf6", size = 996879, upload-time = "2026-07-13T13:56:42.951Z" },
+    { url = "https://files.pythonhosted.org/packages/af/b1/a88e1c7bc883f2767cc09247d395ce530c903a6bd2cc40e1f2000e6eaf85/python_calamine-0.8.2-cp312-cp312-musllinux_1_1_aarch64.whl", hash = "sha256:0cd1c5592b4a2cc7b0a92c5289f0eeef74e8ce3c5af9cc12361ae75625b14926", size = 1098763, upload-time = "2026-07-13T13:56:44.708Z" },
+    { url = "https://files.pythonhosted.org/packages/0c/59/de0fb08a1745d30d85cbef6b474e975a92bb7c6daf6aa37fc63960ae4674/python_calamine-0.8.2-cp312-cp312-musllinux_1_1_armv7l.whl", hash = "sha256:70e6e7aa5e74234254cae87e60a7ccb525e6962f73741f2b1927536aa7cb0093", size = 1192807, upload-time = "2026-07-13T13:56:46.379Z" },
+    { url = "https://files.pythonhosted.org/packages/bf/f5/d6956857490924ab9945d3be6624dadee1918d8630288009a2b7ef7700e1/python_calamine-0.8.2-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:2c2d65dd0cff571ad9463fc964194507e31295387bc1bbc76c2604959d694c0f", size = 1136033, upload-time = "2026-07-13T13:56:48.346Z" },
+    { url = "https://files.pythonhosted.org/packages/12/06/1992fa129dde6f7c5feed07d3b74df37a231782922af2ca6ff89917cc676/python_calamine-0.8.2-cp312-cp312-win32.whl", hash = "sha256:8938f66e6797d9d778504e0f827641d5067784dc68f18d2c2307abe615fcb85e", size = 734193, upload-time = "2026-07-13T13:56:50.546Z" },
+    { url = "https://files.pythonhosted.org/packages/63/31/aed4b50630ba23bb3372b4cebe930a92201dfd2c1c072e2d1570a042542e/python_calamine-0.8.2-cp312-cp312-win_amd64.whl", hash = "sha256:7a64237cc6b3d56b478686adff16502aecaf356543204bd6c70d754023a3d6aa", size = 774599, upload-time = "2026-07-13T13:56:52.278Z" },
+    { url = "https://files.pythonhosted.org/packages/84/c0/23b910999d2d9e1b9145fc3be645add1e1cbc90cba4a39a9ca9d23ebe678/python_calamine-0.8.2-cp312-cp312-win_arm64.whl", hash = "sha256:24607bde04b76f8329dae4963d22830a78f1eaa0622325217e53c5377143384c", size = 731385, upload-time = "2026-07-13T13:56:54.041Z" },
 ]
 
 [[package]]
@@ -5582,14 +5584,14 @@ wheels = [
 
 [[package]]
 name = "python-engineio"
-version = "4.13.1"
+version = "4.13.3"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "simple-websocket" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/34/12/bdef9dbeedbe2cdeba2a2056ad27b1fb081557d34b69a97f574843462cae/python_engineio-4.13.1.tar.gz", hash = "sha256:0a853fcef52f5b345425d8c2b921ac85023a04dfcf75d7b74696c61e940fd066", size = 92348, upload-time = "2026-02-06T23:38:06.12Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/fb/a0/f75491f942184d9960b15e763270f765fe9f239745ca5f9e16289011aed4/python_engineio-4.13.3.tar.gz", hash = "sha256:572b7783e341fed21edbc7cea297ccd378dad79265fdde96aa4664420a7c06c9", size = 79734, upload-time = "2026-06-20T22:53:52.197Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/aa/54/0cce26da03a981f949bb8449c9778537f75f5917c172e1d2992ff25cb57d/python_engineio-4.13.1-py3-none-any.whl", hash = "sha256:f32ad10589859c11053ad7d9bb3c9695cdf862113bfb0d20bc4d890198287399", size = 59847, upload-time = "2026-02-06T23:38:04.861Z" },
+    { url = "https://files.pythonhosted.org/packages/5b/96/82f6328e410515fab21d5602ba35b9377a47b5a141a0c1f9efa00ce21eb4/python_engineio-4.13.3-py3-none-any.whl", hash = "sha256:1f60ecaf1358190f0e26c48c578a60428dc02a8f1295bc3dbf53d1b31116821f", size = 59993, upload-time = "2026-06-20T22:53:50.775Z" },
 ]
 
 [[package]]
@@ -5650,15 +5652,15 @@ wheels = [
 
 [[package]]
 name = "python-socketio"
-version = "5.16.1"
+version = "5.16.3"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "bidict" },
     { name = "python-engineio" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/59/81/cf8284f45e32efa18d3848ed82cdd4dcc1b657b082458fbe01ad3e1f2f8d/python_socketio-5.16.1.tar.gz", hash = "sha256:f863f98eacce81ceea2e742f6388e10ca3cdd0764be21d30d5196470edf5ea89", size = 128508, upload-time = "2026-02-06T23:42:07Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/32/2d/ffce71017c106b75099fea569df6518c63fee5d6202ce0cfe7b01e6f22c3/python_socketio-5.16.3.tar.gz", hash = "sha256:89b136f677ae65607a84cecda9b4d6c5377b40a97582c504c25df89af16d520e", size = 128095, upload-time = "2026-06-15T22:07:04.003Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/07/c7/deb8c5e604404dbf10a3808a858946ca3547692ff6316b698945bb72177e/python_socketio-5.16.1-py3-none-any.whl", hash = "sha256:a3eb1702e92aa2f2b5d3ba00261b61f062cce51f1cfb6900bf3ab4d1934d2d35", size = 82054, upload-time = "2026-02-06T23:42:05.772Z" },
+    { url = "https://files.pythonhosted.org/packages/0a/38/8c5e72d53ff8eb27497c4f268a7f6d9121e727a50b65248288ad79a93053/python_socketio-5.16.3-py3-none-any.whl", hash = "sha256:e7ad14202a5e6448824c7c2f86161d04e13dec05992257df5c709e6a2798c041", size = 82087, upload-time = "2026-06-15T22:07:02.498Z" },
 ]
 
 [[package]]
@@ -6025,11 +6027,11 @@ wheels = [
 
 [[package]]
 name = "setuptools"
-version = "80.9.0"
+version = "80.10.2"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/18/5d/3bf57dcd21979b887f014ea83c24ae194cfcd12b9e0fda66b957c69d1fca/setuptools-80.9.0.tar.gz", hash = "sha256:f36b47402ecde768dbfafc46e8e4207b4360c654f1f3bb84475f0a28628fb19c", size = 1319958, upload-time = "2025-05-27T00:56:51.443Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/76/95/faf61eb8363f26aa7e1d762267a8d602a1b26d4f3a1e758e92cb3cb8b054/setuptools-80.10.2.tar.gz", hash = "sha256:8b0e9d10c784bf7d262c4e5ec5d4ec94127ce206e8738f29a437945fbc219b70", size = 1200343, upload-time = "2026-01-25T22:38:17.252Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/a3/dc/17031897dae0efacfea57dfd3a82fdd2a2aeb58e0ff71b77b87e44edc772/setuptools-80.9.0-py3-none-any.whl", hash = "sha256:062d34222ad13e0cc312a4c02d73f059e86a4acbfbdea8f8f76b28c99f306922", size = 1201486, upload-time = "2025-05-27T00:56:49.664Z" },
+    { url = "https://files.pythonhosted.org/packages/94/b8/f1f62a5e3c0ad2ff1d189590bfa4c46b4f3b6e49cef6f26c6ee4e575394d/setuptools-80.10.2-py3-none-any.whl", hash = "sha256:95b30ddfb717250edb492926c92b5221f7ef3fbcc2b07579bcd4a27da21d0173", size = 1064234, upload-time = "2026-01-25T22:38:15.216Z" },
 ]
 
 [[package]]
@@ -6112,11 +6114,11 @@ wheels = [
 
 [[package]]
 name = "soupsieve"
-version = "2.8"
+version = "2.8.4"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/6d/e6/21ccce3262dd4889aa3332e5a119a3491a95e8f60939870a3a035aabac0d/soupsieve-2.8.tar.gz", hash = "sha256:e2dd4a40a628cb5f28f6d4b0db8800b8f581b65bb380b97de22ba5ca8d72572f", size = 103472, upload-time = "2025-08-27T15:39:51.78Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/47/2c/0a5f6f8ee0d5589e48c7640213ed5175d52cf540a06725b628cc1a45d6ce/soupsieve-2.8.4.tar.gz", hash = "sha256:e121fd02e975c695e4e9e8774a5ee35d74714b59307868dcc5319ad2d9e3328e", size = 121110, upload-time = "2026-05-24T13:55:57.154Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/14/a0/bb38d3b76b8cae341dad93a2dd83ab7462e6dbcdd84d43f54ee60a8dc167/soupsieve-2.8-py3-none-any.whl", hash = "sha256:0cc76456a30e20f5d7f2e14a98a4ae2ee4e5abdc7c5ea0aafe795f344bc7984c", size = 36679, upload-time = "2025-08-27T15:39:50.179Z" },
+    { url = "https://files.pythonhosted.org/packages/5e/f5/0c41cb68dcae6b7de4fac4188a3a9589e21fb31df21ea3a2e888db95e6c9/soupsieve-2.8.4-py3-none-any.whl", hash = "sha256:e7e6b0769c8f51ed59acab6e994b00621096cfb1c640a7509295987388fbaf65", size = 37304, upload-time = "2026-05-24T13:55:55.406Z" },
 ]
 
 [[package]]
@@ -6533,7 +6535,7 @@ wheels = [
 
 [[package]]
 name = "transformers"
-version = "5.3.0"
+version = "5.5.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "huggingface-hub" },
@@ -6546,9 +6548,9 @@ dependencies = [
     { name = "tqdm" },
     { name = "typer" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/fc/1a/70e830d53ecc96ce69cfa8de38f163712d2b43ac52fbd743f39f56025c31/transformers-5.3.0.tar.gz", hash = "sha256:009555b364029da9e2946d41f1c5de9f15e6b1df46b189b7293f33a161b9c557", size = 8830831, upload-time = "2026-03-04T17:41:46.119Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/ff/9d/fb46e729b461985f41a5740167688b924a4019141e5c164bea77548d3d9e/transformers-5.5.0.tar.gz", hash = "sha256:c8db656cf51c600cd8c75f06b20ef85c72e8b8ff9abc880c5d3e8bc70e0ddcbd", size = 8237745, upload-time = "2026-04-02T16:13:08.113Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/b8/88/ae8320064e32679a5429a2c9ebbc05c2bf32cefb6e076f9b07f6d685a9b4/transformers-5.3.0-py3-none-any.whl", hash = "sha256:50ac8c89c3c7033444fb3f9f53138096b997ebb70d4b5e50a2e810bf12d3d29a", size = 10661827, upload-time = "2026-03-04T17:41:42.722Z" },
+    { url = "https://files.pythonhosted.org/packages/e7/28/35f7411ff80a3640c1f4fc907dcbb6a65061ebb82f66950e38bfc9f7f740/transformers-5.5.0-py3-none-any.whl", hash = "sha256:821a9ff0961abbb29eb1eb686d78df1c85929fdf213a3fe49dc6bd94f9efa944", size = 10245591, upload-time = "2026-04-02T16:13:03.462Z" },
 ]
 
 [[package]]
@@ -7286,7 +7288,7 @@ wheels = [
 
 [[package]]
 name = "wandb"
-version = "0.23.1"
+version = "0.28.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "click" },
@@ -7300,17 +7302,17 @@ dependencies = [
     { name = "sentry-sdk" },
     { name = "typing-extensions" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/0a/cc/770ae3aa7ae44f6792f7ecb81c14c0e38b672deb35235719bb1006519487/wandb-0.23.1.tar.gz", hash = "sha256:f6fb1e3717949b29675a69359de0eeb01e67d3360d581947d5b3f98c273567d6", size = 44298053, upload-time = "2025-12-03T02:25:10.79Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/5f/a7/683bfbd6cbade3012bc90d3e9c4cfc72dd62566195bf4c30321946d64b77/wandb-0.28.0.tar.gz", hash = "sha256:b20e5af0fe80e2e2a466b0466a1d60cedcc578dce0f036eca04f4a0adcad95b6", size = 40558332, upload-time = "2026-06-23T00:38:50.115Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/12/0b/c3d7053dfd93fd259a63c7818d9c4ac2ba0642ff8dc8db98662ea0cf9cc0/wandb-0.23.1-py3-none-macosx_12_0_arm64.whl", hash = "sha256:358e15471d19b7d73fc464e37371c19d44d39e433252ac24df107aff993a286b", size = 21527293, upload-time = "2025-12-03T02:24:48.011Z" },
-    { url = "https://files.pythonhosted.org/packages/ee/9f/059420fa0cb6c511dc5c5a50184122b6aca7b178cb2aa210139e354020da/wandb-0.23.1-py3-none-macosx_12_0_x86_64.whl", hash = "sha256:110304407f4b38f163bdd50ed5c5225365e4df3092f13089c30171a75257b575", size = 22745926, upload-time = "2025-12-03T02:24:50.519Z" },
-    { url = "https://files.pythonhosted.org/packages/96/b6/fd465827c14c64d056d30b4c9fcf4dac889a6969dba64489a88fc4ffa333/wandb-0.23.1-py3-none-manylinux_2_28_aarch64.whl", hash = "sha256:6cc984cf85feb2f8ee0451d76bc9fb7f39da94956bb8183e30d26284cf203b65", size = 21212973, upload-time = "2025-12-03T02:24:52.828Z" },
-    { url = "https://files.pythonhosted.org/packages/5c/ee/9a8bb9a39cc1f09c3060456cc79565110226dc4099a719af5c63432da21d/wandb-0.23.1-py3-none-manylinux_2_28_x86_64.whl", hash = "sha256:67431cd3168d79fdb803e503bd669c577872ffd5dadfa86de733b3274b93088e", size = 22887885, upload-time = "2025-12-03T02:24:55.281Z" },
-    { url = "https://files.pythonhosted.org/packages/6d/4d/8d9e75add529142e037b05819cb3ab1005679272950128d69d218b7e5b2e/wandb-0.23.1-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:07be70c0baa97ea25fadc4a9d0097f7371eef6dcacc5ceb525c82491a31e9244", size = 21250967, upload-time = "2025-12-03T02:24:57.603Z" },
-    { url = "https://files.pythonhosted.org/packages/97/72/0b35cddc4e4168f03c759b96d9f671ad18aec8bdfdd84adfea7ecb3f5701/wandb-0.23.1-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:216c95b08e0a2ec6a6008373b056d597573d565e30b43a7a93c35a171485ee26", size = 22988382, upload-time = "2025-12-03T02:25:00.518Z" },
-    { url = "https://files.pythonhosted.org/packages/c0/6d/e78093d49d68afb26f5261a70fc7877c34c114af5c2ee0ab3b1af85f5e76/wandb-0.23.1-py3-none-win32.whl", hash = "sha256:fb5cf0f85692f758a5c36ab65fea96a1284126de64e836610f92ddbb26df5ded", size = 22150756, upload-time = "2025-12-03T02:25:02.734Z" },
-    { url = "https://files.pythonhosted.org/packages/05/27/4f13454b44c9eceaac3d6e4e4efa2230b6712d613ff9bf7df010eef4fd18/wandb-0.23.1-py3-none-win_amd64.whl", hash = "sha256:21c8c56e436eb707b7d54f705652e030d48e5cfcba24cf953823eb652e30e714", size = 22150760, upload-time = "2025-12-03T02:25:05.106Z" },
-    { url = "https://files.pythonhosted.org/packages/30/20/6c091d451e2a07689bfbfaeb7592d488011420e721de170884fedd68c644/wandb-0.23.1-py3-none-win_arm64.whl", hash = "sha256:8aee7f3bb573f2c0acf860f497ca9c684f9b35f2ca51011ba65af3d4592b77c1", size = 20137463, upload-time = "2025-12-03T02:25:08.317Z" },
+    { url = "https://files.pythonhosted.org/packages/c0/47/1723605f76c5d6446b6d0db65b83eda1599721bc8c1e65bd76cc1682b1a7/wandb-0.28.0-py3-none-macosx_12_0_arm64.whl", hash = "sha256:c3dab1205a5aca4abbad1eca08902cdba86add0edfa83d8d61b4429d0e79fa87", size = 24335272, upload-time = "2026-06-23T00:38:26.002Z" },
+    { url = "https://files.pythonhosted.org/packages/81/ff/42b539bc75bc48fc86981dccde89327ba9b71504b805b9ba42cba7c26de9/wandb-0.28.0-py3-none-macosx_12_0_x86_64.whl", hash = "sha256:ae255da18726ee8e731ef82cbc85035b901a28ae14cf91604c361b44b8d44ce0", size = 25557959, upload-time = "2026-06-23T00:38:28.993Z" },
+    { url = "https://files.pythonhosted.org/packages/15/55/c3db03d04aeab3726066a418b2ef6a1f8119774ee510f4fbe992f52b7472/wandb-0.28.0-py3-none-manylinux_2_28_aarch64.whl", hash = "sha256:6dbcba12ab168aa37561f2f32dcdef8713495fc25fa7d30fdc9bfb37989694dd", size = 24878557, upload-time = "2026-06-23T00:38:31.417Z" },
+    { url = "https://files.pythonhosted.org/packages/d8/5d/1385ce3c219cb5bd30d4027687e3f8d25969c7dfd09adad1cbd5080e1a72/wandb-0.28.0-py3-none-manylinux_2_28_x86_64.whl", hash = "sha256:325b2d0bd88be6eda5db10542499bad3710927f2569c81a84dc5eeaffc76825c", size = 26764727, upload-time = "2026-06-23T00:38:33.775Z" },
+    { url = "https://files.pythonhosted.org/packages/00/58/23b6c17a6d3d5422b007707961c4496b2f6f892624d2910c9f7742fcc202/wandb-0.28.0-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:8954bc1c62ae43914dce2bebfd1d9957f72350f8fbb78e5cdfe2ca9b6be8a7b8", size = 25051656, upload-time = "2026-06-23T00:38:36.281Z" },
+    { url = "https://files.pythonhosted.org/packages/89/67/9be00fb2db2281063af24a148636d2dd363d337317642ab5d8e93572c794/wandb-0.28.0-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:9fec6c908554c2dad33110c1312bc3028cc2e430f0679f16b84f82c8ea801e3b", size = 27074113, upload-time = "2026-06-23T00:38:38.737Z" },
+    { url = "https://files.pythonhosted.org/packages/59/b1/f7a96c09cab0c5131b1e6466659b093b401e1653cbe6bb77b462fc1c361d/wandb-0.28.0-py3-none-win32.whl", hash = "sha256:8834ef3a7c8c43b701654162783caa7ad37af48a0ff06fc35d0d65a411f76ccd", size = 24525206, upload-time = "2026-06-23T00:38:42.041Z" },
+    { url = "https://files.pythonhosted.org/packages/c6/c4/c7bed5e981679c74e9fbb22c03ff31c42e95f266199d03d8d325f4d0e6df/wandb-0.28.0-py3-none-win_amd64.whl", hash = "sha256:ac1f82292e2da4f98297b78c3a46726b3a6c5734ecb75fc39b8db2c8a4989159", size = 24525214, upload-time = "2026-06-23T00:38:44.549Z" },
+    { url = "https://files.pythonhosted.org/packages/f0/77/b5ce9696c8cb955521a7941fbc443e78b2f504894c6ae1a2d0b1de6e12ae/wandb-0.28.0-py3-none-win_arm64.whl", hash = "sha256:c5b0faf1b84cf79ebabed77538c1940a4c6053e815f767a4004e877a1354bed1", size = 22378208, upload-time = "2026-06-23T00:38:47.148Z" },
 ]
 
 [[package]]

e2e/features/auth/redirect-security.feature

@@ -0,0 +1,15 @@
+@auth @redirect-security
+Feature: Safe sign-in redirects
+
+  @authenticated
+  Scenario: Ignore an external redirect target for an authenticated user
+    Given I am signed in as the default E2E admin
+    When I open the sign-in page with redirect target "https://google.com"
+    Then I should be on the console home
+
+  @unauthenticated
+  Scenario: Ignore an external redirect target after signing in
+    Given I am not signed in
+    When I open the sign-in page with redirect target "https://google.com"
+    And I sign in as the default E2E admin
+    Then I should be on the console home

e2e/features/step-definitions/auth/sign-in.steps.ts

@@ -7,6 +7,15 @@ When('I open the sign-in page', async function (this: DifyWorld) {
   await this.getPage().goto('/signin?redirect_url=%2Fapps')
 })
 
+When(
+  'I open the sign-in page with redirect target {string}',
+  async function (this: DifyWorld, redirectTarget: string) {
+    const searchParams = new URLSearchParams({ redirect_url: redirectTarget })
+
+    await this.getPage().goto(`/signin?${searchParams}`)
+  },
+)
+
 When('I sign in as the default E2E admin', async function (this: DifyWorld) {
   const page = this.getPage()