Compare commits

...
Author SHA1 Message Date
Jingyi-Dify ad44b01ae1 perf(web): load integration tools by category 2026-07-24 17:40:26 -07:00
Jingyi-Dify baa8d2c358 fix(web): defer model status plugin lookup 2026-07-24 16:39:05 -07:00
Jingyi-Dify b7885fb83a fix(web): optimize data source plugin loading 2026-07-24 16:37:53 -07:00
Jingyi-Dify 46c5093d3e Merge remote-tracking branch 'origin/main' into feat/saas-integrations-initial-load-performance 2026-07-24 15:08:27 -07:00
yyhandGitHub 58f83fa7e7 refactor(web): remove obsolete route prefix handler (#39545) 2026-07-24 15:30:52 +00:00
Stephen ZhouGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2d5161a647 feat(dataset): add document detail (#39361)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-24 15:27:35 +00:00
yyhandGitHub 1227f19c6a fix(web): patch React Server Components DoS (#39539) 2026-07-24 14:00:26 +00:00
EvanandGitHub a0b917425a test: use caplog in ext login tests (#39542) 2026-07-24 13:27:05 +00:00
Stephen ZhouGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
799c7eea3a feat(dataset): add document processing tasks (#39326)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-24 12:58:15 +00:00
wangxiaoleiandGitHub 5277e4b375 fix: fix when tool execute in progress tool_meta miss tool_provider_type (#39513) 2026-07-24 11:35:12 +00:00
a877e1bd7e fix(agent): block unpublished agents in workflows (#39532)
Co-authored-by: yyh <92089059+lyzno1@users.noreply.github.com>
2026-07-24 11:02:43 +00:00
yyhandGitHub a5703762c0 test(e2e): wait for app list before menu interaction (#39538) 2026-07-24 10:37:22 +00:00
yyhandGitHub dcd40d603c fix(agent-v2): link API reference to agent guide (#39534) 2026-07-24 10:14:27 +00:00
JoelandGitHub 8ef002f624 fix: include CSRF token when fetching skill file content (#39527) 2026-07-24 09:35:41 +00:00
yyhandGitHub b1dec5bd42 chore: add web shell code owners (#39526) 2026-07-24 09:17:15 +00:00
yyhandGitHub 4e85acc605 fix(web): hydrate workspace permissions before navigation render (#39524) 2026-07-24 09:06:57 +00:00
Wu TianweiandGitHub a2c67372ed fix(web): allow clipboard writes in embedded apps (#39511) 2026-07-24 08:27:57 +00:00
林玮 (Jade Lin)andGitHub 7bb09ce039 chore(web): improve web accessibility semantics (#39505) 2026-07-24 08:16:04 +00:00
JoelandGitHub e1495c0bae fix: prevent agent preview actions from being clipped (#39510) 2026-07-24 08:02:55 +00:00
JyongandGitHub ededa1f4cd chore: add deploy-knowledge.yml (#39509) 2026-07-24 08:00:11 +00:00
JoelandGitHub b2861b60d8 fix: audio config ui problem (#39504) 2026-07-24 07:42:26 +00:00
3cbe49bf45 fix: don't open the node settings panel when clicking a node in comment mode (#39500)
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-24 07:38:01 +00:00
JoelandGitHub dd19b4ff79 fix: download skill files through authenticated API requests (#39499) 2026-07-24 07:37:27 +00:00
yyhGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>GareArc
66a545fc6d refactor: centralize deployment edition in system features (#39454)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: GareArc <garethcxy@dify.ai>
2026-07-24 07:15:02 +00:00
Stephen ZhouGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
eac9f69ad1 feat(dataset): add crawl source selection (#39325)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-24 07:11:48 +00:00
Yufeng HeandGitHub e4dc29f98e fix(api): make the 10-minute email IP first-strike window take effect (#39479)
Signed-off-by: Yufeng He <40085740+he-yufeng@users.noreply.github.com>
2026-07-24 06:52:30 +00:00
JoelandGitHub 6889974c05 fix: smooth the preview-to-build transition (#39495) 2026-07-24 06:30:07 +00:00
非法操作andGitHub aef1c67721 fix: return disabled auto-upgrade settings when strategy is missing (#39494) 2026-07-24 06:15:02 +00:00
David ParkGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
3ab9e083e0 fix: don't close the active comment panel when a different comment is resolved (#39491)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-24 06:11:25 +00:00
yyhandGitHub 6e2c21f568 fix(web): simplify agent configure mode state (#39492) 2026-07-24 05:30:45 +00:00
JoelandGitHub abbad7b313 fix: prevent build reset from clearing preview chat (#39490) 2026-07-24 04:59:57 +00:00
yyhandGitHub c4bfea6096 fix(web): adapt workflow agent logs (#39488) 2026-07-24 04:55:30 +00:00
Asuka MinatoandGitHub bbb133cd58 test: use sqlite3 session in test_base_trace_instance (#38744) 2026-07-24 03:55:32 +00:00
JoelGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
34e2205efa chore: support preview mode if not in community version (#39399)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-24 03:23:01 +00:00
wangxiaoleiandGitHub db29caff2b feat: knowledge add more trace (#38959) 2026-07-24 02:27:16 +00:00
14c50a9f09 feat(web): locate workflow nodes by node_id in the workflow editor (#38187)
Co-authored-by: Crazywoola <100913391+crazywoola@users.noreply.github.com>
2026-07-24 02:16:41 +00:00
Stephen ZhouGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
ba5b26be03 feat(dataset): add website crawl preview (#39324)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-24 02:10:39 +00:00
samzongGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
922ed4d67d feat(oauth): expose stable account id on provider account endpoint (#39470)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-24 02:09:48 +00:00
zyssyz123andGitHub a8a83a357c fix(agent): show workflow node runs in logs (#39471) 2026-07-24 01:59:57 +00:00
Jingyi-Dify c06fba49eb perf(web): reduce tool marketplace provider loading 2026-07-23 18:15:50 -07:00
Jingyi-Dify 41aa553461 feat(web): paginate installed integration plugins 2026-07-23 18:15:29 -07:00
autofix-ci[bot]andJingyi-Dify f3593fcbd2 [autofix.ci] apply automated fixes 2026-07-23 17:04:29 -07:00
hjlarryandJingyi-Dify 1738693484 feat: add category-scoped installed plugin ids 2026-07-23 17:04:29 -07:00
hjlarryandJingyi-Dify 9134443fda feat: plugin list query improve 2026-07-23 17:04:29 -07:00
Jingyi-Dify e91bb0cc94 Merge remote-tracking branch 'origin/main' into feat/saas-integrations-initial-load-performance 2026-07-23 16:11:14 -07:00
Stephen ZhouGitHubautofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
302b50c4fb feat(dataset): add source connections (#39319)
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
2026-07-23 13:54:12 +00:00
Byron.wangandGitHub 43254c1ded chore: avoid duplicate token counting during dataset indexing (#39466) 2026-07-23 10:56:03 +00:00
JoelandGitHub 2beafdc457 fix: chat not pass doc if not support vision (#39461) 2026-07-23 10:01:43 +00:00
zyssyz123andGitHub 0198e32447 fix(agent): start new Preview conversations without an ID (#39463) 2026-07-23 09:53:08 +00:00
samzongandGitHub b8f91d0e61 fix(web): echo OAuth state on authorize redirect (#39459)
Signed-off-by: samzong <samzong.lu@gmail.com>
2026-07-23 09:30:52 +00:00
非法操作andGitHub 452dff5c37 fix(api): prevent identity logging deadlock (#39449) 2026-07-23 09:08:04 +00:00
Jingyi-Dify 08a10f7593 Revert "docs(perf): record integrations load findings"
This reverts commit 08a7a17573.
2026-07-22 19:04:45 -07:00
Jingyi-Dify 08a7a17573 docs(perf): record integrations load findings 2026-07-22 19:01:33 -07:00
Jingyi-Dify d4a8111c34 perf(model-providers): defer secondary data loading 2026-07-22 19:01:22 -07:00
Jingyi-Dify 2ff450fde5 perf(plugins): defer plugin query loading 2026-07-22 19:01:03 -07:00
Jingyi-Dify 5a148aaeea perf(plugins): retain first page after leaving integrations 2026-07-22 12:48:11 -07:00
Jingyi-Dify 36d36954e6 perf(plugins): lazy load plugin icons 2026-07-22 11:16:49 -07:00
Jingyi-Dify 19b6b936e3 perf(integrations): paginate plugin category lists 2026-07-22 11:16:28 -07:00
645 changed files with 42477 additions and 5643 deletions
+17
View File
@@ -250,5 +250,22 @@
# Frontend - Workspace
/web/app/components/header/account-dropdown/workplace-selector/ @iamjoel @zxhlyh
# Frontend - App Shell and Console Bootstrap
/web/app/layout.tsx @iamjoel @lyzno1
/web/app/error.tsx @iamjoel @lyzno1
/web/app/(commonLayout)/layout.tsx @iamjoel @lyzno1
/web/app/(commonLayout)/providers.tsx @iamjoel @lyzno1
/web/app/(commonLayout)/hydration-boundary.tsx @iamjoel @lyzno1
/web/app/(commonLayout)/profile-bootstrap-gate.tsx @iamjoel @lyzno1
/web/app/(commonLayout)/error.tsx @iamjoel @lyzno1
/web/app/account/(commonLayout)/layout.tsx @iamjoel @lyzno1
/web/app/components/main-nav/* @iamjoel @lyzno1
/web/app/components/main-nav/components/* @iamjoel @lyzno1
/web/context/query-client.tsx @iamjoel @lyzno1
/web/context/query-client-server.ts @iamjoel @lyzno1
/web/proxy.ts @iamjoel @lyzno1
/web/app/auth/refresh/route.ts @iamjoel @lyzno1
/web/service/server.ts @iamjoel @lyzno1
# Docker
/docker/* @laipz8200
+28
View File
@@ -0,0 +1,28 @@
name: Deploy Knowledge
permissions:
contents: read
on:
workflow_run:
workflows: ["Build and Push API & Web"]
branches:
- "deploy/konwledge"
types:
- completed
jobs:
deploy:
runs-on: depot-ubuntu-24.04
if: |
github.event.workflow_run.conclusion == 'success' &&
github.event.workflow_run.head_branch == 'deploy/konwledge'
steps:
- name: Deploy to server
uses: appleboy/ssh-action@0ff4204d59e8e51228ff73bce53f80d53301dee2 # v1.2.5
with:
host: ${{ secrets.SSH_NEW_RAG_HOST }}
username: ${{ secrets.SSH_USER }}
key: ${{ secrets.SSH_PRIVATE_KEY }}
script: |
${{ vars.SSH_SCRIPT || secrets.SSH_SCRIPT }}
+2 -1
View File
@@ -6,6 +6,7 @@ from sqlalchemy import delete, select, update
from sqlalchemy.orm import sessionmaker
from configs import dify_config
from enums.deployment_edition import DeploymentEdition
from events.app_event import app_was_created
from extensions.ext_database import db
from extensions.ext_redis import redis_client
@@ -42,7 +43,7 @@ def reset_encrypt_key_pair():
After the reset, all LLM credentials will become invalid, requiring re-entry.
Only support SELF_HOSTED mode.
"""
if dify_config.EDITION != "SELF_HOSTED":
if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
click.echo(click.style("This command is only for SELF_HOSTED installations.", fg="red"))
return
with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
+9
View File
@@ -5,6 +5,7 @@ from typing import Any, override
from pydantic.fields import FieldInfo
from pydantic_settings import BaseSettings, PydanticBaseSettingsSource, SettingsConfigDict, TomlConfigSettingsSource
from enums.deployment_edition import DeploymentEdition
from libs.file_utils import search_file_upwards
from .deploy import DeploymentConfig
@@ -116,3 +117,11 @@ class DifyConfig(
),
),
)
@property
def DEPLOYMENT_EDITION(self) -> DeploymentEdition:
if self.EDITION == "CLOUD":
return DeploymentEdition.CLOUD
if self.ENTERPRISE_ENABLED:
return DeploymentEdition.ENTERPRISE
return DeploymentEdition.COMMUNITY
+5
View File
@@ -1497,6 +1497,11 @@ class LoginConfig(BaseSettings):
class AccountConfig(BaseSettings):
ENABLE_CHANGE_EMAIL: bool = Field(
description="whether users can change their email address",
default=True,
)
ACCOUNT_DELETION_TOKEN_EXPIRY_MINUTES: PositiveInt = Field(
description="Duration in minutes for which a account deletion token remains valid",
default=5,
+16 -2
View File
@@ -539,9 +539,23 @@ def _parse_observability_time_range(start: str | None, end: str | None, account:
def _query_values(name: str, alias_name: str | None = None) -> list[str]:
values = request.args.getlist(name)
def _get_values(field_name: str) -> list[str]:
values = request.args.getlist(field_name)
indexed_values: list[tuple[int, list[str]]] = []
prefix = f"{field_name}["
for key in request.args:
if not key.startswith(prefix) or not key.endswith("]"):
continue
index = key[len(prefix) : -1]
if index.isdigit():
indexed_values.append((int(index), request.args.getlist(key)))
for _, items in sorted(indexed_values):
values.extend(items)
return values
values = _get_values(name)
if alias_name:
values.extend(request.args.getlist(alias_name))
values.extend(_get_values(alias_name))
return [value.strip() for value in values if value.strip()]
+25 -14
View File
@@ -351,24 +351,31 @@ def _resolve_current_user_agent_debug_conversation_id(
app_model: App,
agent_id: str | None,
draft_type: AgentConfigDraftType,
start_new: bool = False,
) -> str:
"""Resolve the current editor's conversation without crossing draft surfaces."""
"""Resolve or rotate the current editor's conversation within one draft surface.
``start_new`` rotates the scoped mapping through ``AgentRosterService`` so
the old runtime session is retired before the new conversation is used.
Continuations and Build chat keep resolving the existing mapping.
"""
roster_service = AgentRosterService(session)
if agent_id:
return roster_service.get_or_create_agent_app_debug_conversation_id(
tenant_id=current_tenant_id,
agent_id=agent_id,
account_id=current_user.id,
draft_type=draft_type,
)
resolved_agent_id = agent_id
if not resolved_agent_id:
agent = roster_service.get_app_backing_agent(tenant_id=current_tenant_id, app_id=str(app_model.id))
if agent is None:
raise AgentNotFoundError()
resolved_agent_id = agent.id
agent = roster_service.get_app_backing_agent(tenant_id=current_tenant_id, app_id=str(app_model.id))
if agent is None:
raise AgentNotFoundError()
return roster_service.get_or_create_agent_app_debug_conversation_id(
resolve_conversation = (
roster_service.refresh_agent_app_debug_conversation_id
if start_new
else roster_service.get_or_create_agent_app_debug_conversation_id
)
return resolve_conversation(
tenant_id=current_tenant_id,
agent_id=agent.id,
agent_id=resolved_agent_id,
account_id=current_user.id,
draft_type=draft_type,
)
@@ -387,13 +394,17 @@ def _create_chat_message(
args = args_model.model_dump(exclude_none=True, by_alias=True)
if AppMode.value_of(app_model.mode) == AppMode.AGENT:
draft_type = AgentConfigDraftType(args_model.draft_type)
# Preview follows the normal chat contract: an omitted/empty conversation ID starts a new
# conversation. Build chat keeps its stable mapping so build drafts and finalization stay continuous.
debug_conversation_id = _resolve_current_user_agent_debug_conversation_id(
session=session,
current_tenant_id=current_tenant_id or app_model.tenant_id,
current_user=current_user,
app_model=app_model,
agent_id=agent_id,
draft_type=AgentConfigDraftType(args_model.draft_type),
draft_type=draft_type,
start_new=draft_type == AgentConfigDraftType.DRAFT and not args_model.conversation_id,
)
if args_model.conversation_id and args_model.conversation_id != debug_conversation_id:
raise NotFound("Conversation Not Exists.")
+5 -4
View File
@@ -162,9 +162,10 @@ class LoginApi(Resource):
# SELF_HOSTED only have one workspace
tenants = TenantService.get_join_tenants(account, session=db.session())
if len(tenants) == 0:
system_features = FeatureService.get_system_features()
if system_features.is_allow_create_workspace and not system_features.license.workspaces.is_available():
if (
FeatureService.get_system_features().is_allow_create_workspace
and not FeatureService.get_license().workspaces.is_available()
):
raise WorkspacesLimitExceeded()
else:
return SimpleResultOptionalDataResponse(
@@ -310,7 +311,7 @@ class EmailCodeLoginApi(Resource):
if account:
tenants = TenantService.get_join_tenants(account, session=db.session())
if not tenants:
workspaces = FeatureService.get_system_features().license.workspaces
workspaces = FeatureService.get_license().workspaces
if not workspaces.is_available():
raise WorkspacesLimitExceeded()
if not FeatureService.get_system_features().is_allow_create_workspace:
@@ -56,6 +56,7 @@ class OAuthProviderTokenResponse(BaseModel):
class OAuthProviderAccountResponse(BaseModel):
id: str
name: str
email: str
avatar: str | None = None
@@ -251,6 +252,7 @@ class OAuthServerUserAccountApi(Resource):
def post(self, oauth_provider_app: OAuthProviderApp, account: Account):
return jsonable_encoder(
{
"id": account.id,
"name": account.name,
"email": account.email,
"avatar": account.avatar,
+27 -5
View File
@@ -3,10 +3,11 @@ from flask_restx import Resource
from controllers.common.schema import register_response_schema_models
from fields.base import ResponseModel
from libs.helper import dump_response
from libs.login import current_account_with_tenant_optional, login_required
from libs.login import login_required
from services.feature_service import (
FeatureModel,
FeatureService,
LicenseModel,
LimitationModel,
SystemFeatureModel,
)
@@ -32,6 +33,7 @@ register_response_schema_models(
console_ns,
AppDslVersionResponse,
FeatureModel,
LicenseModel,
LimitationModel,
SystemFeatureModel,
TrialModelsResponse,
@@ -135,8 +137,28 @@ class SystemFeatureApi(Resource):
Authentication would create circular dependency (can't login without dashboard loading).
Only non-sensitive configuration data should be returned by this endpoint.
Only non-sensitive configuration data should be returned by this endpoint. Authenticated
license detail is served separately by SystemFeatureLicenseApi.
"""
current_user, _ = current_account_with_tenant_optional()
is_authenticated = current_user is not None
return FeatureService.get_system_features(is_authenticated=is_authenticated).model_dump()
return FeatureService.get_system_features().model_dump()
@console_ns.route("/system-features/license")
class SystemFeatureLicenseApi(Resource):
@console_ns.doc("get_system_license")
@console_ns.doc(description="Get license status and usage detail")
@console_ns.response(
200,
"Success",
console_ns.models[LicenseModel.__name__],
)
@setup_required
@login_required
@account_initialization_required
def get(self):
"""Get full license detail (status, expiry, workspace/seat usage).
Authenticated counterpart to the license *status* exposed on the public
system-features endpoint.
"""
return FeatureService.get_license().model_dump()
+2 -1
View File
@@ -8,6 +8,7 @@ from sqlalchemy.orm import Session
from configs import dify_config
from controllers.fastopenapi import console_router
from enums.deployment_edition import DeploymentEdition
from extensions.ext_database import db
from models.model import DifySetup
from services.account_service import TenantService
@@ -63,7 +64,7 @@ def validate_init_password(payload: InitValidatePayload) -> InitValidateResponse
def get_init_validate_status() -> bool:
if dify_config.EDITION == "SELF_HOSTED":
if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
if os.environ.get("INIT_PASSWORD"):
if session.get("is_init_validated"):
return True
+3 -2
View File
@@ -6,6 +6,7 @@ from sqlalchemy import select
from configs import dify_config
from controllers.fastopenapi import console_router
from enums.deployment_edition import DeploymentEdition
from libs.helper import EmailStr, extract_remote_ip
from libs.password import valid_password
from models.model import DifySetup, db
@@ -52,7 +53,7 @@ def get_setup_status_api() -> SetupStatusResponse:
Only bootstrap-safe status information should be returned by this endpoint.
"""
if dify_config.EDITION == "SELF_HOSTED":
if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
setup_status = get_setup_status()
if setup_status and not isinstance(setup_status, bool):
return SetupStatusResponse(step="finished", setup_at=setup_status.setup_at.isoformat())
@@ -102,7 +103,7 @@ def setup_system(payload: SetupRequestPayload) -> SetupResponse:
def get_setup_status() -> DifySetup | bool | None:
if dify_config.EDITION == "SELF_HOSTED":
if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
return db.session.scalar(select(DifySetup).limit(1))
return True
+2 -1
View File
@@ -46,6 +46,7 @@ from controllers.console.wraps import (
with_current_tenant_id,
with_current_user,
)
from enums.deployment_edition import DeploymentEdition
from extensions.ext_database import db
from fields.base import ResponseModel
from fields.member_fields import AccountResponse
@@ -262,7 +263,7 @@ class AccountInitApi(Resource):
payload = console_ns.payload or {}
args = AccountInitPayload.model_validate(payload)
if dify_config.EDITION == "CLOUD":
if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
if not args.invitation_code:
raise ValueError("invitation_code is required")
+1 -1
View File
@@ -198,7 +198,7 @@ def _check_member_invite_limits(tenant_id: str, new_member_count: int, new_accou
if workspace_members.enabled is True and not workspace_members.is_available(new_member_count):
raise WorkspaceMembersLimitExceeded()
if new_account_count > 0:
seats = FeatureService.get_system_features(is_authenticated=True).license.seats
seats = FeatureService.get_license().seats
if not seats.is_available(new_account_count):
raise SeatsLimitExceeded()
return
+97 -13
View File
@@ -1,5 +1,5 @@
import io
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from datetime import datetime
from typing import Any, Literal, TypedDict
@@ -43,6 +43,7 @@ from core.plugin.entities.plugin_daemon import PluginDecodeResponse, PluginInsta
from core.plugin.impl.exc import PluginDaemonClientSideError
from core.plugin.plugin_service import PluginService
from core.tools.builtin_tool.providers._positions import BuiltinToolProviderSort
from core.tools.entities.api_entities import ToolProviderApiEntity
from core.tools.entities.common_entities import I18nObject
from core.tools.entities.tool_entities import ToolProviderType
from core.tools.tool_manager import ToolManager
@@ -93,6 +94,15 @@ class ParserList(BaseModel):
class PluginCategoryListQuery(BaseModel):
page: int = Field(default=1, ge=1, description="Page number")
page_size: int = Field(default=256, ge=1, le=256, description="Page size (1-256)")
query: str = Field(default="", max_length=256, description="Case-insensitive search query")
tags: list[str] = Field(default_factory=list, max_length=128, description="Match any plugin tag")
language: Literal["en_US", "zh_Hans", "ja_JP", "pt_BR"] = Field(
default="en_US", description="Language used for localized label and description search"
)
class PluginInstalledIdsQuery(BaseModel):
category: PluginCategory = Field(description="Plugin category to include")
class ParserLatest(BaseModel):
@@ -325,6 +335,10 @@ class PluginListResponse(ResponseModel):
total: int
class PluginInstalledIdsResponse(ResponseModel):
plugin_ids: list[str]
class PluginVersionsResponse(ResponseModel):
versions: Mapping[str, PluginService.LatestPluginCache | None]
@@ -384,6 +398,7 @@ register_schema_models(
console_ns,
ParserList,
PluginCategoryListQuery,
PluginInstalledIdsQuery,
PluginAutoUpgradeSettingsPayload,
PluginPermissionSettingsPayload,
ParserLatest,
@@ -420,6 +435,7 @@ register_response_schema_models(
PluginDebuggingKeyResponse,
PluginDynamicOptionsResponse,
PluginInstallationsResponse,
PluginInstalledIdsResponse,
PluginInstallTaskStartResponse,
PluginListResponse,
PluginManifestResponse,
@@ -442,12 +458,10 @@ register_enum_models(
)
def _default_auto_upgrade_settings(
tenant_id: str,
category: TenantPluginAutoUpgradeCategory,
) -> AutoUpgradeSettingsResponse:
def _missing_auto_upgrade_settings(tenant_id: str) -> AutoUpgradeSettingsResponse:
"""Represent a missing persisted strategy as effectively disabled."""
return {
"strategy_setting": PluginAutoUpgradeService.default_strategy_setting_for_category(category),
"strategy_setting": TenantPluginAutoUpgradeStrategySetting.DISABLED,
"upgrade_time_of_day": PluginAutoUpgradeService.default_upgrade_time_of_day(tenant_id),
"upgrade_mode": TenantPluginAutoUpgradeMode.EXCLUDE,
"exclude_plugins": [],
@@ -479,7 +493,39 @@ def _read_upload_content(file: FileStorage, max_size: int) -> bytes:
return content
def _list_hardcoded_builtin_tool_providers(tenant_id: str) -> list[dict[str, Any]]:
def _localized_builtin_tool_text(value: I18nObject, language: str) -> str:
return getattr(value, language, None) or value.en_US
def _builtin_tool_provider_matches_filters(
provider: ToolProviderApiEntity,
*,
query: str,
tags: Sequence[str],
language: str,
) -> bool:
if tags and not any(tag in provider.labels for tag in tags):
return False
if not query:
return True
lower_query = query.lower()
candidates = (
provider.name,
_localized_builtin_tool_text(provider.label, language),
_localized_builtin_tool_text(provider.description, language),
)
return any(lower_query in candidate.lower() for candidate in candidates)
def _list_hardcoded_builtin_tool_providers(
tenant_id: str,
*,
query: str = "",
tags: Sequence[str] = (),
language: str = "en_US",
) -> list[dict[str, Any]]:
"""List builtin providers using the same search and tag semantics as category plugins."""
db_builtin_providers = {
str(ToolProviderID(provider.provider)): provider
for provider in ToolManager.list_default_builtin_providers(tenant_id)
@@ -500,6 +546,13 @@ def _list_hardcoded_builtin_tool_providers(tenant_id: str) -> list[dict[str, Any
db_provider=db_builtin_providers.get(provider.entity.identity.name),
decrypt_credentials=False,
)
if not _builtin_tool_provider_matches_filters(
user_provider,
query=query,
tags=tags,
language=language,
):
continue
ToolTransformService.repack_provider(tenant_id=tenant_id, provider=user_provider)
builtin_providers.append(user_provider)
@@ -554,7 +607,9 @@ class PluginCategoryListApi(Resource):
@account_initialization_required
@with_current_tenant_id
def get(self, tenant_id: str, category: str):
args = PluginCategoryListQuery.model_validate(request.args.to_dict(flat=True))
args = PluginCategoryListQuery.model_validate(
{**request.args.to_dict(flat=True), "tags": request.args.getlist("tags")}
)
try:
plugin_category = PluginCategory(category)
@@ -562,13 +617,26 @@ class PluginCategoryListApi(Resource):
return {"code": "invalid_param", "message": "invalid plugin category"}, 400
try:
plugins = PluginService.list_by_category(tenant_id, plugin_category, args.page, args.page_size)
plugins = PluginService.list_by_category(
tenant_id,
plugin_category,
args.page,
args.page_size,
query=args.query,
tags=args.tags,
language=args.language,
)
except PluginDaemonClientSideError as e:
return {"code": "plugin_error", "message": e.description}, 400
builtin_tools = []
if plugin_category == PluginCategory.Tool:
builtin_tools = _list_hardcoded_builtin_tool_providers(tenant_id)
builtin_tools = _list_hardcoded_builtin_tool_providers(
tenant_id,
query=args.query,
tags=args.tags,
language=args.language,
)
return dump_response(
PluginCategoryListResponse,
@@ -580,6 +648,24 @@ class PluginCategoryListApi(Resource):
)
@console_ns.route("/workspaces/current/plugin/installed-ids")
class PluginInstalledIdsApi(Resource):
@console_ns.doc(params=query_params_from_model(PluginInstalledIdsQuery))
@console_ns.response(200, "Success", console_ns.models[PluginInstalledIdsResponse.__name__])
@setup_required
@login_required
@account_initialization_required
@with_current_tenant_id
def get(self, tenant_id: str):
args = PluginInstalledIdsQuery.model_validate(request.args.to_dict(flat=True))
try:
plugin_ids = PluginService.list_installed_plugin_ids(tenant_id, args.category)
except PluginDaemonClientSideError as e:
return {"code": "plugin_error", "message": e.description}, 400
return dump_response(PluginInstalledIdsResponse, {"plugin_ids": plugin_ids})
@console_ns.route("/workspaces/current/plugin/list/latest-versions")
class PluginListLatestVersionsApi(Resource):
@console_ns.expect(console_ns.models[ParserLatest.__name__])
@@ -1135,9 +1221,7 @@ class PluginFetchAutoUpgradeApi(Resource):
args = ParserAutoUpgradeFetch.model_validate(request.args.to_dict(flat=True))
auto_upgrade = PluginAutoUpgradeService.get_strategy(tenant_id, args.category, session=db.session())
auto_upgrade_dict = (
_auto_upgrade_settings_to_dict(auto_upgrade)
if auto_upgrade
else _default_auto_upgrade_settings(tenant_id, args.category)
_auto_upgrade_settings_to_dict(auto_upgrade) if auto_upgrade else _missing_auto_upgrade_settings(tenant_id)
)
return jsonable_encoder(
@@ -37,6 +37,7 @@ from controllers.console.wraps import (
with_current_user,
)
from enums.cloud_plan import CloudPlan
from enums.deployment_edition import DeploymentEdition
from extensions.ext_database import db
from fields.base import ResponseModel
from libs.helper import dump_response, to_timestamp
@@ -233,7 +234,7 @@ class TenantListApi(Resource):
tenants = [tenant for tenant, _ in tenant_rows]
tenant_dicts = []
is_enterprise_only = dify_config.ENTERPRISE_ENABLED and not dify_config.BILLING_ENABLED
is_saas = dify_config.EDITION == "CLOUD" and dify_config.BILLING_ENABLED
is_saas = dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and dify_config.BILLING_ENABLED
tenant_plans: dict[str, SubscriptionPlan] = {}
if is_saas:
+4 -3
View File
@@ -20,6 +20,7 @@ from controllers.common.wraps import (
from controllers.console.auth.error import AuthenticationFailedError, EmailCodeError
from controllers.console.workspace.error import AccountNotInitializedError
from enums.cloud_plan import CloudPlan
from enums.deployment_edition import DeploymentEdition
from extensions.ext_database import db
from extensions.ext_redis import redis_client
from libs.encryption import FieldEncryption
@@ -129,7 +130,7 @@ def account_initialization_required[R](view: Callable[..., R]) -> Callable[...,
def only_edition_cloud[**P, R](view: Callable[P, R]) -> Callable[P, R]:
@wraps(view)
def decorated(*args: P.args, **kwargs: P.kwargs):
if dify_config.EDITION != "CLOUD":
if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
abort(404)
return view(*args, **kwargs)
@@ -151,7 +152,7 @@ def only_edition_enterprise[**P, R](view: Callable[P, R]) -> Callable[P, R]:
def only_edition_self_hosted[**P, R](view: Callable[P, R]) -> Callable[P, R]:
@wraps(view)
def decorated(*args: P.args, **kwargs: P.kwargs):
if dify_config.EDITION != "SELF_HOSTED":
if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
abort(404)
return view(*args, **kwargs)
@@ -327,7 +328,7 @@ def setup_required[R](view: Callable[..., R]) -> Callable[..., R]:
# The overloads keep Resource methods method-aware for pyrefly while
# preserving support for plain functions used in tests and utilities.
# check setup
if dify_config.EDITION == "SELF_HOSTED" and not _is_setup_completed():
if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD and not _is_setup_completed():
if os.environ.get("INIT_PASSWORD"):
raise NotInitValidateError()
raise NotSetupError()
+2 -1
View File
@@ -8,6 +8,7 @@ from werkzeug.exceptions import InternalServerError
from configs import dify_config
from core.rbac import RBACPermission, RBACResourceScope
from enums.deployment_edition import DeploymentEdition
from libs.oauth_bearer import Scope, TokenType
from models.account import Account, Tenant, TenantAccountRole
from models.model import App, EndUser
@@ -26,7 +27,7 @@ class CallerKind(StrEnum):
def current_edition() -> Edition:
if dify_config.EDITION == "CLOUD":
if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
return Edition.SAAS
if dify_config.ENTERPRISE_ENABLED:
return Edition.EE
+4 -1
View File
@@ -8,6 +8,7 @@ from configs import dify_config
from controllers.common.schema import register_response_schema_models
from controllers.web import web_ns
from controllers.web.wraps import WebApiResource
from enums.deployment_edition import DeploymentEdition
from extensions.ext_database import db
from extensions.storage.storage_type import StorageType
from fields.base import ResponseModel
@@ -127,7 +128,9 @@ def _build_site_icon_url(*, site: Site, tenant_id: str) -> str | None:
"""Use direct S3 URLs only in Cloud Mode and preserve preview URLs elsewhere."""
if site.icon_type != IconType.IMAGE or not site.icon:
return None
if dify_config.EDITION == "CLOUD" and StorageType(dify_config.STORAGE_TYPE) == StorageType.S3:
if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and (
StorageType(dify_config.STORAGE_TYPE) == StorageType.S3
):
return FileService(db.engine).get_file_presigned_url(file_id=site.icon, tenant_id=tenant_id)
return build_icon_url(site.icon_type, site.icon)
+6
View File
@@ -11,6 +11,7 @@ from werkzeug.exceptions import BadRequest, NotFound, Unauthorized
from constants import HEADER_NAME_APP_CODE
from controllers.web.error import WebAppAuthAccessDeniedError, WebAppAuthRequiredError
from core.logging.context import set_identity_context
from extensions.ext_database import db
from libs.passport import PassportService
from libs.token import extract_webapp_passport
@@ -28,6 +29,11 @@ def validate_jwt_token[**P, R](
@wraps(view)
def decorated(*args: P.args, **kwargs: P.kwargs) -> R:
app_model, end_user = decode_jwt_token()
set_identity_context(
tenant_id=end_user.tenant_id,
user_id=end_user.id,
user_type=end_user.type or "end_user",
)
return view(app_model, end_user, *args, **kwargs)
return decorated
+93
View File
@@ -0,0 +1,93 @@
"""Publication visibility rules for calling roster Agents from Workflows.
``Agent.active_config_is_published`` describes whether the editable shared
draft still matches the active snapshot. It is false both before the first
publish and after a published Agent receives new draft edits, so it must not be
used as a runtime availability flag. App-backed Agents are callable from a
Workflow only when the active snapshot has a revision created by a
publish-visible operation. Direct roster Agents are publish-visible by
construction and only need an active snapshot.
"""
from sqlalchemy import and_, or_, select
from sqlalchemy.orm import Session
from sqlalchemy.sql.elements import ColumnElement
from models.agent import Agent, AgentConfigRevision, AgentConfigRevisionOperation, AgentScope, AgentSource
PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS = frozenset(
{
AgentConfigRevisionOperation.PUBLISH_DRAFT,
AgentConfigRevisionOperation.SAVE_CURRENT_VERSION,
AgentConfigRevisionOperation.SAVE_NEW_VERSION,
AgentConfigRevisionOperation.SAVE_NEW_AGENT,
AgentConfigRevisionOperation.SAVE_TO_ROSTER,
AgentConfigRevisionOperation.RESTORE_VERSION,
}
)
def workflow_callable_active_snapshot_filter() -> ColumnElement[bool]:
"""Return the SQL predicate for an Agent with a Workflow-callable active snapshot.
The caller remains responsible for tenant, roster scope, lifecycle status,
and model configuration filters. The correlated revision lookup makes the
predicate safe to compose into roster pagination queries.
"""
app_backed_agent = or_(
Agent.source == AgentSource.AGENT_APP,
and_(
Agent.source == AgentSource.IMPORTED,
Agent.scope == AgentScope.ROSTER,
Agent.app_id.is_not(None),
),
)
publish_visible_revision_exists = (
select(AgentConfigRevision.id)
.where(
AgentConfigRevision.tenant_id == Agent.tenant_id,
AgentConfigRevision.agent_id == Agent.id,
AgentConfigRevision.current_snapshot_id == Agent.active_config_snapshot_id,
AgentConfigRevision.operation.in_(PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS),
)
.correlate(Agent)
.exists()
)
return and_(
Agent.active_config_snapshot_id.is_not(None),
or_(
~app_backed_agent,
publish_visible_revision_exists,
),
)
def agent_has_workflow_callable_active_snapshot(*, session: Session, agent: Agent) -> bool:
"""Return whether ``agent`` has an active snapshot visible to Workflow.
This object-level form is useful after ownership and lifecycle checks have
already loaded an Agent. It intentionally ignores dirty draft state so a
previously published snapshot keeps serving while later edits remain
unpublished.
"""
if not agent.active_config_snapshot_id:
return False
is_app_backed = agent.source == AgentSource.AGENT_APP or (
agent.source == AgentSource.IMPORTED and agent.scope == AgentScope.ROSTER and agent.app_id is not None
)
if not is_app_backed:
return True
return bool(
session.scalar(
select(AgentConfigRevision.id)
.where(
AgentConfigRevision.tenant_id == agent.tenant_id,
AgentConfigRevision.agent_id == agent.id,
AgentConfigRevision.current_snapshot_id == agent.active_config_snapshot_id,
AgentConfigRevision.operation.in_(PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS),
)
.limit(1)
)
)
+2 -1
View File
@@ -6,6 +6,7 @@ from pydantic import BaseModel
from configs import dify_config
from core.entities import DEFAULT_PLUGIN_ID
from core.entities.provider_entities import ProviderQuotaType, QuotaUnit, RestrictModel
from enums.deployment_edition import DeploymentEdition
from graphon.model_runtime.entities.model_entities import ModelType
@@ -49,7 +50,7 @@ class HostingConfiguration:
self.moderation_config = None
def init_app(self, app: Flask):
if dify_config.EDITION != "CLOUD":
if dify_config.DEPLOYMENT_EDITION != DeploymentEdition.CLOUD:
return
self.provider_map[f"{DEFAULT_PLUGIN_ID}/azure_openai/azure_openai"] = self.init_azure_openai()
+46 -44
View File
@@ -21,6 +21,7 @@ from core.model_manager import ModelInstance, ModelManager
from core.rag.cleaner.clean_processor import CleanProcessor
from core.rag.datasource.keyword.keyword_factory import Keyword
from core.rag.docstore.dataset_docstore import DatasetDocumentStore
from core.rag.embedding.token_counter import calculate_segment_token_counts
from core.rag.extractor.entity.datasource_type import DatasourceType
from core.rag.extractor.entity.extract_setting import ExtractSetting, NotionInfo, WebsiteInfo
from core.rag.index_processor.constant.index_type import IndexStructureType, IndexTechniqueType
@@ -113,17 +114,25 @@ class IndexingRunner:
current_user=current_user,
session=session,
)
token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
total_tokens = sum(token_counts)
# save segment
self._load_segments(dataset, requeried_document, documents, session)
self._load_segments(
session=session,
dataset=dataset,
dataset_document=requeried_document,
documents=documents,
token_counts=token_counts,
)
session.commit()
# load
self._load(
index_processor=index_processor,
session=session,
dataset=dataset,
dataset_document=requeried_document,
documents=documents,
session=session,
total_tokens=total_tokens,
)
except DocumentIsPausedError:
raise DocumentIsPausedError(f"Document paused, document id: {document_id}")
@@ -190,17 +199,25 @@ class IndexingRunner:
current_user=current_user,
session=session,
)
token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
total_tokens = sum(token_counts)
# save segment
self._load_segments(dataset, requeried_document, documents, session)
self._load_segments(
session=session,
dataset=dataset,
dataset_document=requeried_document,
documents=documents,
token_counts=token_counts,
)
session.commit()
# load
self._load(
index_processor=index_processor,
session=session,
dataset=dataset,
dataset_document=requeried_document,
documents=documents,
session=session,
total_tokens=total_tokens,
)
except DocumentIsPausedError:
raise DocumentIsPausedError(f"Document paused, document id: {document_id}")
@@ -225,7 +242,7 @@ class IndexingRunner:
if not dataset:
raise ValueError("no dataset found")
# get exist document_segment list and delete
# get existing document segments
document_segments = session.scalars(
select(DocumentSegment).where(
DocumentSegment.dataset_id == dataset.id,
@@ -264,15 +281,15 @@ class IndexingRunner:
child_documents.append(child_document)
document.children = child_documents
documents.append(document)
# Preserve the full document total even when only incomplete segments are re-indexed.
total_tokens = sum(document_segment.tokens for document_segment in document_segments)
# build index
index_type = requeried_document.doc_form
index_processor = IndexProcessorFactory(index_type).init_index_processor()
self._load(
index_processor=index_processor,
session=session,
dataset=dataset,
dataset_document=requeried_document,
documents=documents,
session=session,
total_tokens=total_tokens,
)
except DocumentIsPausedError:
raise DocumentIsPausedError(f"Document paused, document id: {document_id}")
@@ -601,28 +618,16 @@ class IndexingRunner:
def _load(
self,
index_processor: BaseIndexProcessor,
session: Session,
dataset: Dataset,
dataset_document: DatasetDocument,
documents: list[Document],
session: Session,
):
"""
insert index and update document/segment status to completed
"""
total_tokens: int,
) -> None:
"""Build indexes and mark the document complete using the token total computed before hash sharding."""
embedding_model_instance = None
if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
embedding_model_instance = self._get_model_manager(dataset.tenant_id).get_model_instance(
tenant_id=dataset.tenant_id,
provider=dataset.embedding_model_provider,
model_type=ModelType.TEXT_EMBEDDING,
model=dataset.embedding_model,
)
# chunk nodes by chunk size
# Build indexes using the existing hash-based worker groups.
indexing_start_at = time.perf_counter()
tokens = 0
create_keyword_thread = None
if (
dataset_document.doc_form != IndexStructureType.PARENT_CHILD_INDEX
@@ -659,12 +664,11 @@ class IndexingRunner:
chunk_documents,
dataset.id,
dataset_document.id,
embedding_model_instance,
)
)
for future in futures:
tokens += future.result()
future.result()
if (
dataset_document.doc_form != IndexStructureType.PARENT_CHILD_INDEX
and dataset.indexing_technique == IndexTechniqueType.ECONOMY
@@ -679,7 +683,7 @@ class IndexingRunner:
document_id=dataset_document.id,
after_indexing_status=IndexingStatus.COMPLETED,
extra_update_params={
DatasetDocument.tokens: tokens,
DatasetDocument.tokens: total_tokens,
DatasetDocument.completed_at: naive_utc_now(),
DatasetDocument.indexing_latency: indexing_end_at - indexing_start_at,
DatasetDocument.error: None,
@@ -720,8 +724,7 @@ class IndexingRunner:
chunk_documents: list[Document],
dataset_id: str,
dataset_document_id: str,
embedding_model_instance: ModelInstance | None,
):
) -> None:
with flask_app.app_context():
with session_factory.create_session() as session:
dataset = session.get(Dataset, dataset_id)
@@ -735,11 +738,6 @@ class IndexingRunner:
# check document is paused
self._check_document_paused_status(dataset_document.id)
tokens = 0
if embedding_model_instance:
page_content_list = [document.page_content for document in chunk_documents]
tokens += sum(embedding_model_instance.get_text_embedding_num_tokens(page_content_list))
multimodal_documents = []
for document in chunk_documents:
if document.attachments and dataset.is_multimodal:
@@ -773,8 +771,6 @@ class IndexingRunner:
session.commit()
return tokens
@staticmethod
def _check_document_paused_status(document_id: str):
indexing_cache_key = f"document_{document_id}_is_paused"
@@ -864,8 +860,14 @@ class IndexingRunner:
return documents
def _load_segments(
self, dataset: Dataset, dataset_document: DatasetDocument, documents: list[Document], session: Session
):
self,
session: Session,
dataset: Dataset,
dataset_document: DatasetDocument,
documents: list[Document],
token_counts: list[int],
) -> None:
"""Persist transformed documents and their precomputed token counts before indexing starts."""
# save node to document segment
doc_store = DatasetDocumentStore(
dataset=dataset, user_id=dataset_document.created_by, document_id=dataset_document.id
@@ -873,9 +875,10 @@ class IndexingRunner:
# add document segments
doc_store.add_documents(
session=session,
docs=documents,
save_child=dataset_document.doc_form == IndexStructureType.PARENT_CHILD_INDEX,
session=session,
token_counts=token_counts,
)
# update document status to indexing
@@ -900,7 +903,6 @@ class IndexingRunner:
DocumentSegment.indexing_at: naive_utc_now(),
},
)
pass
class DocumentIsPausedError(Exception):
+34 -2
View File
@@ -6,9 +6,21 @@ using Python's contextvars for thread-safe and async-safe storage.
import uuid
from contextvars import ContextVar
from typing import NamedTuple
class IdentityContext(NamedTuple):
"""Immutable identity values captured for logging."""
tenant_id: str
user_id: str
user_type: str
_request_id: ContextVar[str] = ContextVar("log_request_id", default="")
_trace_id: ContextVar[str] = ContextVar("log_trace_id", default="")
_EMPTY_IDENTITY_CONTEXT = IdentityContext(tenant_id="", user_id="", user_type="")
_identity: ContextVar[IdentityContext] = ContextVar("log_identity", default=_EMPTY_IDENTITY_CONTEXT)
def get_request_id() -> str:
@@ -21,15 +33,35 @@ def get_trace_id() -> str:
return _trace_id.get()
def get_identity_context() -> IdentityContext:
"""Get the immutable tenant, user, and user-type snapshot for logging."""
return _identity.get()
def set_identity_context(
*, tenant_id: str | None = None, user_id: str | None = None, user_type: str | None = None
) -> None:
"""Set primitive identity values already resolved by an authentication boundary."""
_identity.set(
IdentityContext(
tenant_id=tenant_id or "",
user_id=user_id or "",
user_type=user_type or "",
)
)
def init_request_context() -> None:
"""Initialize request context. Call at start of each request."""
"""Initialize request context and discard identity left by earlier work."""
req_id = uuid.uuid4().hex[:10]
trace_id = uuid.uuid5(uuid.NAMESPACE_DNS, req_id).hex
_request_id.set(req_id)
_trace_id.set(trace_id)
_identity.set(_EMPTY_IDENTITY_CONTEXT)
def clear_request_context() -> None:
"""Clear request context. Call at end of request (optional)."""
"""Clear request context at a request or task lifecycle boundary."""
_request_id.set("")
_trace_id.set("")
_identity.set(_EMPTY_IDENTITY_CONTEXT)
+9 -45
View File
@@ -4,10 +4,7 @@ import contextlib
import logging
from typing import override
import flask
from core.logging.context import get_request_id, get_trace_id
from core.logging.structured_formatter import IdentityDict
from core.logging.context import get_identity_context, get_request_id, get_trace_id
class TraceContextFilter(logging.Filter):
@@ -51,49 +48,16 @@ class TraceContextFilter(logging.Filter):
class IdentityContextFilter(logging.Filter):
"""
Filter that adds user identity context to log records.
Extracts tenant_id, user_id, and user_type from Flask-Login current_user.
"""Add an identity snapshot without invoking authentication or database work.
Logging can run while other libraries hold internal locks, so this filter must
only read primitive ContextVar values populated by authentication boundaries.
"""
@override
def filter(self, record: logging.LogRecord) -> bool:
identity = self._extract_identity()
record.tenant_id = identity.get("tenant_id", "")
record.user_id = identity.get("user_id", "")
record.user_type = identity.get("user_type", "")
identity = get_identity_context()
record.tenant_id = identity.tenant_id
record.user_id = identity.user_id
record.user_type = identity.user_type
return True
def _extract_identity(self) -> IdentityDict:
"""Extract identity from current_user if in request context."""
try:
if not flask.has_request_context():
return {}
from flask_login import current_user
# Check if user is authenticated using the proxy
if not current_user.is_authenticated:
return {}
# Access the underlying user object
user = current_user
from models import Account
from models.model import EndUser
identity: IdentityDict = {}
match user:
case Account():
if user.current_tenant_id:
identity["tenant_id"] = user.current_tenant_id
identity["user_id"] = user.id
identity["user_type"] = "account"
case EndUser():
identity["tenant_id"] = user.tenant_id
identity["user_id"] = user.id
identity["user_type"] = user.type or "end_user"
return identity
except Exception:
return {}
@@ -207,6 +207,10 @@ class PluginListResponse(BaseModel):
total: int
class PluginInstalledIdsDaemonResponse(BaseModel):
plugin_ids: list[str]
class PluginListWithoutTotalResponse(BaseModel):
list: list[PluginEntity]
has_more: bool
+28 -2
View File
@@ -14,6 +14,7 @@ from core.plugin.entities.plugin import (
)
from core.plugin.entities.plugin_daemon import (
PluginDecodeResponse,
PluginInstalledIdsDaemonResponse,
PluginInstallTask,
PluginInstallTaskStartResponse,
PluginListResponse,
@@ -68,6 +69,16 @@ class PluginInstaller(BasePluginClient):
)
return result.list
def list_installed_plugin_ids(self, tenant_id: str, category: PluginCategory) -> list[str]:
"""List all currently installed plugin IDs in one category."""
result = self._request_with_plugin_daemon_response(
"GET",
f"plugin/{tenant_id}/management/installation/ids",
PluginInstalledIdsDaemonResponse,
params={"category": category.value},
)
return result.plugin_ids
def list_plugins_with_total(self, tenant_id: str, page: int, page_size: int) -> PluginListResponse:
return self._request_with_plugin_daemon_response(
"GET",
@@ -77,13 +88,28 @@ class PluginInstaller(BasePluginClient):
)
def list_plugins_by_category(
self, tenant_id: str, category: PluginCategory, page: int, page_size: int
self,
tenant_id: str,
category: PluginCategory,
page: int,
page_size: int,
*,
query: str = "",
tags: Sequence[str] = (),
language: str = "en_US",
) -> PluginListWithoutTotalResponse:
return self._request_with_plugin_daemon_response(
"GET",
f"plugin/{tenant_id}/management/{category.value}/list",
PluginListWithoutTotalResponse,
params={"page": page, "page_size": page_size, "response_type": "paged"},
params={
"page": page,
"page_size": page_size,
"response_type": "paged",
"query": query,
"tags": list(tags),
"language": language,
},
)
def upload_pkg(
+26 -5
View File
@@ -656,6 +656,12 @@ class PluginService:
plugins = manager.list_plugins(tenant_id)
return plugins
@staticmethod
def list_installed_plugin_ids(tenant_id: str, category: PluginCategory) -> Sequence[str]:
"""List all currently installed plugin IDs in one category through the daemon's lightweight query."""
manager = PluginInstaller()
return manager.list_installed_plugin_ids(tenant_id, category)
@staticmethod
def list_with_total(tenant_id: str, user_id: str, page: int, page_size: int) -> PluginListResponse:
"""List tenant plugins with endpoint counts reconciled from live records.
@@ -672,17 +678,32 @@ class PluginService:
@staticmethod
def list_by_category(
tenant_id: str, category: PluginCategory, page: int, page_size: int
tenant_id: str,
category: PluginCategory,
page: int,
page_size: int,
*,
query: str = "",
tags: Sequence[str] = (),
language: str = "en_US",
) -> PluginListWithoutTotalResponse:
"""
List plugins in one category with a has-more cursor signal and without calculating total.
The daemon scans tenant installations in the existing list order and stops once it finds one extra match.
This keeps pagination usable before category is persisted on installation rows.
The daemon applies category, search, and tag filters before pagination, then stops once it finds one extra
match. Filtered model reads are partial views and therefore do not reconcile the full model-provider cache.
"""
manager = PluginInstaller()
plugins = manager.list_plugins_by_category(tenant_id, category, page, page_size)
if category == PluginCategory.Model:
plugins = manager.list_plugins_by_category(
tenant_id,
category,
page,
page_size,
query=query,
tags=tags,
language=language,
)
if category == PluginCategory.Model and not query and not tags:
should_invalidate_model_provider_cache = (
PluginService._should_invalidate_model_provider_cache_for_remote_model_plugins(
tenant_id,
+3 -2
View File
@@ -34,6 +34,7 @@ from core.entities.provider_entities import (
from core.helper import encrypter
from core.helper.model_provider_cache import ProviderCredentialsCache, ProviderCredentialsCacheType
from core.helper.position_helper import is_filtered
from enums.deployment_edition import DeploymentEdition
from extensions import ext_hosting_provider
from extensions.ext_database import db
from extensions.ext_redis import redis_client
@@ -743,7 +744,7 @@ class ProviderManager:
if preferred_provider_type_record:
preferred_provider_type = preferred_provider_type_record.preferred_provider_type
elif dify_config.EDITION == "CLOUD" and system_configuration.enabled:
elif dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD and system_configuration.enabled:
preferred_provider_type = ProviderType.SYSTEM
elif custom_configuration.provider or custom_configuration.models:
preferred_provider_type = ProviderType.CUSTOM
@@ -1538,7 +1539,7 @@ class ProviderManager:
quota_type_to_provider_records_dict[provider_record.quota_type] = provider_record # type: ignore[index]
quota_configurations = []
if dify_config.EDITION == "CLOUD":
if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
from services.credit_pool_service import CreditPoolService
trail_pool = CreditPoolService.get_pool(
@@ -10,6 +10,7 @@ from core.rag.rerank.entity.weight import KeywordSetting, VectorSetting, Weights
from core.rag.rerank.rerank_base import BaseRerankRunner
from core.rag.rerank.rerank_factory import RerankRunnerFactory
from core.rag.rerank.rerank_type import RerankMode
from extensions.otel import trace_span
from graphon.model_runtime.entities.model_entities import ModelType
from graphon.model_runtime.errors.invoke import InvokeAuthorizationError
@@ -52,6 +53,7 @@ class DataPostProcessor:
)
self.reorder_runner = self._get_reorder_runner(reorder_enabled)
@trace_span()
def invoke(
self,
query: str,
+8 -24
View File
@@ -1,12 +1,10 @@
import concurrent.futures
import functools
import logging
from collections.abc import Callable, Sequence
from collections.abc import Sequence
from concurrent.futures import ThreadPoolExecutor
from typing import Any, NotRequired, TypedDict
from flask import Flask, current_app
from opentelemetry import context as otel_context
from sqlalchemy import select
from sqlalchemy.orm import Session, load_only
@@ -26,7 +24,7 @@ from core.rag.rerank.rerank_type import RerankMode
from core.rag.retrieval.retrieval_methods import RetrievalMethod
from core.tools.signature import sign_upload_file_preview_url
from extensions.ext_database import db
from extensions.otel import trace_span
from extensions.otel import propagate_context, trace_span
from graphon.model_runtime.entities.model_entities import ModelType
from models.dataset import (
ChildChunk,
@@ -92,20 +90,6 @@ default_retrieval_model: DefaultRetrievalModelDict = {
logger = logging.getLogger(__name__)
def _propagate_otel_context[**P, R](func: Callable[P, R]) -> Callable[P, R]:
captured_context = otel_context.get_current()
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
token = otel_context.attach(captured_context)
try:
return func(*args, **kwargs)
finally:
otel_context.detach(token)
return wrapper
class RetrievalService:
# Cache precompiled regular expressions to avoid repeated compilation
@classmethod
@@ -139,7 +123,7 @@ class RetrievalService:
if query:
futures.append(
executor.submit(
_propagate_otel_context(retrieval_service._retrieve),
propagate_context(retrieval_service._retrieve),
flask_app=current_app._get_current_object(), # type: ignore
retrieval_method=retrieval_method,
dataset=dataset,
@@ -159,7 +143,7 @@ class RetrievalService:
for attachment_id in attachment_ids:
futures.append(
executor.submit(
_propagate_otel_context(retrieval_service._retrieve),
propagate_context(retrieval_service._retrieve),
flask_app=current_app._get_current_object(), # type: ignore
retrieval_method=retrieval_method,
dataset=dataset,
@@ -820,7 +804,7 @@ class RetrievalService:
if retrieval_method == RetrievalMethod.KEYWORD_SEARCH and query:
futures.append(
executor.submit(
_propagate_otel_context(self.keyword_search),
propagate_context(self.keyword_search),
flask_app=current_app._get_current_object(), # type: ignore
dataset_id=dataset.id,
query=query,
@@ -834,7 +818,7 @@ class RetrievalService:
if query:
futures.append(
executor.submit(
_propagate_otel_context(self.embedding_search),
propagate_context(self.embedding_search),
flask_app=current_app._get_current_object(), # type: ignore
dataset_id=dataset.id,
query=query,
@@ -851,7 +835,7 @@ class RetrievalService:
if attachment_id:
futures.append(
executor.submit(
_propagate_otel_context(self.embedding_search),
propagate_context(self.embedding_search),
flask_app=current_app._get_current_object(), # type: ignore
dataset_id=dataset.id,
query=attachment_id,
@@ -868,7 +852,7 @@ class RetrievalService:
if RetrievalMethod.is_support_fulltext_search(retrieval_method) and query:
futures.append(
executor.submit(
_propagate_otel_context(self.full_text_index_search),
propagate_context(self.full_text_index_search),
flask_app=current_app._get_current_object(), # type: ignore
dataset_id=dataset.id,
query=query,
+6 -21
View File
@@ -6,10 +6,7 @@ from typing import Any
from sqlalchemy import delete, func, select
from sqlalchemy.orm import Session
from core.model_manager import ModelManager
from core.rag.index_processor.constant.index_type import IndexTechniqueType
from core.rag.models.document import AttachmentDocument, Document
from graphon.model_runtime.entities.model_entities import ModelType
from models.dataset import ChildChunk, Dataset, DocumentSegment, SegmentAttachmentBinding
from models.enums import SegmentType
@@ -69,34 +66,22 @@ class DatasetDocumentStore:
def add_documents(
self,
docs: Sequence[Document],
session: Session,
docs: Sequence[Document],
token_counts: list[int],
allow_update: bool = True,
save_child: bool = False,
):
) -> None:
document_token_pairs = list(zip(docs, token_counts, strict=True))
max_position = session.scalar(
select(func.max(DocumentSegment.position)).where(DocumentSegment.document_id == self._document_id)
)
if max_position is None:
max_position = 0
embedding_model = None
if self._dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
model_manager = ModelManager.for_tenant(tenant_id=self._dataset.tenant_id)
embedding_model = model_manager.get_model_instance(
tenant_id=self._dataset.tenant_id,
provider=self._dataset.embedding_model_provider,
model_type=ModelType.TEXT_EMBEDDING,
model=self._dataset.embedding_model,
)
if embedding_model:
page_content_list = [doc.page_content for doc in docs]
tokens_list = embedding_model.get_text_embedding_num_tokens(page_content_list)
else:
tokens_list = [0] * len(docs)
for doc, tokens in zip(docs, tokens_list):
for doc, tokens in document_token_pairs:
if not isinstance(doc, Document):
raise ValueError("doc must be a Document")
+25
View File
@@ -0,0 +1,25 @@
"""Token counting for document segments."""
from core.model_manager import ModelManager
from core.rag.index_processor.constant.index_type import IndexTechniqueType
from core.rag.models.document import Document
from graphon.model_runtime.entities.model_entities import ModelType
from models.dataset import Dataset
def calculate_segment_token_counts(dataset: Dataset, documents: list[Document]) -> list[int]:
"""Return one token count per document, invoking the embedding model only for high-quality indexes."""
if not documents:
return []
if dataset.indexing_technique != IndexTechniqueType.HIGH_QUALITY:
return [0] * len(documents)
model_manager = ModelManager.for_tenant(tenant_id=dataset.tenant_id)
embedding_model = model_manager.get_model_instance(
tenant_id=dataset.tenant_id,
provider=dataset.embedding_model_provider,
model_type=ModelType.TEXT_EMBEDDING,
model=dataset.embedding_model,
)
return embedding_model.get_text_embedding_num_tokens([document.page_content for document in documents])
@@ -19,6 +19,7 @@ from core.rag.cleaner.clean_processor import CleanProcessor
from core.rag.datasource.keyword.keyword_factory import Keyword
from core.rag.datasource.vdb.vector_factory import Vector
from core.rag.docstore.dataset_docstore import DatasetDocumentStore
from core.rag.embedding.token_counter import calculate_segment_token_counts
from core.rag.entities import Rule
from core.rag.extractor.entity.extract_setting import ExtractSetting
from core.rag.extractor.extract_processor import ExtractProcessor
@@ -244,10 +245,16 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
all_multimodal_documents.extend(doc.attachments)
documents.append(doc)
if documents:
token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
# save node to document segment
doc_store = DatasetDocumentStore(dataset=dataset, user_id=document.created_by, document_id=document.id)
# add document segments
doc_store.add_documents(docs=documents, save_child=False, session=session)
doc_store.add_documents(
session=session,
docs=documents,
token_counts=token_counts,
save_child=False,
)
session.commit()
if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
vector = Vector(dataset, session=session)
@@ -15,6 +15,7 @@ from core.model_manager import ModelInstance
from core.rag.cleaner.clean_processor import CleanProcessor
from core.rag.datasource.vdb.vector_factory import Vector
from core.rag.docstore.dataset_docstore import DatasetDocumentStore
from core.rag.embedding.token_counter import calculate_segment_token_counts
from core.rag.entities import ParentMode, Rule
from core.rag.extractor.entity.extract_setting import ExtractSetting
from core.rag.extractor.extract_processor import ExtractProcessor
@@ -304,6 +305,7 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
doc.attachments = self._get_content_files(doc, current_user=account, session=session)
documents.append(doc)
if documents:
token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
# update document parent mode
dataset_process_rule = DatasetProcessRule(
dataset_id=dataset.id,
@@ -321,7 +323,12 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
# save node to document segment
doc_store = DatasetDocumentStore(dataset=dataset, user_id=document.created_by, document_id=document.id)
# add document segments
doc_store.add_documents(docs=documents, save_child=True, session=session)
doc_store.add_documents(
session=session,
docs=documents,
token_counts=token_counts,
save_child=True,
)
session.commit()
if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
all_child_documents = []
@@ -17,6 +17,7 @@ from core.llm_generator.llm_generator import LLMGenerator
from core.rag.cleaner.clean_processor import CleanProcessor
from core.rag.datasource.vdb.vector_factory import Vector
from core.rag.docstore.dataset_docstore import DatasetDocumentStore
from core.rag.embedding.token_counter import calculate_segment_token_counts
from core.rag.entities import Rule
from core.rag.extractor.entity.extract_setting import ExtractSetting
from core.rag.extractor.extract_processor import ExtractProcessor
@@ -205,9 +206,15 @@ class QAIndexProcessor(BaseIndexProcessor):
doc = Document(page_content=qa_chunk.question, metadata=metadata)
documents.append(doc)
if documents:
token_counts = calculate_segment_token_counts(dataset=dataset, documents=documents)
# save node to document segment
doc_store = DatasetDocumentStore(dataset=dataset, user_id=document.created_by, document_id=document.id)
doc_store.add_documents(docs=documents, save_child=False, session=session)
doc_store.add_documents(
session=session,
docs=documents,
token_counts=token_counts,
save_child=False,
)
session.commit()
if dataset.indexing_technique == IndexTechniqueType.HIGH_QUALITY:
vector = Vector(dataset, session=session)
+2
View File
@@ -9,6 +9,7 @@ from core.rag.index_processor.constant.query_type import QueryType
from core.rag.models.document import Document
from core.rag.rerank.rerank_base import BaseRerankRunner
from extensions.ext_storage import storage
from extensions.otel import trace_span
from graphon.model_runtime.entities.model_entities import ModelType
from graphon.model_runtime.entities.rerank_entities import MultimodalRerankInput, RerankResult
from models.model import UploadFile
@@ -22,6 +23,7 @@ class RerankModelRunner(BaseRerankRunner):
self._session = session
@override
@trace_span()
def run(
self,
query: str,
+101 -24
View File
@@ -65,6 +65,7 @@ from core.workflow.nodes.knowledge_retrieval.retrieval import (
)
from extensions.ext_database import db
from extensions.ext_redis import redis_client
from extensions.otel import propagate_context, trace_span
from graphon.file import File, FileTransferMethod, FileType
from graphon.model_runtime.entities.llm_entities import LLMMode, LLMResult, LLMUsage
from graphon.model_runtime.entities.message_entities import PromptMessage, PromptMessageRole, PromptMessageTool
@@ -116,6 +117,7 @@ class DatasetRetrieval:
else:
self._llm_usage = self._llm_usage.plus(usage)
@trace_span()
def knowledge_retrieval(self, session: Session, request: KnowledgeRetrievalRequest) -> list[Source]:
self._check_knowledge_rate_limit(request.tenant_id)
available_datasets = self._get_available_datasets(request.tenant_id, request.dataset_ids)
@@ -599,6 +601,7 @@ class DatasetRetrieval:
return "\n".join([document_context.content for document_context in document_context_list]), context_files
return "", context_files
@trace_span()
def single_retrieve(
self,
session: Session,
@@ -724,7 +727,7 @@ class DatasetRetrieval:
if results:
thread = threading.Thread(
target=self._on_retrieval_end,
target=propagate_context(self._on_retrieval_end),
kwargs={
"flask_app": current_app._get_current_object(), # type: ignore
"documents": results,
@@ -737,6 +740,7 @@ class DatasetRetrieval:
return results
return []
@trace_span()
def multiple_retrieve(
self,
app_id: str,
@@ -798,7 +802,7 @@ class DatasetRetrieval:
if query:
query_thread = threading.Thread(
target=self._multiple_retrieve_thread,
target=propagate_context(self._multiple_retrieve_thread_safely),
kwargs={
"flask_app": current_app._get_current_object(), # type: ignore
"available_datasets": available_datasets,
@@ -824,7 +828,7 @@ class DatasetRetrieval:
if attachment_ids:
for attachment_id in attachment_ids:
attachment_thread = threading.Thread(
target=self._multiple_retrieve_thread,
target=propagate_context(self._multiple_retrieve_thread_safely),
kwargs={
"flask_app": current_app._get_current_object(), # type: ignore
"available_datasets": available_datasets,
@@ -865,7 +869,7 @@ class DatasetRetrieval:
if all_documents:
# add thread to call _on_retrieval_end
retrieval_end_thread = threading.Thread(
target=self._on_retrieval_end,
target=propagate_context(self._on_retrieval_end),
kwargs={
"flask_app": current_app._get_current_object(), # type: ignore
"documents": all_documents,
@@ -1161,6 +1165,7 @@ class DatasetRetrieval:
all_documents.extend(documents)
@trace_span()
def _run_retriever_thread(
self,
*,
@@ -1172,27 +1177,51 @@ class DatasetRetrieval:
document_ids_filter: list[str] | None,
metadata_condition: MetadataFilteringCondition | None,
attachment_ids: list[str] | None,
) -> None:
with session_factory.create_session() as session:
self._retriever(
flask_app=flask_app,
session=session,
dataset_id=dataset_id,
query=query or "",
top_k=top_k,
all_documents=all_documents,
document_ids_filter=document_ids_filter,
metadata_condition=metadata_condition,
attachment_ids=attachment_ids,
)
def _run_retriever_thread_safely(
self,
*,
flask_app: Flask,
dataset_id: str,
query: str | None,
top_k: int,
all_documents: list[Document],
document_ids_filter: list[str] | None,
metadata_condition: MetadataFilteringCondition | None,
attachment_ids: list[str] | None,
cancel_event: threading.Event | None,
thread_exceptions: list[Exception] | None,
) -> None:
"""Collect errors only after they pass through the traced retrieval method."""
try:
with session_factory.create_session() as session:
self._retriever(
flask_app=flask_app,
session=session,
dataset_id=dataset_id,
query=query or "",
top_k=top_k,
all_documents=all_documents,
document_ids_filter=document_ids_filter,
metadata_condition=metadata_condition,
attachment_ids=attachment_ids,
)
except Exception as e:
self._run_retriever_thread(
flask_app=flask_app,
dataset_id=dataset_id,
query=query,
top_k=top_k,
all_documents=all_documents,
document_ids_filter=document_ids_filter,
metadata_condition=metadata_condition,
attachment_ids=attachment_ids,
)
except Exception as exc:
if cancel_event:
cancel_event.set()
if thread_exceptions is not None:
thread_exceptions.append(e)
thread_exceptions.append(exc)
def to_dataset_retriever_tool(
self,
@@ -1795,6 +1824,7 @@ class DatasetRetrieval:
return full_text, usage
@trace_span()
def _multiple_retrieve_thread(
self,
flask_app: Flask,
@@ -1813,11 +1843,11 @@ class DatasetRetrieval:
attachment_id: str | None,
dataset_count: int,
cancel_event: threading.Event | None = None,
thread_exceptions: list[Exception] | None = None,
):
) -> None:
try:
with flask_app.app_context():
threads = []
retrieval_thread_exceptions: list[Exception] = []
all_documents_item: list[Document] = []
index_type = None
for dataset in available_datasets:
@@ -1836,7 +1866,7 @@ class DatasetRetrieval:
else:
continue
retrieval_thread = threading.Thread(
target=self._run_retriever_thread,
target=propagate_context(self._run_retriever_thread_safely),
kwargs={
"flask_app": flask_app,
"dataset_id": dataset.id,
@@ -1847,7 +1877,7 @@ class DatasetRetrieval:
"metadata_condition": metadata_condition,
"attachment_ids": [attachment_id] if attachment_id else None,
"cancel_event": cancel_event,
"thread_exceptions": thread_exceptions,
"thread_exceptions": retrieval_thread_exceptions,
},
)
threads.append(retrieval_thread)
@@ -1862,6 +1892,9 @@ class DatasetRetrieval:
if cancel_event and cancel_event.is_set():
break
if retrieval_thread_exceptions:
raise retrieval_thread_exceptions[0]
# Skip second reranking when there is only one dataset
if reranking_enable and dataset_count > 1:
# do rerank for searched documents
@@ -1902,11 +1935,55 @@ class DatasetRetrieval:
all_documents_item = all_documents_item[:top_k] if top_k else all_documents_item
if all_documents_item:
all_documents.extend(all_documents_item)
except Exception as e:
except Exception:
raise
def _multiple_retrieve_thread_safely(
self,
*,
flask_app: Flask,
available_datasets: list[Dataset],
metadata_condition: MetadataFilteringCondition | None,
metadata_filter_document_ids: dict[str, list[str]] | None,
all_documents: list[Document],
tenant_id: str,
reranking_enable: bool,
reranking_mode: str,
reranking_model: RerankingModelDict | None,
weights: WeightsDict | None,
top_k: int,
score_threshold: float,
query: str | None,
attachment_id: str | None,
dataset_count: int,
cancel_event: threading.Event | None = None,
thread_exceptions: list[Exception] | None = None,
) -> None:
"""Collect errors only after they pass through the traced multi-retrieval method."""
try:
self._multiple_retrieve_thread(
flask_app=flask_app,
available_datasets=available_datasets,
metadata_condition=metadata_condition,
metadata_filter_document_ids=metadata_filter_document_ids,
all_documents=all_documents,
tenant_id=tenant_id,
reranking_enable=reranking_enable,
reranking_mode=reranking_mode,
reranking_model=reranking_model,
weights=weights,
top_k=top_k,
score_threshold=score_threshold,
query=query,
attachment_id=attachment_id,
dataset_count=dataset_count,
cancel_event=cancel_event,
)
except Exception as exc:
if cancel_event:
cancel_event.set()
if thread_exceptions is not None:
thread_exceptions.append(e)
thread_exceptions.append(exc)
def _get_available_datasets(self, tenant_id: str, dataset_ids: list[str]) -> list[Dataset]:
with session_factory.create_session() as session:
@@ -4,8 +4,16 @@ from dataclasses import dataclass
from sqlalchemy import select
from core.agent.publish_visibility import workflow_callable_active_snapshot_filter
from core.db.session_factory import session_factory
from models.agent import Agent, AgentConfigSnapshot, AgentStatus, WorkflowAgentBindingType, WorkflowAgentNodeBinding
from models.agent import (
Agent,
AgentConfigSnapshot,
AgentScope,
AgentStatus,
WorkflowAgentBindingType,
WorkflowAgentNodeBinding,
)
class WorkflowAgentBindingError(Exception):
@@ -24,7 +32,7 @@ class WorkflowAgentBindingBundle:
class WorkflowAgentBindingResolver:
"""Resolve the Agent binding owned by the current workflow id and node id."""
"""Resolve an owned binding without allowing unpublished roster snapshots to run."""
def resolve(
self,
@@ -53,18 +61,20 @@ class WorkflowAgentBindingResolver:
if binding.agent_id is None:
raise WorkflowAgentBindingError("agent_not_available", "Workflow Agent binding has no agent.")
agent = session.scalar(
select(Agent)
.where(
Agent.tenant_id == tenant_id,
Agent.id == binding.agent_id,
)
.limit(1)
agent_stmt = select(Agent).where(
Agent.tenant_id == tenant_id,
Agent.id == binding.agent_id,
)
if binding.binding_type == WorkflowAgentBindingType.ROSTER_AGENT:
agent_stmt = agent_stmt.where(
Agent.scope == AgentScope.ROSTER,
workflow_callable_active_snapshot_filter(),
)
agent = session.scalar(agent_stmt.limit(1))
if agent is None or agent.status == AgentStatus.ARCHIVED:
raise WorkflowAgentBindingError(
"agent_not_available",
f"Agent {binding.agent_id} is not available.",
f"Agent {binding.agent_id} is not available or has not been published.",
)
snapshot_id = (
+26 -11
View File
@@ -6,9 +6,17 @@ from typing import Any
from sqlalchemy import select
from sqlalchemy.orm import Session
from core.agent.publish_visibility import workflow_callable_active_snapshot_filter
from core.workflow.graph_topology import WorkflowGraphTopology
from graphon.enums import BuiltinNodeTypes
from models.agent import Agent, AgentConfigSnapshot, AgentStatus, WorkflowAgentBindingType, WorkflowAgentNodeBinding
from models.agent import (
Agent,
AgentConfigSnapshot,
AgentScope,
AgentStatus,
WorkflowAgentBindingType,
WorkflowAgentNodeBinding,
)
from models.agent_config_entities import (
AgentFileRefConfig,
AgentHumanContactConfig,
@@ -117,21 +125,28 @@ class WorkflowAgentNodeValidator:
binding: WorkflowAgentNodeBinding,
topology: _WorkflowGraphTopology | None = None,
) -> None:
"""Validate binding ownership, publication state, Agent Soul, and node-job references."""
if binding.agent_id is None:
raise WorkflowAgentNodeValidationError(f"Workflow Agent node {binding.node_id} is missing agent binding.")
agent = session.scalar(
select(Agent)
.where(
Agent.tenant_id == binding.tenant_id,
Agent.id == binding.agent_id,
)
.limit(1)
agent_stmt = select(Agent).where(
Agent.tenant_id == binding.tenant_id,
Agent.id == binding.agent_id,
)
if agent is None or agent.status == AgentStatus.ARCHIVED:
raise WorkflowAgentNodeValidationError(
f"Workflow Agent node {binding.node_id} references an unavailable agent."
if binding.binding_type == WorkflowAgentBindingType.ROSTER_AGENT:
agent_stmt = agent_stmt.where(
Agent.scope == AgentScope.ROSTER,
workflow_callable_active_snapshot_filter(),
)
agent = session.scalar(agent_stmt.limit(1))
if agent is None or agent.status == AgentStatus.ARCHIVED:
availability = (
"an unavailable or unpublished roster agent"
if binding.binding_type == WorkflowAgentBindingType.ROSTER_AGENT
else "an unavailable agent"
)
raise WorkflowAgentNodeValidationError(f"Workflow Agent node {binding.node_id} references {availability}.")
snapshot_id = (
agent.active_config_snapshot_id
+11
View File
@@ -0,0 +1,11 @@
from enum import StrEnum
class DeploymentEdition(StrEnum):
"""
Enum representing the deployment edition of the platform.
"""
COMMUNITY = "COMMUNITY"
ENTERPRISE = "ENTERPRISE"
CLOUD = "CLOUD"
+19 -4
View File
@@ -1,5 +1,6 @@
import json
from typing import cast, override
import logging
from typing import assert_never, cast, override
import flask_login
from flask import Request, Response, request
@@ -11,6 +12,7 @@ from werkzeug.exceptions import NotFound, Unauthorized
from configs import dify_config
from constants import HEADER_NAME_APP_CODE
from core.db.session_factory import session_factory
from core.logging.context import set_identity_context
from dify_app import DifyApp
from libs.passport import PassportService
from libs.token import extract_access_token, extract_console_cookie_token, extract_webapp_passport
@@ -19,6 +21,8 @@ from models.enums import EndUserType
from models.model import AppMCPServer, EndUser
from services.account_service import AccountService
logger = logging.getLogger(__name__)
type LoginUser = Account | EndUser
@@ -156,13 +160,24 @@ def _load_user_from_request(request_from_flask_login: Request, session: Session)
@user_logged_in.connect
@user_loaded_from_request.connect
def on_user_logged_in(_sender: object, user: LoginUser) -> None:
"""Called when a user logged in.
"""Snapshot authenticated identity into the side-effect-free logging context.
Note: AccountService.load_logged_in_account will populate user.current_tenant_id
through the load_user method, which calls account.set_tenant_id_with_session().
"""
# tenant_id context variable removed - using current_user.current_tenant_id directly
pass
set_identity_context()
try:
match user:
case Account():
set_identity_context(tenant_id=user.current_tenant_id, user_id=user.id, user_type="account")
case EndUser():
set_identity_context(tenant_id=user.tenant_id, user_id=user.id, user_type=user.type or "end_user")
case _ as unreachable:
assert_never(unreachable)
except Exception:
# Logging enrichment must never make authentication fail.
logger.exception("Failed to set logging identity context")
return
@login_manager.unauthorized_handler
+2
View File
@@ -1,3 +1,4 @@
from extensions.otel.context import propagate_context
from extensions.otel.decorators.base import trace_span
from extensions.otel.decorators.handler import SpanHandler
from extensions.otel.decorators.handlers.generate_handler import AppGenerateHandler
@@ -7,5 +8,6 @@ __all__ = [
"AppGenerateHandler",
"SpanHandler",
"WorkflowAppRunnerHandler",
"propagate_context",
"trace_span",
]
+21
View File
@@ -0,0 +1,21 @@
"""Utilities for propagating OpenTelemetry context across execution boundaries."""
import functools
from collections.abc import Callable
from opentelemetry import context as otel_context
def propagate_context[**P, R](func: Callable[P, R]) -> Callable[P, R]:
"""Capture the current context and attach it whenever ``func`` executes."""
captured_context = otel_context.get_current()
@functools.wraps(func)
def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
token = otel_context.attach(captured_context)
try:
return func(*args, **kwargs)
finally:
otel_context.detach(token)
return wrapper
+8 -3
View File
@@ -69,12 +69,17 @@ def on_user_loaded(_sender, user: Union["Account", "EndUser"]):
if user:
try:
current_span = get_current_span()
if not current_span.is_recording():
return
tenant_id = extract_tenant_id(user)
if not tenant_id:
return
if current_span:
current_span.set_attribute(DifySpanAttributes.TENANT_ID, tenant_id)
current_span.set_attribute(GenAIAttributes.USER_ID, user.id)
current_span.set_attributes(
{
DifySpanAttributes.TENANT_ID: tenant_id,
GenAIAttributes.USER_ID: user.id,
}
)
except Exception:
logger.exception("Error setting tenant and user attributes")
pass
+63 -2
View File
@@ -9483,7 +9483,8 @@ data required for dashboard initialization.
Authentication would create circular dependency (can't login without dashboard loading).
Only non-sensitive configuration data should be returned by this endpoint.
Only non-sensitive configuration data should be returned by this endpoint. Authenticated
license detail is served separately by SystemFeatureLicenseApi.
#### Responses
@@ -9491,6 +9492,19 @@ Only non-sensitive configuration data should be returned by this endpoint.
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [SystemFeatureModel](#systemfeaturemodel)<br> |
### [GET] /system-features/license
**Get full license detail (status, expiry, workspace/seat usage)**
Get license status and usage detail
Authenticated counterpart to the license *status* exposed on the public
system-features endpoint.
#### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [LicenseModel](#licensemodel)<br> |
### [POST] /tag-bindings
#### Request Body
@@ -11103,6 +11117,19 @@ Returns permission flags that control workspace features like member invitations
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [PluginInstallTaskStartResponse](#plugininstalltaskstartresponse)<br> |
### [GET] /workspaces/current/plugin/installed-ids
#### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| category | query | Plugin category to include | Yes | string, <br>**Available values:** "agent-strategy", "datasource", "extension", "model", "tool", "trigger" |
#### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Success | **application/json**: [PluginInstalledIdsResponse](#plugininstalledidsresponse)<br> |
### [GET] /workspaces/current/plugin/list
#### Parameters
@@ -11361,8 +11388,11 @@ Returns permission flags that control workspace features like member invitations
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| language | query | Language used for localized label and description search | No | string, <br>**Available values:** "en_US", "ja_JP", "pt_BR", "zh_Hans", <br>**Default:** en_US |
| page | query | Page number | No | integer, <br>**Default:** 1 |
| page_size | query | Page size (1-256) | No | integer, <br>**Default:** 256 |
| query | query | Case-insensitive search query | No | string |
| tags | query | Match any plugin tag | No | [ string ] |
| category | path | | Yes | string |
#### Responses
@@ -17318,6 +17348,14 @@ Default model entity.
| tool_name | string | | Yes |
| type | string | | Yes |
#### DeploymentEdition
Enum representing the deployment edition of the platform.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| DeploymentEdition | string | Enum representing the deployment edition of the platform. | |
#### DismissNotificationPayload
| Name | Type | Description | Required |
@@ -18815,6 +18853,12 @@ Enum class for large language model mode.
| ---- | ---- | ----------- | -------- |
| LicenseStatus | string | | |
#### LicenseStatusModel
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| status | [LicenseStatus](#licensestatus) | | Yes |
#### LimitationModel
| Name | Type | Description | Required |
@@ -19559,6 +19603,7 @@ Coarse node-level status used by Inspector to pick a banner.
| ---- | ---- | ----------- | -------- |
| avatar | string | | No |
| email | string | | Yes |
| id | string | | Yes |
| interface_language | string | | Yes |
| name | string | | Yes |
| timezone | string | | Yes |
@@ -20283,8 +20328,11 @@ Shared permission levels for resources (datasets, credentials, etc.)
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| language | string, <br>**Available values:** "en_US", "ja_JP", "pt_BR", "zh_Hans", <br>**Default:** en_US | Language used for localized label and description search<br>*Enum:* `"en_US"`, `"ja_JP"`, `"pt_BR"`, `"zh_Hans"` | No |
| page | integer, <br>**Default:** 1 | Page number | No |
| page_size | integer, <br>**Default:** 256 | Page size (1-256) | No |
| query | string | Case-insensitive search query | No |
| tags | [ string ] | Match any plugin tag | No |
#### PluginCategoryListResponse
@@ -20485,6 +20533,18 @@ Shared permission levels for resources (datasets, credentials, etc.)
| ---- | ---- | ----------- | -------- |
| plugins | [ [PluginInstallationItemResponse](#plugininstallationitemresponse) ] | | Yes |
#### PluginInstalledIdsQuery
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| category | [PluginCategory](#plugincategory) | Plugin category to include | Yes |
#### PluginInstalledIdsResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| plugin_ids | [ string ] | | Yes |
#### PluginListResponse
| Name | Type | Description | Required |
@@ -22026,6 +22086,7 @@ Model class for provider system configuration response.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| branding | [BrandingModel](#brandingmodel) | | Yes |
| deployment_edition | [DeploymentEdition](#deploymentedition) | | Yes |
| enable_app_deploy | boolean | | Yes |
| enable_change_email | boolean, <br>**Default:** true | | Yes |
| enable_collaboration_mode | boolean, <br>**Default:** true | | Yes |
@@ -22042,7 +22103,7 @@ Model class for provider system configuration response.
| is_allow_register | boolean | | Yes |
| is_email_setup | boolean | | Yes |
| knowledge_fs_enabled | boolean | | Yes |
| license | [LicenseModel](#licensemodel) | | Yes |
| license | [LicenseStatusModel](#licensestatusmodel) | | Yes |
| max_plugin_package_size | integer, <br>**Default:** 15728640 | | Yes |
| plugin_installation_permission | [PluginInstallationPermissionModel](#plugininstallationpermissionmodel) | | Yes |
| plugin_manager | [PluginManagerModel](#pluginmanagermodel) | | Yes |
+16 -22
View File
@@ -1058,6 +1058,14 @@ Button styles for user actions.
| auto_generate | boolean | Automatically generate the conversation name. When `true`, the `name` field is ignored. | No |
| name | string | Conversation name. Required when `auto_generate` is `false`. | No |
#### DeploymentEdition
Enum representing the deployment edition of the platform.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| DeploymentEdition | string | Enum representing the deployment edition of the platform. | |
#### EmailCodeLoginSendPayload
| Name | Type | Description | Required |
@@ -1284,33 +1292,18 @@ Parsed multipart form fields for HITL uploads.
| ---- | ---- | ----------- | -------- |
| JsonValue | | | |
#### LicenseLimitationModel
- enabled: whether this limit is enforced
- size: current usage count
- limit: maximum allowed count; 0 means unlimited
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| enabled | boolean | Whether this limit is currently active | Yes |
| limit | integer | Maximum number of resources allowed; 0 means no limit | Yes |
| size | integer | Number of resources already consumed | Yes |
#### LicenseModel
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| expired_at | string | | Yes |
| seats | [LicenseLimitationModel](#licenselimitationmodel) | | Yes |
| status | [LicenseStatus](#licensestatus) | | Yes |
| workspaces | [LicenseLimitationModel](#licenselimitationmodel) | | Yes |
#### LicenseStatus
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| LicenseStatus | string | | |
#### LicenseStatusModel
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| status | [LicenseStatus](#licensestatus) | | Yes |
#### LoginPayload
| Name | Type | Description | Required |
@@ -1567,6 +1560,7 @@ Default configuration for form inputs.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| branding | [BrandingModel](#brandingmodel) | | Yes |
| deployment_edition | [DeploymentEdition](#deploymentedition) | | Yes |
| enable_app_deploy | boolean | | Yes |
| enable_change_email | boolean, <br>**Default:** true | | Yes |
| enable_collaboration_mode | boolean, <br>**Default:** true | | Yes |
@@ -1583,7 +1577,7 @@ Default configuration for form inputs.
| is_allow_register | boolean | | Yes |
| is_email_setup | boolean | | Yes |
| knowledge_fs_enabled | boolean | | Yes |
| license | [LicenseModel](#licensemodel) | | Yes |
| license | [LicenseStatusModel](#licensestatusmodel) | | Yes |
| max_plugin_package_size | integer, <br>**Default:** 15728640 | | Yes |
| plugin_installation_permission | [PluginInstallationPermissionModel](#plugininstallationpermissionmodel) | | Yes |
| plugin_manager | [PluginManagerModel](#pluginmanagermodel) | | Yes |
+10 -10
View File
@@ -442,9 +442,9 @@ class AccountService:
# 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():
# get_license() carries the full license payload that server-side enforcement needs;
# the public system-features endpoint exposes only license status.
if not FeatureService.get_license().seats.is_available():
raise SeatsLimitExceededError("licensed seats limit exceeded")
if dify_config.BILLING_ENABLED and BillingService.is_email_in_freeze(email):
@@ -1229,12 +1229,12 @@ class AccountService:
if hour_limit_count >= 1:
redis_client.setex(freeze_key, 60 * 60, 1)
return True
else:
redis_client.setex(hour_limit_key, 60 * 10, hour_limit_count + 1) # first time limit 10 minutes
# add hour limit count
redis_client.incr(hour_limit_key)
redis_client.expire(hour_limit_key, 60 * 60)
# First strike claims a 10-minute window atomically; a concurrent
# over-limit request that loses the claim is the second strike and
# freezes the IP for an hour.
if not redis_client.set(hour_limit_key, 1, ex=60 * 10, nx=True):
redis_client.setex(freeze_key, 60 * 60, 1)
return True
@@ -1332,7 +1332,7 @@ class TenantService:
):
raise WorkSpaceNotAllowedCreateError()
workspaces = FeatureService.get_system_features().license.workspaces
workspaces = FeatureService.get_license().workspaces
if not workspaces.is_available():
raise WorkspacesLimitExceededError()
@@ -2012,7 +2012,7 @@ class RegisterService:
if (
FeatureService.get_system_features().is_allow_create_workspace
and create_workspace_required
and FeatureService.get_system_features().license.workspaces.is_available()
and FeatureService.get_license().workspaces.is_available()
):
try:
TenantService.create_owner_tenant(account, session=session)
+6 -29
View File
@@ -7,6 +7,7 @@ from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from sqlalchemy.sql.elements import ColumnElement
from core.agent.publish_visibility import agent_has_workflow_callable_active_snapshot
from libs.helper import to_timestamp
from models import Account
from models.agent import (
@@ -271,6 +272,8 @@ class AgentComposerService:
source_snapshot_id: str | None = None,
idempotency_key: str | None = None,
) -> dict[str, Any]:
"""Copy a callable roster Agent snapshot into a workflow-owned inline Agent."""
workflow = cls._get_draft_workflow(session=session, tenant_id=tenant_id, app_id=app_id)
binding = cls._require_binding(
cls._get_workflow_binding(session=session, tenant_id=tenant_id, workflow_id=workflow.id, node_id=node_id)
@@ -296,6 +299,8 @@ class AgentComposerService:
source_agent = cls._require_agent(session=session, tenant_id=tenant_id, agent_id=source_agent_id)
if source_agent.scope != AgentScope.ROSTER or source_agent.status != AgentStatus.ACTIVE:
raise InvalidComposerConfigError("Source agent must be an active roster agent.")
if not agent_has_workflow_callable_active_snapshot(session=session, agent=source_agent):
raise InvalidComposerConfigError("Source agent must have a published config snapshot.")
source_version = cls._require_version(
session=session,
tenant_id=tenant_id,
@@ -529,39 +534,11 @@ class AgentComposerService:
)
if not active_version:
return False
if agent.source in APP_BACKED_AGENT_SOURCES and not cls._has_publish_visible_revision(
session=session,
tenant_id=tenant_id,
agent_id=agent.id,
snapshot_id=agent.active_config_snapshot_id,
):
if not agent_has_workflow_callable_active_snapshot(session=session, agent=agent):
return False
return _agent_soul_config_json(agent_soul) == _agent_soul_config_json(active_version.config_snapshot_dict)
@classmethod
def _has_publish_visible_revision(
cls, *, session: Session, tenant_id: str, agent_id: str, snapshot_id: str
) -> bool:
revisions = session.scalars(
select(AgentConfigRevision.operation).where(
AgentConfigRevision.tenant_id == tenant_id,
AgentConfigRevision.agent_id == agent_id,
AgentConfigRevision.current_snapshot_id == snapshot_id,
)
).all()
return any(
operation
in {
AgentConfigRevisionOperation.PUBLISH_DRAFT,
AgentConfigRevisionOperation.SAVE_NEW_VERSION,
AgentConfigRevisionOperation.SAVE_TO_ROSTER,
AgentConfigRevisionOperation.RESTORE_VERSION,
}
for operation in revisions
)
@classmethod
def publish_agent_app_draft(
cls, *, session: Session, tenant_id: str, agent_id: str, account_id: str, version_note: str | None = None
+252 -52
View File
@@ -1,8 +1,11 @@
from __future__ import annotations
import json
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import Any
import sqlalchemy as sa
@@ -11,6 +14,7 @@ from sqlalchemy.orm import aliased
from configs import dify_config
from core.app.entities.app_invoke_entities import InvokeFrom
from graphon.enums import WorkflowNodeExecutionStatus
from libs.helper import convert_datetime_to_date, escape_like_pattern, to_timestamp
from models.agent import WorkflowAgentNodeBinding
from models.enums import CreatorUserRole, MessageStatus
@@ -194,11 +198,12 @@ class AgentObservabilityService:
self, *, app: App, agent_id: str, conversation_id: str, params: AgentLogQueryParams
) -> dict[str, Any]:
source_filters = self.resolve_source_filters(params.sources)
rows: list[Message] = []
rows: list[dict[str, Any]] = []
for source_filter in source_filters:
if source_filter.kind in {"all", "webapp"}:
rows.extend(
self._list_webapp_messages(
self.serialize_log_message(message)
for message in self._list_webapp_messages(
app=app,
conversation_id=conversation_id,
params=params,
@@ -216,18 +221,18 @@ class AgentObservabilityService:
)
)
deduped = {message.id: message for message in rows}
sort_column = Message.created_at if params.sort_by == "created_at" else Message.updated_at
deduped = {row["id"]: row for row in rows}
sort_key = "created_at" if params.sort_by == "created_at" else "updated_at"
sorted_rows = sorted(
deduped.values(),
key=lambda message: (getattr(message, sort_column.key), message.id),
key=lambda row: (row[sort_key] or 0, row["id"]),
reverse=params.sort_order != "asc",
)
total = len(sorted_rows)
start = (params.page - 1) * params.limit
end = start + params.limit
return {
"data": [self.serialize_log_message(message) for message in sorted_rows[start:end]],
"data": sorted_rows[start:end],
"page": params.page,
"limit": params.limit,
"total": total,
@@ -284,22 +289,20 @@ class AgentObservabilityService:
workflow_app = aliased(App)
stmt = (
select(
Conversation,
WorkflowNodeExecutionModel.id.label("node_execution_id"),
WorkflowNodeExecutionModel.title.label("node_title"),
WorkflowNodeExecutionModel.status.label("node_status"),
WorkflowNodeExecutionModel.created_by_role.label("node_created_by_role"),
WorkflowNodeExecutionModel.created_by.label("node_created_by"),
WorkflowNodeExecutionModel.created_at.label("node_created_at"),
WorkflowNodeExecutionModel.finished_at.label("node_finished_at"),
workflow_app,
WorkflowAgentNodeBinding.workflow_id,
WorkflowAgentNodeBinding.workflow_version,
WorkflowAgentNodeBinding.node_id,
func.count(sa.distinct(Message.id)).label("message_count"),
func.max(Message.created_at).label("created_at"),
func.max(Message.updated_at).label("updated_at"),
func.sum(sa.case((Message.status == MessageStatus.PAUSED, 1), else_=0)).label("paused_count"),
func.sum(
sa.case((or_(Message.error.is_not(None), Message.status == MessageStatus.ERROR), 1), else_=0)
).label("failed_count"),
)
.select_from(Message)
.join(Conversation, Conversation.id == Message.conversation_id)
.join(WorkflowRun, WorkflowRun.id == Message.workflow_run_id)
.select_from(WorkflowNodeExecutionModel)
.join(WorkflowRun, WorkflowRun.id == WorkflowNodeExecutionModel.workflow_run_id)
.join(
WorkflowAgentNodeBinding,
and_(
@@ -310,40 +313,32 @@ class AgentObservabilityService:
WorkflowAgentNodeBinding.workflow_version == WorkflowRun.version,
),
)
.join(
WorkflowNodeExecutionModel,
and_(
WorkflowNodeExecutionModel.workflow_run_id == WorkflowRun.id,
WorkflowNodeExecutionModel.node_id == WorkflowAgentNodeBinding.node_id,
),
)
.join(workflow_app, workflow_app.id == WorkflowAgentNodeBinding.app_id)
.where(Message.workflow_run_id.is_not(None), Conversation.app_id == WorkflowAgentNodeBinding.app_id)
.group_by(
Conversation.id,
workflow_app.id,
WorkflowAgentNodeBinding.workflow_id,
WorkflowAgentNodeBinding.workflow_version,
WorkflowAgentNodeBinding.node_id,
.where(
WorkflowNodeExecutionModel.tenant_id == app.tenant_id,
WorkflowNodeExecutionModel.app_id == WorkflowAgentNodeBinding.app_id,
WorkflowNodeExecutionModel.workflow_id == WorkflowAgentNodeBinding.workflow_id,
WorkflowNodeExecutionModel.node_id == WorkflowAgentNodeBinding.node_id,
)
)
stmt = self._apply_observability_filters(stmt, params=params, source_filter=source_filter)
stmt = self._apply_workflow_node_filters(stmt, params=params, workflow_app=workflow_app)
stmt = self._apply_workflow_source_filter(stmt, source_filter)
rows = list(self._session.execute(stmt).all())
return [
self._serialize_conversation_log(
conversation=row[0],
message_count=row.message_count,
paused_count=row.paused_count,
failed_count=row.failed_count,
self._serialize_workflow_execution_log(
node_execution_id=row.node_execution_id,
title=row.node_title,
status=row.node_status,
created_by_role=row.node_created_by_role,
created_by=row.node_created_by,
created_at=row.node_created_at,
finished_at=row.node_finished_at,
source=self._serialize_workflow_source(
app=row[1],
app=row[7],
workflow_id=row.workflow_id,
workflow_version=row.workflow_version,
node_id=row.node_id,
),
created_at=row.created_at,
updated_at=row.updated_at,
)
for row in rows
]
@@ -363,10 +358,12 @@ class AgentObservabilityService:
conversation_id: str,
params: AgentLogQueryParams,
source_filter: AgentSourceFilter,
) -> list[Message]:
) -> list[dict[str, Any]]:
workflow_app = aliased(App)
stmt = (
select(Message)
.join(WorkflowRun, WorkflowRun.id == Message.workflow_run_id)
select(WorkflowNodeExecutionModel)
.select_from(WorkflowNodeExecutionModel)
.join(WorkflowRun, WorkflowRun.id == WorkflowNodeExecutionModel.workflow_run_id)
.join(
WorkflowAgentNodeBinding,
and_(
@@ -377,18 +374,23 @@ class AgentObservabilityService:
WorkflowAgentNodeBinding.workflow_version == WorkflowRun.version,
),
)
.join(
WorkflowNodeExecutionModel,
and_(
WorkflowNodeExecutionModel.workflow_run_id == WorkflowRun.id,
WorkflowNodeExecutionModel.node_id == WorkflowAgentNodeBinding.node_id,
),
.join(workflow_app, workflow_app.id == WorkflowAgentNodeBinding.app_id)
.where(
WorkflowNodeExecutionModel.id == conversation_id,
WorkflowNodeExecutionModel.tenant_id == app.tenant_id,
WorkflowNodeExecutionModel.app_id == WorkflowAgentNodeBinding.app_id,
WorkflowNodeExecutionModel.workflow_id == WorkflowAgentNodeBinding.workflow_id,
WorkflowNodeExecutionModel.node_id == WorkflowAgentNodeBinding.node_id,
)
.where(Message.conversation_id == conversation_id)
)
stmt = self._apply_message_filters(stmt, params=params, source_filter=source_filter)
stmt = self._apply_workflow_node_filters(stmt, params=params, workflow_app=workflow_app)
stmt = self._apply_workflow_source_filter(stmt, source_filter)
return list(self._session.scalars(stmt.order_by(Message.created_at.desc(), Message.id.desc())).all())
executions = list(
self._session.scalars(
stmt.order_by(WorkflowNodeExecutionModel.created_at.desc(), WorkflowNodeExecutionModel.id.desc())
).all()
)
return [self.serialize_workflow_node_message(execution) for execution in executions]
def _list_workflow_sources(self, *, app: App, agent_id: str) -> list[dict[str, Any]]:
workflow_app = aliased(App)
@@ -443,6 +445,62 @@ class AgentObservabilityService:
stmt = cls._apply_status_filter(stmt, params.statuses)
return stmt
@classmethod
def _apply_workflow_node_filters(cls, stmt, *, params: AgentLogQueryParams, workflow_app):
if params.start:
stmt = stmt.where(WorkflowNodeExecutionModel.created_at >= params.start)
if params.end:
stmt = stmt.where(WorkflowNodeExecutionModel.created_at < params.end)
if params.keyword:
escaped_keyword = escape_like_pattern(params.keyword)
pattern = f"%{escaped_keyword}%"
stmt = stmt.where(
or_(
WorkflowNodeExecutionModel.inputs.ilike(pattern, escape="\\"),
WorkflowNodeExecutionModel.outputs.ilike(pattern, escape="\\"),
WorkflowNodeExecutionModel.error.ilike(pattern, escape="\\"),
WorkflowNodeExecutionModel.title.ilike(pattern, escape="\\"),
workflow_app.name.ilike(pattern, escape="\\"),
)
)
if params.statuses:
stmt = cls._apply_workflow_node_status_filter(stmt, params.statuses)
return stmt
@staticmethod
def _apply_workflow_node_status_filter(stmt, statuses: tuple[str, ...]):
conditions = []
for status in statuses:
normalized = status.strip().lower()
if normalized in {"success", "normal"}:
conditions.append(WorkflowNodeExecutionModel.status == WorkflowNodeExecutionStatus.SUCCEEDED)
elif normalized in {"failed", "error"}:
conditions.append(
WorkflowNodeExecutionModel.status.in_(
(
WorkflowNodeExecutionStatus.FAILED,
WorkflowNodeExecutionStatus.EXCEPTION,
WorkflowNodeExecutionStatus.STOPPED,
)
)
)
elif normalized == "paused":
conditions.append(
WorkflowNodeExecutionModel.status.in_(
(
WorkflowNodeExecutionStatus.PAUSED,
WorkflowNodeExecutionStatus.PENDING,
WorkflowNodeExecutionStatus.RUNNING,
WorkflowNodeExecutionStatus.RETRY,
)
)
)
else:
raise ValueError(f"Unsupported status: {status}")
if not conditions:
return stmt
return stmt.where(or_(*conditions))
@staticmethod
def _apply_workflow_source_filter(stmt, source_filter: AgentSourceFilter):
if source_filter.app_id:
@@ -505,6 +563,148 @@ class AgentObservabilityService:
"updated_at": to_timestamp(updated_at or conversation.updated_at),
}
@classmethod
def _serialize_workflow_execution_log(
cls,
*,
node_execution_id: str,
title: str,
status: object,
created_by_role: object,
created_by: str,
created_at: datetime,
finished_at: datetime | None,
source: dict[str, Any],
) -> dict[str, Any]:
created_by_role_value = cls._enum_value(created_by_role)
return {
"id": node_execution_id,
"conversation_id": node_execution_id,
"title": title,
"end_user_id": created_by if created_by_role_value == CreatorUserRole.END_USER.value else None,
"message_count": 1,
"user_rate": None,
"operation_rate": None,
"unread": False,
"source": source,
"status": cls._workflow_node_status(status),
"created_at": to_timestamp(created_at),
"updated_at": to_timestamp(finished_at or created_at),
}
@classmethod
def serialize_workflow_node_message(cls, node_execution: WorkflowNodeExecutionModel) -> dict[str, Any]:
inputs = cls._json_mapping(node_execution.inputs)
outputs = cls._json_mapping(node_execution.outputs)
metadata = cls._json_mapping(node_execution.execution_metadata)
agent_log = cls._mapping_value(metadata, "agent_log")
agent_backend = cls._mapping_value(agent_log, "agent_backend")
usage = cls._mapping_value(agent_backend, "usage")
prompt_tokens = cls._int_value(usage.get("prompt_tokens"))
completion_tokens = cls._int_value(usage.get("completion_tokens"))
total_tokens = cls._int_value(usage.get("total_tokens") or metadata.get("total_tokens"))
if not total_tokens:
total_tokens = prompt_tokens + completion_tokens
created_by_role = cls._enum_value(node_execution.created_by_role)
return {
"id": node_execution.id,
"message_id": node_execution.id,
"conversation_id": node_execution.id,
"query": cls._workflow_node_query(inputs, fallback=node_execution.title),
"answer": cls._workflow_node_answer(outputs),
"status": cls._workflow_node_status(node_execution.status),
"error": node_execution.error,
"from_end_user_id": (
node_execution.created_by if created_by_role == CreatorUserRole.END_USER.value else None
),
"from_account_id": (
node_execution.created_by if created_by_role == CreatorUserRole.ACCOUNT.value else None
),
"message_tokens": prompt_tokens,
"answer_tokens": completion_tokens,
"total_tokens": total_tokens,
"total_price": str(usage.get("total_price") or metadata.get("total_price") or Decimal(0)),
"currency": str(usage.get("currency") or metadata.get("currency") or ""),
"latency": float(usage.get("latency") or node_execution.elapsed_time or 0),
"created_at": to_timestamp(node_execution.created_at),
"updated_at": to_timestamp(node_execution.finished_at or node_execution.created_at),
}
@staticmethod
def _json_mapping(value: object) -> Mapping[str, Any]:
if isinstance(value, Mapping):
return value
if not isinstance(value, str) or not value:
return {}
try:
parsed = json.loads(value)
except (TypeError, ValueError):
return {}
return parsed if isinstance(parsed, Mapping) else {}
@staticmethod
def _mapping_value(value: Mapping[str, Any], key: str) -> Mapping[str, Any]:
nested = value.get(key)
return nested if isinstance(nested, Mapping) else {}
@staticmethod
def _enum_value(value: object) -> str:
return str(value.value) if isinstance(value, Enum) else str(value)
@staticmethod
def _int_value(value: object) -> int:
if not isinstance(value, (str, int, float, Decimal)):
return 0
try:
return int(value)
except (TypeError, ValueError):
return 0
@classmethod
def _workflow_node_query(cls, inputs: Mapping[str, Any], *, fallback: str) -> str:
request_data = cls._mapping_value(inputs, "agent_backend_request")
composition = cls._mapping_value(request_data, "composition")
layers = composition.get("layers")
prompts: list[str] = []
if isinstance(layers, list):
for layer_name in ("workflow_node_job_prompt", "workflow_user_prompt"):
for layer in layers:
if not isinstance(layer, Mapping) or layer.get("name") != layer_name:
continue
config = cls._mapping_value(layer, "config")
user_prompt = config.get("user")
if isinstance(user_prompt, str) and user_prompt.strip():
prompts.append(user_prompt.strip())
return "\n\n".join(prompts) or fallback
@staticmethod
def _workflow_node_answer(outputs: Mapping[str, Any]) -> str:
for key in ("output", "text", "answer"):
value = outputs.get(key)
if isinstance(value, str):
return value
return json.dumps(outputs, ensure_ascii=False) if outputs else ""
@classmethod
def _workflow_node_status(cls, status: object) -> str:
value = cls._enum_value(status)
if value in {
WorkflowNodeExecutionStatus.FAILED.value,
WorkflowNodeExecutionStatus.EXCEPTION.value,
WorkflowNodeExecutionStatus.STOPPED.value,
}:
return "failed"
if value in {
WorkflowNodeExecutionStatus.PAUSED.value,
WorkflowNodeExecutionStatus.PENDING.value,
WorkflowNodeExecutionStatus.RUNNING.value,
WorkflowNodeExecutionStatus.RETRY.value,
}:
return "paused"
return "success"
@staticmethod
def _conversation_status(*, paused_count: int, failed_count: int) -> str:
if paused_count:
+28 -17
View File
@@ -6,6 +6,7 @@ from sqlalchemy.exc import IntegrityError
from clients.agent_backend.session_cleanup import AgentBackendSessionCleanupPayload
from constants.model_template import default_app_templates
from core.agent.publish_visibility import workflow_callable_active_snapshot_filter
from core.app.apps.agent_app.session_store import AgentAppRuntimeSessionStore
from core.app.entities.app_invoke_entities import InvokeFrom
from libs.datetime_utils import naive_utc_now
@@ -225,8 +226,11 @@ class AgentRosterService:
def list_invite_options(
self, *, tenant_id: str, page: int = 1, limit: int = 20, keyword: str | None = None, app_id: str | None = None
) -> dict[str, Any]:
"""List active roster Agents whose published snapshot can be called by Workflow."""
stmt = self._build_roster_agents_stmt(tenant_id=tenant_id, keyword=keyword).where(
Agent.active_config_has_model.is_(True)
Agent.active_config_has_model.is_(True),
workflow_callable_active_snapshot_filter(),
)
total = self._session.scalar(select(func.count()).select_from(stmt.subquery())) or 0
agents = list(self._session.scalars(stmt.offset((page - 1) * limit).limit(limit)).all())
@@ -660,15 +664,18 @@ class AgentRosterService:
agent_id: str,
account_id: str,
draft_type: AgentConfigDraftType = AgentConfigDraftType.DEBUG_BUILD,
commit: bool = True,
) -> str:
"""Start a new scoped console conversation for the current Agent App editor.
If this account already has a mapping for the requested draft surface, the previous
conversation is abandoned first: any ACTIVE conversation-owned Agent
runtime sessions for that old conversation are sent through best-effort
backend cleanup and then retired locally even when enqueueing fails. The
other draft surface is left untouched.
conversation is abandoned after the replacement mapping is committed: any ACTIVE
conversation-owned Agent runtime sessions for that old conversation are sent through
best-effort backend cleanup and then retired locally even when enqueueing fails. This
order prevents a failed database commit from retiring the still-current runtime session.
The other draft surface is left untouched.
A user and draft surface own one current mapping. If new-conversation requests overlap,
the last committed rotation becomes current and earlier response IDs cannot be continued.
"""
agent = self._session.scalar(
@@ -691,6 +698,7 @@ class AgentRosterService:
app_id=backing_app_id,
account_id=account_id,
)
previous_conversation: tuple[str, str] | None = None
mapping = self._session.scalar(
select(AgentDebugConversation).where(
AgentDebugConversation.tenant_id == tenant_id,
@@ -714,19 +722,22 @@ class AgentRosterService:
previous_app_id = mapping.app_id
previous_conversation_id = mapping.conversation_id
if previous_conversation_id:
self._cleanup_debug_conversation_runtime_sessions(
tenant_id=tenant_id,
agent_id=agent_id,
account_id=account_id,
draft_type=draft_type,
app_id=previous_app_id or backing_app_id,
conversation_id=previous_conversation_id,
)
previous_conversation = (previous_app_id or backing_app_id, previous_conversation_id)
mapping.app_id = backing_app_id
mapping.conversation_id = conversation_id
self._session.flush()
if commit:
self._session.commit()
self._session.commit()
if previous_conversation:
previous_app_id, previous_conversation_id = previous_conversation
self._cleanup_debug_conversation_runtime_sessions(
tenant_id=tenant_id,
agent_id=agent_id,
account_id=account_id,
draft_type=draft_type,
app_id=previous_app_id,
conversation_id=previous_conversation_id,
)
return conversation_id
def _cleanup_debug_conversation_runtime_sessions(
@@ -739,8 +750,8 @@ class AgentRosterService:
app_id: str,
conversation_id: str,
) -> None:
session_store = AgentAppRuntimeSessionStore()
try:
session_store = AgentAppRuntimeSessionStore()
stored_sessions = session_store.list_active_sessions_for_conversation(
tenant_id=tenant_id,
app_id=app_id,
@@ -8,6 +8,7 @@ from pydantic import ValidationError
from sqlalchemy import select
from sqlalchemy.orm import Session
from core.agent.publish_visibility import workflow_callable_active_snapshot_filter
from core.workflow.nodes.agent_v2.validators import WorkflowAgentNodeValidationError, WorkflowAgentNodeValidator
from models.agent import (
Agent,
@@ -447,6 +448,8 @@ class WorkflowAgentPublishService:
node_id: str,
agent_id: str,
) -> tuple[Agent, str]:
"""Resolve an active roster Agent whose published snapshot is callable."""
agent = session.scalar(
select(Agent)
.where(
@@ -454,11 +457,12 @@ class WorkflowAgentPublishService:
Agent.id == agent_id,
Agent.scope == AgentScope.ROSTER,
Agent.status == AgentStatus.ACTIVE,
workflow_callable_active_snapshot_filter(),
)
.limit(1)
)
if agent is None:
raise ValueError(f"Workflow Agent node {node_id} references an unavailable roster agent.")
raise ValueError(f"Workflow Agent node {node_id} references an unavailable or unpublished roster agent.")
if agent.scope != AgentScope.ROSTER:
raise ValueError(f"Workflow Agent node {node_id} roster_agent binding must reference a roster agent.")
if not agent.active_config_snapshot_id:
+15 -3
View File
@@ -9,6 +9,7 @@ import contexts
from core.app.app_config.easy_ui_based_app.agent.manager import AgentConfigManager
from core.plugin.impl.agent import PluginAgentClient
from core.plugin.impl.exc import PluginDaemonClientSideError
from core.tools.entities.tool_entities import EmojiIconDict
from core.tools.tool_manager import ToolManager
from libs.login import current_user
from models import Account
@@ -105,11 +106,22 @@ class AgentService:
tool_output = tool_outputs.get(tool_name, {})
tool_meta_data = tool_meta.get(tool_name, {})
tool_config = tool_meta_data.get("tool_config", {})
if tool_config.get("tool_provider_type", "") != "dataset-retrieval":
tool_provider_type = tool_config.get("tool_provider_type", "")
tool_provider_id = tool_config.get("tool_provider", "")
if not tool_provider_type:
tool_entity = find_agent_tool(tool_name)
if tool_entity:
tool_provider_type = tool_entity.provider_type
tool_provider_id = tool_provider_id or tool_entity.provider_id
tool_icon: str | EmojiIconDict = ""
if tool_provider_type and tool_provider_type != "dataset-retrieval":
tool_icon = ToolManager.get_tool_icon(
tenant_id=app_model.tenant_id,
provider_type=tool_config.get("tool_provider_type", ""),
provider_id=tool_config.get("tool_provider", ""),
provider_type=tool_provider_type,
provider_id=tool_provider_id,
)
if not tool_icon:
tool_entity = find_agent_tool(tool_name)
+50 -23
View File
@@ -5,6 +5,7 @@ from pydantic import BaseModel, ConfigDict, Field
from configs import dify_config
from constants.dsl_version import CURRENT_APP_DSL_VERSION
from enums.cloud_plan import CloudPlan
from enums.deployment_edition import DeploymentEdition
from enums.hosted_provider import HostedTrialProvider
from services.billing_service import BillingInfo, BillingService
from services.enterprise.enterprise_service import EnterpriseService
@@ -75,8 +76,11 @@ class LicenseStatus(StrEnum):
LOST = "lost"
class LicenseModel(FeatureResponseModel):
class LicenseStatusModel(FeatureResponseModel):
status: LicenseStatus = LicenseStatus.NONE
class LicenseModel(LicenseStatusModel):
expired_at: str = ""
workspaces: LicenseLimitationModel = LicenseLimitationModel(enabled=False, size=0, limit=0)
seats: LicenseLimitationModel = LicenseLimitationModel(enabled=False, size=0, limit=0)
@@ -162,6 +166,7 @@ class PluginManagerModel(FeatureResponseModel):
class SystemFeatureModel(FeatureResponseModel):
deployment_edition: DeploymentEdition
enable_app_deploy: bool = False
sso_enforced_for_signin: bool = False
sso_enforced_for_signin_protocol: str = ""
@@ -174,7 +179,7 @@ class SystemFeatureModel(FeatureResponseModel):
is_allow_register: bool = False
is_allow_create_workspace: bool = False
is_email_setup: bool = False
license: LicenseModel = LicenseModel()
license: LicenseStatusModel = LicenseStatusModel()
branding: BrandingModel = BrandingModel()
webapp_auth: WebAppAuthModel = WebAppAuthModel()
plugin_installation_permission: PluginInstallationPermissionModel = PluginInstallationPermissionModel()
@@ -251,8 +256,8 @@ class FeatureService:
)
@classmethod
def get_system_features(cls, is_authenticated: bool = False) -> SystemFeatureModel:
system_features = SystemFeatureModel()
def get_system_features(cls) -> SystemFeatureModel:
system_features = SystemFeatureModel(deployment_edition=dify_config.DEPLOYMENT_EDITION)
system_features.rbac_enabled = dify_config.RBAC_ENABLED
cls._fulfill_system_params_from_env(system_features)
@@ -262,7 +267,7 @@ class FeatureService:
system_features.webapp_auth.enabled = True
system_features.enable_change_email = False
system_features.plugin_manager.enabled = True
cls._fulfill_params_from_enterprise(system_features, is_authenticated)
cls._fulfill_params_from_enterprise(system_features)
if dify_config.MARKETPLACE_ENABLED:
system_features.enable_marketplace = True
@@ -272,6 +277,17 @@ class FeatureService:
return system_features
@classmethod
def get_license(cls) -> LicenseModel:
"""Return full license detail. Enterprise-only; requires an authenticated caller.
Non-enterprise deployments have no license, so an unconstrained default
(unlimited seats/workspaces) is returned.
"""
if not dify_config.ENTERPRISE_ENABLED:
return LicenseModel()
return cls._build_license(EnterpriseService.get_info())
@classmethod
def get_app_dsl_version(cls) -> str:
return CURRENT_APP_DSL_VERSION
@@ -285,6 +301,7 @@ class FeatureService:
system_features.is_allow_register = dify_config.ALLOW_REGISTER
system_features.is_allow_create_workspace = dify_config.ALLOW_CREATE_WORKSPACE
system_features.is_email_setup = dify_config.MAIL_TYPE is not None and dify_config.MAIL_TYPE != ""
system_features.enable_change_email = dify_config.ENABLE_CHANGE_EMAIL
system_features.enable_trial_app = dify_config.ENABLE_TRIAL_APP
system_features.enable_explore_banner = dify_config.ENABLE_EXPLORE_BANNER
system_features.enable_learn_app = dify_config.ENABLE_LEARN_APP
@@ -410,7 +427,27 @@ class FeatureService:
vector_space.limit = billing_info["vector_space"]["limit"]
@classmethod
def _fulfill_params_from_enterprise(cls, features: SystemFeatureModel, is_authenticated: bool = False):
def _build_license(cls, enterprise_info: dict) -> LicenseModel:
license_model = LicenseModel()
if license_info := enterprise_info.get("License"):
license_model.status = LicenseStatus(license_info.get("status", LicenseStatus.INACTIVE))
license_model.expired_at = license_info.get("expiredAt", "")
if workspaces_info := license_info.get("workspaces"):
license_model.workspaces = LicenseLimitationModel(
enabled=workspaces_info.get("enabled", False),
limit=workspaces_info.get("limit", 0),
size=workspaces_info.get("used", 0),
)
if seats_info := license_info.get("licensedSeats"):
license_model.seats = LicenseLimitationModel(
enabled=seats_info.get("enabled", False),
limit=seats_info.get("limit", 0),
size=seats_info.get("used", 0),
)
return license_model
@classmethod
def _fulfill_params_from_enterprise(cls, features: SystemFeatureModel):
enterprise_info = EnterpriseService.get_info()
if "SSOEnforcedForSignin" in enterprise_info:
@@ -450,24 +487,14 @@ class FeatureService:
)
features.webapp_auth.sso_config.protocol = enterprise_info.get("SSOEnforcedForWebProtocol", "")
# SECURITY NOTE: Only license *status* is exposed to unauthenticated callers
# so the login page can detect an expired/inactive license after force-logout.
# All other license details (expiry date, workspace usage) remain auth-gated.
# This behavior reflects prior internal review of information-leakage risks.
# SECURITY NOTE: system-features is unauthenticated, so it exposes only license
# *status* — enough for the login page to detect an expired/inactive license after
# force-logout. Full license detail (expiry, workspace/seat usage) is served
# separately by get_license() behind an authenticated endpoint.
if license_info := enterprise_info.get("License"):
features.license.status = LicenseStatus(license_info.get("status", LicenseStatus.INACTIVE))
if is_authenticated:
features.license.expired_at = license_info.get("expiredAt", "")
if workspaces_info := license_info.get("workspaces"):
features.license.workspaces.enabled = workspaces_info.get("enabled", False)
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)
features.license = LicenseStatusModel(
status=LicenseStatus(license_info.get("status", LicenseStatus.INACTIVE))
)
if "PluginInstallationPermission" in enterprise_info:
plugin_installation_info = enterprise_info["PluginInstallationPermission"]
+5 -7
View File
@@ -99,7 +99,7 @@ def _external_source_operation(
*,
request_headers: tuple[str, ...] = ("x-trace-id",),
) -> KnowledgeFSOperation:
"""Declare a source-connection operation restricted to dataset editors."""
"""Declare an external-source operation restricted to dataset editors."""
return _console_operation(
operation_id,
method,
@@ -347,12 +347,10 @@ KNOWLEDGE_FS_CONSOLE_OPERATIONS: Final[tuple[KnowledgeFSOperation, ...]] = (
rbac_permission=RBACPermission.DATASET_READONLY,
legacy_role="reader",
),
_console_operation(
operation_id="putKnowledgeSpacesByIdSourcesBySourceIdSyncPolicy",
method="PUT",
path="knowledge-spaces/{id}/sources/{sourceId}/sync-policy",
rbac_permission=RBACPermission.DATASET_EDIT,
legacy_role="dataset_editor",
_external_source_operation(
"putKnowledgeSpacesByIdSourcesBySourceIdSyncPolicy",
"PUT",
"knowledge-spaces/{id}/sources/{sourceId}/sync-policy",
),
_dataset_read_operation("getKnowledgeSpacesByIdDocuments", "knowledge-spaces/{id}/documents"),
_dataset_edit_operation("postKnowledgeSpacesByIdDocuments", "POST", "knowledge-spaces/{id}/documents"),
+2 -1
View File
@@ -4,6 +4,7 @@ from sqlalchemy.orm import Session
from configs import dify_config
from enums.cloud_plan import CloudPlan
from enums.deployment_edition import DeploymentEdition
from models.account import Tenant, TenantAccountJoin, TenantAccountRole
from services.account_service import TenantService
from services.feature_service import FeatureService
@@ -60,7 +61,7 @@ class WorkspaceService:
"remove_webapp_brand": remove_webapp_brand,
"replace_webapp_logo": replace_webapp_logo,
}
if dify_config.EDITION == "CLOUD":
if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
tenant_info["next_credit_reset_date"] = feature.next_credit_reset_date
from services.credit_pool_service import CreditPoolBalance, CreditPoolService
+2 -1
View File
@@ -1,10 +1,11 @@
from enum import StrEnum
from configs import dify_config
from enums.deployment_edition import DeploymentEdition
from services.workflow.entities import WorkflowScheduleCFSPlanEntity
# Determine queue names based on edition
if dify_config.EDITION == "CLOUD":
if dify_config.DEPLOYMENT_EDITION == DeploymentEdition.CLOUD:
# Cloud edition: separate queues for different tiers
_professional_queue = "workflow_professional"
_team_queue = "workflow_team"
@@ -318,6 +318,7 @@ def test_oauth_account_successful_retrieval(
assert response.status_code == 200
assert response.get_json() == {
"id": account.id,
"name": "Test User",
"email": account.email,
"avatar": "avatar_url",
@@ -38,7 +38,8 @@ class TestAccountService:
# Setup default mock returns
mock_feature_service.get_system_features.return_value.is_allow_register = True
mock_feature_service.get_system_features.return_value.is_allow_create_workspace = True
mock_feature_service.get_system_features.return_value.license.workspaces.is_available.return_value = True
mock_feature_service.get_license.return_value.workspaces.is_available.return_value = True
mock_feature_service.get_license.return_value.seats.is_available.return_value = True
mock_billing_service.is_email_in_freeze.return_value = False
mock_passport_service.return_value.issue.return_value = "mock_jwt_token"
@@ -405,7 +406,7 @@ class TestAccountService:
].get_system_features.return_value.is_allow_create_workspace = True
mock_external_service_dependencies[
"feature_service"
].get_system_features.return_value.license.workspaces.is_available.return_value = True
].get_license.return_value.workspaces.is_available.return_value = True
mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
account = AccountService.create_account_and_tenant(
email=email,
@@ -466,7 +467,7 @@ class TestAccountService:
].get_system_features.return_value.is_allow_create_workspace = True
mock_external_service_dependencies[
"feature_service"
].get_system_features.return_value.license.workspaces.is_available.return_value = False
].get_license.return_value.workspaces.is_available.return_value = False
mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
with pytest.raises(WorkspacesLimitExceededError):
@@ -492,7 +493,7 @@ class TestAccountService:
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
].get_license.return_value.seats.is_available.return_value = False
mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
with pytest.raises(SeatsLimitExceededError):
@@ -1274,7 +1275,8 @@ class TestTenantService:
):
# Setup default mock returns
mock_feature_service.get_system_features.return_value.is_allow_create_workspace = True
mock_feature_service.get_system_features.return_value.license.workspaces.is_available.return_value = True
mock_feature_service.get_license.return_value.workspaces.is_available.return_value = True
mock_feature_service.get_license.return_value.seats.is_available.return_value = True
mock_billing_service.is_email_in_freeze.return_value = False
yield {
@@ -2244,7 +2246,7 @@ class TestTenantService:
].get_system_features.return_value.is_allow_create_workspace = True
mock_external_service_dependencies[
"feature_service"
].get_system_features.return_value.license.workspaces.is_available.return_value = True
].get_license.return_value.workspaces.is_available.return_value = True
# Create account
account = AccountService.create_account(
@@ -2285,7 +2287,7 @@ class TestTenantService:
].get_system_features.return_value.is_allow_create_workspace = True
mock_external_service_dependencies[
"feature_service"
].get_system_features.return_value.license.workspaces.is_available.return_value = True
].get_license.return_value.workspaces.is_available.return_value = True
# Create account and existing tenant
account = AccountService.create_account(
@@ -2514,7 +2516,8 @@ class TestRegisterService:
# Setup default mock returns
mock_feature_service.get_system_features.return_value.is_allow_register = True
mock_feature_service.get_system_features.return_value.is_allow_create_workspace = True
mock_feature_service.get_system_features.return_value.license.workspaces.is_available.return_value = True
mock_feature_service.get_license.return_value.workspaces.is_available.return_value = True
mock_feature_service.get_license.return_value.seats.is_available.return_value = True
mock_billing_service.is_email_in_freeze.return_value = False
mock_passport_service.return_value.issue.return_value = "mock_jwt_token"
@@ -2631,7 +2634,7 @@ class TestRegisterService:
].get_system_features.return_value.is_allow_create_workspace = True
mock_external_service_dependencies[
"feature_service"
].get_system_features.return_value.license.workspaces.is_available.return_value = True
].get_license.return_value.workspaces.is_available.return_value = True
mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
# Execute registration
@@ -2675,7 +2678,7 @@ class TestRegisterService:
].get_system_features.return_value.is_allow_create_workspace = True
mock_external_service_dependencies[
"feature_service"
].get_system_features.return_value.license.workspaces.is_available.return_value = True
].get_license.return_value.workspaces.is_available.return_value = True
mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
# Execute registration with OAuth
@@ -2724,7 +2727,7 @@ class TestRegisterService:
].get_system_features.return_value.is_allow_create_workspace = True
mock_external_service_dependencies[
"feature_service"
].get_system_features.return_value.license.workspaces.is_available.return_value = True
].get_license.return_value.workspaces.is_available.return_value = True
mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
# Execute registration with pending status
@@ -2809,7 +2812,7 @@ class TestRegisterService:
].get_system_features.return_value.is_allow_create_workspace = True
mock_external_service_dependencies[
"feature_service"
].get_system_features.return_value.license.workspaces.is_available.return_value = False
].get_license.return_value.workspaces.is_available.return_value = False
mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
# with pytest.raises(AccountRegisterError, match="Workspace is not allowed to create."):
@@ -2888,7 +2891,7 @@ class TestRegisterService:
].get_system_features.return_value.is_allow_create_workspace = True
mock_external_service_dependencies[
"feature_service"
].get_system_features.return_value.license.workspaces.is_available.return_value = True
].get_license.return_value.workspaces.is_available.return_value = True
mock_external_service_dependencies["billing_service"].is_email_in_freeze.return_value = False
# Create tenant and inviter account
@@ -5,10 +5,12 @@ from faker import Faker
from sqlalchemy.orm import Session
from enums.cloud_plan import CloudPlan
from enums.deployment_edition import DeploymentEdition
from services.feature_service import (
FeatureModel,
FeatureService,
KnowledgeRateLimitModel,
LicenseModel,
LicenseStatus,
SystemFeatureModel,
)
@@ -273,6 +275,7 @@ class TestFeatureService:
tenant_id = self._create_test_tenant_id()
with patch("services.feature_service.dify_config") as mock_config:
mock_config.DEPLOYMENT_EDITION = DeploymentEdition.ENTERPRISE
mock_config.ENTERPRISE_ENABLED = True
mock_config.MARKETPLACE_ENABLED = True
mock_config.ENABLE_EMAIL_CODE_LOGIN = True
@@ -285,7 +288,7 @@ class TestFeatureService:
mock_config.PLUGIN_MAX_PACKAGE_SIZE = 100
# Act: Execute the method under test
result = FeatureService.get_system_features(is_authenticated=True)
result = FeatureService.get_system_features()
# Assert: Verify the expected outcomes
assert result is not None
@@ -320,12 +323,9 @@ class TestFeatureService:
assert result.webapp_auth.allow_email_password_login is False
assert result.webapp_auth.sso_config.protocol == "oidc"
# Verify license configuration
# Verify license status (public system-features exposes status only; detail lives on get_license)
assert result.license.status.value == "active"
assert result.license.expired_at == "2025-12-31"
assert result.license.workspaces.enabled is True
assert result.license.workspaces.limit == 5
assert result.license.workspaces.size == 2
assert not hasattr(result.license, "expired_at")
# Verify plugin installation permission
assert result.plugin_installation_permission.plugin_installation_scope == "official_only"
@@ -350,6 +350,7 @@ class TestFeatureService:
"""
# Arrange: Setup test data with exact same config as success test
with patch("services.feature_service.dify_config") as mock_config:
mock_config.DEPLOYMENT_EDITION = DeploymentEdition.ENTERPRISE
mock_config.ENTERPRISE_ENABLED = True
mock_config.MARKETPLACE_ENABLED = True
mock_config.ENABLE_EMAIL_CODE_LOGIN = True
@@ -360,20 +361,18 @@ class TestFeatureService:
mock_config.MAIL_TYPE = "smtp"
mock_config.PLUGIN_MAX_PACKAGE_SIZE = 100
# Act: Execute with is_authenticated=False
result = FeatureService.get_system_features(is_authenticated=False)
# Act: Execute the public (unauthenticated) system-features call
result = FeatureService.get_system_features()
# Assert: Basic structure
assert result is not None
assert isinstance(result, SystemFeatureModel)
# --- 1. Verify only license *status* is exposed to unauthenticated clients ---
# Detailed license info (expiry, workspaces) remains auth-gated.
# Detailed license info (expiry, workspaces) is not part of the public model.
assert result.license.status == LicenseStatus.ACTIVE
assert result.license.expired_at == ""
assert result.license.workspaces.enabled is False
assert result.license.workspaces.limit == 0
assert result.license.workspaces.size == 0
assert not hasattr(result.license, "expired_at")
assert not hasattr(result.license, "workspaces")
# --- 2. Verify Public UI Configuration Availability ---
# Ensure that data required for frontend rendering remains accessible.
@@ -394,6 +393,47 @@ class TestFeatureService:
# Marketplace should be visible
assert result.enable_marketplace is True
def test_get_license_returns_full_detail(
self, db_session_with_containers: Session, mock_external_service_dependencies
):
"""
Test the authenticated license accessor.
This test verifies that:
- get_license() returns the full license payload (status, expiry, workspace usage).
- Detail withheld from the public system-features model is present here.
"""
# Arrange
with patch("services.feature_service.dify_config") as mock_config:
mock_config.ENTERPRISE_ENABLED = True
# Act
result = FeatureService.get_license()
# Assert: full license detail is populated
assert isinstance(result, LicenseModel)
assert result.status == LicenseStatus.ACTIVE
assert result.expired_at == "2025-12-31"
assert result.workspaces.enabled is True
assert result.workspaces.limit == 5
assert result.workspaces.size == 2
mock_external_service_dependencies["enterprise_service"].get_info.assert_called_once()
def test_get_license_non_enterprise_is_unconstrained(
self, db_session_with_containers: Session, mock_external_service_dependencies
):
"""Non-enterprise deployments have no license, so limits are unconstrained."""
with patch("services.feature_service.dify_config") as mock_config:
mock_config.ENTERPRISE_ENABLED = False
result = FeatureService.get_license()
assert isinstance(result, LicenseModel)
assert result.status == LicenseStatus.NONE
assert result.workspaces.is_available() is True
assert result.seats.is_available() is True
mock_external_service_dependencies["enterprise_service"].get_info.assert_not_called()
def test_get_system_features_basic_config(
self, db_session_with_containers: Session, mock_external_service_dependencies
):
@@ -408,12 +448,14 @@ class TestFeatureService:
"""
# Arrange: Setup basic config mock (no enterprise)
with patch("services.feature_service.dify_config") as mock_config:
mock_config.DEPLOYMENT_EDITION = DeploymentEdition.COMMUNITY
mock_config.ENTERPRISE_ENABLED = False
mock_config.MARKETPLACE_ENABLED = False
mock_config.ENABLE_EMAIL_CODE_LOGIN = True
mock_config.ENABLE_EMAIL_PASSWORD_LOGIN = True
mock_config.ENABLE_SOCIAL_OAUTH_LOGIN = False
mock_config.ENABLE_COLLABORATION_MODE = False
mock_config.ENABLE_CHANGE_EMAIL = True
mock_config.ALLOW_REGISTER = True
mock_config.ALLOW_CREATE_WORKSPACE = True
mock_config.MAIL_TYPE = "smtp"
@@ -611,11 +653,13 @@ class TestFeatureService:
"""
# Arrange: Setup enterprise disabled mock
with patch("services.feature_service.dify_config") as mock_config:
mock_config.DEPLOYMENT_EDITION = DeploymentEdition.COMMUNITY
mock_config.ENTERPRISE_ENABLED = False
mock_config.MARKETPLACE_ENABLED = True
mock_config.ENABLE_EMAIL_CODE_LOGIN = False
mock_config.ENABLE_EMAIL_PASSWORD_LOGIN = True
mock_config.ENABLE_SOCIAL_OAUTH_LOGIN = True
mock_config.ENABLE_CHANGE_EMAIL = True
mock_config.ALLOW_REGISTER = False
mock_config.ALLOW_CREATE_WORKSPACE = False
mock_config.MAIL_TYPE = None
@@ -650,8 +694,8 @@ class TestFeatureService:
# Verify default license status
assert result.license.status == "none"
assert result.license.expired_at == ""
assert result.license.workspaces.enabled is False
assert not hasattr(result.license, "expired_at")
assert not hasattr(result.license, "workspaces")
# Verify no enterprise service calls
mock_external_service_dependencies["enterprise_service"].get_info.assert_not_called()
@@ -840,6 +884,7 @@ class TestFeatureService:
"""
# Arrange: Setup edge case webapp auth mock with proper config
with patch("services.feature_service.dify_config") as mock_config:
mock_config.DEPLOYMENT_EDITION = DeploymentEdition.ENTERPRISE
mock_config.ENTERPRISE_ENABLED = True
mock_config.MARKETPLACE_ENABLED = False
mock_config.ENABLE_EMAIL_CODE_LOGIN = False
@@ -960,6 +1005,7 @@ class TestFeatureService:
# Test case 1: Official only scope
with patch("services.feature_service.dify_config") as mock_config:
mock_config.DEPLOYMENT_EDITION = DeploymentEdition.ENTERPRISE
mock_config.ENTERPRISE_ENABLED = True
mock_config.MARKETPLACE_ENABLED = False
mock_config.ENABLE_EMAIL_CODE_LOGIN = False
@@ -983,6 +1029,7 @@ class TestFeatureService:
# Test case 2: All plugins scope
with patch("services.feature_service.dify_config") as mock_config:
mock_config.DEPLOYMENT_EDITION = DeploymentEdition.ENTERPRISE
mock_config.ENTERPRISE_ENABLED = True
mock_config.MARKETPLACE_ENABLED = False
mock_config.ENABLE_EMAIL_CODE_LOGIN = False
@@ -1003,6 +1050,7 @@ class TestFeatureService:
# Test case 3: Specific partners scope
with patch("services.feature_service.dify_config") as mock_config:
mock_config.DEPLOYMENT_EDITION = DeploymentEdition.ENTERPRISE
mock_config.ENTERPRISE_ENABLED = True
mock_config.MARKETPLACE_ENABLED = False
mock_config.ENABLE_EMAIL_CODE_LOGIN = False
@@ -1026,6 +1074,7 @@ class TestFeatureService:
# Test case 4: None scope
with patch("services.feature_service.dify_config") as mock_config:
mock_config.DEPLOYMENT_EDITION = DeploymentEdition.ENTERPRISE
mock_config.ENTERPRISE_ENABLED = True
mock_config.MARKETPLACE_ENABLED = False
mock_config.ENABLE_EMAIL_CODE_LOGIN = False
@@ -1118,24 +1167,19 @@ class TestFeatureService:
}
}
# Act: Execute the method under test
result = FeatureService.get_system_features(is_authenticated=True)
# Act: Execute the authenticated license accessor
result = FeatureService.get_license()
# Assert: Verify the expected outcomes
assert result is not None
assert isinstance(result, SystemFeatureModel)
assert isinstance(result, LicenseModel)
# Verify license status
assert result.license.status == "inactive"
assert result.license.expired_at == ""
assert result.license.workspaces.enabled is False
assert result.license.workspaces.size == 0
assert result.license.workspaces.limit == 0
# Verify enterprise features
assert result.branding.enabled is True
assert result.webapp_auth.enabled is True
assert result.enable_change_email is False
# Verify license status and detail
assert result.status == "inactive"
assert result.expired_at == ""
assert result.workspaces.enabled is False
assert result.workspaces.size == 0
assert result.workspaces.limit == 0
# Verify mock interactions
mock_external_service_dependencies["enterprise_service"].get_info.assert_called_once()
@@ -1154,6 +1198,7 @@ class TestFeatureService:
"""
# Arrange: Setup partial enterprise info mock with proper config
with patch("services.feature_service.dify_config") as mock_config:
mock_config.DEPLOYMENT_EDITION = DeploymentEdition.ENTERPRISE
mock_config.ENTERPRISE_ENABLED = True
mock_config.MARKETPLACE_ENABLED = False
mock_config.ENABLE_EMAIL_CODE_LOGIN = False
@@ -1200,8 +1245,8 @@ class TestFeatureService:
# Verify default license status
assert result.license.status == "none"
assert result.license.expired_at == ""
assert result.license.workspaces.enabled is False
assert not hasattr(result.license, "expired_at")
assert not hasattr(result.license, "workspaces")
# Verify default plugin installation permission
assert result.plugin_installation_permission.plugin_installation_scope == "all"
@@ -1283,6 +1328,7 @@ class TestFeatureService:
"""
# Arrange: Setup edge case protocols mock with proper config
with patch("services.feature_service.dify_config") as mock_config:
mock_config.DEPLOYMENT_EDITION = DeploymentEdition.ENTERPRISE
mock_config.ENTERPRISE_ENABLED = True
mock_config.MARKETPLACE_ENABLED = False
mock_config.ENABLE_EMAIL_CODE_LOGIN = False
@@ -1493,24 +1539,19 @@ class TestFeatureService:
}
}
# Act: Execute the method under test
result = FeatureService.get_system_features(is_authenticated=True)
# Act: Execute the authenticated license accessor
result = FeatureService.get_license()
# Assert: Verify the expected outcomes
assert result is not None
assert isinstance(result, SystemFeatureModel)
assert isinstance(result, LicenseModel)
# Verify license status
assert result.license.status == "expired"
assert result.license.expired_at == "2023-12-31"
assert result.license.workspaces.enabled is False
assert result.license.workspaces.size == 0
assert result.license.workspaces.limit == 0
# Verify enterprise features
assert result.branding.enabled is True
assert result.webapp_auth.enabled is True
assert result.enable_change_email is False
# Verify license status and detail
assert result.status == "expired"
assert result.expired_at == "2023-12-31"
assert result.workspaces.enabled is False
assert result.workspaces.size == 0
assert result.workspaces.limit == 0
# Verify mock interactions
mock_external_service_dependencies["enterprise_service"].get_info.assert_called_once()
@@ -1587,6 +1628,7 @@ class TestFeatureService:
"""
# Arrange: Setup edge case branding mock with proper config
with patch("services.feature_service.dify_config") as mock_config:
mock_config.DEPLOYMENT_EDITION = DeploymentEdition.ENTERPRISE
mock_config.ENTERPRISE_ENABLED = True
mock_config.MARKETPLACE_ENABLED = False
mock_config.ENABLE_EMAIL_CODE_LOGIN = False
@@ -1776,6 +1818,7 @@ class TestFeatureService:
"""
# Arrange: Setup lost license mock with proper config
with patch("services.feature_service.dify_config") as mock_config:
mock_config.DEPLOYMENT_EDITION = DeploymentEdition.ENTERPRISE
mock_config.ENTERPRISE_ENABLED = True
mock_config.MARKETPLACE_ENABLED = False
mock_config.ENABLE_EMAIL_CODE_LOGIN = False
@@ -1787,7 +1830,7 @@ class TestFeatureService:
mock_config.PLUGIN_MAX_PACKAGE_SIZE = 100
mock_external_service_dependencies["enterprise_service"].get_info.return_value = {
"license": {"status": "lost", "expired_at": None, "plan": None}
"License": {"status": "lost"}
}
# Act: Execute the method under test
@@ -1809,6 +1852,7 @@ class TestFeatureService:
assert result.enable_email_password_login is True
assert result.is_allow_register is False
assert result.is_allow_create_workspace is False
assert result.license.status == LicenseStatus.LOST
# Verify mock interactions
mock_external_service_dependencies["enterprise_service"].get_info.assert_called_once()
@@ -6,6 +6,7 @@ import pytest
from faker import Faker
from sqlalchemy.orm import Session
from enums.deployment_edition import DeploymentEdition
from models import Account, Tenant, TenantAccountJoin, TenantAccountRole
from services.credit_pool_service import CreditPoolBalance
from services.workspace_service import WorkspaceService
@@ -611,7 +612,7 @@ class TestWorkspaceService:
db_session_with_containers, mock_external_service_dependencies
)
mock_external_service_dependencies["dify_config"].EDITION = "SELF_HOSTED"
mock_external_service_dependencies["dify_config"].DEPLOYMENT_EDITION = DeploymentEdition.COMMUNITY
mock_external_service_dependencies["feature_service"].get_features.return_value.can_replace_logo = False
mock_external_service_dependencies["tenant_service"].has_roles.return_value = False
@@ -632,7 +633,7 @@ class TestWorkspaceService:
db_session_with_containers, mock_external_service_dependencies
)
mock_external_service_dependencies["dify_config"].EDITION = "CLOUD"
mock_external_service_dependencies["dify_config"].DEPLOYMENT_EDITION = DeploymentEdition.CLOUD
feature = mock_external_service_dependencies["feature_service"].get_features.return_value
feature.can_replace_logo = False
feature.next_credit_reset_date = "2025-02-01"
@@ -657,7 +658,7 @@ class TestWorkspaceService:
db_session_with_containers, mock_external_service_dependencies
)
mock_external_service_dependencies["dify_config"].EDITION = "CLOUD"
mock_external_service_dependencies["dify_config"].DEPLOYMENT_EDITION = DeploymentEdition.CLOUD
feature = mock_external_service_dependencies["feature_service"].get_features.return_value
feature.can_replace_logo = False
feature.next_credit_reset_date = "2025-02-01"
@@ -685,7 +686,7 @@ class TestWorkspaceService:
db_session_with_containers, mock_external_service_dependencies
)
mock_external_service_dependencies["dify_config"].EDITION = "CLOUD"
mock_external_service_dependencies["dify_config"].DEPLOYMENT_EDITION = DeploymentEdition.CLOUD
feature = mock_external_service_dependencies["feature_service"].get_features.return_value
feature.can_replace_logo = False
feature.next_credit_reset_date = "2025-02-01"
@@ -713,7 +714,7 @@ class TestWorkspaceService:
db_session_with_containers, mock_external_service_dependencies
)
mock_external_service_dependencies["dify_config"].EDITION = "CLOUD"
mock_external_service_dependencies["dify_config"].DEPLOYMENT_EDITION = DeploymentEdition.CLOUD
feature = mock_external_service_dependencies["feature_service"].get_features.return_value
feature.can_replace_logo = False
feature.next_credit_reset_date = "2025-02-01"
@@ -749,7 +750,7 @@ class TestWorkspaceService:
db_session_with_containers, mock_external_service_dependencies
)
mock_external_service_dependencies["dify_config"].EDITION = "CLOUD"
mock_external_service_dependencies["dify_config"].DEPLOYMENT_EDITION = DeploymentEdition.CLOUD
feature = mock_external_service_dependencies["feature_service"].get_features.return_value
feature.can_replace_logo = False
feature.next_credit_reset_date = "2025-02-01"
@@ -779,7 +780,7 @@ class TestWorkspaceService:
db_session_with_containers, mock_external_service_dependencies
)
mock_external_service_dependencies["dify_config"].EDITION = "CLOUD"
mock_external_service_dependencies["dify_config"].DEPLOYMENT_EDITION = DeploymentEdition.CLOUD
feature = mock_external_service_dependencies["feature_service"].get_features.return_value
feature.can_replace_logo = False
feature.next_credit_reset_date = "2025-02-01"
@@ -808,7 +809,7 @@ class TestWorkspaceService:
db_session_with_containers, mock_external_service_dependencies
)
mock_external_service_dependencies["dify_config"].EDITION = "CLOUD"
mock_external_service_dependencies["dify_config"].DEPLOYMENT_EDITION = DeploymentEdition.CLOUD
feature = mock_external_service_dependencies["feature_service"].get_features.return_value
feature.can_replace_logo = False
feature.next_credit_reset_date = "2025-02-01"
@@ -69,6 +69,20 @@ def _version_response(version_id: str = "version-1") -> dict:
}
def test_query_values_accepts_repeated_and_indexed_arrays() -> None:
app = Flask(__name__)
with app.test_request_context("/?sources=webapp:app-1&sources%5B1%5D=workflow:app-2&sources%5B0%5D=workflow:app-1"):
assert roster_controller._query_values("sources", "source") == [
"webapp:app-1",
"workflow:app-1",
"workflow:app-2",
]
with app.test_request_context("/?source%5B0%5D=workflow:app-3"):
assert roster_controller._query_values("sources", "source") == ["workflow:app-3"]
def _workflow_composer_response(**overrides) -> dict:
response = {
"variant": "workflow",
@@ -1530,8 +1544,35 @@ def test_drain_streaming_generate_response_raises_when_stream_ends_early() -> No
completion_controller._drain_streaming_generate_response(response)
def test_agent_chat_helper_forces_agent_streaming_and_external_trace(
app: Flask, monkeypatch: pytest.MonkeyPatch, account_id: str
@pytest.mark.parametrize(
("payload_extra", "expected_draft_type", "expected_start_new"),
[
({}, AgentConfigDraftType.DRAFT, True),
({"conversation_id": None}, AgentConfigDraftType.DRAFT, True),
({"conversation_id": ""}, AgentConfigDraftType.DRAFT, True),
(
{"conversation_id": "00000000-0000-0000-0000-000000000001"},
AgentConfigDraftType.DRAFT,
False,
),
({"draft_type": "debug_build"}, AgentConfigDraftType.DEBUG_BUILD, False),
(
{
"draft_type": "debug_build",
"conversation_id": "00000000-0000-0000-0000-000000000001",
},
AgentConfigDraftType.DEBUG_BUILD,
False,
),
],
)
def test_agent_chat_helper_resolves_scoped_conversation_and_forces_streaming(
app: Flask,
monkeypatch: pytest.MonkeyPatch,
account_id: str,
payload_extra: dict[str, str | None],
expected_draft_type: AgentConfigDraftType,
expected_start_new: bool,
) -> None:
app_model = SimpleNamespace(id="app-1", tenant_id="tenant-1", mode="agent")
current_user = SimpleNamespace(id=account_id)
@@ -1543,7 +1584,7 @@ def test_agent_chat_helper_forces_agent_streaming_and_external_trace(
def resolve_debug_conversation(**kwargs: object) -> str:
captured["resolve_debug_conversation"] = kwargs
return "debug-conversation-1"
return "00000000-0000-0000-0000-000000000001"
monkeypatch.setattr(completion_controller.AppGenerateService, "generate", generate)
monkeypatch.setattr(
@@ -1555,7 +1596,8 @@ def test_agent_chat_helper_forces_agent_streaming_and_external_trace(
completion_controller.helper, "compact_generate_response", lambda response: {"response": response}
)
with app.test_request_context(
json={"inputs": {}, "query": "hello", "response_mode": "streaming"}, headers={"X-Trace-Id": "trace-1"}
json={"inputs": {}, "query": "hello", "response_mode": "streaming", **payload_extra},
headers={"X-Trace-Id": "trace-1"},
):
result = completion_controller._create_chat_message(
current_user=current_user, app_model=app_model, session=Mock()
@@ -1566,10 +1608,12 @@ def test_agent_chat_helper_forces_agent_streaming_and_external_trace(
assert captured["streaming"] is True
args = cast(dict[str, object], captured["args"])
assert args["response_mode"] == "streaming"
assert args["conversation_id"] == "debug-conversation-1"
assert args["conversation_id"] == "00000000-0000-0000-0000-000000000001"
assert args["auto_generate_name"] is False
assert args["external_trace_id"] == "trace-1"
assert cast(dict[str, object], captured["resolve_debug_conversation"])["draft_type"] == AgentConfigDraftType.DRAFT
resolve_call = cast(dict[str, object], captured["resolve_debug_conversation"])
assert resolve_call["draft_type"] == expected_draft_type
assert resolve_call["start_new"] is expected_start_new
def test_agent_chat_helper_ignores_private_exit_intent_payload_key(
@@ -1617,14 +1661,28 @@ def test_agent_chat_helper_ignores_private_exit_intent_payload_key(
assert completion_controller.AGENT_RUNTIME_EXIT_INTENT_ARG not in args
def test_agent_chat_helper_rejects_foreign_debug_conversation(
app: Flask, monkeypatch: pytest.MonkeyPatch, account_id: str
@pytest.mark.parametrize(
("payload_extra", "expected_draft_type"),
[
({}, AgentConfigDraftType.DRAFT),
({"draft_type": "debug_build"}, AgentConfigDraftType.DEBUG_BUILD),
],
)
def test_agent_chat_helper_rejects_foreign_debug_conversation_before_generation(
app: Flask,
monkeypatch: pytest.MonkeyPatch,
account_id: str,
payload_extra: dict[str, str],
expected_draft_type: AgentConfigDraftType,
) -> None:
app_model = SimpleNamespace(id="app-1", tenant_id="tenant-1", mode="agent")
generate = MagicMock()
resolve_debug_conversation = MagicMock(return_value="owned-conversation")
monkeypatch.setattr(completion_controller.AppGenerateService, "generate", generate)
monkeypatch.setattr(
completion_controller,
"_resolve_current_user_agent_debug_conversation_id",
lambda **kwargs: "owned-conversation",
resolve_debug_conversation,
)
with app.test_request_context(
json={
@@ -1632,6 +1690,7 @@ def test_agent_chat_helper_rejects_foreign_debug_conversation(
"query": "hello",
"response_mode": "streaming",
"conversation_id": "00000000-0000-0000-0000-000000000001",
**payload_extra,
}
):
with pytest.raises(NotFound):
@@ -1643,6 +1702,12 @@ def test_agent_chat_helper_rejects_foreign_debug_conversation(
session=Mock(),
)
resolve_debug_conversation.assert_called_once()
resolve_call = resolve_debug_conversation.call_args.kwargs
assert resolve_call["draft_type"] == expected_draft_type
assert resolve_call["start_new"] is False
generate.assert_not_called()
def test_resolve_current_user_agent_debug_conversation_uses_agent_or_backing_app(
monkeypatch: pytest.MonkeyPatch,
@@ -1657,6 +1722,10 @@ def test_resolve_current_user_agent_debug_conversation_uses_agent_or_backing_app
calls.append({"get_or_create": kwargs})
return f"debug-{kwargs['agent_id']}"
def refresh_agent_app_debug_conversation_id(self, **kwargs: object) -> str:
calls.append({"refresh": kwargs})
return f"new-{kwargs['agent_id']}"
def get_app_backing_agent(self, **kwargs: object) -> object:
calls.append({"get_app_backing_agent": kwargs})
return SimpleNamespace(id="backing-agent")
@@ -1669,6 +1738,7 @@ def test_resolve_current_user_agent_debug_conversation_uses_agent_or_backing_app
app_model=SimpleNamespace(id="app-1"),
agent_id="agent-1",
draft_type=AgentConfigDraftType.DRAFT,
start_new=True,
)
fallback_id = completion_controller._resolve_current_user_agent_debug_conversation_id(
session="session-1", # type: ignore[arg-type]
@@ -1678,10 +1748,20 @@ def test_resolve_current_user_agent_debug_conversation_uses_agent_or_backing_app
agent_id=None,
draft_type=AgentConfigDraftType.DEBUG_BUILD,
)
assert explicit_id == "debug-agent-1"
fallback_preview_id = completion_controller._resolve_current_user_agent_debug_conversation_id(
session="session-1", # type: ignore[arg-type]
current_tenant_id="tenant-1",
current_user=SimpleNamespace(id="account-1"),
app_model=SimpleNamespace(id="app-1"),
agent_id=None,
draft_type=AgentConfigDraftType.DRAFT,
start_new=True,
)
assert explicit_id == "new-agent-1"
assert fallback_id == "debug-backing-agent"
assert fallback_preview_id == "new-backing-agent"
assert calls[1] == {
"get_or_create": {
"refresh": {
"tenant_id": "tenant-1",
"agent_id": "agent-1",
"account_id": "account-1",
@@ -1697,6 +1777,15 @@ def test_resolve_current_user_agent_debug_conversation_uses_agent_or_backing_app
"draft_type": AgentConfigDraftType.DEBUG_BUILD,
}
}
assert calls[6] == {"get_app_backing_agent": {"tenant_id": "tenant-1", "app_id": "app-1"}}
assert calls[7] == {
"refresh": {
"tenant_id": "tenant-1",
"agent_id": "backing-agent",
"account_id": "account-1",
"draft_type": AgentConfigDraftType.DRAFT,
}
}
@pytest.mark.parametrize(
@@ -13,6 +13,7 @@ from sqlalchemy import Engine, event
from sqlalchemy.orm import Session
from controllers.console.app import app_import as app_import_module
from enums.deployment_edition import DeploymentEdition
from models.account import Account
from models.base import TypeBase
from models.engine import db
@@ -47,7 +48,10 @@ class _Result:
def _install_features(monkeypatch: pytest.MonkeyPatch, enabled: bool) -> None:
features = SystemFeatureModel(webapp_auth=WebAppAuthModel(enabled=enabled))
features = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
webapp_auth=WebAppAuthModel(enabled=enabled),
)
monkeypatch.setattr(app_import_module.FeatureService, "get_system_features", lambda: features)
@@ -11,6 +11,7 @@ from controllers.console.auth.email_register import (
EmailRegisterResetApi,
EmailRegisterSendEmailApi,
)
from enums.deployment_edition import DeploymentEdition
from services.feature_service import SystemFeatureModel
@@ -34,7 +35,11 @@ class TestEmailRegisterSendEmailApi:
mock_account = MagicMock()
mock_get_account.return_value = mock_account
feature_flags = SystemFeatureModel(enable_email_password_login=True, is_allow_register=True)
feature_flags = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
is_allow_register=True,
)
with (
patch("controllers.console.auth.email_register.dify_config.BILLING_ENABLED", True),
patch("controllers.console.wraps.dify_config.EDITION", "CLOUD"),
@@ -75,7 +80,11 @@ class TestEmailRegisterCheckApi:
mock_get_data.return_value = {"email": "User@Example.com", "code": "4321"}
mock_generate_token.return_value = (None, "new-token")
feature_flags = SystemFeatureModel(enable_email_password_login=True, is_allow_register=True)
feature_flags = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
is_allow_register=True,
)
with (
patch("controllers.console.wraps.dify_config.EDITION", "CLOUD"),
patch("controllers.console.wraps.FeatureService.get_system_features", return_value=feature_flags),
@@ -123,7 +132,11 @@ class TestEmailRegisterResetApi:
mock_login.return_value = token_pair
mock_get_account.return_value = None
feature_flags = SystemFeatureModel(enable_email_password_login=True, is_allow_register=True)
feature_flags = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
is_allow_register=True,
)
with (
patch("controllers.console.wraps.dify_config.EDITION", "CLOUD"),
patch("controllers.console.wraps.FeatureService.get_system_features", return_value=feature_flags),
@@ -171,7 +184,11 @@ class TestEmailRegisterResetApi:
mock_login.return_value = token_pair
mock_get_account.return_value = None
feature_flags = SystemFeatureModel(enable_email_password_login=True, is_allow_register=True)
feature_flags = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
is_allow_register=True,
)
with (
patch("controllers.console.wraps.dify_config.EDITION", "CLOUD"),
patch("controllers.console.wraps.FeatureService.get_system_features", return_value=feature_flags),
@@ -224,7 +241,11 @@ class TestEmailRegisterResetApi:
mock_login.return_value = token_pair
mock_get_account.return_value = None
feature_flags = SystemFeatureModel(enable_email_password_login=True, is_allow_register=True)
feature_flags = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
is_allow_register=True,
)
with (
patch("controllers.console.wraps.dify_config.EDITION", "CLOUD"),
patch("controllers.console.wraps.FeatureService.get_system_features", return_value=feature_flags),
@@ -485,10 +485,12 @@ class TestEmailCodeLoginApi:
@patch("controllers.console.auth.login.AccountService.revoke_email_code_login_token")
@patch("controllers.console.auth.login.AccountService.get_user_through_email")
@patch("controllers.console.auth.login.TenantService.get_join_tenants")
@patch("controllers.console.auth.login.FeatureService.get_license")
@patch("controllers.console.auth.login.FeatureService.get_system_features")
def test_email_code_login_workspace_limit_exceeded(
self,
mock_get_features,
mock_get_license,
mock_get_tenants,
mock_get_user,
mock_revoke_token,
@@ -507,9 +509,7 @@ class TestEmailCodeLoginApi:
mock_get_data.return_value = {"email": "test@example.com", "code": "123456"}
mock_get_user.return_value = mock_account
mock_get_tenants.return_value = []
mock_features = MagicMock()
mock_features.license.workspaces.is_available.return_value = False
mock_get_features.return_value = mock_features
mock_get_license.return_value.workspaces.is_available.return_value = False
# Act & Assert
with app.test_request_context(
@@ -13,6 +13,7 @@ from controllers.console.auth.forgot_password import (
ForgotPasswordResetApi,
ForgotPasswordSendEmailApi,
)
from enums.deployment_edition import DeploymentEdition
from models.account import Account
from models.engine import db
from services.feature_service import SystemFeatureModel
@@ -46,8 +47,15 @@ class TestForgotPasswordSendEmailApi:
mock_get_account.return_value = mock_account
mock_send_email.return_value = "token-123"
wraps_features = SystemFeatureModel(enable_email_password_login=True, is_allow_register=True)
controller_features = SystemFeatureModel(is_allow_register=True)
wraps_features = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
is_allow_register=True,
)
controller_features = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
is_allow_register=True,
)
with (
patch(
"controllers.console.auth.forgot_password.FeatureService.get_system_features",
@@ -95,7 +103,10 @@ class TestForgotPasswordCheckApi:
mock_get_data.return_value = {"email": "Admin@Example.com", "code": "4321"}
mock_generate_token.return_value = (None, "new-token")
wraps_features = SystemFeatureModel(enable_email_password_login=True)
wraps_features = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
)
with (
patch("controllers.console.wraps.dify_config.EDITION", "CLOUD"),
patch("controllers.console.wraps.FeatureService.get_system_features", return_value=wraps_features),
@@ -138,7 +149,10 @@ class TestForgotPasswordResetApi:
db.session.commit()
mock_get_account.return_value = account
wraps_features = SystemFeatureModel(enable_email_password_login=True)
wraps_features = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
)
with (
patch("controllers.console.wraps.dify_config.EDITION", "CLOUD"),
patch("controllers.console.wraps.FeatureService.get_system_features", return_value=wraps_features),
@@ -343,10 +343,12 @@ class TestLoginApi:
@patch("controllers.console.auth.login.RegisterService.get_invitation_with_case_fallback")
@patch("controllers.console.auth.login.AccountService.authenticate")
@patch("controllers.console.auth.login.TenantService.get_join_tenants")
@patch("controllers.console.auth.login.FeatureService.get_license")
@patch("controllers.console.auth.login.FeatureService.get_system_features")
def test_login_fails_when_no_workspace_and_limit_exceeded(
self,
mock_get_features: MagicMock,
mock_get_license: MagicMock,
mock_get_tenants: MagicMock,
mock_authenticate: MagicMock,
mock_get_invitation: MagicMock,
@@ -370,8 +372,8 @@ class TestLoginApi:
mock_features = MagicMock()
mock_features.is_allow_create_workspace = True
mock_features.license.workspaces.is_available.return_value = False
mock_get_features.return_value = mock_features
mock_get_license.return_value.workspaces.is_available.return_value = False
# Act & Assert
with app.test_request_context(
@@ -3,7 +3,7 @@ from __future__ import annotations
from inspect import unwrap
from unittest.mock import patch
from controllers.console.auth.oauth_server import OAuthServerUserAuthorizeApi
from controllers.console.auth.oauth_server import OAuthServerUserAccountApi, OAuthServerUserAuthorizeApi
from models import Account
from models.account import AccountStatus, TenantAccountRole
from models.model import OAuthProviderApp
@@ -45,3 +45,13 @@ def test_oauth_authorize_uses_injected_current_user() -> None:
sign_oauth_authorization_code.assert_called_once_with("client-1", "account-1")
assert response == {"code": "authorization-code"}
def test_oauth_account_returns_stable_account_id() -> None:
api = OAuthServerUserAccountApi()
method = unwrap(api.post)
account = _make_account()
response = method(api, _make_oauth_provider_app(), account)
assert response["id"] == "account-1"
@@ -23,6 +23,7 @@ from controllers.console.auth.forgot_password import (
ForgotPasswordSendEmailApi,
)
from controllers.console.error import AccountNotFound, EmailSendIpLimitError
from enums.deployment_edition import DeploymentEdition
from models.account import Account, Tenant, TenantAccountJoin
from services.feature_service import SystemFeatureModel
@@ -48,7 +49,10 @@ def enable_password_login_wrappers(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr("controllers.console.wraps.dify_config.EDITION", "CLOUD")
monkeypatch.setattr(
"controllers.console.wraps.FeatureService.get_system_features",
lambda: SystemFeatureModel(enable_email_password_login=True),
lambda: SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
),
)
@@ -2,14 +2,15 @@ from inspect import unwrap
from pytest_mock import MockerFixture
from models import Account
from services.feature_service import FeatureModel, LimitationModel, SystemFeatureModel
def make_account() -> Account:
account = Account(name="Alice", email="alice@example.com")
account.id = "account-1"
return account
from enums.deployment_edition import DeploymentEdition
from services.feature_service import (
FeatureModel,
LicenseLimitationModel,
LicenseModel,
LicenseStatus,
LimitationModel,
SystemFeatureModel,
)
class TestFeatureApi:
@@ -82,19 +83,16 @@ class TestAppDslVersionApi:
class TestSystemFeatureApi:
def test_get_system_features_authenticated(self, mocker: MockerFixture):
"""
current_user.is_authenticated == True
"""
def test_get_system_features_public(self, mocker: MockerFixture):
"""The public endpoint returns system features without any authentication input."""
from controllers.console.feature import SystemFeatureApi
account = make_account()
current_account = mocker.patch(
"controllers.console.feature.current_account_with_tenant_optional",
return_value=(account, "tenant-123"),
system_features = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
is_allow_register=True,
enable_learn_app=True,
)
system_features = SystemFeatureModel(is_allow_register=True, enable_learn_app=True)
get_system_features = mocker.patch(
"controllers.console.feature.FeatureService.get_system_features",
return_value=system_features,
@@ -105,29 +103,28 @@ class TestSystemFeatureApi:
assert result == system_features.model_dump()
assert result["enable_learn_app"] is True
current_account.assert_called_once_with()
get_system_features.assert_called_once_with(is_authenticated=True)
assert result["license"] == {"status": LicenseStatus.NONE}
get_system_features.assert_called_once_with()
def test_get_system_features_unauthenticated(self, mocker: MockerFixture):
"""
current_user.is_authenticated raises Unauthorized
"""
from controllers.console.feature import SystemFeatureApi
class TestSystemFeatureLicenseApi:
def test_get_license_success(self, mocker: MockerFixture):
from controllers.console.feature import SystemFeatureLicenseApi
current_account = mocker.patch(
"controllers.console.feature.current_account_with_tenant_optional",
return_value=(None, None),
license_model = LicenseModel(
status=LicenseStatus.ACTIVE,
expired_at="2025-12-31",
seats=LicenseLimitationModel(enabled=True, limit=5, size=2),
)
system_features = SystemFeatureModel(is_allow_register=False)
get_system_features = mocker.patch(
"controllers.console.feature.FeatureService.get_system_features",
return_value=system_features,
get_license = mocker.patch(
"controllers.console.feature.FeatureService.get_license",
return_value=license_model,
)
api = SystemFeatureApi()
result = api.get()
api = SystemFeatureLicenseApi()
raw_get = unwrap(SystemFeatureLicenseApi.get)
result = raw_get(api)
assert result == system_features.model_dump()
current_account.assert_called_once_with()
get_system_features.assert_called_once_with(is_authenticated=False)
assert result == license_model.model_dump()
assert result["seats"] == {"enabled": True, "limit": 5, "size": 2}
get_license.assert_called_once_with()
@@ -323,8 +323,8 @@ class TestMemberInviteEmailApi:
features = MagicMock()
features.billing.enabled = False
features.workspace_members.enabled = False
system_features = MagicMock()
system_features.license.seats.is_available.return_value = False
license_info = MagicMock()
license_info.seats.is_available.return_value = False
payload = {
"emails": ["a@test.com", "b@test.com"],
@@ -336,9 +336,9 @@ class TestMemberInviteEmailApi:
patch("controllers.console.workspace.members.FeatureService.get_features", return_value=features),
patch("controllers.console.workspace.members._count_new_member_invites", return_value=(2, 2)),
patch(
"controllers.console.workspace.members.FeatureService.get_system_features",
return_value=system_features,
) as mock_get_system_features,
"controllers.console.workspace.members.FeatureService.get_license",
return_value=license_info,
) as mock_get_license,
patch("controllers.console.workspace.members.RegisterService.invite_new_member") as mock_invite,
patch("controllers.console.workspace.members.dify_config.ENTERPRISE_ENABLED", True),
patch("controllers.console.workspace.members.dify_config.BILLING_ENABLED", False),
@@ -346,8 +346,8 @@ class TestMemberInviteEmailApi:
with pytest.raises(SeatsLimitExceeded):
method(api, user)
mock_get_system_features.assert_called_once_with(is_authenticated=True)
system_features.license.seats.is_available.assert_called_once_with(2)
mock_get_license.assert_called_once_with()
license_info.seats.is_available.assert_called_once_with(2)
mock_invite.assert_not_called()
def test_invite_existing_accounts_do_not_consume_seats(self, app: Flask):
@@ -359,8 +359,8 @@ class TestMemberInviteEmailApi:
features = MagicMock()
features.billing.enabled = False
features.workspace_members.enabled = False
system_features = MagicMock()
system_features.license.seats.is_available.return_value = False
license_info = MagicMock()
license_info.seats.is_available.return_value = False
payload = {
"emails": ["a@test.com", "b@test.com"],
@@ -372,9 +372,9 @@ class TestMemberInviteEmailApi:
patch("controllers.console.workspace.members.FeatureService.get_features", return_value=features),
patch("controllers.console.workspace.members._count_new_member_invites", return_value=(2, 0)),
patch(
"controllers.console.workspace.members.FeatureService.get_system_features",
return_value=system_features,
) as mock_get_system_features,
"controllers.console.workspace.members.FeatureService.get_license",
return_value=license_info,
) as mock_get_license,
patch(
"controllers.console.workspace.members.RegisterService.invite_new_member", return_value="token"
) as mock_invite,
@@ -386,8 +386,8 @@ class TestMemberInviteEmailApi:
assert status == 201
assert len(result["invitation_results"]) == 2
mock_get_system_features.assert_not_called()
system_features.license.seats.is_available.assert_not_called()
mock_get_license.assert_not_called()
license_info.seats.is_available.assert_not_called()
assert mock_invite.call_count == 2
def test_invite_mixed_accounts_with_available_seats(self, app: Flask):
@@ -399,8 +399,8 @@ class TestMemberInviteEmailApi:
features = MagicMock()
features.billing.enabled = False
features.workspace_members.enabled = False
system_features = MagicMock()
system_features.license.seats.is_available.return_value = True
license_info = MagicMock()
license_info.seats.is_available.return_value = True
payload = {
"emails": ["a@test.com", "b@test.com"],
@@ -412,9 +412,9 @@ class TestMemberInviteEmailApi:
patch("controllers.console.workspace.members.FeatureService.get_features", return_value=features),
patch("controllers.console.workspace.members._count_new_member_invites", return_value=(2, 1)),
patch(
"controllers.console.workspace.members.FeatureService.get_system_features",
return_value=system_features,
) as mock_get_system_features,
"controllers.console.workspace.members.FeatureService.get_license",
return_value=license_info,
) as mock_get_license,
patch(
"controllers.console.workspace.members.RegisterService.invite_new_member", return_value="token"
) as mock_invite,
@@ -426,8 +426,8 @@ class TestMemberInviteEmailApi:
assert status == 201
assert len(result["invitation_results"]) == 2
mock_get_system_features.assert_called_once_with(is_authenticated=True)
system_features.license.seats.is_available.assert_called_once_with(1)
mock_get_license.assert_called_once_with()
license_info.seats.is_available.assert_called_once_with(1)
assert mock_invite.call_count == 2
def test_invite_skips_seats_limit_when_enterprise_disabled(self, app: Flask):
@@ -439,8 +439,8 @@ class TestMemberInviteEmailApi:
features = MagicMock()
features.billing.enabled = False
features.workspace_members.enabled = False
system_features = MagicMock()
system_features.license.seats.is_available.return_value = False
license_info = MagicMock()
license_info.seats.is_available.return_value = False
payload = {
"emails": ["a@test.com"],
@@ -452,9 +452,9 @@ class TestMemberInviteEmailApi:
patch("controllers.console.workspace.members.FeatureService.get_features", return_value=features),
patch("controllers.console.workspace.members._count_new_member_invites", return_value=(1, 1)),
patch(
"controllers.console.workspace.members.FeatureService.get_system_features",
return_value=system_features,
) as mock_get_system_features,
"controllers.console.workspace.members.FeatureService.get_license",
return_value=license_info,
) as mock_get_license,
patch("controllers.console.workspace.members.RegisterService.invite_new_member", return_value="token"),
patch("controllers.console.workspace.members.dify_config.CONSOLE_WEB_URL", "http://x"),
patch("controllers.console.workspace.members.dify_config.ENTERPRISE_ENABLED", False),
@@ -464,8 +464,8 @@ class TestMemberInviteEmailApi:
assert status == 201
assert result["invitation_results"][0]["status"] == "success"
mock_get_system_features.assert_not_called()
system_features.license.seats.is_available.assert_not_called()
mock_get_license.assert_not_called()
license_info.seats.is_available.assert_not_called()
def test_invite_seats_error_is_reported_as_failed_result(self, app: Flask):
api = MemberInviteEmailApi()
@@ -476,8 +476,8 @@ class TestMemberInviteEmailApi:
features = MagicMock()
features.billing.enabled = False
features.workspace_members.enabled = False
system_features = MagicMock()
system_features.license.seats.is_available.return_value = True
license_info = MagicMock()
license_info.seats.is_available.return_value = True
payload = {
"emails": ["a@test.com"],
@@ -489,8 +489,8 @@ class TestMemberInviteEmailApi:
patch("controllers.console.workspace.members.FeatureService.get_features", return_value=features),
patch("controllers.console.workspace.members._count_new_member_invites", return_value=(1, 1)),
patch(
"controllers.console.workspace.members.FeatureService.get_system_features",
return_value=system_features,
"controllers.console.workspace.members.FeatureService.get_license",
return_value=license_info,
),
patch(
"controllers.console.workspace.members.RegisterService.invite_new_member",
@@ -28,6 +28,7 @@ from controllers.console.workspace.plugin import (
PluginFetchMarketplacePkgApi,
PluginFetchPermissionApi,
PluginIconApi,
PluginInstalledIdsApi,
PluginInstallFromGithubApi,
PluginInstallFromMarketplaceApi,
PluginInstallFromPkgApi,
@@ -41,12 +42,16 @@ from controllers.console.workspace.plugin import (
PluginUploadFromBundleApi,
PluginUploadFromGithubApi,
PluginUploadFromPkgApi,
_list_hardcoded_builtin_tool_providers,
)
from core.plugin.entities.parameters import PluginParameterOption
from core.plugin.entities.plugin import PluginDeclaration, PluginEntity, PluginInstallation
from core.plugin.entities.plugin import PluginCategory, PluginDeclaration, PluginEntity, PluginInstallation
from core.plugin.entities.plugin_daemon import PluginInstallTask
from core.plugin.impl.exc import PluginDaemonClientSideError
from core.plugin.plugin_service import PluginService
from core.tools.entities.api_entities import ToolProviderApiEntity
from core.tools.entities.common_entities import I18nObject
from core.tools.entities.tool_entities import ToolProviderType
from models.account import (
Account,
TenantAccountRole,
@@ -445,7 +450,7 @@ class TestPluginCategoryListApi:
mock_list = MagicMock(list=[plugin_item], has_more=True)
with (
app.test_request_context("/?page=2&page_size=10"),
app.test_request_context("/?page=2&page_size=10&query=weather&tags=search&tags=rag&language=zh_Hans"),
patch(
"controllers.console.workspace.plugin.PluginService.list_by_category", return_value=mock_list
) as list_mock,
@@ -456,18 +461,75 @@ class TestPluginCategoryListApi:
):
result = method(api, "t1", "tool")
list_mock.assert_called_once()
assert list_mock.call_args.args[0] == "t1"
assert list_mock.call_args.args[1] == "tool"
assert list_mock.call_args.args[2] == 2
assert list_mock.call_args.args[3] == 10
list_mock.assert_called_once_with(
"t1",
"tool",
2,
10,
query="weather",
tags=["search", "rag"],
language="zh_Hans",
)
assert result["plugins"][0]["id"] == "entity-1"
assert result["plugins"][0]["plugin_unique_identifier"] == "test-author/test-plugin:1.0.0@checksum"
assert result["builtin_tools"][0]["id"] == "builtin"
assert result["builtin_tools"][0]["type"] == "builtin"
assert result["has_more"] is True
assert "total" not in result
builtin_mock.assert_called_once_with("t1")
builtin_mock.assert_called_once_with(
"t1",
query="weather",
tags=["search", "rag"],
language="zh_Hans",
)
def test_builtin_tool_providers_use_the_category_list_filters(self):
search_provider = ToolProviderApiEntity(
id="search-provider",
author="dify",
name="search-provider",
description=I18nObject(en_US="Search provider", zh_Hans="搜索工具"),
icon="icon.svg",
label=I18nObject(en_US="Search", zh_Hans="搜索"),
type=ToolProviderType.BUILT_IN,
labels=["search"],
)
rag_provider = ToolProviderApiEntity(
id="rag-provider",
author="dify",
name="rag-provider",
description=I18nObject(en_US="RAG provider", zh_Hans="知识库工具"),
icon="icon.svg",
label=I18nObject(en_US="RAG", zh_Hans="知识库"),
type=ToolProviderType.BUILT_IN,
labels=["rag"],
)
with (
patch("controllers.console.workspace.plugin.ToolManager.list_default_builtin_providers", return_value=[]),
patch(
"controllers.console.workspace.plugin.ToolManager.list_hardcoded_providers",
return_value=[MagicMock(), MagicMock()],
),
patch("controllers.console.workspace.plugin.is_filtered", return_value=False),
patch(
"controllers.console.workspace.plugin.ToolTransformService.builtin_provider_to_user_provider",
side_effect=[search_provider, rag_provider],
),
patch("controllers.console.workspace.plugin.ToolTransformService.repack_provider"),
patch(
"controllers.console.workspace.plugin.BuiltinToolProviderSort.sort",
side_effect=lambda providers: providers,
),
):
result = _list_hardcoded_builtin_tool_providers(
"t1",
query="搜索",
tags=["search", "weather"],
language="zh_Hans",
)
assert [provider["id"] for provider in result] == ["search-provider"]
def test_non_tool_category_does_not_include_builtin_tools(self, app: Flask):
api = PluginCategoryListApi()
@@ -730,6 +792,39 @@ class TestPluginListInstallationsFromIdsApi:
assert result == ({"code": "plugin_error", "message": "error"}, 400)
class TestPluginInstalledIdsApi:
def test_success(self, app: Flask):
api = PluginInstalledIdsApi()
method = unwrap(api.get)
with (
app.test_request_context("/?category=tool"),
patch(
"controllers.console.workspace.plugin.PluginService.list_installed_plugin_ids",
return_value=["langgenius/openai", "langgenius/anthropic"],
) as list_installed_plugin_ids,
):
result = method(api, "t1")
assert result == {"plugin_ids": ["langgenius/openai", "langgenius/anthropic"]}
list_installed_plugin_ids.assert_called_once_with("t1", PluginCategory.Tool)
def test_daemon_error(self, app: Flask):
api = PluginInstalledIdsApi()
method = unwrap(api.get)
with (
app.test_request_context("/?category=tool"),
patch(
"controllers.console.workspace.plugin.PluginService.list_installed_plugin_ids",
side_effect=PluginDaemonClientSideError("error"),
),
):
result = method(api, "t1")
assert result == ({"code": "plugin_error", "message": "error"}, 400)
class TestPluginUploadFromGithubApi:
def test_success(self, app: Flask, user):
api = PluginUploadFromGithubApi()
@@ -1343,6 +1438,34 @@ class TestPluginFetchAutoUpgradeApi:
assert result["category"] == TenantPluginAutoUpgradeCategory.TOOL
assert result["auto_upgrade"]["upgrade_time_of_day"] == 1
def test_returns_disabled_settings_when_strategy_is_missing(self, app: Flask):
api = PluginFetchAutoUpgradeApi()
method = unwrap(api.get)
with (
app.test_request_context(f"/?category={TenantPluginAutoUpgradeCategory.MODEL.value}"),
patch(
"controllers.console.workspace.plugin.PluginAutoUpgradeService.get_strategy",
return_value=None,
),
patch(
"controllers.console.workspace.plugin.PluginAutoUpgradeService.default_upgrade_time_of_day",
return_value=78300,
),
):
result = method(api, "t1")
assert result == {
"category": TenantPluginAutoUpgradeCategory.MODEL,
"auto_upgrade": {
"strategy_setting": TenantPluginAutoUpgradeStrategySetting.DISABLED,
"upgrade_time_of_day": 78300,
"upgrade_mode": TenantPluginAutoUpgradeMode.EXCLUDE,
"exclude_plugins": [],
"include_plugins": [],
},
}
class TestPluginAutoUpgradeExcludePluginApi:
def test_success(self, app: Flask):
@@ -11,26 +11,27 @@ from controllers.openapi.auth.data import (
RequestContext,
current_edition,
)
from enums.deployment_edition import DeploymentEdition
from libs.oauth_bearer import Scope, TokenType
def test_current_edition_saas():
with patch("controllers.openapi.auth.data.dify_config") as cfg:
cfg.EDITION = "CLOUD"
cfg.DEPLOYMENT_EDITION = DeploymentEdition.CLOUD
cfg.ENTERPRISE_ENABLED = True
assert current_edition() == Edition.SAAS
def test_current_edition_ee():
with patch("controllers.openapi.auth.data.dify_config") as cfg:
cfg.EDITION = "SELF_HOSTED"
cfg.DEPLOYMENT_EDITION = DeploymentEdition.ENTERPRISE
cfg.ENTERPRISE_ENABLED = True
assert current_edition() == Edition.EE
def test_current_edition_ce():
with patch("controllers.openapi.auth.data.dify_config") as cfg:
cfg.EDITION = "SELF_HOSTED"
cfg.DEPLOYMENT_EDITION = DeploymentEdition.COMMUNITY
cfg.ENTERPRISE_ENABLED = False
assert current_edition() == Edition.CE
@@ -634,6 +634,11 @@ def test_console_plugin_category_list_exported_schema_uses_typed_items(tmp_path)
console_openapi_path = next(path for path in written_paths if path.name == "console-openapi.json")
payload = json.loads(console_openapi_path.read_text(encoding="utf-8"))
operation = payload["paths"]["/workspaces/current/plugin/{category}/list"]["get"]
parameters = {parameter["name"]: parameter for parameter in operation["parameters"]}
assert parameters["query"]["in"] == "query"
assert parameters["tags"]["in"] == "query"
assert parameters["tags"]["schema"]["type"] == "array"
assert parameters["language"]["in"] == "query"
response_ref = operation["responses"]["200"]["content"]["application/json"]["schema"]["$ref"].removeprefix(
"#/components/schemas/"
)
@@ -661,3 +666,36 @@ def test_console_plugin_category_list_exported_schema_uses_typed_items(tmp_path)
builtin_tool_schema = schemas["PluginCategoryBuiltinToolProviderResponse"]
for field in ("plugin_unique_identifier", "team_credentials", "type", "tools"):
assert field in builtin_tool_schema["properties"]
def test_console_installed_plugin_ids_exported_schema_is_lightweight(tmp_path):
from dev.generate_swagger_specs import generate_specs
written_paths = generate_specs(tmp_path)
console_openapi_path = next(path for path in written_paths if path.name == "console-openapi.json")
payload = json.loads(console_openapi_path.read_text(encoding="utf-8"))
operation = payload["paths"]["/workspaces/current/plugin/installed-ids"]["get"]
parameters = {parameter["name"]: parameter for parameter in operation["parameters"]}
assert parameters["category"]["in"] == "query"
assert parameters["category"]["required"] is True
assert parameters["category"]["schema"]["enum"] == [
"agent-strategy",
"datasource",
"extension",
"model",
"tool",
"trigger",
]
response_ref = operation["responses"]["200"]["content"]["application/json"]["schema"]["$ref"].removeprefix(
"#/components/schemas/"
)
response_schema = payload["components"]["schemas"][response_ref]
assert response_schema["required"] == ["plugin_ids"]
assert response_schema["properties"] == {
"plugin_ids": {
"items": {"type": "string"},
"title": "Plugin Ids",
"type": "array",
}
}
@@ -14,6 +14,7 @@ from controllers.web.forgot_password import (
ForgotPasswordResetApi,
ForgotPasswordSendEmailApi,
)
from enums.deployment_edition import DeploymentEdition
from models.account import Account
from models.engine import db
from services.feature_service import SystemFeatureModel
@@ -32,7 +33,10 @@ def database_app() -> Iterator[Flask]:
@pytest.fixture(autouse=True)
def _patch_wraps():
wraps_features = SystemFeatureModel(enable_email_password_login=True)
wraps_features = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
)
with (
patch("controllers.console.wraps.db") as mock_db,
patch("controllers.console.wraps.dify_config.ENTERPRISE_ENABLED", True),
@@ -10,6 +10,7 @@ from werkzeug.exceptions import Unauthorized
import services.errors.account
from controllers.web.login import EmailCodeLoginApi, EmailCodeLoginSendEmailApi, LoginApi, LoginStatusApi, LogoutApi
from enums.deployment_edition import DeploymentEdition
from services.entities.auth_entities import LoginFailureReason
@@ -34,7 +35,7 @@ def app():
@pytest.fixture(autouse=True)
def _patch_wraps():
wraps_features = SimpleNamespace(enable_email_password_login=True)
console_dify = SimpleNamespace(ENTERPRISE_ENABLED=True, EDITION="CLOUD")
console_dify = SimpleNamespace(ENTERPRISE_ENABLED=True, DEPLOYMENT_EDITION=DeploymentEdition.CLOUD)
web_dify = SimpleNamespace(ENTERPRISE_ENABLED=True)
with (
patch("controllers.console.wraps.db") as mock_db,
@@ -0,0 +1,49 @@
from unittest import mock
import pytest
from werkzeug.exceptions import Unauthorized
from core.logging.context import clear_request_context, get_identity_context
@pytest.fixture(autouse=True)
def _reset_logging_context():
clear_request_context()
yield
clear_request_context()
def test_validate_jwt_token_sets_logging_identity_before_view() -> None:
from controllers.web import wraps
app_model = mock.Mock()
end_user = mock.Mock(id="end-user-id", tenant_id="tenant-id", type=None)
clear_request_context()
@wraps.validate_jwt_token
def protected_view(received_app, received_user):
assert get_identity_context() == ("tenant-id", "end-user-id", "end_user")
return received_app, received_user
with mock.patch.object(wraps, "decode_jwt_token", return_value=(app_model, end_user)):
result = protected_view()
assert result == (app_model, end_user)
def test_validate_jwt_token_does_not_set_identity_when_authentication_fails() -> None:
from controllers.web import wraps
clear_request_context()
@wraps.validate_jwt_token
def protected_view(_app, _user):
raise AssertionError("view must not be called")
with (
mock.patch.object(wraps, "decode_jwt_token", side_effect=Unauthorized()),
pytest.raises(Unauthorized),
):
protected_view()
assert get_identity_context() == ("", "", "")
@@ -0,0 +1,341 @@
from datetime import datetime
import pytest
from sqlalchemy import select
from sqlalchemy.orm import Session
from core.agent.publish_visibility import (
PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS,
agent_has_workflow_callable_active_snapshot,
workflow_callable_active_snapshot_filter,
)
from models.agent import (
Agent,
AgentConfigRevision,
AgentConfigRevisionOperation,
AgentConfigSnapshot,
AgentKind,
AgentScope,
AgentSource,
AgentStatus,
WorkflowAgentNodeBinding,
)
from models.agent_config_entities import AgentSoulConfig, AgentSoulModelConfig
from services.agent.roster_service import AgentRosterService
def _agent_soul() -> AgentSoulConfig:
return AgentSoulConfig(
model=AgentSoulModelConfig(
plugin_id="langgenius/openai",
model_provider="openai",
model="gpt-test",
)
)
def _add_agent(
session: Session,
*,
agent_id: str,
snapshot_id: str | None,
name: str,
source: AgentSource,
operation: AgentConfigRevisionOperation | None,
app_id: str | None = None,
scope: AgentScope = AgentScope.ROSTER,
status: AgentStatus = AgentStatus.ACTIVE,
has_model: bool | None = None,
) -> Agent:
agent = Agent(
id=agent_id,
tenant_id="tenant-1",
name=name,
description="",
agent_kind=AgentKind.DIFY_AGENT,
scope=scope,
source=source,
app_id=app_id,
status=status,
active_config_snapshot_id=snapshot_id,
active_config_has_model=snapshot_id is not None if has_model is None else has_model,
# A dirty draft must not hide an already published active snapshot.
active_config_is_published=False,
)
session.add(agent)
if snapshot_id is None:
return agent
session.add(
AgentConfigSnapshot(
id=snapshot_id,
tenant_id="tenant-1",
agent_id=agent_id,
version=1,
config_snapshot=_agent_soul(),
)
)
if operation is not None:
session.add(
AgentConfigRevision(
id=f"revision-{agent_id}",
tenant_id="tenant-1",
agent_id=agent_id,
current_snapshot_id=snapshot_id,
revision=1,
operation=operation,
)
)
return agent
def test_publish_visible_operation_contract() -> None:
assert {
AgentConfigRevisionOperation.PUBLISH_DRAFT,
AgentConfigRevisionOperation.SAVE_CURRENT_VERSION,
AgentConfigRevisionOperation.SAVE_NEW_VERSION,
AgentConfigRevisionOperation.SAVE_NEW_AGENT,
AgentConfigRevisionOperation.SAVE_TO_ROSTER,
AgentConfigRevisionOperation.RESTORE_VERSION,
} == PUBLISH_VISIBLE_APP_BACKED_REVISION_OPERATIONS
@pytest.mark.parametrize(
"sqlite_session",
[(Agent, AgentConfigSnapshot, AgentConfigRevision, WorkflowAgentNodeBinding)],
indirect=True,
)
def test_workflow_callable_filter_distinguishes_never_published_from_dirty_drafts(
sqlite_session: Session,
) -> None:
imported_draft = _add_agent(
sqlite_session,
agent_id="agent-imported-draft",
snapshot_id="snapshot-imported-draft",
name="Imported draft",
source=AgentSource.IMPORTED,
operation=AgentConfigRevisionOperation.IMPORT_PACKAGE,
app_id="app-imported-draft",
)
app_draft = _add_agent(
sqlite_session,
agent_id="agent-app-draft",
snapshot_id="snapshot-app-draft",
name="App draft",
source=AgentSource.AGENT_APP,
operation=AgentConfigRevisionOperation.CREATE_VERSION,
)
published_with_dirty_draft = _add_agent(
sqlite_session,
agent_id="agent-published-dirty",
snapshot_id="snapshot-published-dirty",
name="Published with dirty draft",
source=AgentSource.AGENT_APP,
operation=AgentConfigRevisionOperation.PUBLISH_DRAFT,
)
saved_to_current_version = _add_agent(
sqlite_session,
agent_id="agent-saved-current",
snapshot_id="snapshot-saved-current",
name="Saved to current version",
source=AgentSource.AGENT_APP,
operation=AgentConfigRevisionOperation.SAVE_CURRENT_VERSION,
)
saved_as_new_agent = _add_agent(
sqlite_session,
agent_id="agent-saved-new",
snapshot_id="snapshot-saved-new",
name="Saved as new agent",
source=AgentSource.AGENT_APP,
operation=AgentConfigRevisionOperation.SAVE_NEW_AGENT,
)
saved_as_new_version = _add_agent(
sqlite_session,
agent_id="agent-saved-new-version",
snapshot_id="snapshot-saved-new-version",
name="Saved as new version",
source=AgentSource.AGENT_APP,
operation=AgentConfigRevisionOperation.SAVE_NEW_VERSION,
)
saved_to_roster = _add_agent(
sqlite_session,
agent_id="agent-saved-to-roster",
snapshot_id="snapshot-saved-to-roster",
name="Saved to roster",
source=AgentSource.AGENT_APP,
operation=AgentConfigRevisionOperation.SAVE_TO_ROSTER,
)
restored_version = _add_agent(
sqlite_session,
agent_id="agent-restored-version",
snapshot_id="snapshot-restored-version",
name="Restored version",
source=AgentSource.AGENT_APP,
operation=AgentConfigRevisionOperation.RESTORE_VERSION,
)
direct_roster_agent = _add_agent(
sqlite_session,
agent_id="agent-direct-roster",
snapshot_id="snapshot-direct-roster",
name="Direct roster",
source=AgentSource.ROSTER,
operation=AgentConfigRevisionOperation.CREATE_VERSION,
)
direct_imported_roster_agent = _add_agent(
sqlite_session,
agent_id="agent-direct-imported",
snapshot_id="snapshot-direct-imported",
name="Direct imported roster",
source=AgentSource.IMPORTED,
operation=AgentConfigRevisionOperation.IMPORT_PACKAGE,
)
no_snapshot = _add_agent(
sqlite_session,
agent_id="agent-no-snapshot",
snapshot_id=None,
name="No snapshot",
source=AgentSource.ROSTER,
operation=None,
)
published_without_model = _add_agent(
sqlite_session,
agent_id="agent-published-without-model",
snapshot_id="snapshot-published-without-model",
name="Published without model",
source=AgentSource.AGENT_APP,
operation=AgentConfigRevisionOperation.PUBLISH_DRAFT,
has_model=False,
)
archived_agent = _add_agent(
sqlite_session,
agent_id="agent-archived",
snapshot_id="snapshot-archived",
name="Archived",
source=AgentSource.ROSTER,
operation=AgentConfigRevisionOperation.CREATE_VERSION,
status=AgentStatus.ARCHIVED,
)
workflow_only_agent = _add_agent(
sqlite_session,
agent_id="agent-workflow-only",
snapshot_id="snapshot-workflow-only",
name="Workflow only",
source=AgentSource.WORKFLOW,
operation=AgentConfigRevisionOperation.CREATE_VERSION,
scope=AgentScope.WORKFLOW_ONLY,
)
stale_published_snapshot = _add_agent(
sqlite_session,
agent_id="agent-stale-published-snapshot",
snapshot_id="snapshot-current-unpublished",
name="Stale published snapshot",
source=AgentSource.AGENT_APP,
operation=AgentConfigRevisionOperation.CREATE_VERSION,
)
cross_owner_revision = _add_agent(
sqlite_session,
agent_id="agent-cross-owner-revision",
snapshot_id="snapshot-cross-owner-revision",
name="Cross owner revision",
source=AgentSource.AGENT_APP,
operation=None,
)
sqlite_session.add(
AgentConfigSnapshot(
id="snapshot-old-published",
tenant_id="tenant-1",
agent_id=stale_published_snapshot.id,
version=2,
config_snapshot=_agent_soul(),
)
)
sqlite_session.add(
AgentConfigRevision(
id="revision-old-published",
tenant_id="tenant-1",
agent_id=stale_published_snapshot.id,
current_snapshot_id="snapshot-old-published",
revision=2,
operation=AgentConfigRevisionOperation.PUBLISH_DRAFT,
)
)
sqlite_session.add(
AgentConfigRevision(
id="revision-published-dirty-2",
tenant_id="tenant-1",
agent_id=published_with_dirty_draft.id,
current_snapshot_id=published_with_dirty_draft.active_config_snapshot_id,
revision=2,
operation=AgentConfigRevisionOperation.RESTORE_VERSION,
)
)
sqlite_session.add(
AgentConfigRevision(
id="revision-cross-owner",
tenant_id="tenant-other",
agent_id="agent-other",
current_snapshot_id=cross_owner_revision.active_config_snapshot_id,
revision=1,
operation=AgentConfigRevisionOperation.PUBLISH_DRAFT,
)
)
imported_draft.updated_at = datetime(2031, 7, 24, 12, 0, 0)
published_with_dirty_draft.updated_at = datetime(2030, 7, 24, 11, 0, 0)
sqlite_session.commit()
callable_agent_ids = set(
sqlite_session.scalars(select(Agent.id).where(workflow_callable_active_snapshot_filter())).all()
)
assert callable_agent_ids == {
published_with_dirty_draft.id,
saved_to_current_version.id,
saved_as_new_agent.id,
saved_as_new_version.id,
saved_to_roster.id,
restored_version.id,
direct_roster_agent.id,
direct_imported_roster_agent.id,
published_without_model.id,
archived_agent.id,
workflow_only_agent.id,
}
assert agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=published_with_dirty_draft)
assert agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=direct_roster_agent)
assert agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=direct_imported_roster_agent)
assert not agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=imported_draft)
assert not agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=app_draft)
assert not agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=no_snapshot)
assert not agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=stale_published_snapshot)
assert not agent_has_workflow_callable_active_snapshot(session=sqlite_session, agent=cross_owner_revision)
invite_options = AgentRosterService(sqlite_session).list_invite_options(
tenant_id="tenant-1",
page=1,
limit=20,
)
expected_invite_ids = callable_agent_ids - {
published_without_model.id,
archived_agent.id,
workflow_only_agent.id,
}
assert invite_options["total"] == len(expected_invite_ids)
assert {item["id"] for item in invite_options["data"]} == expected_invite_ids
first_page = AgentRosterService(sqlite_session).list_invite_options(
tenant_id="tenant-1",
page=1,
limit=1,
)
assert first_page["total"] == len(expected_invite_ids)
assert first_page["has_more"] is True
assert [item["id"] for item in first_page["data"]] == [published_with_dirty_draft.id]
unpublished_keyword = AgentRosterService(sqlite_session).list_invite_options(
tenant_id="tenant-1",
page=1,
limit=20,
keyword="Imported draft",
)
assert unpublished_keyword["total"] == 0
assert unpublished_keyword["data"] == []
@@ -1,15 +1,27 @@
"""Tests for logging context module."""
import uuid
from contextvars import copy_context
import pytest
from core.logging.context import (
clear_request_context,
get_identity_context,
get_request_id,
get_trace_id,
init_request_context,
set_identity_context,
)
@pytest.fixture(autouse=True)
def _reset_logging_context():
clear_request_context()
yield
clear_request_context()
class TestLoggingContext:
"""Tests for the logging context functions."""
@@ -77,3 +89,41 @@ class TestLoggingContext:
# IDs should be different
assert id1 != id2
def test_set_identity_context(self):
set_identity_context(tenant_id="tenant-1", user_id="user-1", user_type="end_user")
identity = get_identity_context()
assert identity.tenant_id == "tenant-1"
assert identity.user_id == "user-1"
assert identity.user_type == "end_user"
def test_set_identity_context_replaces_all_fields(self):
set_identity_context(tenant_id="tenant-1", user_id="user-1", user_type="account")
set_identity_context(user_id="user-2", user_type="end_user")
assert get_identity_context() == ("", "user-2", "end_user")
def test_identity_context_is_copied_as_primitive_values(self):
set_identity_context(tenant_id="tenant-1", user_id="user-1", user_type="end_user")
copied_context = copy_context()
clear_request_context()
assert get_identity_context() == ("", "", "")
assert copied_context.run(get_identity_context) == ("tenant-1", "user-1", "end_user")
def test_init_clears_existing_identity_context(self):
set_identity_context(tenant_id="tenant-1", user_id="user-1", user_type="end_user")
init_request_context()
assert get_identity_context() == ("", "", "")
def test_clear_resets_identity_context(self):
set_identity_context(tenant_id="tenant-1", user_id="user-1", user_type="end_user")
clear_request_context()
assert get_identity_context() == ("", "", "")
+83 -110
View File
@@ -1,5 +1,6 @@
"""Tests for logging filters."""
import io
import logging
from unittest import mock
@@ -19,6 +20,15 @@ def log_record():
)
@pytest.fixture(autouse=True)
def _reset_logging_context():
from core.logging.context import clear_request_context
clear_request_context()
yield
clear_request_context()
class TestTraceContextFilter:
def test_sets_empty_trace_id_without_context(self, log_record):
from core.logging.context import clear_request_context
@@ -147,8 +157,10 @@ class TestTraceContextFilter:
class TestIdentityContextFilter:
def test_sets_empty_identity_without_request_context(self, log_record):
from core.logging.context import clear_request_context
from core.logging.filters import IdentityContextFilter
clear_request_context()
filter = IdentityContextFilter()
result = filter.filter(log_record)
@@ -164,131 +176,92 @@ class TestIdentityContextFilter:
result = filter.filter(log_record)
assert result is True
def test_handles_exception_gracefully(self, log_record):
def test_uses_explicit_identity_context_without_flask_context(self, log_record):
from core.logging.context import set_identity_context
from core.logging.filters import IdentityContextFilter
set_identity_context(tenant_id="tenant_id", user_id="end_user_id", user_type="end_user")
filter = IdentityContextFilter()
filter.filter(log_record)
# Should not raise even if something goes wrong
with mock.patch(
"core.logging.filters.flask.has_request_context", side_effect=Exception("Test error"), autospec=True
):
result = filter.filter(log_record)
assert result is True
assert log_record.tenant_id == ""
assert log_record.tenant_id == "tenant_id"
assert log_record.user_id == "end_user_id"
assert log_record.user_type == "end_user"
def test_sets_empty_identity_unauthenticated(self, log_record):
def test_does_not_trigger_flask_login_request_loader(self, log_record):
from flask import Flask
from flask_login import LoginManager
from core.logging.context import clear_request_context
from core.logging.filters import IdentityContextFilter
mock_user = mock.MagicMock()
mock_user.is_authenticated = False
app = Flask(__name__)
app.secret_key = "test"
login_manager = LoginManager(app)
request_loader = mock.Mock(return_value=None)
login_manager.request_loader(request_loader)
clear_request_context()
with (
mock.patch("flask.has_request_context", return_value=True),
mock.patch("flask_login.current_user", mock_user),
):
filter = IdentityContextFilter()
filter.filter(log_record)
assert log_record.user_id == ""
with app.test_request_context("/"):
from flask import g
def test_sets_identity_for_account(self, log_record):
assert "_login_user" not in g
IdentityContextFilter().filter(log_record)
assert "_login_user" not in g
request_loader.assert_not_called()
assert log_record.tenant_id == ""
assert log_record.user_id == ""
assert log_record.user_type == ""
def test_ended_otel_span_warning_does_not_trigger_request_loader(self):
from flask import Flask, g
from flask_login import LoginManager
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from core.logging.context import clear_request_context
from core.logging.filters import IdentityContextFilter
class MockAccount:
pass
app = Flask(__name__)
app.secret_key = "test"
login_manager = LoginManager(app)
request_loader = mock.Mock(return_value=None)
login_manager.request_loader(request_loader)
mock_user = MockAccount()
mock_user.id = "account_id"
mock_user.current_tenant_id = "tenant_id"
mock_user.is_authenticated = True
span = TracerProvider().get_tracer(__name__).start_span("ended")
span.end()
with (
mock.patch("flask.has_request_context", return_value=True),
mock.patch("models.Account", MockAccount),
mock.patch("flask_login.current_user", mock_user),
):
filter = IdentityContextFilter()
filter.filter(log_record)
stream = io.StringIO()
handler = logging.StreamHandler(stream)
handler.addFilter(IdentityContextFilter())
handler.setFormatter(logging.Formatter("%(tenant_id)s %(user_id)s %(user_type)s %(message)s"))
assert log_record.tenant_id == "tenant_id"
assert log_record.user_id == "account_id"
assert log_record.user_type == "account"
sdk_logger = logging.getLogger("opentelemetry.sdk.trace")
previous_level = sdk_logger.level
previous_propagate = sdk_logger.propagate
previous_disabled = sdk_logger.disabled
sdk_logger.addHandler(handler)
sdk_logger.setLevel(logging.WARNING)
sdk_logger.propagate = False
sdk_logger.disabled = False
clear_request_context()
def test_sets_identity_for_account_no_tenant(self, log_record):
from core.logging.filters import IdentityContextFilter
try:
with app.test_request_context("/"), trace.use_span(span, end_on_exit=False):
assert "_login_user" not in g
class MockAccount:
pass
span.set_attribute("test.key", "test-value")
mock_user = MockAccount()
mock_user.id = "account_id"
mock_user.current_tenant_id = None
mock_user.is_authenticated = True
assert "_login_user" not in g
finally:
clear_request_context()
sdk_logger.removeHandler(handler)
sdk_logger.setLevel(previous_level)
sdk_logger.propagate = previous_propagate
sdk_logger.disabled = previous_disabled
handler.close()
with (
mock.patch("flask.has_request_context", return_value=True),
mock.patch("models.Account", MockAccount),
mock.patch("flask_login.current_user", mock_user),
):
filter = IdentityContextFilter()
filter.filter(log_record)
assert log_record.tenant_id == ""
assert log_record.user_id == "account_id"
assert log_record.user_type == "account"
def test_sets_identity_for_end_user(self, log_record):
from core.logging.filters import IdentityContextFilter
class MockEndUser:
pass
class AnotherClass:
pass
mock_user = MockEndUser()
mock_user.id = "end_user_id"
mock_user.tenant_id = "tenant_id"
mock_user.type = "custom_type"
mock_user.is_authenticated = True
with (
mock.patch("flask.has_request_context", return_value=True),
mock.patch("models.model.EndUser", MockEndUser),
mock.patch("models.Account", AnotherClass),
mock.patch("flask_login.current_user", mock_user),
):
filter = IdentityContextFilter()
filter.filter(log_record)
assert log_record.tenant_id == "tenant_id"
assert log_record.user_id == "end_user_id"
assert log_record.user_type == "custom_type"
def test_sets_identity_for_end_user_default_type(self, log_record):
from core.logging.filters import IdentityContextFilter
class MockEndUser:
pass
class AnotherClass:
pass
mock_user = MockEndUser()
mock_user.id = "end_user_id"
mock_user.tenant_id = "tenant_id"
mock_user.type = None
mock_user.is_authenticated = True
with (
mock.patch("flask.has_request_context", return_value=True),
mock.patch("models.model.EndUser", MockEndUser),
mock.patch("models.Account", AnotherClass),
mock.patch("flask_login.current_user", mock_user),
):
filter = IdentityContextFilter()
filter.filter(log_record)
assert log_record.tenant_id == "tenant_id"
assert log_record.user_id == "end_user_id"
assert log_record.user_type == "end_user"
request_loader.assert_not_called()
assert "Setting attribute on ended span" in stream.getvalue()
@@ -1,12 +1,18 @@
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
from sqlalchemy import Engine
from sqlalchemy.orm import Session
from core.ops.base_trace_instance import BaseTraceInstance
from core.ops.entities.config_entity import BaseTracingConfig
from core.ops.entities.trace_entity import BaseTraceInfo
from models import Account, App, TenantAccountJoin
from models import Account, App, Tenant, TenantAccountJoin
from models.account import TenantAccountRole
from models.model import AppMode
TABLES = (App, Account, Tenant, TenantAccountJoin)
class ConcreteTraceInstance(BaseTraceInstance):
@@ -17,91 +23,101 @@ class ConcreteTraceInstance(BaseTraceInstance):
super().trace(trace_info)
@pytest.fixture
def mock_db_session(monkeypatch: pytest.MonkeyPatch):
mock_session = MagicMock(spec=Session)
mock_session.__enter__.return_value = mock_session
mock_session.__exit__.return_value = None
mock_session_class = MagicMock(return_value=mock_session)
monkeypatch.setattr("core.ops.base_trace_instance.Session", mock_session_class)
monkeypatch.setattr("core.ops.base_trace_instance.db", MagicMock())
return mock_session
@pytest.fixture(autouse=True)
def _bind_production_sessions(monkeypatch: pytest.MonkeyPatch, sqlite_engine: Engine) -> None:
"""Bind both service-owned ORM sessions to the test's SQLite engine."""
database = SimpleNamespace(engine=sqlite_engine)
monkeypatch.setattr("core.ops.base_trace_instance.db", database)
monkeypatch.setattr("models.account.db", database)
def test_get_service_account_with_tenant_app_not_found(mock_db_session):
mock_db_session.scalar.return_value = None
def _persist_app(session: Session, *, created_by: str | None) -> App:
app = App(
id="some_app_id",
tenant_id="tenant_id",
name="Trace App",
description="",
mode=AppMode.CHAT,
icon_type=None,
icon=None,
icon_background=None,
app_model_config_id=None,
workflow_id=None,
enable_site=True,
enable_api=True,
max_active_requests=None,
created_by=created_by,
)
session.add(app)
session.commit()
return app
config = MagicMock(spec=BaseTracingConfig)
instance = ConcreteTraceInstance(config)
def _persist_account(session: Session) -> Account:
account = Account(name="Creator", email="creator@example.com")
account.id = "creator_id"
session.add(account)
session.commit()
return account
def _trace_instance() -> ConcreteTraceInstance:
# Tracing configuration is a domain collaborator, not an ORM session.
return ConcreteTraceInstance(MagicMock(spec=BaseTracingConfig))
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_get_service_account_with_tenant_app_not_found(sqlite_session: Session):
with pytest.raises(ValueError, match="App with id some_app_id not found"):
instance.get_service_account_with_tenant("some_app_id")
_trace_instance().get_service_account_with_tenant("some_app_id")
def test_get_service_account_with_tenant_no_creator(mock_db_session):
mock_app = MagicMock(spec=App)
mock_app.id = "some_app_id"
mock_app.created_by = None
mock_db_session.scalar.return_value = mock_app
config = MagicMock(spec=BaseTracingConfig)
instance = ConcreteTraceInstance(config)
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_get_service_account_with_tenant_no_creator(sqlite_session: Session):
_persist_app(sqlite_session, created_by=None)
with pytest.raises(ValueError, match="App with id some_app_id has no creator"):
instance.get_service_account_with_tenant("some_app_id")
_trace_instance().get_service_account_with_tenant("some_app_id")
def test_get_service_account_with_tenant_creator_not_found(mock_db_session):
mock_app = MagicMock(spec=App)
mock_app.id = "some_app_id"
mock_app.created_by = "creator_id"
# First call to scalar returns app, second returns None (for account)
mock_db_session.scalar.side_effect = [mock_app, None]
config = MagicMock(spec=BaseTracingConfig)
instance = ConcreteTraceInstance(config)
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_get_service_account_with_tenant_creator_not_found(sqlite_session: Session):
_persist_app(sqlite_session, created_by="creator_id")
with pytest.raises(ValueError, match="Creator account with id creator_id not found for app some_app_id"):
instance.get_service_account_with_tenant("some_app_id")
_trace_instance().get_service_account_with_tenant("some_app_id")
def test_get_service_account_with_tenant_tenant_not_found(mock_db_session):
mock_app = MagicMock(spec=App)
mock_app.id = "some_app_id"
mock_app.created_by = "creator_id"
mock_account = MagicMock(spec=Account)
mock_account.id = "creator_id"
mock_db_session.scalar.side_effect = [mock_app, mock_account, None]
config = MagicMock(spec=BaseTracingConfig)
instance = ConcreteTraceInstance(config)
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_get_service_account_with_tenant_tenant_not_found(sqlite_session: Session):
_persist_app(sqlite_session, created_by="creator_id")
_persist_account(sqlite_session)
with pytest.raises(ValueError, match="Current tenant not found for account creator_id"):
instance.get_service_account_with_tenant("some_app_id")
_trace_instance().get_service_account_with_tenant("some_app_id")
def test_get_service_account_with_tenant_success(mock_db_session):
mock_app = MagicMock(spec=App)
mock_app.id = "some_app_id"
mock_app.created_by = "creator_id"
@pytest.mark.parametrize("sqlite_session", [TABLES], indirect=True)
def test_get_service_account_with_tenant_success(sqlite_session: Session):
_persist_app(sqlite_session, created_by="creator_id")
_persist_account(sqlite_session)
tenant = Tenant(name="Workspace")
tenant.id = "tenant_id"
sqlite_session.add_all(
[
tenant,
TenantAccountJoin(
tenant_id=tenant.id,
account_id="creator_id",
current=True,
role=TenantAccountRole.OWNER,
),
]
)
sqlite_session.commit()
mock_account = MagicMock(spec=Account)
mock_account.id = "creator_id"
result = _trace_instance().get_service_account_with_tenant("some_app_id")
mock_tenant_join = MagicMock(spec=TenantAccountJoin)
mock_tenant_join.tenant_id = "tenant_id"
mock_db_session.scalar.side_effect = [mock_app, mock_account, mock_tenant_join]
config = MagicMock(spec=BaseTracingConfig)
instance = ConcreteTraceInstance(config)
result = instance.get_service_account_with_tenant("some_app_id")
assert result == mock_account
mock_account.set_tenant_id_with_session.assert_called_once_with("tenant_id", session=mock_db_session)
assert result.id == "creator_id"
assert result.current_tenant_id == "tenant_id"
assert result.current_role == TenantAccountRole.OWNER
@@ -29,6 +29,7 @@ from core.plugin.entities.plugin import (
)
from core.plugin.entities.plugin_daemon import (
PluginDecodeResponse,
PluginInstalledIdsDaemonResponse,
PluginInstallTask,
PluginInstallTaskStartResponse,
PluginInstallTaskStatus,
@@ -132,7 +133,13 @@ class TestPluginDiscovery:
plugin_installer, "_request_with_plugin_daemon_response", return_value=mock_response
) as mock_request:
result = plugin_installer.list_plugins_by_category(
"test-tenant", category=PluginCategory.Tool, page=2, page_size=10
"test-tenant",
category=PluginCategory.Tool,
page=2,
page_size=10,
query="weather",
tags=["search", "rag"],
language="zh_Hans",
)
mock_request.assert_called_once()
@@ -141,6 +148,9 @@ class TestPluginDiscovery:
assert call_args.args[2] is PluginListWithoutTotalResponse
assert call_args.kwargs["params"]["page"] == 2
assert call_args.kwargs["params"]["page_size"] == 10
assert call_args.kwargs["params"]["query"] == "weather"
assert call_args.kwargs["params"]["tags"] == ["search", "rag"]
assert call_args.kwargs["params"]["language"] == "zh_Hans"
assert result.list == [mock_plugin_entity]
assert result.has_more is True
@@ -156,6 +166,23 @@ class TestPluginDiscovery:
# Assert: Verify empty list is returned
assert len(result) == 0
def test_list_installed_plugin_ids(self, plugin_installer):
"""The lightweight ID endpoint is unpaginated and does not request plugin details."""
mock_response = PluginInstalledIdsDaemonResponse(plugin_ids=["langgenius/openai", "langgenius/anthropic"])
with patch.object(
plugin_installer, "_request_with_plugin_daemon_response", return_value=mock_response
) as mock_request:
result = plugin_installer.list_installed_plugin_ids("test-tenant", PluginCategory.Tool)
mock_request.assert_called_once_with(
"GET",
"plugin/test-tenant/management/installation/ids",
PluginInstalledIdsDaemonResponse,
params={"category": "tool"},
)
assert result == ["langgenius/openai", "langgenius/anthropic"]
def test_fetch_plugin_by_identifier_found(self, plugin_installer):
"""Test fetching a plugin by its unique identifier when it exists."""
# Arrange: Mock successful fetch
@@ -8,10 +8,49 @@ which provides document storage and retrieval functionality for datasets in the
from unittest.mock import MagicMock, patch
import pytest
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from core.rag.docstore.dataset_docstore import DatasetDocumentStore, DocumentSegment
from core.rag.models.document import AttachmentDocument, Document
from models.dataset import Dataset
from core.rag.models.document import AttachmentDocument, ChildDocument, Document
from models.dataset import ChildChunk, Dataset, SegmentAttachmentBinding
TENANT_ID = "00000000-0000-0000-0000-000000000001"
DATASET_ID = "00000000-0000-0000-0000-000000000002"
DOCUMENT_ID = "00000000-0000-0000-0000-000000000003"
USER_ID = "00000000-0000-0000-0000-000000000004"
def _dataset() -> Dataset:
dataset = MagicMock(spec=Dataset)
dataset.id = DATASET_ID
dataset.tenant_id = TENANT_ID
return dataset
def _persist_segment(
session: Session,
*,
index_node_id: str = "doc-1",
index_node_hash: str = "hash-1",
content: str = "Test content",
tokens: int = 5,
) -> DocumentSegment:
segment = DocumentSegment(
tenant_id=TENANT_ID,
dataset_id=DATASET_ID,
document_id=DOCUMENT_ID,
position=1,
content=content,
word_count=len(content),
tokens=tokens,
created_by=USER_ID,
index_node_id=index_node_id,
index_node_hash=index_node_hash,
)
session.add(segment)
session.flush()
return segment
class TestDatasetDocumentStoreInit:
@@ -132,228 +171,153 @@ class TestDatasetDocumentStoreDocs:
assert result == {}
@pytest.mark.parametrize(
"sqlite_session",
[(DocumentSegment, ChildChunk, SegmentAttachmentBinding)],
indirect=True,
)
class TestDatasetDocumentStoreAddDocuments:
"""Tests for add_documents method."""
def test_add_documents_new_document_with_embedding(self):
"""Test adding new documents with embedding model."""
def test_add_documents_new_document_with_token_count(self, sqlite_session: Session):
"""Test adding a new document with a precomputed token count."""
mock_dataset = MagicMock(spec=Dataset)
mock_dataset.id = "test-dataset-id"
mock_dataset.tenant_id = "tenant-1"
mock_dataset.indexing_technique = "high_quality"
mock_dataset.embedding_model_provider = "provider"
mock_dataset.embedding_model = "model"
document = Document(
page_content="Test content",
metadata={"doc_id": "doc-1", "doc_hash": "hash-1"},
)
store = DatasetDocumentStore(dataset=_dataset(), user_id=USER_ID, document_id=DOCUMENT_ID)
mock_doc = MagicMock(spec=Document)
mock_doc.page_content = "Test content"
mock_doc.metadata = {
"doc_id": "doc-1",
"doc_hash": "hash-1",
}
mock_doc.attachments = None
mock_doc.children = None
store.add_documents(session=sqlite_session, docs=[document], token_counts=[10])
sqlite_session.expire_all()
mock_model_instance = MagicMock()
mock_model_instance.get_text_embedding_num_tokens.return_value = [10]
segment = sqlite_session.scalar(
select(DocumentSegment).where(
DocumentSegment.dataset_id == DATASET_ID,
DocumentSegment.index_node_id == "doc-1",
)
)
assert segment is not None
assert segment.content == "Test content"
assert segment.tokens == 10
assert segment.position == 1
with (
patch("core.rag.docstore.dataset_docstore.ModelManager.for_tenant") as mock_manager_class,
):
mock_session = MagicMock()
mock_session.scalar.return_value = None
mock_manager = MagicMock()
mock_manager.get_model_instance.return_value = mock_model_instance
mock_manager_class.return_value = mock_manager
with patch.object(DatasetDocumentStore, "get_document_segment", return_value=None):
with patch.object(DatasetDocumentStore, "add_multimodel_documents_binding"):
store = DatasetDocumentStore(
dataset=mock_dataset,
user_id="test-user-id",
document_id="test-doc-id",
)
store.add_documents([mock_doc], session=mock_session)
mock_session.add.assert_called()
mock_session.flush.assert_called()
def test_add_documents_update_existing_document(self):
def test_add_documents_update_existing_document(self, sqlite_session: Session):
"""Test updating existing document with allow_update=True."""
mock_dataset = MagicMock(spec=Dataset)
mock_dataset.id = "test-dataset-id"
mock_dataset.tenant_id = "tenant-1"
mock_dataset.indexing_technique = "economy"
mock_dataset.embedding_model_provider = None
mock_dataset.embedding_model = None
existing_segment = _persist_segment(sqlite_session)
document = Document(
page_content="Updated content",
metadata={"doc_id": "doc-1", "doc_hash": "new-hash"},
)
store = DatasetDocumentStore(dataset=_dataset(), user_id=USER_ID, document_id=DOCUMENT_ID)
mock_doc = MagicMock(spec=Document)
mock_doc.page_content = "Updated content"
mock_doc.metadata = {
"doc_id": "doc-1",
"doc_hash": "new-hash",
}
mock_doc.attachments = None
mock_doc.children = None
store.add_documents(session=sqlite_session, docs=[document], token_counts=[0])
sqlite_session.expire_all()
mock_existing_segment = MagicMock()
mock_existing_segment.id = "seg-1"
updated_segment = sqlite_session.get(DocumentSegment, existing_segment.id)
assert updated_segment is not None
assert updated_segment.content == "Updated content"
assert updated_segment.index_node_hash == "new-hash"
assert updated_segment.tokens == 0
mock_session = MagicMock()
mock_session.scalar.return_value = 5
with patch.object(DatasetDocumentStore, "get_document_segment", return_value=mock_existing_segment):
with patch.object(DatasetDocumentStore, "add_multimodel_documents_binding"):
store = DatasetDocumentStore(
dataset=mock_dataset,
user_id="test-user-id",
document_id="test-doc-id",
)
store.add_documents([mock_doc], session=mock_session)
mock_session.flush.assert_called()
def test_add_documents_raises_when_not_allowed(self):
def test_add_documents_raises_when_not_allowed(self, sqlite_session: Session):
"""Test that adding existing doc without allow_update raises ValueError."""
mock_dataset = MagicMock(spec=Dataset)
mock_dataset.id = "test-dataset-id"
mock_dataset.tenant_id = "tenant-1"
mock_dataset.indexing_technique = "economy"
_persist_segment(sqlite_session)
document = Document(
page_content="Test content",
metadata={"doc_id": "doc-1", "doc_hash": "hash-1"},
)
store = DatasetDocumentStore(dataset=_dataset(), user_id=USER_ID, document_id=DOCUMENT_ID)
mock_doc = MagicMock(spec=Document)
mock_doc.page_content = "Test content"
mock_doc.metadata = {
"doc_id": "doc-1",
"doc_hash": "hash-1",
}
mock_doc.attachments = None
mock_doc.children = None
mock_existing_segment = MagicMock()
mock_session = MagicMock()
with patch.object(DatasetDocumentStore, "get_document_segment", return_value=mock_existing_segment):
store = DatasetDocumentStore(
dataset=mock_dataset,
user_id="test-user-id",
document_id="test-doc-id",
with pytest.raises(ValueError, match="already exists"):
store.add_documents(
session=sqlite_session,
docs=[document],
token_counts=[0],
allow_update=False,
)
with pytest.raises(ValueError, match="already exists"):
store.add_documents([mock_doc], session=mock_session, allow_update=False)
assert sqlite_session.scalar(select(func.count()).select_from(DocumentSegment)) == 1
def test_add_documents_with_answer_metadata(self):
def test_add_documents_with_answer_metadata(self, sqlite_session: Session):
"""Test adding document with answer in metadata."""
mock_dataset = MagicMock(spec=Dataset)
mock_dataset.id = "test-dataset-id"
mock_dataset.tenant_id = "tenant-1"
mock_dataset.indexing_technique = "economy"
document = Document(
page_content="Test content",
metadata={
"doc_id": "doc-1",
"doc_hash": "hash-1",
"answer": "Test answer",
},
)
store = DatasetDocumentStore(dataset=_dataset(), user_id=USER_ID, document_id=DOCUMENT_ID)
mock_doc = MagicMock(spec=Document)
mock_doc.page_content = "Test content"
mock_doc.metadata = {
"doc_id": "doc-1",
"doc_hash": "hash-1",
"answer": "Test answer",
}
mock_doc.attachments = None
mock_doc.children = None
store.add_documents(session=sqlite_session, docs=[document], token_counts=[0])
sqlite_session.expire_all()
mock_session = MagicMock()
mock_session.scalar.return_value = None
segment = sqlite_session.scalar(select(DocumentSegment))
assert segment is not None
assert segment.answer == "Test answer"
with patch.object(DatasetDocumentStore, "get_document_segment", return_value=None):
with patch.object(DatasetDocumentStore, "add_multimodel_documents_binding"):
store = DatasetDocumentStore(
dataset=mock_dataset,
user_id="test-user-id",
document_id="test-doc-id",
)
store.add_documents([mock_doc], session=mock_session)
mock_session.add.assert_called()
def test_add_documents_with_invalid_document_type(self):
def test_add_documents_with_invalid_document_type(self, sqlite_session: Session):
"""Test that non-Document raises ValueError."""
mock_dataset = MagicMock(spec=Dataset)
mock_dataset.id = "test-dataset-id"
mock_session = MagicMock()
store = DatasetDocumentStore(
dataset=mock_dataset,
user_id="test-user-id",
document_id="test-doc-id",
)
store = DatasetDocumentStore(dataset=_dataset(), user_id=USER_ID, document_id=DOCUMENT_ID)
with pytest.raises(ValueError, match="must be a Document"):
store.add_documents(["not a document"], session=mock_session)
store.add_documents(session=sqlite_session, docs=["not a document"], token_counts=[0]) # type: ignore[list-item]
def test_add_documents_with_none_metadata(self):
def test_add_documents_with_none_metadata(self, sqlite_session: Session):
"""Test that document with None metadata raises ValueError."""
mock_dataset = MagicMock(spec=Dataset)
mock_dataset.id = "test-dataset-id"
mock_doc = MagicMock(spec=Document)
mock_doc.page_content = "Test content"
mock_doc.metadata = None
mock_session = MagicMock()
store = DatasetDocumentStore(
dataset=mock_dataset,
user_id="test-user-id",
document_id="test-doc-id",
)
document = MagicMock(spec=Document)
document.metadata = None
store = DatasetDocumentStore(dataset=_dataset(), user_id=USER_ID, document_id=DOCUMENT_ID)
with pytest.raises(ValueError, match="metadata must be a dict"):
store.add_documents([mock_doc], session=mock_session)
store.add_documents(session=sqlite_session, docs=[document], token_counts=[0])
def test_add_documents_with_save_child(self):
def test_add_documents_with_save_child(self, sqlite_session: Session):
"""Test adding documents with save_child=True."""
mock_dataset = MagicMock(spec=Dataset)
mock_dataset.id = "test-dataset-id"
mock_dataset.tenant_id = "tenant-1"
mock_dataset.indexing_technique = "economy"
mock_child = MagicMock(spec=Document)
mock_child.page_content = "Child content"
mock_child.metadata = {
"doc_id": "child-1",
"doc_hash": "child-hash",
}
mock_doc = MagicMock(spec=Document)
mock_doc.page_content = "Test content"
mock_doc.metadata = {
"doc_id": "doc-1",
"doc_hash": "hash-1",
}
mock_doc.attachments = None
mock_doc.children = [mock_child]
mock_session = MagicMock()
mock_session.scalar.return_value = None
with patch.object(DatasetDocumentStore, "get_document_segment", return_value=None):
with patch.object(DatasetDocumentStore, "add_multimodel_documents_binding"):
store = DatasetDocumentStore(
dataset=mock_dataset,
user_id="test-user-id",
document_id="test-doc-id",
document = Document(
page_content="Test content",
metadata={"doc_id": "doc-1", "doc_hash": "hash-1"},
children=[
ChildDocument(
page_content="Child content",
metadata={"doc_id": "child-1", "doc_hash": "child-hash"},
)
],
)
store = DatasetDocumentStore(dataset=_dataset(), user_id=USER_ID, document_id=DOCUMENT_ID)
store.add_documents([mock_doc], session=mock_session, save_child=True)
store.add_documents(
session=sqlite_session,
docs=[document],
token_counts=[0],
save_child=True,
)
sqlite_session.expire_all()
mock_session.add.assert_called()
child = sqlite_session.scalar(select(ChildChunk))
assert child is not None
assert child.content == "Child content"
assert child.index_node_id == "child-1"
def test_add_documents_rejects_mismatched_token_counts(self, sqlite_session: Session):
document = Document(
page_content="Test content",
metadata={"doc_id": "doc-1", "doc_hash": "hash-1"},
)
store = DatasetDocumentStore(dataset=_dataset(), user_id=USER_ID, document_id=DOCUMENT_ID)
with pytest.raises(ValueError):
store.add_documents(session=sqlite_session, docs=[document], token_counts=[])
assert sqlite_session.scalar(select(func.count()).select_from(DocumentSegment)) == 0
class TestDatasetDocumentStoreExists:
@@ -722,88 +686,85 @@ class TestDatasetDocumentStoreMultimodelBinding:
mock_session.add.assert_not_called()
@pytest.mark.parametrize(
"sqlite_session",
[(DocumentSegment, ChildChunk, SegmentAttachmentBinding)],
indirect=True,
)
class TestDatasetDocumentStoreAddDocumentsUpdateChild:
"""Tests for add_documents when updating existing documents with children."""
def test_add_documents_update_existing_with_children(self):
def test_add_documents_update_existing_with_children(self, sqlite_session: Session):
"""Test updating existing document with save_child=True and children."""
mock_dataset = MagicMock(spec=Dataset)
mock_dataset.id = "test-dataset-id"
mock_dataset.tenant_id = "tenant-1"
mock_dataset.indexing_technique = "economy"
mock_child = MagicMock(spec=Document)
mock_child.page_content = "Updated child content"
mock_child.metadata = {
"doc_id": "child-1",
"doc_hash": "new-child-hash",
}
mock_doc = MagicMock(spec=Document)
mock_doc.page_content = "Updated content"
mock_doc.metadata = {
"doc_id": "doc-1",
"doc_hash": "new-hash",
}
mock_doc.attachments = None
mock_doc.children = [mock_child]
mock_existing_segment = MagicMock()
mock_existing_segment.id = "seg-1"
mock_session = MagicMock()
mock_session.scalar.return_value = 5
with patch.object(DatasetDocumentStore, "get_document_segment", return_value=mock_existing_segment):
with patch.object(DatasetDocumentStore, "add_multimodel_documents_binding"):
store = DatasetDocumentStore(
dataset=mock_dataset,
user_id="test-user-id",
document_id="test-doc-id",
segment = _persist_segment(sqlite_session)
sqlite_session.add(
ChildChunk(
tenant_id=TENANT_ID,
dataset_id=DATASET_ID,
document_id=DOCUMENT_ID,
segment_id=segment.id,
position=1,
index_node_id="old-child",
index_node_hash="old-child-hash",
content="Old child content",
word_count=len("Old child content"),
created_by=USER_ID,
)
)
sqlite_session.flush()
document = Document(
page_content="Updated content",
metadata={"doc_id": "doc-1", "doc_hash": "new-hash"},
children=[
ChildDocument(
page_content="Updated child content",
metadata={"doc_id": "child-1", "doc_hash": "new-child-hash"},
)
],
)
store = DatasetDocumentStore(dataset=_dataset(), user_id=USER_ID, document_id=DOCUMENT_ID)
store.add_documents([mock_doc], session=mock_session, save_child=True)
store.add_documents(
session=sqlite_session,
docs=[document],
token_counts=[0],
save_child=True,
)
sqlite_session.expire_all()
mock_session.execute.assert_called()
mock_session.flush.assert_called()
children = sqlite_session.scalars(select(ChildChunk).order_by(ChildChunk.position)).all()
assert len(children) == 1
assert children[0].content == "Updated child content"
assert children[0].index_node_id == "child-1"
@pytest.mark.parametrize(
"sqlite_session",
[(DocumentSegment, ChildChunk, SegmentAttachmentBinding)],
indirect=True,
)
class TestDatasetDocumentStoreAddDocumentsUpdateAnswer:
"""Tests for add_documents when updating existing documents with answer metadata."""
def test_add_documents_update_existing_with_answer(self):
def test_add_documents_update_existing_with_answer(self, sqlite_session: Session):
"""Test updating existing document with answer in metadata."""
mock_dataset = MagicMock(spec=Dataset)
mock_dataset.id = "test-dataset-id"
mock_dataset.tenant_id = "tenant-1"
mock_dataset.indexing_technique = "economy"
existing_segment = _persist_segment(sqlite_session)
document = Document(
page_content="Updated content",
metadata={
"doc_id": "doc-1",
"doc_hash": "new-hash",
"answer": "Updated answer",
},
)
store = DatasetDocumentStore(dataset=_dataset(), user_id=USER_ID, document_id=DOCUMENT_ID)
mock_doc = MagicMock(spec=Document)
mock_doc.page_content = "Updated content"
mock_doc.metadata = {
"doc_id": "doc-1",
"doc_hash": "new-hash",
"answer": "Updated answer",
}
mock_doc.attachments = None
mock_doc.children = None
store.add_documents(session=sqlite_session, docs=[document], token_counts=[0])
sqlite_session.expire_all()
mock_existing_segment = MagicMock()
mock_existing_segment.id = "seg-1"
mock_session = MagicMock()
mock_session.scalar.return_value = 5
with patch.object(DatasetDocumentStore, "get_document_segment", return_value=mock_existing_segment):
with patch.object(DatasetDocumentStore, "add_multimodel_documents_binding"):
store = DatasetDocumentStore(
dataset=mock_dataset,
user_id="test-user-id",
document_id="test-doc-id",
)
store.add_documents([mock_doc], session=mock_session)
mock_session.flush.assert_called()
updated_segment = sqlite_session.get(DocumentSegment, existing_segment.id)
assert updated_segment is not None
assert updated_segment.answer == "Updated answer"
assert updated_segment.tokens == 0
@@ -0,0 +1,56 @@
from unittest.mock import Mock, patch
from core.rag.embedding.token_counter import calculate_segment_token_counts
from core.rag.index_processor.constant.index_type import IndexTechniqueType
from core.rag.models.document import Document
from models.dataset import Dataset
def test_high_quality_counts_each_document_once() -> None:
dataset = Mock(spec=Dataset)
dataset.tenant_id = "tenant-1"
dataset.indexing_technique = IndexTechniqueType.HIGH_QUALITY
dataset.embedding_model_provider = "provider"
dataset.embedding_model = "model"
documents = [
Document(page_content="first", metadata={}),
Document(page_content="second", metadata={}),
Document(page_content="third", metadata={}),
]
with patch("core.rag.embedding.token_counter.ModelManager.for_tenant") as model_manager_factory:
embedding_model = model_manager_factory.return_value.get_model_instance.return_value
embedding_model.get_text_embedding_num_tokens.return_value = [11, 22, 33]
result = calculate_segment_token_counts(dataset=dataset, documents=documents)
assert result == [11, 22, 33]
model_manager_factory.assert_called_once_with(tenant_id=dataset.tenant_id)
model_manager_factory.return_value.get_model_instance.assert_called_once()
embedding_model.get_text_embedding_num_tokens.assert_called_once_with(["first", "second", "third"])
def test_economy_returns_zero_without_loading_model() -> None:
dataset = Mock(spec=Dataset)
dataset.indexing_technique = IndexTechniqueType.ECONOMY
documents = [
Document(page_content="first", metadata={}),
Document(page_content="second", metadata={}),
]
with patch("core.rag.embedding.token_counter.ModelManager.for_tenant") as model_manager_factory:
result = calculate_segment_token_counts(dataset=dataset, documents=documents)
assert result == [0, 0]
model_manager_factory.assert_not_called()
def test_empty_documents_return_without_loading_model() -> None:
dataset = Mock(spec=Dataset)
dataset.indexing_technique = IndexTechniqueType.HIGH_QUALITY
with patch("core.rag.embedding.token_counter.ModelManager.for_tenant") as model_manager_factory:
result = calculate_segment_token_counts(dataset=dataset, documents=[])
assert result == []
model_manager_factory.assert_not_called()
@@ -271,14 +271,26 @@ class TestParagraphIndexProcessor:
patch(
"core.rag.index_processor.processor.paragraph_index_processor.DatasetDocumentStore"
) as mock_store_cls,
patch(
"core.rag.index_processor.processor.paragraph_index_processor.calculate_segment_token_counts"
) as mock_token_counter,
patch("core.rag.index_processor.processor.paragraph_index_processor.Vector") as mock_vector_cls,
):
mock_token_counter.side_effect = lambda **_kwargs: phase_events.append("count") or [11, 22]
mock_store_cls.return_value.add_documents.side_effect = lambda **_kwargs: phase_events.append("store")
mock_vector_cls.return_value.create.side_effect = lambda _documents: phase_events.append("vector")
processor.index(dataset, dataset_document, ["chunk-1", "chunk-2"], session)
assert phase_events == ["store", "commit", "vector"]
mock_store_cls.return_value.add_documents.assert_called_once()
assert phase_events == ["count", "store", "commit", "vector"]
documents = mock_token_counter.call_args.kwargs["documents"]
assert [document.page_content for document in documents] == ["chunk-1", "chunk-2"]
mock_token_counter.assert_called_once_with(dataset=dataset, documents=documents)
mock_store_cls.return_value.add_documents.assert_called_once_with(
session=session,
docs=documents,
token_counts=[11, 22],
save_child=False,
)
mock_vector_cls.assert_called_once_with(dataset, session=session)
mock_vector_cls.return_value.create.assert_called_once()
mock_vector_cls.return_value.create_multimodal.assert_called_once()
@@ -299,13 +311,18 @@ class TestParagraphIndexProcessor:
patch(
"core.rag.index_processor.processor.paragraph_index_processor.DatasetDocumentStore"
) as mock_store_cls,
patch(
"core.rag.index_processor.processor.paragraph_index_processor.calculate_segment_token_counts"
) as mock_token_counter,
patch("core.rag.index_processor.processor.paragraph_index_processor.Keyword") as mock_keyword_cls,
):
mock_token_counter.side_effect = lambda **_kwargs: phase_events.append("count") or [0]
mock_store_cls.return_value.add_documents.side_effect = lambda **_kwargs: phase_events.append("store")
mock_keyword_cls.return_value.add_texts.side_effect = lambda *_args: phase_events.append("keyword")
processor.index(dataset, dataset_document, ["chunk-3"], session)
assert phase_events == ["store", "commit", "keyword"]
assert phase_events == ["count", "store", "commit", "keyword"]
mock_token_counter.assert_called_once()
mock_keyword_cls.return_value.add_texts.assert_called_once()
def test_index_multimodal_structure_handles_files_and_account_lookup(
@@ -341,6 +358,10 @@ class TestParagraphIndexProcessor:
processor, "_get_content_files", return_value=[AttachmentDocument(page_content="img", metadata={})]
) as mock_files,
patch("core.rag.index_processor.processor.paragraph_index_processor.DatasetDocumentStore"),
patch(
"core.rag.index_processor.processor.paragraph_index_processor.calculate_segment_token_counts",
return_value=[11, 22],
),
patch("core.rag.index_processor.processor.paragraph_index_processor.Vector"),
):
processor.index(dataset, dataset_document, {"general_chunks": []}, session)
@@ -362,17 +362,29 @@ class TestParentChildIndexProcessor:
patch(
"core.rag.index_processor.processor.parent_child_index_processor.DatasetDocumentStore"
) as mock_store_cls,
patch(
"core.rag.index_processor.processor.parent_child_index_processor.calculate_segment_token_counts"
) as mock_token_counter,
patch("core.rag.index_processor.processor.parent_child_index_processor.Vector") as mock_vector_cls,
):
mock_token_counter.side_effect = lambda **_kwargs: phase_events.append("count") or [11]
mock_store_cls.return_value.add_documents.side_effect = lambda **_kwargs: phase_events.append("store")
mock_vector_cls.return_value.create.side_effect = lambda _documents: phase_events.append("vector")
processor.index(dataset, dataset_document, {"parent_child_chunks": []}, session)
assert phase_events == ["store", "commit", "vector"]
assert phase_events == ["count", "store", "commit", "vector"]
assert dataset_document.dataset_process_rule_id == "rule-1"
session.add.assert_called_once_with(dataset_rule)
session.flush.assert_called_once()
mock_store_cls.return_value.add_documents.assert_called_once()
documents = mock_token_counter.call_args.kwargs["documents"]
assert [document.page_content for document in documents] == ["parent text"]
mock_token_counter.assert_called_once_with(dataset=dataset, documents=documents)
mock_store_cls.return_value.add_documents.assert_called_once_with(
session=session,
docs=documents,
token_counts=[11],
save_child=True,
)
mock_vector_cls.assert_called_once_with(dataset, session=session)
assert mock_vector_cls.return_value.create.call_count == 1
mock_vector_cls.return_value.create_multimodal.assert_called_once()
@@ -413,6 +425,10 @@ class TestParentChildIndexProcessor:
processor, "_get_content_files", return_value=[AttachmentDocument(page_content="image", metadata={})]
) as mock_files,
patch("core.rag.index_processor.processor.parent_child_index_processor.DatasetDocumentStore"),
patch(
"core.rag.index_processor.processor.parent_child_index_processor.calculate_segment_token_counts",
return_value=[11],
),
patch("core.rag.index_processor.processor.parent_child_index_processor.Vector"),
):
processor.index(dataset, dataset_document, {"parent_child_chunks": []}, session)
@@ -292,14 +292,26 @@ class TestQAIndexProcessor:
"core.rag.index_processor.processor.qa_index_processor.helper.generate_text_hash", return_value="hash"
),
patch("core.rag.index_processor.processor.qa_index_processor.DatasetDocumentStore") as mock_store_cls,
patch(
"core.rag.index_processor.processor.qa_index_processor.calculate_segment_token_counts"
) as mock_token_counter,
patch("core.rag.index_processor.processor.qa_index_processor.Vector") as mock_vector_cls,
):
mock_token_counter.side_effect = lambda **_kwargs: phase_events.append("count") or [11, 22]
mock_store_cls.return_value.add_documents.side_effect = lambda **_kwargs: phase_events.append("store")
mock_vector_cls.return_value.create.side_effect = lambda _documents: phase_events.append("vector")
processor.index(dataset, dataset_document, {"qa_chunks": []}, session)
assert phase_events == ["store", "commit", "vector"]
mock_store_cls.return_value.add_documents.assert_called_once()
assert phase_events == ["count", "store", "commit", "vector"]
documents = mock_token_counter.call_args.kwargs["documents"]
assert [document.page_content for document in documents] == ["Q1", "Q2"]
mock_token_counter.assert_called_once_with(dataset=dataset, documents=documents)
mock_store_cls.return_value.add_documents.assert_called_once_with(
session=session,
docs=documents,
token_counts=[11, 22],
save_child=False,
)
mock_vector_cls.return_value.create.assert_called_once()
def test_index_requires_high_quality(
@@ -318,6 +330,10 @@ class TestQAIndexProcessor:
"core.rag.index_processor.processor.qa_index_processor.helper.generate_text_hash", return_value="hash"
),
patch("core.rag.index_processor.processor.qa_index_processor.DatasetDocumentStore"),
patch(
"core.rag.index_processor.processor.qa_index_processor.calculate_segment_token_counts",
return_value=[0],
),
):
with pytest.raises(ValueError, match="must be high quality"):
processor.index(dataset, dataset_document, {"qa_chunks": []}, session)
@@ -69,6 +69,7 @@ from graphon.model_runtime.entities.model_entities import ModelType
from libs.datetime_utils import naive_utc_now
from models.dataset import Dataset, DatasetProcessRule, DocumentSegment
from models.dataset import Document as DatasetDocument
from models.enums import SegmentStatus
from models.model import Account
# ============================================================================
@@ -611,7 +612,7 @@ class TestIndexingRunnerLoad:
- Keyword index creation
- Multi-threaded processing
- Document segment status updates
- Token counting
- Precomputed token totals
- Error handling during loading
"""
@@ -677,16 +678,10 @@ class TestIndexingRunnerLoad:
"""Test loading with high quality indexing (vector embeddings)."""
# Arrange
runner = IndexingRunner()
mock_embedding_instance = MagicMock()
mock_embedding_instance.get_text_embedding_num_tokens.return_value = 100
model_manager = mock_dependencies["model_manager"].return_value
model_manager.get_model_instance.return_value = mock_embedding_instance
mock_processor = MagicMock()
# Mock ThreadPoolExecutor
mock_future = MagicMock()
mock_future.result.return_value = 300 # Total tokens
mock_future.result.return_value = None
mock_executor_instance = MagicMock()
mock_executor_instance.__enter__.return_value = mock_executor_instance
mock_executor_instance.__exit__.return_value = None
@@ -694,20 +689,51 @@ class TestIndexingRunnerLoad:
mock_dependencies["executor"].return_value = mock_executor_instance
# Mock update_document_index_status to avoid database calls
with patch.object(runner, "_update_document_index_status"):
with patch.object(runner, "_update_document_index_status") as mock_update_status:
# Act
runner._load(
mock_processor,
sample_dataset,
sample_dataset_document,
sample_documents,
mock_dependencies["session"],
session=mock_dependencies["session"],
dataset=sample_dataset,
dataset_document=sample_dataset_document,
documents=sample_documents,
total_tokens=300,
)
# Assert
model_manager.get_model_instance.assert_called_once()
mock_dependencies["model_manager"].assert_not_called()
# Verify executor was used for parallel processing
assert mock_executor_instance.submit.called
for submit_call in mock_executor_instance.submit.call_args_list:
assert submit_call.args[0] == runner._process_chunk
assert len(submit_call.args) == 6
mock_future.result.assert_called()
assert mock_update_status.call_args.kwargs["extra_update_params"][DatasetDocument.tokens] == 300
def test_load_propagates_worker_errors(
self, mock_dependencies, sample_dataset, sample_dataset_document, sample_documents
):
runner = IndexingRunner()
mock_future = MagicMock()
mock_future.result.side_effect = RuntimeError("index failed")
mock_executor_instance = MagicMock()
mock_executor_instance.__enter__.return_value = mock_executor_instance
mock_executor_instance.__exit__.return_value = None
mock_executor_instance.submit.return_value = mock_future
mock_dependencies["executor"].return_value = mock_executor_instance
with (
patch.object(runner, "_update_document_index_status") as mock_update_status,
pytest.raises(RuntimeError, match="index failed"),
):
runner._load(
session=mock_dependencies["session"],
dataset=sample_dataset,
dataset_document=sample_dataset_document,
documents=sample_documents,
total_tokens=300,
)
mock_update_status.assert_not_called()
def test_load_with_economy_indexing(
self, mock_dependencies, sample_dataset, sample_dataset_document, sample_documents
@@ -717,8 +743,6 @@ class TestIndexingRunnerLoad:
runner = IndexingRunner()
sample_dataset.indexing_technique = IndexTechniqueType.ECONOMY
mock_processor = MagicMock()
# Mock thread for keyword indexing
mock_thread_instance = MagicMock()
mock_thread_instance.join = MagicMock()
@@ -728,11 +752,11 @@ class TestIndexingRunnerLoad:
with patch.object(runner, "_update_document_index_status"):
# Act
runner._load(
mock_processor,
sample_dataset,
sample_dataset_document,
sample_documents,
mock_dependencies["session"],
session=mock_dependencies["session"],
dataset=sample_dataset,
dataset_document=sample_dataset_document,
documents=sample_documents,
total_tokens=0,
)
# Assert
@@ -759,16 +783,9 @@ class TestIndexingRunnerLoad:
)
]
mock_embedding_instance = MagicMock()
mock_embedding_instance.get_text_embedding_num_tokens.return_value = 50
model_manager = mock_dependencies["model_manager"].return_value
model_manager.get_model_instance.return_value = mock_embedding_instance
mock_processor = MagicMock()
# Mock ThreadPoolExecutor
mock_future = MagicMock()
mock_future.result.return_value = 150
mock_future.result.return_value = None
mock_executor_instance = MagicMock()
mock_executor_instance.__enter__.return_value = mock_executor_instance
mock_executor_instance.__exit__.return_value = None
@@ -779,14 +796,15 @@ class TestIndexingRunnerLoad:
with patch.object(runner, "_update_document_index_status"):
# Act
runner._load(
mock_processor,
sample_dataset,
sample_dataset_document,
sample_documents,
mock_dependencies["session"],
session=mock_dependencies["session"],
dataset=sample_dataset,
dataset_document=sample_dataset_document,
documents=sample_documents,
total_tokens=150,
)
# Assert
mock_dependencies["model_manager"].assert_not_called()
# Verify no keyword thread for parent-child index
mock_dependencies["thread"].assert_not_called()
@@ -850,6 +868,7 @@ class TestIndexingRunnerRun:
segment.index_node_hash = "parent-hash"
segment.document_id = dataset_document.id
segment.dataset_id = dataset_document.dataset_id
segment.tokens = 12
segment.get_child_chunks.return_value = [
SimpleNamespace(content="child", index_node_id="child-node", index_node_hash="child-hash")
]
@@ -862,6 +881,32 @@ class TestIndexingRunnerRun:
segment.get_child_chunks.assert_called_once_with(session=session)
assert load.call_args.kwargs["documents"][0].children[0].page_content == "child"
assert load.call_args.kwargs["total_tokens"] == 12
def test_run_in_indexing_status_uses_tokens_from_all_segments(self, mock_dependencies, sample_dataset_documents):
runner = IndexingRunner()
dataset_document = sample_dataset_documents[0]
dataset = Mock(spec=Dataset)
completed_segment = Mock(spec=DocumentSegment)
completed_segment.status = SegmentStatus.COMPLETED
completed_segment.tokens = 10
incomplete_segment = Mock(spec=DocumentSegment)
incomplete_segment.status = SegmentStatus.WAITING
incomplete_segment.tokens = 20
incomplete_segment.content = "pending"
incomplete_segment.index_node_id = "pending-node"
incomplete_segment.index_node_hash = "pending-hash"
incomplete_segment.document_id = dataset_document.id
incomplete_segment.dataset_id = dataset_document.dataset_id
session = mock_dependencies["session"]
session.get.side_effect = lambda model, _: dataset_document if model is DatasetDocument else dataset
session.scalars.return_value.all.return_value = [completed_segment, incomplete_segment]
with patch.object(runner, "_load") as load:
runner.run_in_indexing_status(dataset_document, session)
assert load.call_args.kwargs["documents"][0].page_content == "pending"
assert load.call_args.kwargs["total_tokens"] == 30
def test_run_success_single_document(self, mock_dependencies, sample_dataset_documents):
"""Test successful run with single document."""
@@ -953,6 +998,98 @@ class TestIndexingRunnerRun:
with pytest.raises(DocumentIsPausedError):
runner.run([doc], mock_dependencies["session"])
def test_run_counts_each_transformed_document_once(self, mock_dependencies, sample_dataset_documents):
runner = IndexingRunner()
dataset_document = sample_dataset_documents[0]
dataset = Mock(spec=Dataset)
dataset.id = dataset_document.dataset_id
dataset.tenant_id = dataset_document.tenant_id
dataset.indexing_technique = IndexTechniqueType.HIGH_QUALITY
current_user = Mock(spec=Account)
transformed_documents = [
Document(page_content="first", metadata={"doc_id": "first", "doc_hash": "hash-first"}),
Document(page_content="second", metadata={"doc_id": "second", "doc_hash": "hash-second"}),
]
model_dispatch = {
DatasetDocument: dataset_document,
Dataset: dataset,
Account: current_user,
}
mock_dependencies["session"].get.side_effect = lambda model, _: model_dispatch.get(model)
process_rule = Mock(spec=DatasetProcessRule)
process_rule.to_dict.return_value = {"mode": "automatic", "rules": {}}
mock_dependencies["session"].scalar.return_value = process_rule
with (
patch.object(runner, "_extract", return_value=[Document(page_content="source", metadata={})]),
patch.object(runner, "_transform", return_value=transformed_documents),
patch.object(runner, "_load_segments") as load_segments,
patch.object(runner, "_load") as load,
patch(
"core.indexing_runner.calculate_segment_token_counts",
return_value=[11, 22],
) as calculate_token_counts,
):
runner.run([dataset_document], mock_dependencies["session"])
calculate_token_counts.assert_called_once_with(dataset=dataset, documents=transformed_documents)
load_segments.assert_called_once_with(
session=mock_dependencies["session"],
dataset=dataset,
dataset_document=dataset_document,
documents=transformed_documents,
token_counts=[11, 22],
)
assert load.call_args.kwargs["total_tokens"] == 33
def test_run_in_splitting_status_counts_each_transformed_document_once(
self, mock_dependencies, sample_dataset_documents
):
runner = IndexingRunner()
dataset_document = sample_dataset_documents[0]
dataset_document.created_by = "user-1"
dataset = Mock(spec=Dataset)
dataset.id = dataset_document.dataset_id
dataset.tenant_id = dataset_document.tenant_id
dataset.indexing_technique = IndexTechniqueType.HIGH_QUALITY
current_user = Mock(spec=Account)
transformed_documents = [
Document(page_content="first", metadata={"doc_id": "first", "doc_hash": "hash-first"}),
Document(page_content="second", metadata={"doc_id": "second", "doc_hash": "hash-second"}),
]
model_dispatch = {
DatasetDocument: dataset_document,
Dataset: dataset,
Account: current_user,
}
mock_dependencies["session"].get.side_effect = lambda model, _: model_dispatch.get(model)
mock_dependencies["session"].scalars.return_value.all.return_value = []
process_rule = Mock(spec=DatasetProcessRule)
process_rule.to_dict.return_value = {"mode": "automatic", "rules": {}}
mock_dependencies["session"].scalar.return_value = process_rule
with (
patch.object(runner, "_extract", return_value=[Document(page_content="source", metadata={})]),
patch.object(runner, "_transform", return_value=transformed_documents),
patch.object(runner, "_load_segments") as load_segments,
patch.object(runner, "_load") as load,
patch(
"core.indexing_runner.calculate_segment_token_counts",
return_value=[11, 22],
) as calculate_token_counts,
):
runner.run_in_splitting_status(dataset_document, mock_dependencies["session"])
calculate_token_counts.assert_called_once_with(dataset=dataset, documents=transformed_documents)
load_segments.assert_called_once_with(
session=mock_dependencies["session"],
dataset=dataset,
dataset_document=dataset_document,
documents=transformed_documents,
token_counts=[11, 22],
)
assert load.call_args.kwargs["total_tokens"] == 33
def test_run_handles_provider_token_error(self, mock_dependencies, sample_dataset_documents):
"""Test run handles ProviderTokenNotInitError and updates document status."""
# Arrange
@@ -1395,7 +1532,11 @@ class TestIndexingRunnerLoadSegments:
):
# Act
runner._load_segments(
sample_dataset, sample_dataset_document, sample_documents, mock_dependencies["session"]
session=mock_dependencies["session"],
dataset=sample_dataset,
dataset_document=sample_dataset_document,
documents=sample_documents,
token_counts=[10, 20],
)
# Assert
@@ -1405,7 +1546,10 @@ class TestIndexingRunnerLoadSegments:
document_id=sample_dataset_document.id,
)
mock_docstore_instance.add_documents.assert_called_once_with(
docs=sample_documents, save_child=False, session=mock_dependencies["session"]
session=mock_dependencies["session"],
docs=sample_documents,
save_child=False,
token_counts=[10, 20],
)
def test_load_segments_parent_child_index(
@@ -1435,12 +1579,19 @@ class TestIndexingRunnerLoadSegments:
):
# Act
runner._load_segments(
sample_dataset, sample_dataset_document, sample_documents, mock_dependencies["session"]
session=mock_dependencies["session"],
dataset=sample_dataset,
dataset_document=sample_dataset_document,
documents=sample_documents,
token_counts=[10, 20],
)
# Assert
mock_docstore_instance.add_documents.assert_called_once_with(
docs=sample_documents, save_child=True, session=mock_dependencies["session"]
session=mock_dependencies["session"],
docs=sample_documents,
save_child=True,
token_counts=[10, 20],
)
def test_load_segments_updates_word_count(
@@ -1462,7 +1613,11 @@ class TestIndexingRunnerLoadSegments:
):
# Act
runner._load_segments(
sample_dataset, sample_dataset_document, sample_documents, mock_dependencies["session"]
session=mock_dependencies["session"],
dataset=sample_dataset,
dataset_document=sample_dataset_document,
documents=sample_documents,
token_counts=[10, 20],
)
# Assert
@@ -1565,7 +1720,6 @@ class TestIndexingRunnerProcessChunk:
"""Unit tests for chunk processing in parallel.
Tests cover:
- Token counting
- Vector index creation
- Segment status updates
- Pause detection during processing
@@ -1590,16 +1744,12 @@ class TestIndexingRunnerProcessChunk:
app.app_context.return_value.__exit__ = MagicMock()
return app
def test_process_chunk_counts_tokens(self, mock_dependencies, mock_flask_app):
"""Test process chunk correctly counts tokens."""
def test_process_chunk_loads_index_and_completes_segments(self, mock_dependencies, mock_flask_app):
"""Test process chunk loads the index and completes segments without counting tokens."""
# Arrange
from core.indexing_runner import IndexingRunner
runner = IndexingRunner()
mock_embedding_instance = MagicMock()
# Mock to return an iterable that sums to 150 tokens
mock_embedding_instance.get_text_embedding_num_tokens.return_value = [75, 75]
mock_processor = MagicMock()
chunk_documents = [
Document(page_content="Chunk 1", metadata={"doc_id": "c1"}),
@@ -1638,18 +1788,19 @@ class TestIndexingRunnerProcessChunk:
mock_factory.return_value.init_index_processor.return_value = mock_processor
# Act - the method creates its own app_context and session
tokens = runner._process_chunk(
result = runner._process_chunk(
mock_flask_app,
IndexStructureType.PARAGRAPH_INDEX,
chunk_documents,
mock_dataset.id,
mock_dataset_document.id,
mock_embedding_instance,
)
# Assert
assert tokens == 150
assert result is None
mock_processor.load.assert_called_once()
mock_dependencies["session"].execute.assert_called_once()
mock_dependencies["session"].commit.assert_called_once()
def test_process_chunk_detects_pause(self, mock_dependencies, mock_flask_app):
"""Test process chunk detects document pause."""
@@ -1657,8 +1808,6 @@ class TestIndexingRunnerProcessChunk:
from core.indexing_runner import IndexingRunner
runner = IndexingRunner()
mock_embedding_instance = MagicMock()
mock_processor = MagicMock()
chunk_documents = [Document(page_content="Chunk", metadata={"doc_id": "c1"})]
mock_dataset = Mock(spec=Dataset)
@@ -1691,5 +1840,4 @@ class TestIndexingRunnerProcessChunk:
chunk_documents,
mock_dataset.id,
mock_dataset_document.id,
mock_embedding_instance,
)
@@ -3752,8 +3752,8 @@ class TestKnowledgeRetrievalRegression:
"""
Repro test for current bug:
reranking runs after `with flask_app.app_context():` exits.
`_multiple_retrieve_thread` catches exceptions and stores them into `thread_exceptions`,
so we must assert from that list (not from an outer try/except).
The outer thread entry point catches exceptions from the traced retrieval method
and stores them in `thread_exceptions`.
"""
dataset_retrieval = DatasetRetrieval()
flask_app = Flask(__name__)
@@ -3806,7 +3806,6 @@ class TestKnowledgeRetrievalRegression:
# output list from _multiple_retrieve_thread
all_documents: list[Document] = []
# IMPORTANT: _multiple_retrieve_thread swallows exceptions and appends them here
thread_exceptions: list[Exception] = []
def target():
@@ -3818,7 +3817,7 @@ class TestKnowledgeRetrievalRegression:
),
_patched_retriever_session(),
):
dataset_retrieval._multiple_retrieve_thread(
dataset_retrieval._multiple_retrieve_thread_safely(
flask_app=flask_app,
available_datasets=[mock_dataset, secondary_dataset],
metadata_condition=None,
@@ -3847,7 +3846,6 @@ class TestKnowledgeRetrievalRegression:
# Ensure reranking branch was actually executed
assert called["init"] >= 1, "DataPostProcessor was never constructed; reranking branch may not have run."
# Current buggy code should record an exception (not raise it)
assert not thread_exceptions, thread_exceptions
def test_run_retriever_thread_provides_session_to_retriever(self):
@@ -3865,14 +3863,12 @@ class TestKnowledgeRetrievalRegression:
document_ids_filter=None,
metadata_condition=None,
attachment_ids=None,
cancel_event=None,
thread_exceptions=[],
)
mock_retriever.assert_called_once()
assert mock_retriever.call_args.kwargs["session"] is session
def test_run_retriever_thread_records_retriever_exception(self):
def test_run_retriever_thread_safely_records_retriever_exception(self):
dataset_retrieval = DatasetRetrieval()
all_documents: list[Document] = []
cancel_event = threading.Event()
@@ -3881,7 +3877,7 @@ class TestKnowledgeRetrievalRegression:
with _patched_retriever_session():
with patch.object(dataset_retrieval, "_retriever", side_effect=expected_error):
dataset_retrieval._run_retriever_thread(
dataset_retrieval._run_retriever_thread_safely(
flask_app=_FakeFlaskApp(),
dataset_id="dataset-1",
query="test query",
@@ -5139,7 +5135,7 @@ class TestSingleAndMultipleRetrieveCoverage:
app = Flask(__name__)
def failing_thread(**kwargs):
kwargs["thread_exceptions"].append(RuntimeError("thread boom"))
raise RuntimeError("thread boom")
with app.app_context():
with (
@@ -1,10 +1,21 @@
import pytest
from sqlalchemy.orm import Session
from core.workflow.nodes.agent_v2.binding_resolver import (
WorkflowAgentBindingError,
WorkflowAgentBindingResolver,
)
from models.agent import Agent, AgentConfigSnapshot, AgentStatus, WorkflowAgentBindingType, WorkflowAgentNodeBinding
from models.agent import (
Agent,
AgentConfigRevision,
AgentConfigRevisionOperation,
AgentConfigSnapshot,
AgentScope,
AgentSource,
AgentStatus,
WorkflowAgentBindingType,
WorkflowAgentNodeBinding,
)
from models.agent_config_entities import AgentSoulConfig, AgentSoulModelConfig, WorkflowNodeJobConfig
@@ -104,6 +115,89 @@ def test_binding_resolver_uses_active_snapshot_for_roster_agent(monkeypatch: pyt
assert bundle.snapshot.id == "active-snapshot"
def test_binding_resolver_rejects_unpublished_roster_agent(monkeypatch: pytest.MonkeyPatch):
binding = _binding()
binding.binding_type = WorkflowAgentBindingType.ROSTER_AGENT
monkeypatch.setattr(
"core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session",
lambda: FakeSession([binding, None]),
)
with pytest.raises(WorkflowAgentBindingError) as exc_info:
WorkflowAgentBindingResolver().resolve(**_resolve())
assert exc_info.value.error_code == "agent_not_available"
assert "not been published" in str(exc_info.value)
@pytest.mark.parametrize(
"sqlite_session",
[(Agent, AgentConfigSnapshot, AgentConfigRevision, WorkflowAgentNodeBinding)],
indirect=True,
)
def test_binding_resolver_requires_publish_provenance_for_active_roster_snapshot(
monkeypatch: pytest.MonkeyPatch,
sqlite_session: Session,
) -> None:
binding = _binding()
binding.binding_type = WorkflowAgentBindingType.ROSTER_AGENT
binding.workflow_version = "draft"
agent = Agent(
id="agent-1",
tenant_id="tenant-1",
name="Imported Agent",
scope=AgentScope.ROSTER,
source=AgentSource.IMPORTED,
app_id="agent-app-1",
status=AgentStatus.ACTIVE,
active_config_snapshot_id="snapshot-1",
active_config_has_model=True,
# Dirty draft state must not hide a snapshot after it has publish provenance.
active_config_is_published=False,
)
sqlite_session.add_all(
[
binding,
agent,
_snapshot(),
AgentConfigRevision(
id="revision-import",
tenant_id="tenant-1",
agent_id="agent-1",
current_snapshot_id="snapshot-1",
revision=1,
operation=AgentConfigRevisionOperation.IMPORT_PACKAGE,
),
]
)
sqlite_session.commit()
monkeypatch.setattr(
"core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session",
lambda: sqlite_session,
)
with pytest.raises(WorkflowAgentBindingError) as exc_info:
WorkflowAgentBindingResolver().resolve(**_resolve())
assert exc_info.value.error_code == "agent_not_available"
sqlite_session.add(
AgentConfigRevision(
id="revision-publish",
tenant_id="tenant-1",
agent_id="agent-1",
current_snapshot_id="snapshot-1",
revision=2,
operation=AgentConfigRevisionOperation.PUBLISH_DRAFT,
)
)
sqlite_session.commit()
bundle = WorkflowAgentBindingResolver().resolve(**_resolve())
assert bundle.agent.id == agent.id
assert bundle.snapshot.id == "snapshot-1"
def test_binding_resolver_raises_when_binding_missing(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(
"core.workflow.nodes.agent_v2.binding_resolver.session_factory.create_session",
@@ -156,6 +156,19 @@ def test_publish_validation_uses_active_snapshot_for_roster_agent():
)
def test_publish_validation_rejects_unpublished_roster_agent():
binding = _binding(WorkflowNodeJobConfig())
binding.binding_type = WorkflowAgentBindingType.ROSTER_AGENT
session = Mock()
session.scalar.side_effect = [binding, None]
with pytest.raises(WorkflowAgentNodeValidationError, match="unpublished roster agent"):
WorkflowAgentNodeValidator.validate_published_workflow(
session=session,
workflow=_workflow(_graph([{"source": "start", "target": "agent-node"}])),
)
def test_publish_validation_rejects_non_upstream_previous_output_ref():
node_job = WorkflowNodeJobConfig.model_validate(
{"previous_node_output_refs": [{"node_id": "later-node", "output": "text"}]}
@@ -0,0 +1,51 @@
from concurrent.futures import ThreadPoolExecutor
import pytest
from opentelemetry import context as otel_context
from extensions.otel.context import propagate_context
def test_propagate_context_captures_context_when_wrapped() -> None:
context_key = otel_context.create_key("test-context")
captured_context = otel_context.set_value(context_key, "captured")
token = otel_context.attach(captured_context)
try:
wrapped = propagate_context(lambda: otel_context.get_value(context_key))
finally:
otel_context.detach(token)
with ThreadPoolExecutor(max_workers=1) as executor:
assert executor.submit(wrapped).result() == "captured"
def test_propagate_context_detaches_context_after_exception() -> None:
context_key = otel_context.create_key("test-context")
captured_context = otel_context.set_value(context_key, "captured")
def raise_error() -> None:
raise RuntimeError("retrieval failed")
token = otel_context.attach(captured_context)
try:
wrapped = propagate_context(raise_error)
finally:
otel_context.detach(token)
def invoke_and_read_context() -> str | None:
with pytest.raises(RuntimeError, match="retrieval failed"):
wrapped()
return otel_context.get_value(context_key)
with ThreadPoolExecutor(max_workers=1) as executor:
assert executor.submit(invoke_and_read_context).result() is None
def test_propagate_context_preserves_function_metadata() -> None:
def retrieve() -> None:
pass
wrapped = propagate_context(retrieve)
assert wrapped.__name__ == "retrieve"
@@ -0,0 +1,122 @@
import threading
from unittest.mock import MagicMock, patch
from uuid import uuid4
from opentelemetry.trace import StatusCode, get_current_span, get_tracer
from core.rag.rerank.rerank_type import RerankMode
from core.rag.retrieval.dataset_retrieval import DatasetRetrieval
from core.workflow.nodes.knowledge_retrieval.retrieval import KnowledgeRetrievalRequest
from models.dataset import Dataset
def test_knowledge_retrieval_creates_a_child_otel_span(
memory_span_exporter,
tracer_provider_with_memory_exporter,
) -> None:
"""The retrieval entry point must be visible beneath its workflow node span."""
request = KnowledgeRetrievalRequest(
tenant_id=str(uuid4()),
user_id=str(uuid4()),
app_id=str(uuid4()),
user_from="account",
dataset_ids=[str(uuid4())],
retrieval_mode="multiple",
query="test query",
)
retrieval = DatasetRetrieval()
with (
patch("extensions.otel.decorators.base.dify_config.ENABLE_OTEL", True),
patch.object(retrieval, "_check_knowledge_rate_limit"),
patch.object(retrieval, "_get_available_datasets", return_value=[]),
get_tracer(__name__).start_as_current_span("knowledge-retrieval-node") as node_span,
):
assert retrieval.knowledge_retrieval(MagicMock(), request) == []
retrieval_span = next(
span
for span in memory_span_exporter.get_finished_spans()
if span.name == "core.rag.retrieval.dataset_retrieval.DatasetRetrieval.knowledge_retrieval"
)
node_span_context = node_span.get_span_context()
assert retrieval_span.context.trace_id == node_span_context.trace_id
assert retrieval_span.parent is not None
assert retrieval_span.parent.span_id == node_span_context.span_id
def test_multiple_retrieve_preserves_otel_context_in_dataset_thread(
app,
tracer_provider_with_memory_exporter,
) -> None:
"""Per-dataset retrieval spans must remain in the workflow node trace."""
retrieval = DatasetRetrieval()
dataset = MagicMock(spec=Dataset)
dataset.id = str(uuid4())
dataset.indexing_technique = "high_quality"
dataset.embedding_model = "text-embedding-3-small"
dataset.embedding_model_provider = "openai"
observed_trace_ids: list[int] = []
def record_active_trace(**_kwargs: object) -> None:
observed_trace_ids.append(get_current_span().get_span_context().trace_id)
with (
app.app_context(),
patch("extensions.otel.decorators.base.dify_config.ENABLE_OTEL", True),
patch.object(retrieval, "_multiple_retrieve_thread", side_effect=record_active_trace),
patch.object(retrieval, "_on_query"),
get_tracer(__name__).start_as_current_span("knowledge-retrieval-node") as node_span,
):
retrieval.multiple_retrieve(
app_id=str(uuid4()),
tenant_id=str(uuid4()),
user_id=str(uuid4()),
user_from="account",
available_datasets=[dataset],
query="test query",
top_k=4,
score_threshold=0.0,
reranking_mode=RerankMode.RERANKING_MODEL,
reranking_enable=False,
)
assert observed_trace_ids == [node_span.get_span_context().trace_id]
def test_retriever_thread_exception_sets_error_span_and_is_collected(
app,
memory_span_exporter,
tracer_provider_with_memory_exporter,
) -> None:
retrieval = DatasetRetrieval()
cancel_event = threading.Event()
thread_exceptions: list[Exception] = []
expected_error = RuntimeError("retrieval failed")
with (
patch("extensions.otel.decorators.base.dify_config.ENABLE_OTEL", True),
patch("core.rag.retrieval.dataset_retrieval.session_factory.create_session"),
patch.object(retrieval, "_retriever", side_effect=expected_error),
):
retrieval._run_retriever_thread_safely(
flask_app=app,
dataset_id=str(uuid4()),
query="test query",
top_k=4,
all_documents=[],
document_ids_filter=None,
metadata_condition=None,
attachment_ids=None,
cancel_event=cancel_event,
thread_exceptions=thread_exceptions,
)
retrieval_span = next(
span
for span in memory_span_exporter.get_finished_spans()
if span.name.endswith("DatasetRetrieval._run_retriever_thread")
)
assert retrieval_span.status.status_code == StatusCode.ERROR
assert cancel_event.is_set()
assert thread_exceptions == [expected_error]
@@ -0,0 +1,75 @@
from unittest import mock
import pytest
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from core.logging.context import clear_request_context
@pytest.fixture(autouse=True)
def _reset_logging_context():
clear_request_context()
yield
clear_request_context()
def test_on_user_loaded_does_not_write_to_non_recording_span() -> None:
from extensions.otel import runtime
span = mock.MagicMock()
span.is_recording.return_value = False
user = mock.Mock(id="user-id")
with (
mock.patch.object(runtime.dify_config, "ENABLE_OTEL", True),
mock.patch("opentelemetry.trace.get_current_span", return_value=span),
mock.patch.object(runtime, "extract_tenant_id", return_value="tenant-id"),
):
runtime.on_user_loaded(None, user)
span.is_recording.assert_called_once_with()
span.set_attribute.assert_not_called()
span.set_attributes.assert_not_called()
def test_on_user_loaded_sets_attributes_on_recording_span() -> None:
from extensions.otel import runtime
from extensions.otel.semconv import DifySpanAttributes, GenAIAttributes
span = mock.MagicMock()
span.is_recording.return_value = True
user = mock.Mock(id="user-id")
with (
mock.patch.object(runtime.dify_config, "ENABLE_OTEL", True),
mock.patch("opentelemetry.trace.get_current_span", return_value=span),
mock.patch.object(runtime, "extract_tenant_id", return_value="tenant-id"),
):
runtime.on_user_loaded(None, user)
span.set_attributes.assert_called_once_with(
{
DifySpanAttributes.TENANT_ID: "tenant-id",
GenAIAttributes.USER_ID: "user-id",
}
)
def test_on_user_loaded_ignores_ended_sdk_span(caplog) -> None:
from extensions.otel import runtime
tracer_provider = TracerProvider()
span = tracer_provider.get_tracer(__name__).start_span("ended")
span.end()
user = mock.Mock(id="user-id")
with (
trace.use_span(span, end_on_exit=False),
mock.patch.object(runtime.dify_config, "ENABLE_OTEL", True),
mock.patch.object(runtime, "extract_tenant_id", return_value="tenant-id"),
caplog.at_level("WARNING", logger="opentelemetry.sdk.trace"),
):
runtime.on_user_loaded(None, user)
assert "Setting attribute on ended span" not in caplog.text
@@ -1,10 +1,22 @@
import json
from typing import cast
from unittest import mock
import pytest
from flask import Response
from core.logging.context import clear_request_context, get_identity_context
from extensions import ext_login
from extensions.ext_login import unauthorized_handler
@pytest.fixture(autouse=True)
def _reset_logging_context():
clear_request_context()
yield
clear_request_context()
def test_unauthorized_handler_returns_json_response() -> None:
response = unauthorized_handler()
@@ -15,3 +27,50 @@ def test_unauthorized_handler_returns_json_response() -> None:
"code": "unauthorized",
"message": "Unauthorized.",
}
def test_on_user_logged_in_sets_account_logging_identity() -> None:
account = mock.Mock(spec=ext_login.Account)
account.id = "account-id"
account.current_tenant_id = "tenant-id"
clear_request_context()
ext_login.on_user_logged_in(None, account)
assert get_identity_context() == ("tenant-id", "account-id", "account")
def test_on_user_logged_in_sets_end_user_logging_identity() -> None:
end_user = mock.Mock(spec=ext_login.EndUser)
end_user.id = "end-user-id"
end_user.tenant_id = "tenant-id"
end_user.type = "browser"
clear_request_context()
ext_login.on_user_logged_in(None, end_user)
assert get_identity_context() == ("tenant-id", "end-user-id", "browser")
def test_on_user_logged_in_does_not_break_auth_when_identity_is_unavailable(caplog: pytest.LogCaptureFixture) -> None:
account = mock.Mock(spec=ext_login.Account)
type(account).current_tenant_id = mock.PropertyMock(side_effect=RuntimeError("unavailable"))
account.id = "account-id"
clear_request_context()
with caplog.at_level("ERROR", logger=ext_login.logger.name):
ext_login.on_user_logged_in(None, account)
assert get_identity_context() == ("", "", "")
assert "Failed to set logging identity context" in caplog.text
def test_on_user_logged_in_logs_unsupported_user_type(caplog: pytest.LogCaptureFixture) -> None:
unsupported_user = cast(ext_login.LoginUser, object())
clear_request_context()
with caplog.at_level("ERROR", logger=ext_login.logger.name):
ext_login.on_user_logged_in(None, unsupported_user)
assert get_identity_context() == ("", "", "")
assert "Failed to set logging identity context" in caplog.text
@@ -5,7 +5,8 @@ from types import SimpleNamespace
import pytest
from core.app.entities.app_invoke_entities import InvokeFrom
from models.enums import ConversationFromSource, MessageStatus
from graphon.enums import WorkflowNodeExecutionStatus
from models.enums import ConversationFromSource, CreatorUserRole, MessageStatus
from services.agent import observability_service as observability_service_module
from services.agent.observability_service import AgentLogQueryParams, AgentObservabilityService
@@ -268,6 +269,183 @@ def test_list_logs_sorts_by_requested_field(monkeypatch: pytest.MonkeyPatch) ->
assert [item["id"] for item in payload["data"]] == ["old", "new"]
def test_list_log_messages_merges_deduplicates_and_sorts_sources(monkeypatch: pytest.MonkeyPatch) -> None:
service = AgentObservabilityService(session=None)
webapp_message = SimpleNamespace(id="shared")
webapp_row = {"id": "shared", "created_at": 10, "updated_at": 30}
workflow_rows = [
{"id": "shared", "created_at": 10, "updated_at": 20},
{"id": "workflow-only", "created_at": 20, "updated_at": 10},
]
monkeypatch.setattr(service, "_list_webapp_messages", lambda **kwargs: [webapp_message])
monkeypatch.setattr(service, "serialize_log_message", lambda message: webapp_row)
monkeypatch.setattr(service, "_list_workflow_messages", lambda **kwargs: workflow_rows)
payload = service.list_log_messages(
app=SimpleNamespace(id="agent-app"), # type: ignore[arg-type]
agent_id="agent-1",
conversation_id="execution-1",
params=AgentLogQueryParams(
sources=("webapp:agent-app", "workflow:workflow-app"),
sort_by="created_at",
sort_order="asc",
),
)
assert payload == {
"data": [workflow_rows[0], workflow_rows[1]],
"page": 1,
"limit": 20,
"total": 2,
"has_more": False,
}
def test_list_workflow_logs_uses_node_executions_without_messages() -> None:
created_at = datetime(2026, 7, 23, 7, 0, 19, tzinfo=UTC)
node_execution = SimpleNamespace(
id="node-execution-1",
title="Agent",
status=WorkflowNodeExecutionStatus.SUCCEEDED,
created_by_role=CreatorUserRole.END_USER,
created_by="end-user-1",
created_at=created_at,
finished_at=None,
)
workflow_app = SimpleNamespace(
id="workflow-app-1",
name="Marketing Department",
icon_type=None,
icon=None,
icon_background=None,
)
class FakeRow:
node_execution_id = node_execution.id
node_title = node_execution.title
node_status = node_execution.status
node_created_by_role = node_execution.created_by_role
node_created_by = node_execution.created_by
node_created_at = node_execution.created_at
node_finished_at = node_execution.finished_at
workflow_id = "workflow-1"
workflow_version = "v1"
node_id = "node-1"
def __getitem__(self, index: int):
return (None, None, None, None, None, None, None, workflow_app)[index]
class FakeResult:
def all(self):
return [FakeRow()]
class FakeSession:
def __init__(self):
self.query = ""
def execute(self, stmt):
self.query = str(stmt)
return FakeResult()
session = FakeSession()
service = AgentObservabilityService(session)
rows = service._list_workflow_conversation_logs(
app=SimpleNamespace(tenant_id="tenant-1"), # type: ignore[arg-type]
agent_id="agent-1",
params=AgentLogQueryParams(),
source_filter=AgentObservabilityService.resolve_source_filter("workflow:workflow-app-1"),
)
assert "FROM workflow_node_executions" in session.query
assert "JOIN messages" not in session.query
assert rows[0]["id"] == "node-execution-1"
assert rows[0]["source"]["app_name"] == "Marketing Department"
def test_list_workflow_messages_uses_node_execution_identity(monkeypatch: pytest.MonkeyPatch) -> None:
node_execution = SimpleNamespace(id="node-execution-1")
class FakeScalarResult:
def all(self):
return [node_execution]
class FakeSession:
def __init__(self):
self.query = ""
def scalars(self, stmt):
self.query = str(stmt)
return FakeScalarResult()
session = FakeSession()
service = AgentObservabilityService(session)
serialized = {"id": "node-execution-1", "conversation_id": "node-execution-1"}
monkeypatch.setattr(service, "serialize_workflow_node_message", lambda execution: serialized)
rows = service._list_workflow_messages(
app=SimpleNamespace(tenant_id="tenant-1"), # type: ignore[arg-type]
agent_id="agent-1",
conversation_id="node-execution-1",
params=AgentLogQueryParams(),
source_filter=AgentObservabilityService.resolve_source_filter("workflow:workflow-app-1"),
)
assert rows == [serialized]
assert "FROM workflow_node_executions" in session.query
assert "workflow_node_executions.id =" in session.query
assert "JOIN messages" not in session.query
def test_apply_workflow_node_filters_supports_time_keyword_and_status() -> None:
class FakeStmt:
def __init__(self):
self.conditions = []
def where(self, *conditions):
self.conditions.extend(conditions)
return self
stmt = FakeStmt()
params = AgentLogQueryParams(
start=datetime(2026, 7, 1, tzinfo=UTC),
end=datetime(2026, 8, 1, tzinfo=UTC),
keyword="meeting_100%",
statuses=("success",),
)
result = AgentObservabilityService._apply_workflow_node_filters(
stmt,
params=params,
workflow_app=SimpleNamespace(name=observability_service_module.App.name),
)
assert result is stmt
assert len(stmt.conditions) == 4
def test_apply_workflow_node_status_filter_supports_all_status_groups() -> None:
class FakeStmt:
def __init__(self):
self.conditions = []
def where(self, *conditions):
self.conditions.extend(conditions)
return self
stmt = FakeStmt()
result = AgentObservabilityService._apply_workflow_node_status_filter(stmt, ("normal", "error", "paused"))
assert result is stmt
assert len(stmt.conditions) == 1
empty_stmt = FakeStmt()
assert AgentObservabilityService._apply_workflow_node_status_filter(empty_stmt, ()) is empty_stmt
assert empty_stmt.conditions == []
with pytest.raises(ValueError, match="Unsupported status"):
AgentObservabilityService._apply_workflow_node_status_filter(FakeStmt(), ("unknown",))
def test_source_serializers_return_structured_frontend_shape() -> None:
app = SimpleNamespace(
id="app-1",
@@ -400,6 +578,152 @@ def test_serialize_log_message_returns_frontend_log_shape() -> None:
}
def test_serialize_workflow_node_message_returns_frontend_log_shape() -> None:
created_at = datetime(2026, 7, 23, 7, 0, 19, tzinfo=UTC)
finished_at = datetime(2026, 7, 23, 7, 0, 28, tzinfo=UTC)
node_execution = SimpleNamespace(
id="node-execution-1",
title="Agent",
inputs=(
'{"agent_backend_request":{"composition":{"layers":['
'{"name":"workflow_node_job_prompt","config":{"user":"Summarize the meeting"}},'
'{"name":"workflow_user_prompt","config":{"user":"Focus on action items"}}]}}}'
),
outputs='{"output":"Alice owns the follow-up."}',
execution_metadata=(
'{"agent_log":{"agent_backend":{"usage":{"prompt_tokens":10,"completion_tokens":5,'
'"total_tokens":15,"total_price":"0.0015","currency":"USD","latency":1.25}}}}'
),
status=WorkflowNodeExecutionStatus.SUCCEEDED,
error=None,
elapsed_time=1.5,
created_by_role=CreatorUserRole.END_USER,
created_by="end-user-1",
created_at=created_at,
finished_at=finished_at,
)
payload = AgentObservabilityService.serialize_workflow_node_message(node_execution) # type: ignore[arg-type]
assert payload == {
"id": "node-execution-1",
"message_id": "node-execution-1",
"conversation_id": "node-execution-1",
"query": "Summarize the meeting\n\nFocus on action items",
"answer": "Alice owns the follow-up.",
"status": "success",
"error": None,
"from_end_user_id": "end-user-1",
"from_account_id": None,
"message_tokens": 10,
"answer_tokens": 5,
"total_tokens": 15,
"total_price": "0.0015",
"currency": "USD",
"latency": 1.25,
"created_at": int(created_at.timestamp()),
"updated_at": int(finished_at.timestamp()),
}
def test_serialize_workflow_node_message_handles_sparse_runtime_data() -> None:
created_at = datetime(2026, 7, 23, 7, 0, 19, tzinfo=UTC)
node_execution = SimpleNamespace(
id="node-execution-2",
title="Fallback prompt",
inputs={
"agent_backend_request": {
"composition": {
"layers": [
None,
{"name": "unrelated", "config": {"user": "ignored"}},
{"name": "workflow_user_prompt", "config": {"user": " "}},
]
}
}
},
outputs={"output": {"structured": True}},
execution_metadata={
"agent_log": {
"agent_backend": {
"usage": {
"prompt_tokens": "2",
"completion_tokens": 3,
}
}
}
},
status=WorkflowNodeExecutionStatus.PAUSED,
error=None,
elapsed_time=2,
created_by_role=CreatorUserRole.ACCOUNT.value,
created_by="account-1",
created_at=created_at,
finished_at=None,
)
payload = AgentObservabilityService.serialize_workflow_node_message(node_execution) # type: ignore[arg-type]
assert payload["query"] == "Fallback prompt"
assert payload["answer"] == '{"output": {"structured": true}}'
assert payload["status"] == "paused"
assert payload["from_end_user_id"] is None
assert payload["from_account_id"] == "account-1"
assert payload["total_tokens"] == 5
assert payload["total_price"] == "0"
assert payload["currency"] == ""
assert payload["latency"] == 2.0
assert payload["updated_at"] == int(created_at.timestamp())
def test_workflow_node_serialization_helpers_handle_invalid_values() -> None:
assert AgentObservabilityService._json_mapping(None) == {}
assert AgentObservabilityService._json_mapping("not-json") == {}
assert AgentObservabilityService._json_mapping("[]") == {}
assert AgentObservabilityService._mapping_value({"value": []}, "value") == {}
assert AgentObservabilityService._int_value(None) == 0
assert AgentObservabilityService._int_value("not-a-number") == 0
assert (
AgentObservabilityService._workflow_node_query(
{"agent_backend_request": {"composition": {"layers": "invalid"}}}, fallback="fallback"
)
== "fallback"
)
assert AgentObservabilityService._workflow_node_answer({"output": 1, "text": "fallback text"}) == "fallback text"
assert AgentObservabilityService._workflow_node_answer({}) == ""
def test_serialize_workflow_execution_log_uses_node_execution_identity() -> None:
created_at = datetime(2026, 7, 23, 7, 0, 19, tzinfo=UTC)
node_execution = SimpleNamespace(
id="node-execution-1",
title="Agent",
status=WorkflowNodeExecutionStatus.FAILED,
created_by_role=CreatorUserRole.ACCOUNT,
created_by="account-1",
created_at=created_at,
finished_at=None,
)
payload = AgentObservabilityService._serialize_workflow_execution_log(
node_execution_id=node_execution.id,
title=node_execution.title,
status=node_execution.status,
created_by_role=node_execution.created_by_role,
created_by=node_execution.created_by,
created_at=node_execution.created_at,
finished_at=node_execution.finished_at,
source={"id": "workflow:app-1:workflow-1:v1:node-1"},
)
assert payload["id"] == "node-execution-1"
assert payload["conversation_id"] == "node-execution-1"
assert payload["message_count"] == 1
assert payload["end_user_id"] is None
assert payload["status"] == "failed"
assert payload["unread"] is False
def test_build_charts_and_summary_match_monitoring_metrics() -> None:
rows = [
{
@@ -592,6 +592,7 @@ def test_save_agent_app_composer_rejects_version_save_strategy():
def test_save_agent_app_composer_updates_normal_draft(monkeypatch: pytest.MonkeyPatch):
agent = SimpleNamespace(
id="agent-1",
tenant_id="tenant-1",
source=AgentSource.AGENT_APP,
active_config_snapshot_id="version-1",
active_config_is_published=True,
@@ -639,13 +640,14 @@ def test_save_agent_app_composer_keeps_published_when_draft_matches_active_snaps
agent_soul = _agent_soul_with_model()
agent = SimpleNamespace(
id="agent-1",
tenant_id="tenant-1",
source=AgentSource.AGENT_APP,
active_config_snapshot_id="version-1",
active_config_is_published=False,
updated_by=None,
)
active_version = SimpleNamespace(config_snapshot_dict=agent_soul.model_dump(mode="json"))
fake_session = FakeSession(scalar=[agent], scalars=[[AgentConfigRevisionOperation.PUBLISH_DRAFT]])
fake_session = FakeSession(scalar=[agent, "publish-revision-1"])
session = fake_session
monkeypatch.setattr(composer_service.ComposerConfigValidator, "validate_draft_save_payload", lambda payload: None)
@@ -822,8 +824,7 @@ def test_agent_app_build_draft_checkout_and_apply_use_user_isolated_draft(monkey
active_version = SimpleNamespace(config_snapshot_dict=build_draft.config_snapshot_dict)
fake_session = FakeSession(
scalar=[agent, build_draft, normal_draft, active_version],
scalars=[[AgentConfigRevisionOperation.PUBLISH_DRAFT]],
scalar=[agent, build_draft, normal_draft, active_version, "publish-revision-1"],
)
session = fake_session
@@ -1612,7 +1613,7 @@ def test_node_job_only_rejects_inline_binding_pointing_to_roster_agent(monkeypat
def test_copy_workflow_composer_from_roster_creates_inline_agent_and_preserves_node_job(
monkeypatch: pytest.MonkeyPatch,
):
fake_session = FakeSession()
fake_session = FakeSession(scalar=["publish-revision-1"])
session = fake_session
workflow = SimpleNamespace(id="workflow-1")
node_job = WorkflowNodeJobConfig(workflow_prompt="keep this node task")
@@ -1710,7 +1711,7 @@ def test_copy_workflow_composer_from_roster_creates_inline_agent_and_preserves_n
def test_copy_workflow_composer_from_roster_rejects_stale_source_snapshot(monkeypatch: pytest.MonkeyPatch):
session = FakeSession()
session = FakeSession(scalar=["publish-revision-1"])
monkeypatch.setattr(AgentComposerService, "_get_draft_workflow", lambda **kwargs: SimpleNamespace(id="workflow-1"))
monkeypatch.setattr(
AgentComposerService,
@@ -1758,6 +1759,49 @@ def test_copy_workflow_composer_from_roster_rejects_stale_source_snapshot(monkey
)
def test_copy_workflow_composer_from_roster_rejects_unpublished_source(monkeypatch: pytest.MonkeyPatch):
session = FakeSession()
binding = WorkflowAgentNodeBinding(
tenant_id="tenant-1",
app_id="app-1",
workflow_id="workflow-1",
workflow_version="draft",
node_id="node-1",
binding_type=WorkflowAgentBindingType.ROSTER_AGENT,
agent_id="roster-agent-1",
current_snapshot_id="roster-version-1",
node_job_config=WorkflowNodeJobConfig(),
)
source_agent = Agent(
id="roster-agent-1",
tenant_id="tenant-1",
name="Unpublished import",
scope=AgentScope.ROSTER,
source=AgentSource.IMPORTED,
app_id="agent-app-1",
status=AgentStatus.ACTIVE,
active_config_snapshot_id="roster-version-1",
)
monkeypatch.setattr(AgentComposerService, "_get_draft_workflow", lambda **kwargs: SimpleNamespace(id="workflow-1"))
monkeypatch.setattr(AgentComposerService, "_get_workflow_binding", lambda **kwargs: binding)
monkeypatch.setattr(AgentComposerService, "_require_agent", lambda **kwargs: source_agent)
require_version = MagicMock(side_effect=AssertionError("unpublished source must fail before loading snapshot"))
monkeypatch.setattr(AgentComposerService, "_require_version", require_version)
with pytest.raises(InvalidComposerConfigError, match="published config snapshot"):
AgentComposerService.copy_workflow_composer_from_roster(
session=session,
tenant_id="tenant-1",
app_id="app-1",
node_id="node-1",
account_id="account-1",
source_agent_id="roster-agent-1",
)
require_version.assert_not_called()
assert session.flushes == 0
def test_copy_workflow_composer_from_roster_is_idempotent_when_already_inline(monkeypatch: pytest.MonkeyPatch):
inline_binding = WorkflowAgentNodeBinding(
tenant_id="tenant-1",
@@ -2203,7 +2247,7 @@ def test_agent_app_draft_match_does_not_mark_create_version_as_published(monkeyp
active_config_snapshot_id="snapshot-1",
)
snapshot = SimpleNamespace(config_snapshot_dict=agent_soul)
fake_session = FakeSession(scalars=[[AgentConfigRevisionOperation.CREATE_VERSION]])
fake_session = FakeSession()
session = fake_session
monkeypatch.setattr(AgentComposerService, "_get_version_if_present", lambda **kwargs: snapshot)
@@ -2227,7 +2271,7 @@ def test_agent_app_draft_match_marks_publish_visible_revision_as_published(monke
active_config_snapshot_id="snapshot-1",
)
snapshot = SimpleNamespace(config_snapshot_dict=agent_soul)
fake_session = FakeSession(scalars=[[AgentConfigRevisionOperation.PUBLISH_DRAFT]])
fake_session = FakeSession(scalar=["publish-revision-1"])
session = fake_session
monkeypatch.setattr(AgentComposerService, "_get_version_if_present", lambda **kwargs: snapshot)
@@ -3496,8 +3540,61 @@ class TestAgentAppBackingAgent:
assert session.deleted == []
assert session.commits == 1
def test_refresh_agent_app_debug_conversation_enqueues_cleanup_for_old_runtime_sessions(
def test_refresh_agent_app_debug_conversation_rotates_preview_mapping_each_time(
self, monkeypatch: pytest.MonkeyPatch
):
agent = Agent(
id="agent-1",
tenant_id="tenant-1",
name="Iris",
description="",
agent_kind=AgentKind.DIFY_AGENT,
scope=AgentScope.ROSTER,
source=AgentSource.AGENT_APP,
status=AgentStatus.ACTIVE,
app_id="app-1",
)
session = FakeSession(scalar=[agent, None])
service = AgentRosterService(session)
cleanup = MagicMock()
monkeypatch.setattr(service, "_cleanup_debug_conversation_runtime_sessions", cleanup)
first_conversation_id = service.refresh_agent_app_debug_conversation_id(
tenant_id="tenant-1",
agent_id="agent-1",
account_id="account-1",
draft_type=AgentConfigDraftType.DRAFT,
)
mapping = next(value for value in session.added if isinstance(value, AgentDebugConversation))
session._scalar.extend([agent, mapping])
second_conversation_id = service.refresh_agent_app_debug_conversation_id(
tenant_id="tenant-1",
agent_id="agent-1",
account_id="account-1",
draft_type=AgentConfigDraftType.DRAFT,
)
assert first_conversation_id != second_conversation_id
assert mapping.draft_type == AgentConfigDraftType.DRAFT
assert mapping.conversation_id == second_conversation_id
assert len([value for value in session.added if isinstance(value, Conversation)]) == 2
cleanup.assert_called_once_with(
tenant_id="tenant-1",
agent_id="agent-1",
account_id="account-1",
draft_type=AgentConfigDraftType.DRAFT,
app_id="app-1",
conversation_id=first_conversation_id,
)
@pytest.mark.parametrize(
"draft_type",
[AgentConfigDraftType.DRAFT, AgentConfigDraftType.DEBUG_BUILD],
)
def test_refresh_agent_app_debug_conversation_enqueues_cleanup_for_old_runtime_sessions(
self,
monkeypatch: pytest.MonkeyPatch,
draft_type: AgentConfigDraftType,
):
agent = Agent(
id="agent-1",
@@ -3525,9 +3622,21 @@ class TestAgentAppBackingAgent:
)
session = FakeSession(scalar=[agent, mapping])
service = AgentRosterService(session)
events: list[str] = []
original_commit = session.commit
def record_commit() -> None:
events.append("commit")
original_commit()
def list_active_sessions(**kwargs: object) -> list[object]:
events.append("cleanup")
return [stored_session]
cleanup_delay = MagicMock()
cleanup_store = MagicMock()
cleanup_store.list_active_sessions_for_conversation.return_value = [stored_session]
cleanup_store.list_active_sessions_for_conversation.side_effect = list_active_sessions
monkeypatch.setattr(session, "commit", record_commit)
monkeypatch.setattr(roster_service, "AgentAppRuntimeSessionStore", lambda: cleanup_store)
monkeypatch.setattr(roster_service.cleanup_conversation_agent_runtime_session, "delay", cleanup_delay)
@@ -3535,6 +3644,7 @@ class TestAgentAppBackingAgent:
tenant_id="tenant-1",
agent_id="agent-1",
account_id="account-1",
draft_type=draft_type,
)
cleanup_store.list_active_sessions_for_conversation.assert_called_once_with(
@@ -3546,10 +3656,10 @@ class TestAgentAppBackingAgent:
payload = cleanup_delay.call_args.args[0]
assert payload["metadata"]["conversation_id"] == "old-conversation"
assert payload["metadata"]["agent_id"] == "agent-9"
assert payload["metadata"]["draft_type"] == "debug_build"
assert payload["metadata"]["draft_type"] == draft_type.value
assert (
payload["idempotency_key"]
== "tenant-1:agent-1:account-1:debug_build:old-conversation:debug-session-cleanup:"
== f"tenant-1:agent-1:account-1:{draft_type.value}:old-conversation:debug-session-cleanup:"
"agent-9:snap-9:run-old"
)
cleanup_store.mark_cleaned.assert_called_once_with(
@@ -3558,6 +3668,41 @@ class TestAgentAppBackingAgent:
)
assert mapping.app_id == "app-1"
assert mapping.conversation_id == conversation_id
assert events == ["commit", "cleanup"]
def test_refresh_agent_app_debug_conversation_does_not_cleanup_when_commit_fails(
self, monkeypatch: pytest.MonkeyPatch
):
agent = Agent(
id="agent-1",
tenant_id="tenant-1",
name="Iris",
description="",
agent_kind=AgentKind.DIFY_AGENT,
scope=AgentScope.ROSTER,
source=AgentSource.AGENT_APP,
status=AgentStatus.ACTIVE,
app_id="app-1",
)
mapping = SimpleNamespace(app_id="old-app", conversation_id="old-conversation")
session = FakeSession(scalar=[agent, mapping])
service = AgentRosterService(session)
runtime_store_factory = MagicMock()
cleanup_delay = MagicMock()
monkeypatch.setattr(session, "commit", MagicMock(side_effect=RuntimeError("database unavailable")))
monkeypatch.setattr(roster_service, "AgentAppRuntimeSessionStore", runtime_store_factory)
monkeypatch.setattr(roster_service.cleanup_conversation_agent_runtime_session, "delay", cleanup_delay)
with pytest.raises(RuntimeError, match="database unavailable"):
service.refresh_agent_app_debug_conversation_id(
tenant_id="tenant-1",
agent_id="agent-1",
account_id="account-1",
draft_type=AgentConfigDraftType.DRAFT,
)
runtime_store_factory.assert_not_called()
cleanup_delay.assert_not_called()
def test_refresh_agent_app_debug_conversation_marks_old_runtime_sessions_clean_when_enqueue_fails(
self, monkeypatch: pytest.MonkeyPatch
@@ -4994,6 +5139,7 @@ def test_composer_save_rejects_missing_or_out_of_scope_knowledge_datasets(monkey
def test_save_agent_composer_allows_incomplete_knowledge_draft(monkeypatch: pytest.MonkeyPatch):
agent = SimpleNamespace(
id="agent-1",
tenant_id="tenant-1",
source=AgentSource.AGENT_APP,
active_config_snapshot_id="version-1",
active_config_is_published=True,
@@ -196,6 +196,19 @@ def test_resolve_existing_inline_binding_agent_returns_valid_agent_or_none(monke
)
def test_resolve_roster_binding_rejects_unpublished_agent() -> None:
session = Mock()
session.scalar.return_value = None
with pytest.raises(ValueError, match="unavailable or unpublished roster agent"):
WorkflowAgentPublishService._resolve_roster_agent_graph_binding(
session=session,
draft_workflow=_workflow(),
node_id="agent-node",
agent_id="agent-1",
)
def test_clone_inline_graph_binding_for_node_clones_source(monkeypatch) -> None:
session = Mock()
source_agent = SimpleNamespace(id="source-agent")
@@ -806,6 +806,79 @@ class TestPluginListEndpointCounts:
assert tool_plugin.endpoints_active == 0
class TestPluginCategoryList:
def test_list_by_category_forwards_search_and_tag_filters(self) -> None:
plugins = SimpleNamespace(list=[], has_more=False)
with patch(f"{MODULE}.PluginInstaller") as installer_cls:
installer_cls.return_value.list_plugins_by_category.return_value = plugins
from core.plugin.plugin_service import PluginService
result = PluginService.list_by_category(
"tenant-1",
PluginCategory.Tool,
2,
25,
query="weather",
tags=["search", "rag"],
language="zh_Hans",
)
assert result is plugins
installer_cls.return_value.list_plugins_by_category.assert_called_once_with(
"tenant-1",
PluginCategory.Tool,
2,
25,
query="weather",
tags=["search", "rag"],
language="zh_Hans",
)
def test_filtered_model_category_does_not_reconcile_from_a_partial_result(self) -> None:
plugins = SimpleNamespace(list=[], has_more=False)
with (
patch(f"{MODULE}.PluginInstaller") as installer_cls,
patch(f"{MODULE}.PluginService.invalidate_plugin_model_providers_cache") as invalidate_cache,
patch(f"{MODULE}.PluginService._store_cached_remote_model_plugin_marker") as store_marker,
):
installer_cls.return_value.list_plugins_by_category.return_value = plugins
from core.plugin.plugin_service import PluginService
result = PluginService.list_by_category(
"tenant-1",
PluginCategory.Model,
1,
100,
query="openai",
tags=[],
language="en_US",
)
assert result is plugins
invalidate_cache.assert_not_called()
store_marker.assert_not_called()
class TestInstalledPluginIds:
def test_list_installed_plugin_ids_uses_lightweight_daemon_endpoint(self) -> None:
with patch(f"{MODULE}.PluginInstaller") as installer_cls:
installer_cls.return_value.list_installed_plugin_ids.return_value = [
"langgenius/openai",
"langgenius/anthropic",
]
from core.plugin.plugin_service import PluginService
result = PluginService.list_installed_plugin_ids("tenant-1", PluginCategory.Tool)
assert result == ["langgenius/openai", "langgenius/anthropic"]
installer_cls.return_value.list_installed_plugin_ids.assert_called_once_with("tenant-1", PluginCategory.Tool)
class TestPluginModelProviderCacheInvalidation:
def test_get_debugging_key_does_not_invalidate_model_provider_cache(self) -> None:
"""Reading a debug key does not mean a debug runtime has registered a model provider."""
@@ -850,7 +923,13 @@ class TestPluginModelProviderCacheInvalidation:
assert result is plugins
installer_cls.return_value.list_plugins_by_category.assert_called_once_with(
"tenant-1", PluginCategory.Model, 1, 100
"tenant-1",
PluginCategory.Model,
1,
100,
query="",
tags=(),
language="en_US",
)
invalidate_cache.assert_called_once_with("tenant-1")
store_marker.assert_called_once_with("tenant-1", remote_plugin_marker)
@@ -19,6 +19,7 @@ from sqlalchemy.orm import Session
from core.plugin.entities.plugin import PluginInstallationSource
from core.plugin.entities.plugin_daemon import PluginVerification
from core.plugin.plugin_service import PluginService
from enums.deployment_edition import DeploymentEdition
from models import ProviderType
from models.engine import db
from models.provider import Provider, ProviderCredential, TenantPreferredModelProvider
@@ -35,10 +36,11 @@ def _make_features(
scope: PluginInstallationScope = PluginInstallationScope.ALL,
) -> SystemFeatureModel:
return SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
plugin_installation_permission=PluginInstallationPermissionModel(
restrict_to_marketplace_only=restrict_to_marketplace,
plugin_installation_scope=scope,
)
),
)

Some files were not shown because too many files have changed in this diff Show More