Compare commits

..
444 changed files with 4548 additions and 18917 deletions
-111
View File
@@ -110,114 +110,3 @@ updates:
github-actions-dependencies:
patterns:
- "*"
- package-ecosystem: "uv"
directory: "/api"
target-branch: "lts/1.13.x"
open-pull-requests-limit: 10
schedule:
interval: "weekly"
groups:
flask:
patterns:
- "flask"
- "flask-*"
- "werkzeug"
- "gunicorn"
google:
patterns:
- "google-*"
- "googleapis-*"
opentelemetry:
patterns:
- "opentelemetry-*"
pydantic:
patterns:
- "pydantic"
- "pydantic-*"
llm:
patterns:
- "langfuse"
- "langsmith"
- "litellm"
- "mlflow*"
- "opik"
- "weave*"
- "arize*"
- "tiktoken"
- "transformers"
database:
patterns:
- "sqlalchemy"
- "psycopg2*"
- "psycogreen"
- "redis*"
- "alembic*"
storage:
patterns:
- "boto3*"
- "botocore*"
- "azure-*"
- "bce-*"
- "cos-python-*"
- "esdk-obs-*"
- "google-cloud-storage"
- "opendal"
- "oss2"
- "supabase*"
- "tos*"
vdb:
patterns:
- "alibabacloud*"
- "chromadb"
- "clickhouse-*"
- "clickzetta-*"
- "couchbase"
- "elasticsearch"
- "opensearch-py"
- "oracledb"
- "pgvect*"
- "pymilvus"
- "pymochow"
- "pyobvector"
- "qdrant-client"
- "intersystems-*"
- "tablestore"
- "tcvectordb"
- "tidb-vector"
- "upstash-*"
- "volcengine-*"
- "weaviate-*"
- "xinference-*"
- "mo-vector"
- "mysql-connector-*"
dev:
patterns:
- "coverage"
- "dotenv-linter"
- "faker"
- "lxml-stubs"
- "basedpyright"
- "ruff"
- "pytest*"
- "types-*"
- "boto3-stubs"
- "hypothesis"
- "pandas-stubs"
- "scipy-stubs"
- "import-linter"
- "celery-types"
- "mypy*"
- "pyrefly"
python-packages:
patterns:
- "*"
- package-ecosystem: "github-actions"
directory: "/"
target-branch: "lts/1.13.x"
open-pull-requests-limit: 5
schedule:
interval: "weekly"
groups:
github-actions-dependencies:
patterns:
- "*"
+3 -7
View File
@@ -15,12 +15,8 @@ concurrency:
jobs:
test:
name: CLI Tests (${{ matrix.os }})
runs-on: ${{ matrix.os }}
strategy:
fail-fast: false
matrix:
os: [depot-ubuntu-24.04, windows-latest, macos-latest]
name: CLI Tests
runs-on: depot-ubuntu-24.04
env:
CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
defaults:
@@ -41,7 +37,7 @@ jobs:
run: pnpm ci
- name: Report coverage
if: ${{ env.CODECOV_TOKEN != '' && matrix.os == 'depot-ubuntu-24.04' }}
if: ${{ env.CODECOV_TOKEN != '' }}
uses: codecov/codecov-action@57e3a136b779b570ffcdbf80b3bdc90e7fab3de2 # v6.0.0
with:
directory: cli/coverage
+1 -1
View File
@@ -257,5 +257,5 @@ scripts/stress-test/reports/
# Code Agent Folder
.qoder/*
.context/
.context/*
.eslintcache
+2 -4
View File
@@ -30,8 +30,7 @@ from clients.agent_backend.factory import create_agent_backend_run_client
from clients.agent_backend.fake_client import FakeAgentBackendRunClient, FakeAgentBackendScenario
from clients.agent_backend.request_builder import (
AGENT_SOUL_PROMPT_LAYER_ID,
DIFY_EXECUTION_CONTEXT_LAYER_ID,
DIFY_PLUGIN_TOOLS_LAYER_ID,
DIFY_PLUGIN_CONTEXT_LAYER_ID,
WORKFLOW_NODE_JOB_PROMPT_LAYER_ID,
WORKFLOW_USER_PROMPT_LAYER_ID,
AgentBackendModelConfig,
@@ -43,8 +42,7 @@ from clients.agent_backend.request_builder import (
__all__ = [
"AGENT_SOUL_PROMPT_LAYER_ID",
"DIFY_EXECUTION_CONTEXT_LAYER_ID",
"DIFY_PLUGIN_TOOLS_LAYER_ID",
"DIFY_PLUGIN_CONTEXT_LAYER_ID",
"WORKFLOW_NODE_JOB_PROMPT_LAYER_ID",
"WORKFLOW_USER_PROMPT_LAYER_ID",
"AgentBackendError",
+17 -29
View File
@@ -4,9 +4,7 @@ This module is intentionally an adapter, not a wire DTO package. The emitted
object is always ``dify_agent.protocol.CreateRunRequest`` so the Agent backend
protocol has a single owner. API-only context such as Agent Soul vs workflow job
prompt is preserved in layer names and metadata until the dedicated product
schemas land in later phases. Dify-owned execution identifiers are emitted as an
explicit ``dify.execution_context`` layer so the run request stays fully
composition-driven.
schemas land in later phases.
"""
from __future__ import annotations
@@ -17,21 +15,18 @@ from agenton.compositor import CompositorSessionSnapshot
from agenton.layers import ExitIntent
from agenton_collections.layers.plain import PLAIN_PROMPT_LAYER_TYPE_ID, PromptLayerConfig
from dify_agent.layers.dify_plugin import (
DIFY_PLUGIN_LAYER_TYPE_ID,
DIFY_PLUGIN_LLM_LAYER_TYPE_ID,
DIFY_PLUGIN_TOOLS_LAYER_TYPE_ID,
DifyPluginCredentialValue,
DifyPluginLayerConfig,
DifyPluginLLMLayerConfig,
DifyPluginToolsLayerConfig,
)
from dify_agent.layers.execution_context import (
DIFY_EXECUTION_CONTEXT_LAYER_TYPE_ID,
DifyExecutionContextLayerConfig,
)
from dify_agent.layers.output import DIFY_OUTPUT_LAYER_TYPE_ID, DifyOutputLayerConfig
from dify_agent.protocol import (
DIFY_AGENT_MODEL_LAYER_ID,
DIFY_AGENT_OUTPUT_LAYER_ID,
CreateRunRequest,
ExecutionContext,
LayerExitSignals,
RunComposition,
RunLayerSpec,
@@ -42,16 +37,17 @@ from pydantic import BaseModel, ConfigDict, Field, JsonValue, field_validator
AGENT_SOUL_PROMPT_LAYER_ID = "agent_soul_prompt"
WORKFLOW_NODE_JOB_PROMPT_LAYER_ID = "workflow_node_job_prompt"
WORKFLOW_USER_PROMPT_LAYER_ID = "workflow_user_prompt"
DIFY_EXECUTION_CONTEXT_LAYER_ID = "execution_context"
DIFY_PLUGIN_TOOLS_LAYER_ID = "tools"
DIFY_PLUGIN_CONTEXT_LAYER_ID = "plugin"
class AgentBackendModelConfig(BaseModel):
"""API-side model/plugin selection before it is converted to Dify Agent layers."""
tenant_id: str
plugin_id: str
model_provider: str
model: str
user_id: str | None = None
credentials: dict[str, DifyPluginCredentialValue] = Field(default_factory=dict)
model_settings: dict[str, JsonValue] = Field(default_factory=dict)
@@ -77,14 +73,13 @@ class AgentBackendWorkflowNodeRunInput(BaseModel):
"""Inputs needed to build the first workflow-node-oriented Agent backend run request."""
model: AgentBackendModelConfig
execution_context: DifyExecutionContextLayerConfig
execution_context: ExecutionContext
workflow_node_job_prompt: str
user_prompt: str
agent_soul_prompt: str | None = None
purpose: RunPurpose = "workflow_node"
idempotency_key: str | None = None
output: AgentBackendOutputConfig | None = None
tools: DifyPluginToolsLayerConfig | None = None
session_snapshot: CompositorSessionSnapshot | None = None
suspend_on_exit: bool = False
metadata: dict[str, JsonValue] = Field(default_factory=dict)
@@ -130,18 +125,21 @@ class AgentBackendRunRequestBuilder:
config=PromptLayerConfig(user=run_input.user_prompt),
),
RunLayerSpec(
name=DIFY_EXECUTION_CONTEXT_LAYER_ID,
type=DIFY_EXECUTION_CONTEXT_LAYER_TYPE_ID,
name=DIFY_PLUGIN_CONTEXT_LAYER_ID,
type=DIFY_PLUGIN_LAYER_TYPE_ID,
metadata=run_input.metadata,
config=run_input.execution_context,
config=DifyPluginLayerConfig(
tenant_id=run_input.model.tenant_id,
plugin_id=run_input.model.plugin_id,
user_id=run_input.model.user_id,
),
),
RunLayerSpec(
name=DIFY_AGENT_MODEL_LAYER_ID,
type=DIFY_PLUGIN_LLM_LAYER_TYPE_ID,
deps={"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID},
deps={"plugin": DIFY_PLUGIN_CONTEXT_LAYER_ID},
metadata=run_input.metadata,
config=DifyPluginLLMLayerConfig(
plugin_id=run_input.model.plugin_id,
model_provider=run_input.model.model_provider,
model=run_input.model.model,
credentials=run_input.model.credentials,
@@ -151,17 +149,6 @@ class AgentBackendRunRequestBuilder:
]
)
if run_input.tools is not None and run_input.tools.tools:
layers.append(
RunLayerSpec(
name=DIFY_PLUGIN_TOOLS_LAYER_ID,
type=DIFY_PLUGIN_TOOLS_LAYER_TYPE_ID,
deps={"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID},
metadata=run_input.metadata,
config=run_input.tools,
)
)
if run_input.output is not None:
layers.append(
RunLayerSpec(
@@ -178,6 +165,7 @@ class AgentBackendRunRequestBuilder:
return CreateRunRequest(
composition=RunComposition(layers=layers),
execution_context=run_input.execution_context,
purpose=run_input.purpose,
idempotency_key=run_input.idempotency_key,
metadata=run_input.metadata,
+6 -16
View File
@@ -79,14 +79,10 @@ def data_migrate() -> None:
multiple=True,
type=str,
help=(
"Limit migration to specific tables. Accepts comma-separated values or repeated flags.\n"
"\n"
"Options: load_balancing_model_configs, provider_model_credentials, "
"provider_model_settings, provider_models, tenant_default_models.\n\n"
"Limit model_type migration to specific tables. Accepts comma-separated values or repeated flags. "
"When provider_model_credentials is selected, provider_models and "
"load_balancing_model_configs may also be updated for credential reference rewrites.\n"
"\n"
"If unspecified, all relevant tables are migrated."
"load_balancing_model_configs may also be updated for credential reference rewrites."
"Default to: "
),
)
@click.option(
@@ -95,11 +91,8 @@ def data_migrate() -> None:
multiple=True,
type=str,
help=(
"Canonical model types to migrate. Accepts comma-separated values or repeated flags.\n"
"\n"
"Options: llm,text-embedding,rerank\n"
"\n"
"If unspecified, all relevant legacy model types are migrated."
"Canonical model types to migrate. Accepts comma-separated values or repeated flags. "
"Defaults to: `llm,text-embedding,rerank`"
),
)
@click.option(
@@ -110,10 +103,7 @@ def data_migrate() -> None:
@click.option(
"--output",
type=click.Path(dir_okay=False, resolve_path=True, path_type=Path),
help=(
"Optional file path for JSON lines event logs. Defaults to stdout.\n"
"It's highly recommended to save the event logs to a file and preserve it for a period of time."
),
help="Optional file path for JSON lines event logs. Defaults to stdout.",
)
@click.option(
"--concurrency",
@@ -41,21 +41,3 @@ class MilvusConfig(BaseSettings):
description='Milvus text analyzer parameters, e.g., {"type": "chinese"} for Chinese segmentation support.',
default=None,
)
MILVUS_SECURE: bool = Field(
description="Enable TLS for the Milvus connection (one-way TLS). When True, the client uses gRPC over TLS "
"and verifies the server certificate. Equivalent to passing secure=True to pymilvus.",
default=False,
)
MILVUS_SERVER_PEM_PATH: str | None = Field(
description="Filesystem path inside the container to the Milvus server certificate (PEM). Mount this via "
"a Kubernetes secret. Used as pymilvus's server_pem_path when MILVUS_SECURE is True.",
default=None,
)
MILVUS_SERVER_NAME: str | None = Field(
description="Server name (TLS SNI / certificate CN or SAN) to verify against the Milvus server certificate. "
"Required when MILVUS_SERVER_PEM_PATH is set.",
default=None,
)
-2
View File
@@ -68,7 +68,6 @@ from .app import (
workflow_app_log,
workflow_comment,
workflow_draft_variable,
workflow_node_output_inspector,
workflow_run,
workflow_statistic,
workflow_trigger,
@@ -219,7 +218,6 @@ __all__ = [
"workflow_app_log",
"workflow_comment",
"workflow_draft_variable",
"workflow_node_output_inspector",
"workflow_run",
"workflow_statistic",
"workflow_trigger",
+11 -11
View File
@@ -5,7 +5,7 @@ from controllers.console import console_ns
from controllers.console.app.wraps import get_app_model
from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
from libs.login import current_account_with_tenant, login_required
from models.model import App, AppMode
from models.model import AppMode
from services.agent.composer_service import AgentComposerService
from services.agent.composer_validator import ComposerConfigValidator
from services.entities.agent_entities import ComposerSavePayload
@@ -19,7 +19,7 @@ class WorkflowAgentComposerApi(Resource):
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.WORKFLOW, AppMode.ADVANCED_CHAT])
def get(self, app_model: App, node_id: str):
def get(self, app_model, node_id: str):
_, tenant_id = current_account_with_tenant()
return AgentComposerService.load_workflow_composer(
tenant_id=tenant_id,
@@ -33,7 +33,7 @@ class WorkflowAgentComposerApi(Resource):
@account_initialization_required
@edit_permission_required
@get_app_model(mode=[AppMode.WORKFLOW, AppMode.ADVANCED_CHAT])
def put(self, app_model: App, node_id: str):
def put(self, app_model, node_id: str):
account, tenant_id = current_account_with_tenant()
payload = ComposerSavePayload.model_validate(console_ns.payload or {})
return AgentComposerService.save_workflow_composer(
@@ -52,7 +52,7 @@ class WorkflowAgentComposerValidateApi(Resource):
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.WORKFLOW, AppMode.ADVANCED_CHAT])
def post(self, app_model: App, node_id: str):
def post(self, app_model, node_id: str):
payload = ComposerSavePayload.model_validate(console_ns.payload or {})
ComposerConfigValidator.validate_save_payload(payload)
return {"result": "success", "errors": []}
@@ -64,7 +64,7 @@ class WorkflowAgentComposerCandidatesApi(Resource):
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.WORKFLOW, AppMode.ADVANCED_CHAT])
def get(self, app_model: App, node_id: str):
def get(self, app_model, node_id: str):
return AgentComposerService.get_workflow_candidates(app_id=app_model.id)
@@ -74,7 +74,7 @@ class WorkflowAgentComposerImpactApi(Resource):
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.WORKFLOW, AppMode.ADVANCED_CHAT])
def post(self, app_model: App, node_id: str):
def post(self, app_model, node_id: str):
_, tenant_id = current_account_with_tenant()
payload = ComposerSavePayload.model_validate(console_ns.payload or {})
current_snapshot_id = payload.binding.current_snapshot_id if payload.binding else None
@@ -91,7 +91,7 @@ class WorkflowAgentComposerSaveToRosterApi(Resource):
@account_initialization_required
@edit_permission_required
@get_app_model(mode=[AppMode.WORKFLOW, AppMode.ADVANCED_CHAT])
def post(self, app_model: App, node_id: str):
def post(self, app_model, node_id: str):
account, tenant_id = current_account_with_tenant()
payload = ComposerSavePayload.model_validate(console_ns.payload or {})
return AgentComposerService.save_workflow_composer(
@@ -109,7 +109,7 @@ class AgentAppComposerApi(Resource):
@login_required
@account_initialization_required
@get_app_model()
def get(self, app_model: App):
def get(self, app_model):
_, tenant_id = current_account_with_tenant()
return AgentComposerService.load_agent_app_composer(tenant_id=tenant_id, app_id=app_model.id)
@@ -119,7 +119,7 @@ class AgentAppComposerApi(Resource):
@account_initialization_required
@edit_permission_required
@get_app_model()
def put(self, app_model: App):
def put(self, app_model):
account, tenant_id = current_account_with_tenant()
payload = ComposerSavePayload.model_validate(console_ns.payload or {})
return AgentComposerService.save_agent_app_composer(
@@ -137,7 +137,7 @@ class AgentAppComposerValidateApi(Resource):
@login_required
@account_initialization_required
@get_app_model()
def post(self, app_model: App):
def post(self, app_model):
payload = ComposerSavePayload.model_validate(console_ns.payload or {})
ComposerConfigValidator.validate_save_payload(payload)
return {"result": "success", "errors": []}
@@ -149,5 +149,5 @@ class AgentAppComposerCandidatesApi(Resource):
@login_required
@account_initialization_required
@get_app_model()
def get(self, app_model: App):
def get(self, app_model):
return AgentComposerService.get_agent_app_candidates(app_id=app_model.id)
+29 -63
View File
@@ -9,25 +9,18 @@ from sqlalchemy import delete, func, select
from sqlalchemy.orm import sessionmaker
from werkzeug.exceptions import Forbidden
from controllers.common.schema import register_response_schema_models
from controllers.common.schema import register_schema_models
from extensions.ext_database import db
from fields.base import ResponseModel
from libs.helper import dump_response, to_timestamp
from libs.login import login_required
from models import Account
from libs.helper import to_timestamp
from libs.login import current_account_with_tenant, login_required
from models.dataset import Dataset
from models.enums import ApiTokenType
from models.model import ApiToken, App
from services.api_token_service import ApiTokenCache
from . import console_ns
from .wraps import (
account_initialization_required,
edit_permission_required,
setup_required,
with_current_tenant_id,
with_current_user,
)
from .wraps import account_initialization_required, edit_permission_required, setup_required
class ApiKeyItem(ResponseModel):
@@ -47,7 +40,7 @@ class ApiKeyList(ResponseModel):
data: list[ApiKeyItem]
register_response_schema_models(console_ns, ApiKeyItem, ApiKeyList)
register_schema_models(console_ns, ApiKeyItem, ApiKeyList)
def _get_resource(resource_id, tenant_id, resource_model):
@@ -71,11 +64,10 @@ class BaseApiKeyListResource(Resource):
token_prefix: str | None = None
max_keys = 10
def get(self, resource_id: str, current_tenant_id: str) -> dict[str, object]:
return dump_response(ApiKeyList, self._get_api_key_list(resource_id, current_tenant_id))
def _get_api_key_list(self, resource_id: str, current_tenant_id: str) -> ApiKeyList:
def get(self, resource_id):
assert self.resource_id_field is not None, "resource_id_field must be set"
resource_id = str(resource_id)
_, current_tenant_id = current_account_with_tenant()
_get_resource(resource_id, current_tenant_id, self.resource_model)
keys = db.session.scalars(
@@ -83,14 +75,13 @@ class BaseApiKeyListResource(Resource):
ApiToken.type == self.resource_type, getattr(ApiToken, self.resource_id_field) == resource_id
)
).all()
return ApiKeyList.model_validate({"data": keys}, from_attributes=True)
return ApiKeyList.model_validate({"data": keys}, from_attributes=True).model_dump(mode="json")
@edit_permission_required
def post(self, resource_id: str, current_tenant_id: str) -> tuple[dict[str, object], int]:
return dump_response(ApiKeyItem, self._create_api_key(resource_id, current_tenant_id)), 201
def _create_api_key(self, resource_id: str, current_tenant_id: str) -> ApiToken:
def post(self, resource_id):
assert self.resource_id_field is not None, "resource_id_field must be set"
resource_id = str(resource_id)
_, current_tenant_id = current_account_with_tenant()
_get_resource(resource_id, current_tenant_id, self.resource_model)
current_key_count: int = (
db.session.scalar(
@@ -117,7 +108,7 @@ class BaseApiKeyListResource(Resource):
api_token.type = self.resource_type
db.session.add(api_token)
db.session.commit()
return api_token
return ApiKeyItem.model_validate(api_token, from_attributes=True).model_dump(mode="json"), 201
class BaseApiKeyResource(Resource):
@@ -127,20 +118,9 @@ class BaseApiKeyResource(Resource):
resource_model: type | None = None
resource_id_field: str | None = None
def delete(
self, resource_id: str, api_key_id: str, current_tenant_id: str, current_user: Account
) -> tuple[str, int]:
self._delete_api_key(resource_id, api_key_id, current_tenant_id, current_user)
return "", 204
def _delete_api_key(
self,
resource_id: str,
api_key_id: str,
current_tenant_id: str,
current_user: Account,
) -> None:
def delete(self, resource_id: str, api_key_id: str):
assert self.resource_id_field is not None, "resource_id_field must be set"
current_user, current_tenant_id = current_account_with_tenant()
_get_resource(resource_id, current_tenant_id, self.resource_model)
if not current_user.is_admin_or_owner:
@@ -167,6 +147,8 @@ class BaseApiKeyResource(Resource):
db.session.execute(delete(ApiToken).where(ApiToken.id == api_key_id))
db.session.commit()
return "", 204
@console_ns.route("/apps/<uuid:resource_id>/api-keys")
class AppApiKeyListResource(BaseApiKeyListResource):
@@ -174,21 +156,18 @@ class AppApiKeyListResource(BaseApiKeyListResource):
@console_ns.doc(description="Get all API keys for an app")
@console_ns.doc(params={"resource_id": "App ID"})
@console_ns.response(200, "API keys retrieved successfully", console_ns.models[ApiKeyList.__name__])
@with_current_tenant_id
def get(self, current_tenant_id: str, resource_id: UUID) -> dict[str, object]:
def get(self, resource_id: UUID):
"""Get all API keys for an app"""
return dump_response(ApiKeyList, self._get_api_key_list(str(resource_id), current_tenant_id))
return super().get(resource_id)
@console_ns.doc("create_app_api_key")
@console_ns.doc(description="Create a new API key for an app")
@console_ns.doc(params={"resource_id": "App ID"})
@console_ns.response(201, "API key created successfully", console_ns.models[ApiKeyItem.__name__])
@console_ns.response(400, "Maximum keys exceeded")
@with_current_tenant_id
@edit_permission_required
def post(self, current_tenant_id: str, resource_id: UUID) -> tuple[dict[str, object], int]:
def post(self, resource_id: UUID):
"""Create a new API key for an app"""
return dump_response(ApiKeyItem, self._create_api_key(str(resource_id), current_tenant_id)), 201
return super().post(resource_id)
resource_type = ApiTokenType.APP
resource_model = App
@@ -202,14 +181,9 @@ class AppApiKeyResource(BaseApiKeyResource):
@console_ns.doc(description="Delete an API key for an app")
@console_ns.doc(params={"resource_id": "App ID", "api_key_id": "API key ID"})
@console_ns.response(204, "API key deleted successfully")
@with_current_user
@with_current_tenant_id
def delete(
self, current_tenant_id: str, current_user: Account, resource_id: UUID, api_key_id: UUID
) -> tuple[str, int]:
def delete(self, resource_id: UUID, api_key_id: UUID):
"""Delete an API key for an app"""
self._delete_api_key(str(resource_id), str(api_key_id), current_tenant_id, current_user)
return "", 204
return super().delete(str(resource_id), str(api_key_id))
resource_type = ApiTokenType.APP
resource_model = App
@@ -222,21 +196,18 @@ class DatasetApiKeyListResource(BaseApiKeyListResource):
@console_ns.doc(description="Get all API keys for a dataset")
@console_ns.doc(params={"resource_id": "Dataset ID"})
@console_ns.response(200, "API keys retrieved successfully", console_ns.models[ApiKeyList.__name__])
@with_current_tenant_id
def get(self, current_tenant_id: str, resource_id: UUID) -> dict[str, object]:
def get(self, resource_id: UUID):
"""Get all API keys for a dataset"""
return dump_response(ApiKeyList, self._get_api_key_list(str(resource_id), current_tenant_id))
return super().get(resource_id)
@console_ns.doc("create_dataset_api_key")
@console_ns.doc(description="Create a new API key for a dataset")
@console_ns.doc(params={"resource_id": "Dataset ID"})
@console_ns.response(201, "API key created successfully", console_ns.models[ApiKeyItem.__name__])
@console_ns.response(400, "Maximum keys exceeded")
@with_current_tenant_id
@edit_permission_required
def post(self, current_tenant_id: str, resource_id: UUID) -> tuple[dict[str, object], int]:
def post(self, resource_id: UUID):
"""Create a new API key for a dataset"""
return dump_response(ApiKeyItem, self._create_api_key(str(resource_id), current_tenant_id)), 201
return super().post(resource_id)
resource_type = ApiTokenType.DATASET
resource_model = Dataset
@@ -250,14 +221,9 @@ class DatasetApiKeyResource(BaseApiKeyResource):
@console_ns.doc(description="Delete an API key for a dataset")
@console_ns.doc(params={"resource_id": "Dataset ID", "api_key_id": "API key ID"})
@console_ns.response(204, "API key deleted successfully")
@with_current_user
@with_current_tenant_id
def delete(
self, current_tenant_id: str, current_user: Account, resource_id: UUID, api_key_id: UUID
) -> tuple[str, int]:
def delete(self, resource_id: UUID, api_key_id: UUID):
"""Delete an API key for a dataset"""
self._delete_api_key(str(resource_id), str(api_key_id), current_tenant_id, current_user)
return "", 204
return super().delete(str(resource_id), str(api_key_id))
resource_type = ApiTokenType.DATASET
resource_model = Dataset
+2 -2
View File
@@ -8,7 +8,7 @@ from controllers.console.app.wraps import get_app_model
from controllers.console.wraps import account_initialization_required, setup_required
from libs.helper import uuid_value
from libs.login import login_required
from models.model import App, AppMode
from models.model import AppMode
from services.agent_service import AgentService
@@ -39,7 +39,7 @@ class AgentLogApi(Resource):
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.AGENT_CHAT])
def get(self, app_model: App):
def get(self, app_model):
"""Get agent logs"""
args = AgentLogQuery.model_validate(request.args.to_dict(flat=True))
+12 -12
View File
@@ -573,7 +573,7 @@ class AppApi(Resource):
@account_initialization_required
@enterprise_license_required
@get_app_model(mode=None)
def get(self, app_model: App):
def get(self, app_model):
"""Get app detail"""
app_service = AppService()
@@ -581,7 +581,7 @@ class AppApi(Resource):
if FeatureService.get_system_features().webapp_auth.enabled:
app_setting = EnterpriseService.WebAppAuth.get_app_access_mode_by_id(app_id=str(app_model.id))
app_model.access_mode = app_setting.access_mode # type: ignore[attr-defined]
app_model.access_mode = app_setting.access_mode
response_model = AppDetailWithSite.model_validate(app_model, from_attributes=True)
return response_model.model_dump(mode="json")
@@ -598,7 +598,7 @@ class AppApi(Resource):
@account_initialization_required
@get_app_model(mode=None)
@edit_permission_required
def put(self, app_model: App):
def put(self, app_model):
"""Update app"""
args = UpdateAppPayload.model_validate(console_ns.payload)
@@ -627,7 +627,7 @@ class AppApi(Resource):
@login_required
@account_initialization_required
@edit_permission_required
def delete(self, app_model: App):
def delete(self, app_model):
"""Delete app"""
app_service = AppService()
app_service.delete_app(app_model)
@@ -648,7 +648,7 @@ class AppCopyApi(Resource):
@account_initialization_required
@get_app_model(mode=None)
@edit_permission_required
def post(self, app_model: App):
def post(self, app_model):
"""Copy app"""
# The role of the current user in the ta table must be admin, owner, or editor
current_user, _ = current_account_with_tenant()
@@ -709,7 +709,7 @@ class AppExportApi(Resource):
@login_required
@account_initialization_required
@edit_permission_required
def get(self, app_model: App):
def get(self, app_model):
"""Export app"""
args = AppExportQuery.model_validate(request.args.to_dict(flat=True))
@@ -731,7 +731,7 @@ class AppPublishToCreatorsPlatformApi(Resource):
@account_initialization_required
@get_app_model(mode=None)
@edit_permission_required
def post(self, app_model: App):
def post(self, app_model):
"""Publish app to Creators Platform"""
from configs import dify_config
from core.helper.creators import get_redirect_url, upload_dsl
@@ -762,7 +762,7 @@ class AppNameApi(Resource):
@account_initialization_required
@get_app_model(mode=None)
@edit_permission_required
def post(self, app_model: App):
def post(self, app_model):
args = AppNamePayload.model_validate(console_ns.payload)
app_service = AppService()
@@ -784,7 +784,7 @@ class AppIconApi(Resource):
@account_initialization_required
@get_app_model(mode=None)
@edit_permission_required
def post(self, app_model: App):
def post(self, app_model):
args = AppIconPayload.model_validate(console_ns.payload or {})
app_service = AppService()
@@ -811,7 +811,7 @@ class AppSiteStatus(Resource):
@account_initialization_required
@get_app_model(mode=None)
@edit_permission_required
def post(self, app_model: App):
def post(self, app_model):
args = AppSiteStatusPayload.model_validate(console_ns.payload)
app_service = AppService()
@@ -833,7 +833,7 @@ class AppApiStatus(Resource):
@is_admin_or_owner_required
@account_initialization_required
@get_app_model(mode=None)
def post(self, app_model: App):
def post(self, app_model):
args = AppApiStatusPayload.model_validate(console_ns.payload)
app_service = AppService()
@@ -874,7 +874,7 @@ class AppTraceApi(Resource):
@account_initialization_required
@edit_permission_required
@get_app_model
def post(self, app_model: App):
def post(self, app_model):
# add app trace
args = AppTracePayload.model_validate(console_ns.payload)
+2 -2
View File
@@ -70,7 +70,7 @@ class ChatMessageAudioApi(Resource):
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT])
def post(self, app_model: App):
def post(self, app_model):
file = request.files["file"]
try:
@@ -171,7 +171,7 @@ class TextModesApi(Resource):
@setup_required
@login_required
@account_initialization_required
def get(self, app_model: App):
def get(self, app_model):
try:
args = TextToSpeechVoiceQuery.model_validate(request.args.to_dict(flat=True))
+5 -5
View File
@@ -33,7 +33,7 @@ from libs import helper
from libs.helper import uuid_value
from libs.login import current_user, login_required
from models import Account
from models.model import App, AppMode
from models.model import AppMode
from services.app_generate_service import AppGenerateService
from services.app_task_service import AppTaskService
from services.errors.llm import InvokeRateLimitError
@@ -84,7 +84,7 @@ class CompletionMessageApi(Resource):
@login_required
@account_initialization_required
@get_app_model(mode=AppMode.COMPLETION)
def post(self, app_model: App):
def post(self, app_model):
args_model = CompletionMessagePayload.model_validate(console_ns.payload)
args = args_model.model_dump(exclude_none=True, by_alias=True)
@@ -131,7 +131,7 @@ class CompletionMessageStopApi(Resource):
@login_required
@account_initialization_required
@get_app_model(mode=AppMode.COMPLETION)
def post(self, app_model: App, task_id: str):
def post(self, app_model, task_id: str):
if not isinstance(current_user, Account):
raise ValueError("current_user must be an Account instance")
@@ -159,7 +159,7 @@ class ChatMessageApi(Resource):
@account_initialization_required
@get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT])
@edit_permission_required
def post(self, app_model: App):
def post(self, app_model):
args_model = ChatMessagePayload.model_validate(console_ns.payload)
args = args_model.model_dump(exclude_none=True, by_alias=True)
@@ -212,7 +212,7 @@ class ChatMessageStopApi(Resource):
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT])
def post(self, app_model: App, task_id: str):
def post(self, app_model, task_id: str):
if not isinstance(current_user, Account):
raise ValueError("current_user must be an Account instance")
+7 -7
View File
@@ -33,7 +33,7 @@ from fields.conversation_fields import (
from libs.datetime_utils import naive_utc_now, parse_time_range
from libs.login import current_account_with_tenant, login_required
from models import Conversation, EndUser, Message, MessageAnnotation
from models.model import App, AppMode
from models.model import AppMode
from services.conversation_service import ConversationService
from services.errors.conversation import ConversationNotExistsError
@@ -93,7 +93,7 @@ class CompletionConversationApi(Resource):
@account_initialization_required
@get_app_model(mode=AppMode.COMPLETION)
@edit_permission_required
def get(self, app_model: App):
def get(self, app_model):
current_user, _ = current_account_with_tenant()
args = CompletionConversationQuery.model_validate(request.args.to_dict(flat=True))
@@ -165,7 +165,7 @@ class CompletionConversationDetailApi(Resource):
@account_initialization_required
@get_app_model(mode=AppMode.COMPLETION)
@edit_permission_required
def get(self, app_model: App, conversation_id: UUID):
def get(self, app_model, conversation_id: UUID):
conversation_id_str = str(conversation_id)
return ConversationMessageDetailResponse.model_validate(
_get_conversation(app_model, conversation_id_str), from_attributes=True
@@ -182,7 +182,7 @@ class CompletionConversationDetailApi(Resource):
@account_initialization_required
@get_app_model(mode=AppMode.COMPLETION)
@edit_permission_required
def delete(self, app_model: App, conversation_id: UUID):
def delete(self, app_model, conversation_id: UUID):
current_user, _ = current_account_with_tenant()
conversation_id_str = str(conversation_id)
@@ -207,7 +207,7 @@ class ChatConversationApi(Resource):
@account_initialization_required
@get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT])
@edit_permission_required
def get(self, app_model: App):
def get(self, app_model):
current_user, _ = current_account_with_tenant()
args = ChatConversationQuery.model_validate(request.args.to_dict(flat=True))
@@ -318,7 +318,7 @@ class ChatConversationDetailApi(Resource):
@account_initialization_required
@get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT])
@edit_permission_required
def get(self, app_model: App, conversation_id: UUID):
def get(self, app_model, conversation_id: UUID):
conversation_id_str = str(conversation_id)
return ConversationDetailResponse.model_validate(
_get_conversation(app_model, conversation_id_str), from_attributes=True
@@ -335,7 +335,7 @@ class ChatConversationDetailApi(Resource):
@get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT])
@account_initialization_required
@edit_permission_required
def delete(self, app_model: App, conversation_id: UUID):
def delete(self, app_model, conversation_id: UUID):
current_user, _ = current_account_with_tenant()
conversation_id_str = str(conversation_id)
@@ -19,7 +19,7 @@ from fields.base import ResponseModel
from libs.helper import to_timestamp
from libs.login import login_required
from models import ConversationVariable
from models.model import App, AppMode
from models.model import AppMode
class ConversationVariablesQuery(BaseModel):
@@ -94,7 +94,7 @@ class ConversationVariablesApi(Resource):
@login_required
@account_initialization_required
@get_app_model(mode=AppMode.ADVANCED_CHAT)
def get(self, app_model: App):
def get(self, app_model):
args = ConversationVariablesQuery.model_validate(request.args.to_dict(flat=True))
stmt = (
+4 -4
View File
@@ -17,7 +17,7 @@ from fields.base import ResponseModel
from libs.helper import to_timestamp
from libs.login import current_account_with_tenant, login_required
from models.enums import AppMCPServerStatus
from models.model import App, AppMCPServer
from models.model import AppMCPServer
class MCPServerCreatePayload(BaseModel):
@@ -73,7 +73,7 @@ class AppMCPServerController(Resource):
@account_initialization_required
@setup_required
@get_app_model
def get(self, app_model: App):
def get(self, app_model):
server = db.session.scalar(select(AppMCPServer).where(AppMCPServer.app_id == app_model.id).limit(1))
if server is None:
return {}
@@ -92,7 +92,7 @@ class AppMCPServerController(Resource):
@login_required
@setup_required
@edit_permission_required
def post(self, app_model: App):
def post(self, app_model):
_, current_tenant_id = current_account_with_tenant()
payload = MCPServerCreatePayload.model_validate(console_ns.payload or {})
@@ -127,7 +127,7 @@ class AppMCPServerController(Resource):
@setup_required
@account_initialization_required
@edit_permission_required
def put(self, app_model: App):
def put(self, app_model):
payload = MCPServerUpdatePayload.model_validate(console_ns.payload or {})
server = db.session.get(AppMCPServer, payload.id)
if not server:
+7 -7
View File
@@ -45,7 +45,7 @@ from libs.helper import to_timestamp, uuid_value
from libs.infinite_scroll_pagination import InfiniteScrollPagination
from libs.login import current_account_with_tenant, login_required
from models.enums import FeedbackFromSource, FeedbackRating
from models.model import App, AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
from models.model import AppMode, Conversation, Message, MessageAnnotation, MessageFeedback
from services.errors.conversation import ConversationNotExistsError
from services.errors.message import MessageNotExistsError, SuggestedQuestionsAfterAnswerDisabledError
from services.message_service import MessageService, attach_message_extra_contents
@@ -180,7 +180,7 @@ class ChatMessageListApi(Resource):
@setup_required
@get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT])
@edit_permission_required
def get(self, app_model: App):
def get(self, app_model):
args = ChatMessagesQuery.model_validate(request.args.to_dict())
conversation = db.session.scalar(
@@ -257,7 +257,7 @@ class MessageFeedbackApi(Resource):
@setup_required
@login_required
@account_initialization_required
def post(self, app_model: App):
def post(self, app_model):
current_user, _ = current_account_with_tenant()
args = MessageFeedbackPayload.model_validate(console_ns.payload)
@@ -314,7 +314,7 @@ class MessageAnnotationCountApi(Resource):
@setup_required
@login_required
@account_initialization_required
def get(self, app_model: App):
def get(self, app_model):
count = db.session.scalar(
select(func.count(MessageAnnotation.id)).where(MessageAnnotation.app_id == app_model.id)
)
@@ -337,7 +337,7 @@ class MessageSuggestedQuestionApi(Resource):
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT])
def get(self, app_model: App, message_id: UUID):
def get(self, app_model, message_id: UUID):
current_user, _ = current_account_with_tenant()
message_id_str = str(message_id)
@@ -379,7 +379,7 @@ class MessageFeedbackExportApi(Resource):
@setup_required
@login_required
@account_initialization_required
def get(self, app_model: App):
def get(self, app_model):
args = FeedbackExportQuery.model_validate(request.args.to_dict())
# Import the service function
@@ -417,7 +417,7 @@ class MessageApi(Resource):
@setup_required
@login_required
@account_initialization_required
def get(self, app_model: App, message_id: UUID):
def get(self, app_model, message_id: UUID):
message_id_str = str(message_id)
message = db.session.scalar(
+2 -2
View File
@@ -16,7 +16,7 @@ from events.app_event import app_model_config_was_updated
from extensions.ext_database import db
from libs.datetime_utils import naive_utc_now
from libs.login import current_account_with_tenant, login_required
from models.model import App, AppMode, AppModelConfig
from models.model import AppMode, AppModelConfig
from services.app_model_config_service import AppModelConfigService
@@ -52,7 +52,7 @@ class ModelConfigResource(Resource):
@edit_permission_required
@account_initialization_required
@get_app_model(mode=[AppMode.AGENT_CHAT, AppMode.CHAT, AppMode.COMPLETION])
def post(self, app_model: App):
def post(self, app_model):
"""Modify app model config"""
current_user, current_tenant_id = current_account_with_tenant()
# validate config
+2 -3
View File
@@ -20,7 +20,6 @@ from fields.base import ResponseModel
from libs.datetime_utils import naive_utc_now
from libs.login import current_account_with_tenant, login_required
from models import Site
from models.model import App
class AppSiteUpdatePayload(BaseModel):
@@ -85,7 +84,7 @@ class AppSite(Resource):
@edit_permission_required
@account_initialization_required
@get_app_model
def post(self, app_model: App):
def post(self, app_model):
args = AppSiteUpdatePayload.model_validate(console_ns.payload or {})
current_user, _ = current_account_with_tenant()
site = db.session.scalar(select(Site).where(Site.app_id == app_model.id).limit(1))
@@ -134,7 +133,7 @@ class AppSiteAccessTokenReset(Resource):
@is_admin_or_owner_required
@account_initialization_required
@get_app_model
def post(self, app_model: App):
def post(self, app_model):
current_user, _ = current_account_with_tenant()
site = db.session.scalar(select(Site).where(Site.app_id == app_model.id).limit(1))
+16 -49
View File
@@ -15,7 +15,6 @@ from libs.datetime_utils import parse_time_range
from libs.helper import convert_datetime_to_date
from libs.login import current_account_with_tenant, login_required
from models import AppMode
from models.model import App
class StatisticTimeRangeQuery(BaseModel):
@@ -48,7 +47,7 @@ class DailyMessageStatistic(Resource):
@setup_required
@login_required
@account_initialization_required
def get(self, app_model: App):
def get(self, app_model):
account, _ = current_account_with_tenant()
args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
@@ -62,12 +61,8 @@ FROM
WHERE
app_id = :app_id
AND invoke_from != :invoke_from"""
arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
assert account.timezone is not None
arg_dict: dict[str, object] = {
"tz": account.timezone,
"app_id": app_model.id,
"invoke_from": InvokeFrom.DEBUGGER,
}
try:
start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
@@ -109,7 +104,7 @@ class DailyConversationStatistic(Resource):
@setup_required
@login_required
@account_initialization_required
def get(self, app_model: App):
def get(self, app_model):
account, _ = current_account_with_tenant()
args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
@@ -123,12 +118,8 @@ FROM
WHERE
app_id = :app_id
AND invoke_from != :invoke_from"""
arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
assert account.timezone is not None
arg_dict: dict[str, object] = {
"tz": account.timezone,
"app_id": app_model.id,
"invoke_from": InvokeFrom.DEBUGGER,
}
try:
start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
@@ -169,7 +160,7 @@ class DailyTerminalsStatistic(Resource):
@setup_required
@login_required
@account_initialization_required
def get(self, app_model: App):
def get(self, app_model):
account, _ = current_account_with_tenant()
args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
@@ -183,12 +174,8 @@ FROM
WHERE
app_id = :app_id
AND invoke_from != :invoke_from"""
arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
assert account.timezone is not None
arg_dict: dict[str, object] = {
"tz": account.timezone,
"app_id": app_model.id,
"invoke_from": InvokeFrom.DEBUGGER,
}
try:
start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
@@ -230,7 +217,7 @@ class DailyTokenCostStatistic(Resource):
@setup_required
@login_required
@account_initialization_required
def get(self, app_model: App):
def get(self, app_model):
account, _ = current_account_with_tenant()
args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
@@ -245,12 +232,8 @@ FROM
WHERE
app_id = :app_id
AND invoke_from != :invoke_from"""
arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
assert account.timezone is not None
arg_dict: dict[str, object] = {
"tz": account.timezone,
"app_id": app_model.id,
"invoke_from": InvokeFrom.DEBUGGER,
}
try:
start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
@@ -294,7 +277,7 @@ class AverageSessionInteractionStatistic(Resource):
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT])
def get(self, app_model: App):
def get(self, app_model):
account, _ = current_account_with_tenant()
args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
@@ -316,12 +299,8 @@ FROM
WHERE
c.app_id = :app_id
AND m.invoke_from != :invoke_from"""
arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
assert account.timezone is not None
arg_dict: dict[str, object] = {
"tz": account.timezone,
"app_id": app_model.id,
"invoke_from": InvokeFrom.DEBUGGER,
}
try:
start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
@@ -374,7 +353,7 @@ class UserSatisfactionRateStatistic(Resource):
@setup_required
@login_required
@account_initialization_required
def get(self, app_model: App):
def get(self, app_model):
account, _ = current_account_with_tenant()
args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
@@ -392,12 +371,8 @@ LEFT JOIN
WHERE
m.app_id = :app_id
AND m.invoke_from != :invoke_from"""
arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
assert account.timezone is not None
arg_dict: dict[str, object] = {
"tz": account.timezone,
"app_id": app_model.id,
"invoke_from": InvokeFrom.DEBUGGER,
}
try:
start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
@@ -444,7 +419,7 @@ class AverageResponseTimeStatistic(Resource):
@login_required
@account_initialization_required
@get_app_model(mode=AppMode.COMPLETION)
def get(self, app_model: App):
def get(self, app_model):
account, _ = current_account_with_tenant()
args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
@@ -458,12 +433,8 @@ FROM
WHERE
app_id = :app_id
AND invoke_from != :invoke_from"""
arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
assert account.timezone is not None
arg_dict: dict[str, object] = {
"tz": account.timezone,
"app_id": app_model.id,
"invoke_from": InvokeFrom.DEBUGGER,
}
try:
start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
@@ -505,7 +476,7 @@ class TokensPerSecondStatistic(Resource):
@setup_required
@login_required
@account_initialization_required
def get(self, app_model: App):
def get(self, app_model):
account, _ = current_account_with_tenant()
args = StatisticTimeRangeQuery.model_validate(request.args.to_dict(flat=True))
@@ -521,12 +492,8 @@ FROM
WHERE
app_id = :app_id
AND invoke_from != :invoke_from"""
arg_dict = {"tz": account.timezone, "app_id": app_model.id, "invoke_from": InvokeFrom.DEBUGGER}
assert account.timezone is not None
arg_dict: dict[str, object] = {
"tz": account.timezone,
"app_id": app_model.id,
"invoke_from": InvokeFrom.DEBUGGER,
}
try:
start_datetime_utc, end_datetime_utc = parse_time_range(args.start, args.end, account.timezone)
@@ -1,415 +0,0 @@
"""Console REST endpoints for the Node Output Inspector (Stage 4 §8 / §10.3).
PRD §Node Output Inspector replaces the consumer-organized Variable Inspector
with a producer-organized view of each node's declared outputs and their
per-run status. This module exposes two parallel sets of three read-only
endpoints — one for ``/workflows/draft/runs/...`` (Composer test runs) and one
for ``/workflows/published/runs/...`` (real App API / webapp / webhook /
schedule / plugin triggers). Both sets share the same service code, the same
response shapes, and the same error codes; the URL is the *only* difference,
so the frontend can pick the right prefix based on which run-detail page the
user is on.
Decision D-1 (published Inspector deferred) was lifted 2026-05-26 — the
``published_run_inspector_not_implemented`` 404 code is therefore no longer
produced.
URLs follow the design doc and reuse the existing
``/apps/<uuid:app_id>/workflows/draft/...`` prefix from
:mod:`controllers.console.app.workflow_draft_variable`. The
``published`` prefix mirrors it shape-for-shape.
"""
from __future__ import annotations
import json
import logging
from collections.abc import Iterator
from uuid import UUID
from flask import Response
from flask_restx import Resource
from controllers.console import console_ns
from controllers.console.app.wraps import get_app_model
from controllers.console.wraps import account_initialization_required, setup_required
from libs.exception import BaseHTTPException
from libs.login import login_required
from models import App, AppMode
from services.workflow import inspector_events
from services.workflow.node_output_inspector_service import (
NodeOutputInspectorError,
NodeOutputInspectorService,
)
logger = logging.getLogger(__name__)
# Heartbeat cadence — every N empty subscribe ticks emit a SSE comment so
# intervening proxies (nginx, ingress) don't reap the idle connection.
# ``inspector_events.subscribe`` ticks at 1s, so 15 → 15s heartbeat.
_HEARTBEAT_EVERY_TICKS = 15
# Hard ceiling on a single stream — if we never see a terminal workflow
# event (engine crashed, redis dropped the message), force-close after this
# many ticks (= seconds).
_STREAM_HARD_TIMEOUT_TICKS = 1800 # 30 min
def _service() -> NodeOutputInspectorService:
"""One-line factory so tests can monkeypatch a stub if needed."""
return NodeOutputInspectorService()
def _serve_snapshot(app_model: App, run_id: UUID) -> dict:
"""Resource-body shared by draft + published snapshot endpoints.
Pulled out so the 6 REST routes don't duplicate the same 6-line try/except
+ ``model_dump`` ritual — the routes shrink to one-liners and the actual
behaviour lives here, where unit tests can hit it without spinning up
Flask request context.
"""
try:
snapshot = _service().snapshot_workflow_run(app_model=app_model, workflow_run_id=str(run_id))
except NodeOutputInspectorError as error:
raise _InspectorNotFound(error) from error
return snapshot.model_dump(mode="json")
def _serve_node_detail(app_model: App, run_id: UUID, node_id: str) -> dict:
"""Resource-body shared by draft + published node-detail endpoints."""
try:
view = _service().node_detail(
app_model=app_model,
workflow_run_id=str(run_id),
node_id=node_id,
)
except NodeOutputInspectorError as error:
raise _InspectorNotFound(error) from error
return view.model_dump(mode="json")
def _serve_output_preview(app_model: App, run_id: UUID, node_id: str, output_name: str) -> dict:
"""Resource-body shared by draft + published output-preview endpoints."""
try:
preview = _service().output_preview(
app_model=app_model,
workflow_run_id=str(run_id),
node_id=node_id,
output_name=output_name,
)
except NodeOutputInspectorError as error:
raise _InspectorNotFound(error) from error
return preview.model_dump(mode="json")
class _InspectorNotFound(BaseHTTPException):
"""404 that preserves the inspector's specific error code.
Without this the response body collapses to a generic ``not_found`` code
and clients lose the ability to distinguish, e.g.,
``workflow_run_not_found`` from ``published_run_inspector_not_implemented``.
"""
code = 404
def __init__(self, error: NodeOutputInspectorError) -> None:
self.error_code = error.code
super().__init__(description=str(error))
@console_ns.route("/apps/<uuid:app_id>/workflows/draft/runs/<uuid:run_id>/node-outputs")
class WorkflowDraftRunNodeOutputsApi(Resource):
"""Whole-run snapshot organized by producer node."""
@console_ns.doc("get_workflow_draft_run_node_outputs")
@console_ns.doc(description="Snapshot of every node's declared outputs for a draft workflow run.")
@console_ns.doc(params={"app_id": "Application ID", "run_id": "Workflow run ID"})
@console_ns.response(404, "Workflow run not found")
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
def get(self, app_model: App, run_id: UUID):
return _serve_snapshot(app_model, run_id)
@console_ns.route("/apps/<uuid:app_id>/workflows/draft/runs/<uuid:run_id>/node-outputs/<string:node_id>")
class WorkflowDraftRunNodeOutputDetailApi(Resource):
"""One node's declared outputs + per-output status."""
@console_ns.doc("get_workflow_draft_run_node_output_detail")
@console_ns.doc(description="One node's declared outputs for a draft workflow run.")
@console_ns.doc(
params={
"app_id": "Application ID",
"run_id": "Workflow run ID",
"node_id": "Node ID inside the workflow graph",
}
)
@console_ns.response(404, "Workflow run / node not found")
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
def get(self, app_model: App, run_id: UUID, node_id: str):
return _serve_node_detail(app_model, run_id, node_id)
@console_ns.route(
"/apps/<uuid:app_id>/workflows/draft/runs/<uuid:run_id>/node-outputs/<string:node_id>/<string:output_name>/preview"
)
class WorkflowDraftRunNodeOutputPreviewApi(Resource):
"""Full value for one declared output (with signed URL for file refs)."""
@console_ns.doc("get_workflow_draft_run_node_output_preview")
@console_ns.doc(description="Full value for one declared output, including signed download URL for files.")
@console_ns.doc(
params={
"app_id": "Application ID",
"run_id": "Workflow run ID",
"node_id": "Node ID inside the workflow graph",
"output_name": "Declared output name as exposed by Composer",
}
)
@console_ns.response(404, "Workflow run / node / output not found")
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
def get(self, app_model: App, run_id: UUID, node_id: str, output_name: str):
return _serve_output_preview(app_model, run_id, node_id, output_name)
# ──────────────────────────────────────────────────────────────────────────────
# SSE event stream — shared generator used by draft + published variants
# ──────────────────────────────────────────────────────────────────────────────
def _sse_envelope(event: str, data: dict | str, event_id: int) -> str:
"""Format one SSE record per D-5 ``{event, data, id}`` envelope.
``data`` is JSON-serialized when given as a dict; raw strings are
forwarded unchanged so we can also emit ``:keepalive`` comment lines.
"""
payload = data if isinstance(data, str) else json.dumps(data, ensure_ascii=False)
return f"event: {event}\nid: {event_id}\ndata: {payload}\n\n"
def _stream_inspector_events(app_model: App, run_id: UUID) -> Iterator[str]:
"""Yield SSE-framed strings for one workflow run.
The stream begins with a full ``snapshot`` event so the client has a
starting state without needing a separate REST GET. Then for every
``node_changed`` message from the pub/sub channel we re-read that node
from DB and push a fresh ``node_changed`` event. When the workflow run
reaches a terminal state we push one final ``workflow_run_completed``
event and close the stream.
Failures inside the loop are caught and surfaced as ``error`` events so
the frontend can show a banner rather than seeing the connection drop
silently. The Inspector never raises across the SSE boundary.
"""
service = _service()
run_id_str = str(run_id)
# Initial snapshot — also flushes a 404 back at the client right away
# if the run is gone (raised before yielding any bytes, so Flask turns it
# into the normal HTTP 404 path).
try:
snapshot = service.snapshot_workflow_run(app_model=app_model, workflow_run_id=run_id_str)
except NodeOutputInspectorError as error:
raise _InspectorNotFound(error) from error
event_id = 0
yield _sse_envelope("snapshot", snapshot.model_dump(mode="json"), event_id)
# If the run already finished by the time the client connected, emit
# the terminal envelope synchronously and close — no point subscribing.
# The enum value for partial success is the hyphenated ``partial-succeeded``
# (graphon.enums.WorkflowExecutionStatus), not ``partial_succeeded``.
if snapshot.workflow_run_status.value in {"succeeded", "failed", "stopped", "partial-succeeded"}:
event_id += 1
yield _sse_envelope(
"workflow_run_completed",
{"workflow_run_id": run_id_str, "workflow_run_status": snapshot.workflow_run_status.value},
event_id,
)
return
# Live subscription
ticks_since_heartbeat = 0
total_ticks = 0
for message in inspector_events.subscribe(run_id_str, timeout_seconds=1.0):
total_ticks += 1
if total_ticks > _STREAM_HARD_TIMEOUT_TICKS:
logger.warning(
"Inspector SSE: forcing close after %ds without terminal event for run %s",
_STREAM_HARD_TIMEOUT_TICKS,
run_id_str,
)
return
# Heartbeat sentinel — ``inspector_events.subscribe`` synthesizes a
# ``node_changed`` message with both fields ``None`` on every redis
# timeout. Real ``workflow_completed`` messages keep their kind even
# when status couldn't be resolved (publisher race), so checking kind
# first makes the heartbeat branch safe.
if message.kind == "node_changed" and message.node_id is None and message.status is None:
ticks_since_heartbeat += 1
if ticks_since_heartbeat >= _HEARTBEAT_EVERY_TICKS:
yield ":keepalive\n\n"
ticks_since_heartbeat = 0
continue
ticks_since_heartbeat = 0
if message.kind == "workflow_completed":
event_id += 1
yield _sse_envelope(
"workflow_run_completed",
{"workflow_run_id": run_id_str, "workflow_run_status": message.status or "unknown"},
event_id,
)
return
# node_changed: recompute the node slice from DB
if not message.node_id:
continue
try:
node_view = service.node_detail(
app_model=app_model,
workflow_run_id=run_id_str,
node_id=message.node_id,
)
except NodeOutputInspectorError:
# Node may not appear in the graph yet (race with persistence); skip.
continue
except Exception:
logger.warning(
"Inspector SSE: node_detail failed for run %s node %s",
run_id_str,
message.node_id,
exc_info=True,
)
event_id += 1
yield _sse_envelope(
"error",
{"node_id": message.node_id, "message": "failed to refresh node detail"},
event_id,
)
continue
event_id += 1
yield _sse_envelope("node_changed", node_view.model_dump(mode="json"), event_id)
@console_ns.route("/apps/<uuid:app_id>/workflows/draft/runs/<uuid:run_id>/node-outputs/events")
class WorkflowDraftRunNodeOutputEventsApi(Resource):
"""SSE stream of inspector deltas for a draft run."""
@console_ns.doc("stream_workflow_draft_run_node_output_events")
@console_ns.doc(description="Server-Sent Events stream of inspector deltas for a draft workflow run.")
@console_ns.doc(params={"app_id": "Application ID", "run_id": "Workflow run ID"})
@console_ns.response(404, "Workflow run not found")
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
def get(self, app_model: App, run_id: UUID):
return Response(
_stream_inspector_events(app_model, run_id),
mimetype="text/event-stream",
headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
)
# ──────────────────────────────────────────────────────────────────────────────
# Published-run endpoints — symmetric to the draft trio above
# ──────────────────────────────────────────────────────────────────────────────
@console_ns.route("/apps/<uuid:app_id>/workflows/published/runs/<uuid:run_id>/node-outputs")
class WorkflowPublishedRunNodeOutputsApi(Resource):
"""Whole-run snapshot for a *published* workflow run.
Same response shape as the ``/draft/`` variant — frontend can multiplex
based on which page (Composer test-run vs. Run History) is mounted.
"""
@console_ns.doc("get_workflow_published_run_node_outputs")
@console_ns.doc(description="Snapshot of every node's declared outputs for a published workflow run.")
@console_ns.doc(params={"app_id": "Application ID", "run_id": "Workflow run ID"})
@console_ns.response(404, "Workflow run not found")
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
def get(self, app_model: App, run_id: UUID):
return _serve_snapshot(app_model, run_id)
@console_ns.route("/apps/<uuid:app_id>/workflows/published/runs/<uuid:run_id>/node-outputs/<string:node_id>")
class WorkflowPublishedRunNodeOutputDetailApi(Resource):
"""One node's declared outputs + per-output status (published run)."""
@console_ns.doc("get_workflow_published_run_node_output_detail")
@console_ns.doc(description="One node's declared outputs for a published workflow run.")
@console_ns.doc(
params={
"app_id": "Application ID",
"run_id": "Workflow run ID",
"node_id": "Node ID inside the workflow graph",
}
)
@console_ns.response(404, "Workflow run / node not found")
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
def get(self, app_model: App, run_id: UUID, node_id: str):
return _serve_node_detail(app_model, run_id, node_id)
@console_ns.route(
"/apps/<uuid:app_id>/workflows/published/runs/<uuid:run_id>"
"/node-outputs/<string:node_id>/<string:output_name>/preview"
)
class WorkflowPublishedRunNodeOutputPreviewApi(Resource):
"""Full value for one declared output of a published run."""
@console_ns.doc("get_workflow_published_run_node_output_preview")
@console_ns.doc(description="Full value for one declared output of a published run.")
@console_ns.doc(
params={
"app_id": "Application ID",
"run_id": "Workflow run ID",
"node_id": "Node ID inside the workflow graph",
"output_name": "Declared output name as exposed by Composer",
}
)
@console_ns.response(404, "Workflow run / node / output not found")
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
def get(self, app_model: App, run_id: UUID, node_id: str, output_name: str):
return _serve_output_preview(app_model, run_id, node_id, output_name)
@console_ns.route("/apps/<uuid:app_id>/workflows/published/runs/<uuid:run_id>/node-outputs/events")
class WorkflowPublishedRunNodeOutputEventsApi(Resource):
"""SSE stream of inspector deltas for a published run."""
@console_ns.doc("stream_workflow_published_run_node_output_events")
@console_ns.doc(description="Server-Sent Events stream of inspector deltas for a published workflow run.")
@console_ns.doc(params={"app_id": "Application ID", "run_id": "Workflow run ID"})
@console_ns.response(404, "Workflow run not found")
@setup_required
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.ADVANCED_CHAT, AppMode.WORKFLOW])
def get(self, app_model: App, run_id: UUID):
return Response(
_stream_inspector_events(app_model, run_id),
mimetype="text/event-stream",
headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
)
@@ -11,7 +11,7 @@ from extensions.ext_database import db
from libs.datetime_utils import parse_time_range
from libs.login import current_account_with_tenant, login_required
from models.enums import WorkflowRunTriggeredFrom
from models.model import App, AppMode
from models.model import AppMode
from repositories.factory import DifyAPIRepositoryFactory
@@ -46,7 +46,7 @@ class WorkflowDailyRunsStatistic(Resource):
@setup_required
@login_required
@account_initialization_required
def get(self, app_model: App):
def get(self, app_model):
account, _ = current_account_with_tenant()
args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))
@@ -86,7 +86,7 @@ class WorkflowDailyTerminalsStatistic(Resource):
@setup_required
@login_required
@account_initialization_required
def get(self, app_model: App):
def get(self, app_model):
account, _ = current_account_with_tenant()
args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))
@@ -126,7 +126,7 @@ class WorkflowDailyTokenCostStatistic(Resource):
@setup_required
@login_required
@account_initialization_required
def get(self, app_model: App):
def get(self, app_model):
account, _ = current_account_with_tenant()
args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))
@@ -166,7 +166,7 @@ class WorkflowAverageAppInteractionStatistic(Resource):
@login_required
@account_initialization_required
@get_app_model(mode=[AppMode.WORKFLOW])
def get(self, app_model: App):
def get(self, app_model):
account, _ = current_account_with_tenant()
args = WorkflowStatisticQuery.model_validate(request.args.to_dict(flat=True))
@@ -5,12 +5,12 @@ from pydantic import BaseModel, Field
from controllers.common.schema import register_response_schema_models, register_schema_models
from fields.base import ResponseModel
from libs.login import login_required
from libs.login import current_account_with_tenant, login_required
from services.auth.api_key_auth_service import ApiKeyAuthService
from .. import console_ns
from ..auth.error import ApiKeyAuthFailedError
from ..wraps import account_initialization_required, is_admin_or_owner_required, setup_required, with_current_tenant_id
from ..wraps import account_initialization_required, is_admin_or_owner_required, setup_required
class ApiKeyAuthBindingPayload(BaseModel):
@@ -42,8 +42,8 @@ class ApiKeyAuthDataSource(Resource):
@setup_required
@login_required
@account_initialization_required
@with_current_tenant_id
def get(self, current_tenant_id: str):
def get(self):
_, current_tenant_id = current_account_with_tenant()
data_source_api_key_bindings = ApiKeyAuthService.get_provider_auth_list(current_tenant_id)
if data_source_api_key_bindings:
return {
@@ -69,9 +69,9 @@ class ApiKeyAuthDataSourceBinding(Resource):
@account_initialization_required
@is_admin_or_owner_required
@console_ns.expect(console_ns.models[ApiKeyAuthBindingPayload.__name__])
@with_current_tenant_id
def post(self, current_tenant_id: str):
def post(self):
# The role of the current user in the table must be admin or owner
_, current_tenant_id = current_account_with_tenant()
payload = ApiKeyAuthBindingPayload.model_validate(console_ns.payload)
data = payload.model_dump()
ApiKeyAuthService.validate_api_key_auth_args(data)
@@ -89,9 +89,10 @@ class ApiKeyAuthDataSourceBindingDelete(Resource):
@account_initialization_required
@is_admin_or_owner_required
@console_ns.response(204, "Binding deleted successfully")
@with_current_tenant_id
def delete(self, current_tenant_id: str, binding_id: UUID):
def delete(self, binding_id: UUID):
# The role of the current user in the table must be admin or owner
_, current_tenant_id = current_account_with_tenant()
ApiKeyAuthService.delete_provider_auth(current_tenant_id, str(binding_id))
return "", 204
+7 -5
View File
@@ -8,9 +8,9 @@ from flask_restx import Resource
from pydantic import BaseModel
from werkzeug.exceptions import BadRequest, NotFound
from controllers.console.wraps import account_initialization_required, setup_required, with_current_user
from controllers.console.wraps import account_initialization_required, setup_required
from graphon.model_runtime.utils.encoders import jsonable_encoder
from libs.login import login_required
from libs.login import current_account_with_tenant, login_required
from models import Account
from models.model import OAuthProviderApp
from services.oauth_server import OAUTH_ACCESS_TOKEN_EXPIRES_IN, OAuthGrantType, OAuthServerService
@@ -133,10 +133,12 @@ class OAuthServerUserAuthorizeApi(Resource):
@setup_required
@login_required
@account_initialization_required
@with_current_user
@oauth_server_client_id_required
def post(self, oauth_provider_app: OAuthProviderApp, current_user: Account):
user_account_id = current_user.id
def post(self, oauth_provider_app: OAuthProviderApp):
current_user, _ = current_account_with_tenant()
account = current_user
user_account_id = account.id
code = OAuthServerService.sign_oauth_authorization_code(oauth_provider_app.client_id, user_account_id)
return jsonable_encoder(
{
@@ -1,12 +1,15 @@
from __future__ import annotations
from datetime import datetime
from typing import Any
from uuid import UUID
from flask_restx import Resource
from pydantic import Field, field_validator
from controllers.common.schema import register_response_schema_models, register_schema_models
from fields.hit_testing_fields import HitTestingResponse
from libs.helper import dump_response
from controllers.common.schema import register_schema_models
from fields.base import ResponseModel
from libs.helper import to_timestamp
from libs.login import login_required
from .. import console_ns
@@ -17,8 +20,86 @@ from ..wraps import (
setup_required,
)
register_schema_models(console_ns, HitTestingPayload)
register_response_schema_models(console_ns, HitTestingResponse)
class HitTestingDocument(ResponseModel):
id: str | None = None
data_source_type: str | None = None
name: str | None = None
doc_type: str | None = None
doc_metadata: Any | None = None
class HitTestingSegment(ResponseModel):
id: str | None = None
position: int | None = None
document_id: str | None = None
content: str | None = None
sign_content: str | None = None
answer: str | None = None
word_count: int | None = None
tokens: int | None = None
keywords: list[str] = Field(default_factory=list)
index_node_id: str | None = None
index_node_hash: str | None = None
hit_count: int | None = None
enabled: bool | None = None
disabled_at: int | None = None
disabled_by: str | None = None
status: str | None = None
created_by: str | None = None
created_at: int | None = None
indexing_at: int | None = None
completed_at: int | None = None
error: str | None = None
stopped_at: int | None = None
document: HitTestingDocument | None = None
@field_validator("disabled_at", "created_at", "indexing_at", "completed_at", "stopped_at", mode="before")
@classmethod
def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
return to_timestamp(value)
class HitTestingChildChunk(ResponseModel):
id: str | None = None
content: str | None = None
position: int | None = None
score: float | None = None
class HitTestingFile(ResponseModel):
id: str | None = None
name: str | None = None
size: int | None = None
extension: str | None = None
mime_type: str | None = None
source_url: str | None = None
class HitTestingRecord(ResponseModel):
segment: HitTestingSegment | None = None
child_chunks: list[HitTestingChildChunk] = Field(default_factory=list)
score: float | None = None
tsne_position: Any | None = None
files: list[HitTestingFile] = Field(default_factory=list)
summary: str | None = None
class HitTestingResponse(ResponseModel):
query: str
records: list[HitTestingRecord] = Field(default_factory=list)
register_schema_models(
console_ns,
HitTestingPayload,
HitTestingDocument,
HitTestingSegment,
HitTestingChildChunk,
HitTestingFile,
HitTestingRecord,
HitTestingResponse,
)
@console_ns.route("/datasets/<uuid:dataset_id>/hit-testing")
@@ -38,11 +119,12 @@ class HitTestingApi(Resource, DatasetsHitTestingBase):
@login_required
@account_initialization_required
@cloud_edition_billing_rate_limit_check("knowledge")
def post(self, dataset_id: UUID) -> dict[str, object]:
def post(self, dataset_id: UUID):
dataset_id_str = str(dataset_id)
dataset = self.get_and_validate_dataset(dataset_id_str)
args = self.parse_args(console_ns.payload)
payload = HitTestingPayload.model_validate(console_ns.payload or {})
args = payload.model_dump(exclude_none=True)
self.hit_testing_args_check(args)
return dump_response(HitTestingResponse, self.perform_hit_testing(dataset, args))
return HitTestingResponse.model_validate(self.perform_hit_testing(dataset, args)).model_dump(mode="json")
@@ -1,6 +1,7 @@
import logging
from typing import Any, cast
from typing import Any
from flask_restx import marshal
from pydantic import BaseModel, Field
from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
@@ -18,10 +19,10 @@ from core.errors.error import (
ProviderTokenNotInitError,
QuotaExceededError,
)
from fields.hit_testing_fields import hit_testing_record_fields
from graphon.model_runtime.errors.invoke import InvokeError
from libs.login import current_user
from models.account import Account
from models.dataset import Dataset
from services.dataset_service import DatasetService
from services.entities.knowledge_entities.knowledge_entities import RetrievalModel
from services.hit_testing_service import HitTestingService
@@ -37,6 +38,16 @@ class HitTestingPayload(BaseModel):
class DatasetsHitTestingBase:
@staticmethod
def _extract_hit_testing_query(query: Any) -> str:
"""Return the query string from the service response shape."""
if isinstance(query, dict):
content = query.get("content")
if isinstance(content, str):
return content
raise ValueError("Invalid hit testing query response")
@staticmethod
def _prepare_hit_testing_records(records: Any) -> list[dict[str, Any]]:
"""Ensure collection fields match the API schema before response validation."""
@@ -52,7 +63,6 @@ class DatasetsHitTestingBase:
segment = normalized_record.get("segment")
if isinstance(segment, dict):
normalized_segment = dict(segment)
normalized_segment.setdefault("sign_content", None)
if normalized_segment.get("keywords") is None:
normalized_segment["keywords"] = []
normalized_record["segment"] = normalized_segment
@@ -63,15 +73,12 @@ class DatasetsHitTestingBase:
if normalized_record.get("files") is None:
normalized_record["files"] = []
normalized_record.setdefault("tsne_position", None)
normalized_record.setdefault("summary", None)
normalized_records.append(normalized_record)
return normalized_records
@staticmethod
def get_and_validate_dataset(dataset_id: str) -> Dataset:
def get_and_validate_dataset(dataset_id: str):
assert isinstance(current_user, Account)
dataset = DatasetService.get_dataset(dataset_id)
if dataset is None:
@@ -85,35 +92,33 @@ class DatasetsHitTestingBase:
return dataset
@staticmethod
def hit_testing_args_check(args: dict[str, Any]) -> None:
def hit_testing_args_check(args: dict[str, Any]):
HitTestingService.hit_testing_args_check(args)
@staticmethod
def parse_args(payload: dict[str, Any] | None) -> dict[str, Any]:
def parse_args(payload: dict[str, Any]) -> dict[str, Any]:
"""Validate and return hit-testing arguments from an incoming payload."""
hit_testing_payload = HitTestingPayload.model_validate(payload or {})
return hit_testing_payload.model_dump(exclude_none=True)
@staticmethod
def perform_hit_testing(dataset: Dataset, args: dict[str, Any]) -> dict[str, Any]:
def perform_hit_testing(dataset, args):
assert isinstance(current_user, Account)
try:
response = HitTestingService.retrieve(
dataset=dataset,
query=cast(str, args.get("query")),
query=args.get("query"),
account=current_user,
retrieval_model=args.get("retrieval_model"),
external_retrieval_model=cast(dict[str, Any], args.get("external_retrieval_model")),
external_retrieval_model=args.get("external_retrieval_model"),
attachment_ids=args.get("attachment_ids"),
limit=10,
)
query = response.get("query")
if not isinstance(query, dict) or not isinstance(query.get("content"), str):
raise ValueError("Invalid hit testing query response")
return {
"query": {"content": query["content"]},
"records": DatasetsHitTestingBase._prepare_hit_testing_records(response.get("records", [])),
"query": DatasetsHitTestingBase._extract_hit_testing_query(response.get("query")),
"records": DatasetsHitTestingBase._prepare_hit_testing_records(
marshal(response.get("records", []), hit_testing_record_fields)
),
}
except services.errors.index.IndexNotInitializedError:
raise DatasetNotInitializedError()
+2 -7
View File
@@ -20,7 +20,6 @@ from controllers.console.app.error import (
from controllers.console.explore.wraps import InstalledAppResource
from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
from graphon.model_runtime.errors.invoke import InvokeError
from models.model import InstalledApp
from services.audio_service import AudioService
from services.errors.audio import (
AudioTooLargeServiceError,
@@ -41,10 +40,8 @@ register_schema_model(console_ns, TextToAudioPayload)
endpoint="installed_app_audio",
)
class ChatAudioApi(InstalledAppResource):
def post(self, installed_app: InstalledApp):
def post(self, installed_app):
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
file = request.files["file"]
@@ -84,10 +81,8 @@ class ChatAudioApi(InstalledAppResource):
)
class ChatTextApi(InstalledAppResource):
@console_ns.expect(console_ns.models[TextToAudioPayload.__name__])
def post(self, installed_app: InstalledApp):
def post(self, installed_app):
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
try:
payload = TextToAudioPayload.model_validate(console_ns.payload or {})
+5 -13
View File
@@ -31,7 +31,7 @@ from libs import helper
from libs.datetime_utils import naive_utc_now
from libs.login import current_user
from models import Account
from models.model import AppMode, InstalledApp
from models.model import AppMode
from services.app_generate_service import AppGenerateService
from services.app_task_service import AppTaskService
from services.errors.llm import InvokeRateLimitError
@@ -83,10 +83,8 @@ register_response_schema_models(console_ns, SimpleResultResponse)
)
class CompletionApi(InstalledAppResource):
@console_ns.expect(console_ns.models[CompletionMessageExplorePayload.__name__])
def post(self, installed_app: InstalledApp):
def post(self, installed_app):
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
if app_model.mode != AppMode.COMPLETION:
raise NotCompletionAppError()
@@ -135,10 +133,8 @@ class CompletionApi(InstalledAppResource):
)
class CompletionStopApi(InstalledAppResource):
@console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
def post(self, installed_app: InstalledApp, task_id: str):
def post(self, installed_app, task_id: str):
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
if app_model.mode != AppMode.COMPLETION:
raise NotCompletionAppError()
@@ -161,10 +157,8 @@ class CompletionStopApi(InstalledAppResource):
)
class ChatApi(InstalledAppResource):
@console_ns.expect(console_ns.models[ChatMessagePayload.__name__])
def post(self, installed_app: InstalledApp):
def post(self, installed_app):
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
@@ -215,10 +209,8 @@ class ChatApi(InstalledAppResource):
)
class ChatStopApi(InstalledAppResource):
@console_ns.response(200, "Success", console_ns.models[SimpleResultResponse.__name__])
def post(self, installed_app: InstalledApp, task_id: str):
def post(self, installed_app, task_id: str):
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
@@ -8,7 +8,6 @@ from werkzeug.exceptions import NotFound
from controllers.common.controller_schemas import ConversationRenamePayload
from controllers.common.schema import register_response_schema_models, register_schema_models
from controllers.console.app.error import AppUnavailableError
from controllers.console.explore.error import NotChatAppError
from controllers.console.explore.wraps import InstalledAppResource
from core.app.entities.app_invoke_entities import InvokeFrom
@@ -21,7 +20,7 @@ from fields.conversation_fields import (
from libs.helper import UUIDStrOrEmpty
from libs.login import current_user
from models import Account
from models.model import AppMode, InstalledApp
from models.model import AppMode
from services.conversation_service import ConversationService
from services.errors.conversation import ConversationNotExistsError, LastConversationNotExistsError
from services.web_conversation_service import WebConversationService
@@ -45,10 +44,8 @@ register_response_schema_models(console_ns, ResultResponse)
)
class ConversationListApi(InstalledAppResource):
@console_ns.expect(console_ns.models[ConversationListQuery.__name__])
def get(self, installed_app: InstalledApp):
def get(self, installed_app):
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
@@ -95,10 +92,8 @@ class ConversationListApi(InstalledAppResource):
)
class ConversationApi(InstalledAppResource):
@console_ns.response(204, "Conversation deleted successfully")
def delete(self, installed_app: InstalledApp, c_id: UUID):
def delete(self, installed_app, c_id: UUID):
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
@@ -120,10 +115,8 @@ class ConversationApi(InstalledAppResource):
)
class ConversationRenameApi(InstalledAppResource):
@console_ns.expect(console_ns.models[ConversationRenamePayload.__name__])
def post(self, installed_app: InstalledApp, c_id: UUID):
def post(self, installed_app, c_id: UUID):
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
@@ -153,10 +146,8 @@ class ConversationRenameApi(InstalledAppResource):
)
class ConversationPinApi(InstalledAppResource):
@console_ns.response(200, "Success", console_ns.models[ResultResponse.__name__])
def patch(self, installed_app: InstalledApp, c_id: UUID):
def patch(self, installed_app, c_id: UUID):
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
@@ -179,10 +170,8 @@ class ConversationPinApi(InstalledAppResource):
)
class ConversationUnPinApi(InstalledAppResource):
@console_ns.response(200, "Success", console_ns.models[ResultResponse.__name__])
def patch(self, installed_app: InstalledApp, c_id: UUID):
def patch(self, installed_app, c_id: UUID):
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
@@ -262,7 +262,7 @@ class InstalledAppApi(InstalledAppResource):
"""
@console_ns.response(204, "App uninstalled successfully")
def delete(self, installed_app: InstalledApp):
def delete(self, installed_app):
_, current_tenant_id = current_account_with_tenant()
if installed_app.app_owner_tenant_id == current_tenant_id:
raise BadRequest("You can't uninstall an app owned by the current tenant")
@@ -273,7 +273,7 @@ class InstalledAppApi(InstalledAppResource):
return "", 204
@console_ns.response(200, "Success", console_ns.models[SimpleResultMessageResponse.__name__])
def patch(self, installed_app: InstalledApp):
def patch(self, installed_app):
payload = InstalledAppUpdatePayload.model_validate(console_ns.payload or {})
commit_args = False
+10 -20
View File
@@ -10,7 +10,6 @@ from controllers.common.controller_schemas import MessageFeedbackPayload, Messag
from controllers.common.schema import register_response_schema_models, register_schema_models
from controllers.console.app.error import (
AppMoreLikeThisDisabledError,
AppUnavailableError,
CompletionRequestError,
ProviderModelCurrentlyNotSupportError,
ProviderNotInitializeError,
@@ -22,16 +21,15 @@ from controllers.console.explore.error import (
NotCompletionAppError,
)
from controllers.console.explore.wraps import InstalledAppResource
from controllers.console.wraps import with_current_user
from core.app.entities.app_invoke_entities import InvokeFrom
from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotInitError, QuotaExceededError
from fields.conversation_fields import ResultResponse
from fields.message_fields import MessageInfiniteScrollPagination, MessageListItem, SuggestedQuestionsResponse
from graphon.model_runtime.errors.invoke import InvokeError
from libs import helper
from models import Account
from libs.login import current_account_with_tenant
from models.enums import FeedbackRating
from models.model import AppMode, InstalledApp
from models.model import AppMode
from services.app_generate_service import AppGenerateService
from services.errors.app import MoreLikeThisDisabledError
from services.errors.conversation import ConversationNotExistsError
@@ -61,11 +59,9 @@ register_response_schema_models(console_ns, ResultResponse, SuggestedQuestionsRe
)
class MessageListApi(InstalledAppResource):
@console_ns.expect(console_ns.models[MessageListQuery.__name__])
@with_current_user
def get(self, current_user: Account, installed_app: InstalledApp):
def get(self, installed_app):
current_user, _ = current_account_with_tenant()
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
@@ -100,11 +96,9 @@ class MessageListApi(InstalledAppResource):
class MessageFeedbackApi(InstalledAppResource):
@console_ns.expect(console_ns.models[MessageFeedbackPayload.__name__])
@console_ns.response(200, "Feedback submitted successfully", console_ns.models[ResultResponse.__name__])
@with_current_user
def post(self, current_user: Account, installed_app: InstalledApp, message_id: UUID):
def post(self, installed_app, message_id: UUID):
current_user, _ = current_account_with_tenant()
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
message_id_str = str(message_id)
@@ -130,11 +124,9 @@ class MessageFeedbackApi(InstalledAppResource):
)
class MessageMoreLikeThisApi(InstalledAppResource):
@console_ns.expect(console_ns.models[MoreLikeThisQuery.__name__])
@with_current_user
def get(self, current_user: Account, installed_app: InstalledApp, message_id: UUID):
def get(self, installed_app, message_id: UUID):
current_user, _ = current_account_with_tenant()
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
if app_model.mode != "completion":
raise NotCompletionAppError()
@@ -178,11 +170,9 @@ class MessageMoreLikeThisApi(InstalledAppResource):
)
class MessageSuggestedQuestionApi(InstalledAppResource):
@console_ns.response(200, "Success", console_ns.models[SuggestedQuestionsResponse.__name__])
@with_current_user
def get(self, current_user: Account, installed_app: InstalledApp, message_id: UUID):
def get(self, installed_app, message_id: UUID):
current_user, _ = current_account_with_tenant()
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
@@ -7,14 +7,12 @@ from werkzeug.exceptions import NotFound
from controllers.common.controller_schemas import SavedMessageCreatePayload, SavedMessageListQuery
from controllers.common.schema import register_response_schema_models, register_schema_models
from controllers.console import console_ns
from controllers.console.app.error import AppUnavailableError
from controllers.console.explore.error import NotCompletionAppError
from controllers.console.explore.wraps import InstalledAppResource
from controllers.console.wraps import with_current_user
from fields.conversation_fields import ResultResponse
from fields.message_fields import SavedMessageInfiniteScrollPagination, SavedMessageItem
from models import Account
from models.model import InstalledApp
from services.errors.message import MessageNotExistsError
from services.saved_message_service import SavedMessageService
@@ -26,10 +24,8 @@ register_response_schema_models(console_ns, ResultResponse)
class SavedMessageListApi(InstalledAppResource):
@console_ns.expect(console_ns.models[SavedMessageListQuery.__name__])
@with_current_user
def get(self, current_user: Account, installed_app: InstalledApp):
def get(self, current_user: Account, installed_app):
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
if app_model.mode != "completion":
raise NotCompletionAppError()
@@ -52,10 +48,8 @@ class SavedMessageListApi(InstalledAppResource):
@console_ns.expect(console_ns.models[SavedMessageCreatePayload.__name__])
@console_ns.response(200, "Success", console_ns.models[ResultResponse.__name__])
@with_current_user
def post(self, current_user: Account, installed_app: InstalledApp):
def post(self, current_user: Account, installed_app):
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
if app_model.mode != "completion":
raise NotCompletionAppError()
@@ -75,10 +69,8 @@ class SavedMessageListApi(InstalledAppResource):
class SavedMessageApi(InstalledAppResource):
@console_ns.response(204, "Saved message deleted successfully")
@with_current_user
def delete(self, current_user: Account, installed_app: InstalledApp, message_id: UUID):
def delete(self, current_user: Account, installed_app, message_id: UUID):
app_model = installed_app.app
if app_model is None:
raise AppUnavailableError()
message_id_str = str(message_id)
+14 -14
View File
@@ -9,14 +9,14 @@ from pydantic import BaseModel, Field, TypeAdapter, field_validator
from constants import HIDDEN_VALUE
from fields.base import ResponseModel
from libs.helper import to_timestamp
from libs.login import login_required
from libs.login import current_account_with_tenant, login_required
from models.api_based_extension import APIBasedExtension
from services.api_based_extension_service import APIBasedExtensionService
from services.code_based_extension_service import CodeBasedExtensionService
from ..common.schema import DEFAULT_REF_TEMPLATE_SWAGGER_2_0, register_schema_models
from . import console_ns
from .wraps import account_initialization_required, setup_required, with_current_tenant_id
from .wraps import account_initialization_required, setup_required
class CodeBasedExtensionQuery(BaseModel):
@@ -116,11 +116,11 @@ class APIBasedExtensionAPI(Resource):
@setup_required
@login_required
@account_initialization_required
@with_current_tenant_id
def get(self, current_tenant_id: str):
def get(self):
_, tenant_id = current_account_with_tenant()
return [
_serialize_api_based_extension(extension)
for extension in APIBasedExtensionService.get_all_by_tenant_id(current_tenant_id)
for extension in APIBasedExtensionService.get_all_by_tenant_id(tenant_id)
]
@console_ns.doc("create_api_based_extension")
@@ -130,9 +130,9 @@ class APIBasedExtensionAPI(Resource):
@setup_required
@login_required
@account_initialization_required
@with_current_tenant_id
def post(self, current_tenant_id: str):
def post(self):
payload = APIBasedExtensionPayload.model_validate(console_ns.payload or {})
_, current_tenant_id = current_account_with_tenant()
extension_data = APIBasedExtension(
tenant_id=current_tenant_id,
@@ -153,12 +153,12 @@ class APIBasedExtensionDetailAPI(Resource):
@setup_required
@login_required
@account_initialization_required
@with_current_tenant_id
def get(self, current_tenant_id: str, id: UUID):
def get(self, id: UUID):
api_based_extension_id = str(id)
_, tenant_id = current_account_with_tenant()
return _serialize_api_based_extension(
APIBasedExtensionService.get_with_tenant_id(current_tenant_id, api_based_extension_id)
APIBasedExtensionService.get_with_tenant_id(tenant_id, api_based_extension_id)
)
@console_ns.doc("update_api_based_extension")
@@ -169,9 +169,9 @@ class APIBasedExtensionDetailAPI(Resource):
@setup_required
@login_required
@account_initialization_required
@with_current_tenant_id
def post(self, current_tenant_id: str, id: UUID):
def post(self, id: UUID):
api_based_extension_id = str(id)
_, current_tenant_id = current_account_with_tenant()
extension_data_from_db = APIBasedExtensionService.get_with_tenant_id(current_tenant_id, api_based_extension_id)
@@ -197,9 +197,9 @@ class APIBasedExtensionDetailAPI(Resource):
@setup_required
@login_required
@account_initialization_required
@with_current_tenant_id
def delete(self, current_tenant_id: str, id: UUID):
def delete(self, id: UUID):
api_based_extension_id = str(id)
_, current_tenant_id = current_account_with_tenant()
extension_data_from_db = APIBasedExtensionService.get_with_tenant_id(current_tenant_id, api_based_extension_id)
+8 -6
View File
@@ -2,11 +2,11 @@ from flask_restx import Resource
from werkzeug.exceptions import Unauthorized
from controllers.common.schema import register_response_schema_models
from libs.login import current_user, login_required
from libs.login import current_account_with_tenant, current_user, login_required
from services.feature_service import FeatureModel, FeatureService, LimitationModel, SystemFeatureModel
from . import console_ns
from .wraps import account_initialization_required, cloud_utm_record, setup_required, with_current_tenant_id
from .wraps import account_initialization_required, cloud_utm_record, setup_required
register_response_schema_models(console_ns, FeatureModel, LimitationModel, SystemFeatureModel)
@@ -24,9 +24,10 @@ class FeatureApi(Resource):
@login_required
@account_initialization_required
@cloud_utm_record
@with_current_tenant_id
def get(self, current_tenant_id: str):
def get(self):
"""Get feature configuration for current tenant"""
_, current_tenant_id = current_account_with_tenant()
payload = FeatureService.get_features(
current_tenant_id,
exclude_vector_space=True,
@@ -48,9 +49,10 @@ class FeatureVectorSpaceApi(Resource):
@login_required
@account_initialization_required
@cloud_utm_record
@with_current_tenant_id
def get(self, current_tenant_id: str):
def get(self):
"""Get vector-space usage and limit for current tenant"""
_, current_tenant_id = current_account_with_tenant()
return FeatureService.get_vector_space(current_tenant_id).model_dump()
+6 -9
View File
@@ -22,13 +22,10 @@ from controllers.console.wraps import (
account_initialization_required,
cloud_edition_billing_resource_check,
setup_required,
with_current_tenant_id,
with_current_user,
)
from extensions.ext_database import db
from fields.file_fields import FileResponse, UploadConfig
from libs.login import login_required
from models import Account
from libs.login import current_account_with_tenant, login_required
from services.file_service import FileService
from . import console_ns
@@ -65,8 +62,8 @@ class FileApi(Resource):
@account_initialization_required
@cloud_edition_billing_resource_check("documents")
@console_ns.response(201, "File uploaded successfully", console_ns.models[FileResponse.__name__])
@with_current_user
def post(self, current_user: Account):
def post(self):
current_user, _ = current_account_with_tenant()
source_str = request.form.get("source")
source: Literal["datasets"] | None = "datasets" if source_str == "datasets" else None
@@ -110,10 +107,10 @@ class FilePreviewApi(Resource):
@login_required
@account_initialization_required
@console_ns.response(200, "Success", console_ns.models[TextContentResponse.__name__])
@with_current_tenant_id
def get(self, current_tenant_id: str, file_id: UUID):
def get(self, file_id: UUID):
file_id_str = str(file_id)
text = FileService(db.engine).get_file_preview(file_id_str, current_tenant_id)
_, tenant_id = current_account_with_tenant()
text = FileService(db.engine).get_file_preview(file_id_str, tenant_id)
return {"content": text}
+4 -6
View File
@@ -12,13 +12,11 @@ from controllers.common.errors import (
)
from controllers.common.schema import register_response_schema_models, register_schema_models
from controllers.console import console_ns
from controllers.console.wraps import with_current_user
from core.helper import ssrf_proxy
from extensions.ext_database import db
from fields.file_fields import FileWithSignedUrl, RemoteFileInfo
from graphon.file import helpers as file_helpers
from libs.login import login_required
from models import Account
from libs.login import current_account_with_tenant, login_required
from services.file_service import FileService
@@ -51,8 +49,7 @@ class RemoteFileUpload(Resource):
@console_ns.expect(console_ns.models[RemoteFileUploadPayload.__name__])
@console_ns.response(201, "File uploaded successfully", console_ns.models[FileWithSignedUrl.__name__])
@login_required
@with_current_user
def post(self, current_user: Account):
def post(self):
payload = RemoteFileUploadPayload.model_validate(console_ns.payload)
url = payload.url
@@ -77,11 +74,12 @@ class RemoteFileUpload(Resource):
content = resp.content if resp.request.method == "GET" else ssrf_proxy.get(url).content
try:
user, _ = current_account_with_tenant()
upload_file = FileService(db.engine).upload_file(
filename=file_info.filename,
content=content,
mimetype=file_info.mimetype,
user=current_user,
user=user,
source_url=url,
)
except services.errors.file.FileTooLargeError as file_too_large_error:
+19 -27
View File
@@ -9,16 +9,9 @@ from werkzeug.exceptions import Forbidden
from controllers.common.fields import SimpleResultResponse
from controllers.common.schema import register_response_schema_models, register_schema_models
from controllers.console import console_ns
from controllers.console.wraps import (
account_initialization_required,
edit_permission_required,
setup_required,
with_current_tenant_id,
with_current_user,
)
from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
from fields.base import ResponseModel
from libs.login import login_required
from models import Account
from libs.login import current_account_with_tenant, login_required
from models.enums import TagType
from services.tag_service import (
SaveTagPayload,
@@ -99,8 +92,8 @@ class TagListApi(Resource):
params={"type": 'Tag type filter. Can be "knowledge" or "app".', "keyword": "Search keyword for tag name."}
)
@console_ns.doc(responses={200: ("Success", [console_ns.models[TagResponse.__name__]])})
@with_current_tenant_id
def get(self, current_tenant_id: str):
def get(self):
_, current_tenant_id = current_account_with_tenant()
raw_args = request.args.to_dict()
param = TagListQueryParam.model_validate(raw_args)
tags = TagService.get_tags(param.type, current_tenant_id, param.keyword)
@@ -116,9 +109,9 @@ class TagListApi(Resource):
@setup_required
@login_required
@account_initialization_required
@with_current_user
def post(self, current_user: Account):
# Allow users with edit permission, or dataset editors (including dataset operators).
def post(self):
current_user, _ = current_account_with_tenant()
# The role of the current user in the ta table must be admin, owner, or editor
if not (current_user.has_edit_permission or current_user.is_dataset_editor):
raise Forbidden()
@@ -139,8 +132,8 @@ class TagUpdateDeleteApi(Resource):
@setup_required
@login_required
@account_initialization_required
@with_current_user
def patch(self, current_user: Account, tag_id: UUID):
def patch(self, tag_id: UUID):
current_user, _ = current_account_with_tenant()
tag_id_str = str(tag_id)
# The role of the current user in the ta table must be admin, owner, or editor
if not (current_user.has_edit_permission or current_user.is_dataset_editor):
@@ -170,19 +163,20 @@ class TagUpdateDeleteApi(Resource):
return "", 204
def _require_tag_binding_edit_permission(current_user: Account) -> None:
def _require_tag_binding_edit_permission() -> None:
"""
Ensure the current account can edit tag bindings.
Tag binding operations are allowed for users who can edit resources (app/dataset) within the current tenant.
"""
current_user, _ = current_account_with_tenant()
# The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
if not (current_user.has_edit_permission or current_user.is_dataset_editor):
raise Forbidden()
def _create_tag_bindings(current_user: Account) -> tuple[dict[str, str], int]:
_require_tag_binding_edit_permission(current_user)
def _create_tag_bindings() -> tuple[dict[str, str], int]:
_require_tag_binding_edit_permission()
payload = TagBindingPayload.model_validate(console_ns.payload or {})
TagService.save_tag_binding(
@@ -195,8 +189,8 @@ def _create_tag_bindings(current_user: Account) -> tuple[dict[str, str], int]:
return {"result": "success"}, 200
def _remove_tag_bindings(current_user: Account) -> tuple[dict[str, str], int]:
_require_tag_binding_edit_permission(current_user)
def _remove_tag_bindings() -> tuple[dict[str, str], int]:
_require_tag_binding_edit_permission()
payload = TagBindingRemovePayload.model_validate(console_ns.payload or {})
TagService.delete_tag_binding(
@@ -219,9 +213,8 @@ class TagBindingCollectionApi(Resource):
@setup_required
@login_required
@account_initialization_required
@with_current_user
def post(self, current_user: Account):
return _create_tag_bindings(current_user)
def post(self):
return _create_tag_bindings()
@console_ns.route("/tag-bindings/remove")
@@ -235,6 +228,5 @@ class TagBindingRemoveApi(Resource):
@setup_required
@login_required
@account_initialization_required
@with_current_user
def post(self, current_user: Account):
return _remove_tag_bindings(current_user)
def post(self):
return _remove_tag_bindings()
+2 -2
View File
@@ -77,7 +77,7 @@ register_response_schema_models(console_ns, SimpleResultDataResponse, Verificati
def _is_role_enabled(role: TenantAccountRole | str, tenant_id: str) -> bool:
if role != TenantAccountRole.DATASET_OPERATOR:
return True
return FeatureService.get_features(tenant_id=tenant_id, exclude_vector_space=True).dataset_operator_enabled
return FeatureService.get_features(tenant_id=tenant_id).dataset_operator_enabled
def _normalize_invitee_emails(emails: list[str]) -> list[str]:
@@ -113,7 +113,7 @@ def _check_member_invite_limits(tenant_id: str, new_member_count: int) -> None:
if new_member_count <= 0:
return
features = FeatureService.get_features(tenant_id=tenant_id, exclude_vector_space=True)
features = FeatureService.get_features(tenant_id=tenant_id)
if dify_config.ENTERPRISE_ENABLED:
workspace_members = features.workspace_members
@@ -166,10 +166,10 @@ class TenantListApi(Resource):
if tenant_plan:
plan = tenant_plan["plan"] or CloudPlan.SANDBOX
else:
features = FeatureService.get_features(tenant.id, exclude_vector_space=True)
features = FeatureService.get_features(tenant.id)
plan = features.billing.subscription.plan or CloudPlan.SANDBOX
elif not is_enterprise_only:
features = FeatureService.get_features(tenant.id, exclude_vector_space=True)
features = FeatureService.get_features(tenant.id)
plan = features.billing.subscription.plan or CloudPlan.SANDBOX
# Create a dictionary with tenant attributes
+8 -15
View File
@@ -96,28 +96,21 @@ def cloud_edition_billing_resource_check[**P, R](resource: str) -> Callable[[Cal
@wraps(view)
def decorated(*args: P.args, **kwargs: P.kwargs):
_, current_tenant_id = current_account_with_tenant()
if resource == "vector_space":
if not dify_config.BILLING_ENABLED:
return view(*args, **kwargs)
vector_space = FeatureService.get_vector_space(current_tenant_id)
if 0 < vector_space.limit <= vector_space.size:
abort(
403,
"The capacity of the knowledge storage space has reached the limit of your subscription.",
)
return view(*args, **kwargs)
features = FeatureService.get_features(current_tenant_id, exclude_vector_space=True)
features = FeatureService.get_features(current_tenant_id)
if features.billing.enabled:
members = features.members
apps = features.apps
vector_space = features.vector_space
documents_upload_quota = features.documents_upload_quota
annotation_quota_limit = features.annotation_quota_limit
if resource == "members" and 0 < members.limit <= members.size:
abort(403, "The number of members has reached the limit of your subscription.")
elif resource == "apps" and 0 < apps.limit <= apps.size:
abort(403, "The number of apps has reached the limit of your subscription.")
elif resource == "vector_space" and 0 < vector_space.limit <= vector_space.size:
abort(
403, "The capacity of the knowledge storage space has reached the limit of your subscription."
)
elif resource == "documents" and 0 < documents_upload_quota.limit <= documents_upload_quota.size:
# The api of file upload is used in the multiple places,
# so we need to check the source of the request from datasets
@@ -147,7 +140,7 @@ def cloud_edition_billing_knowledge_limit_check[**P, R](
@wraps(view)
def decorated(*args: P.args, **kwargs: P.kwargs):
_, current_tenant_id = current_account_with_tenant()
features = FeatureService.get_features(current_tenant_id, exclude_vector_space=True)
features = FeatureService.get_features(current_tenant_id)
if features.billing.enabled:
if resource == "add_segment":
if features.billing.subscription.plan == CloudPlan.SANDBOX:
@@ -298,7 +291,7 @@ def knowledge_pipeline_publish_enabled[**P, R](view: Callable[P, R]) -> Callable
@wraps(view)
def decorated(*args: P.args, **kwargs: P.kwargs):
_, current_tenant_id = current_account_with_tenant()
features = FeatureService.get_features(current_tenant_id, exclude_vector_space=True)
features = FeatureService.get_features(current_tenant_id)
if features.knowledge_pipeline.publish_enabled:
return view(*args, **kwargs)
abort(403)
-14
View File
@@ -37,13 +37,6 @@ from controllers.openapi._models import (
DeviceMutateRequest,
DeviceMutateResponse,
DevicePollRequest,
MemberActionResponse,
MemberInvitePayload,
MemberInviteResponse,
MemberListQuery,
MemberListResponse,
MemberResponse,
MemberRoleUpdatePayload,
MessageMetadata,
PermittedExternalAppsListQuery,
PermittedExternalAppsListResponse,
@@ -70,9 +63,6 @@ register_schema_models(
DevicePollRequest,
DeviceLookupQuery,
DeviceMutateRequest,
MemberInvitePayload,
MemberListQuery,
MemberRoleUpdatePayload,
PermittedExternalAppsListQuery,
)
register_response_schema_models(
@@ -96,10 +86,6 @@ register_response_schema_models(
WorkspaceSummaryResponse,
WorkspaceListResponse,
WorkspaceDetailResponse,
MemberResponse,
MemberListResponse,
MemberInviteResponse,
MemberActionResponse,
DeviceCodeResponse,
DeviceLookupResponse,
DeviceMutateResponse,
+1 -59
View File
@@ -6,7 +6,7 @@ from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator
from libs.helper import EmailStr, UUIDStrOrEmpty, uuid_value
from libs.helper import UUIDStrOrEmpty, uuid_value
from models.model import AppMode
# Server-side cap on `limit` query param for /openapi/v1/* list endpoints.
@@ -342,61 +342,3 @@ class ApprovalGrantClaimsPayload(BaseModel):
user_code: str = Field(min_length=1, max_length=32)
nonce: str = Field(min_length=1, max_length=128)
csrf_token: str = Field(min_length=1, max_length=128)
# Closed enum for invite/update-role payloads. Owner is intentionally not
# assignable through these endpoints — ownership transfer goes through the
# console's three-step email-verification flow.
MemberAssignableRole = Literal["normal", "admin"]
class MemberResponse(BaseModel):
id: str
name: str
email: str
role: str
status: str
avatar: str | None = None
class MemberListResponse(BaseModel):
page: int
limit: int
total: int
has_more: bool
data: list[MemberResponse]
class MemberListQuery(BaseModel):
"""Strict (extra='forbid')."""
model_config = ConfigDict(extra="forbid")
page: int = Field(1, ge=1)
limit: int = Field(20, ge=1, le=MAX_PAGE_LIMIT)
class MemberInvitePayload(BaseModel):
model_config = ConfigDict(extra="forbid")
email: EmailStr
role: MemberAssignableRole
class MemberRoleUpdatePayload(BaseModel):
model_config = ConfigDict(extra="forbid")
role: MemberAssignableRole
class MemberInviteResponse(BaseModel):
result: Literal["success"] = "success"
email: str
role: str
member_id: str
invite_url: str
tenant_id: str
class MemberActionResponse(BaseModel):
result: Literal["success"] = "success"
-77
View File
@@ -1,77 +0,0 @@
"""Workspace role gate.
Layered on top of `validate_bearer` + `accept_subjects(SubjectType.ACCOUNT)`
for routes whose access depends on the caller's `TenantAccountJoin.role`
in the workspace named by the `workspace_id` path parameter.
Usage::
@openapi_ns.route("/workspaces/<string:workspace_id>/members")
class Members(Resource):
@validate_bearer(accept=ACCEPT_USER_ANY)
@accept_subjects(SubjectType.ACCOUNT)
@require_workspace_role() # any member
def get(self, workspace_id: str): ...
@validate_bearer(accept=ACCEPT_USER_ANY)
@accept_subjects(SubjectType.ACCOUNT)
@require_workspace_role(TenantAccountRole.OWNER, TenantAccountRole.ADMIN)
def post(self, workspace_id: str): ...
Non-member callers get 404 (matching `GET /openapi/v1/workspaces/<id>`)
so workspace IDs do not leak across tenants. A member without one of the
allowed roles gets 403.
"""
from __future__ import annotations
from collections.abc import Callable
from functools import wraps
from typing import TypeVar
from werkzeug.exceptions import Forbidden, NotFound
from extensions.ext_database import db
from libs.oauth_bearer import try_get_auth_ctx
from models.account import TenantAccountRole
from services.account_service import TenantService
F = TypeVar("F", bound=Callable[..., object])
def require_workspace_role(*allowed_roles: TenantAccountRole) -> Callable[[F], F]:
"""Gate a route on the caller's role in ``workspace_id``.
Pass no roles to require only membership. Pass one or more roles to
require the caller's role be in that set.
"""
allowed = frozenset(allowed_roles)
def deco(fn: F) -> F:
@wraps(fn)
def wrapper(*args: object, **kwargs: object) -> object:
ctx = try_get_auth_ctx()
if ctx is None or ctx.account_id is None:
raise RuntimeError(
"require_workspace_role called without account-bearer context; "
"stack validate_bearer + accept_subjects(SubjectType.ACCOUNT) above it"
)
workspace_id = kwargs.get("workspace_id")
if not workspace_id:
raise RuntimeError("require_workspace_role expects a 'workspace_id' route parameter")
role = TenantService.get_account_role_in_tenant(db.session, str(ctx.account_id), str(workspace_id))
if role is None:
raise NotFound("workspace not found")
if allowed and role not in allowed:
raise Forbidden("insufficient workspace role")
return fn(*args, **kwargs)
return wrapper # type: ignore[return-value]
return deco
+9 -308
View File
@@ -1,41 +1,20 @@
"""User-scoped workspace reads and member management under /openapi/v1/workspaces.
"""User-scoped workspace reads under /openapi/v1/workspaces. Bearer-authed
counterparts to the cookie-authed /console/api/workspaces endpoints.
Bearer-authed counterparts to the cookie-authed /console/api/workspaces
endpoints. Account bearers (dfoa_) see every tenant they're a member of.
External SSO bearers (dfoe_) have no account_id and so see an empty list —
that matches /openapi/v1/account.
Member-management endpoints are gated by both `accept_subjects` (SSO out)
and `require_workspace_role` (membership / role lookup against the path's
``workspace_id``).
Account bearers (dfoa_) see every tenant they're a member of. External
SSO bearers (dfoe_) have no account_id and so see an empty list — that
matches /openapi/v1/account.
"""
from __future__ import annotations
from itertools import starmap
from urllib import parse
from flask import jsonify, make_response, request
from flask_restx import Resource
from pydantic import BaseModel, ValidationError
from werkzeug.exceptions import BadRequest, Forbidden, NotFound
from werkzeug.exceptions import NotFound
from configs import dify_config
from controllers.common.schema import query_params_from_model
from controllers.openapi import openapi_ns
from controllers.openapi._models import (
MemberActionResponse,
MemberInvitePayload,
MemberInviteResponse,
MemberListQuery,
MemberListResponse,
MemberResponse,
MemberRoleUpdatePayload,
WorkspaceDetailResponse,
WorkspaceListResponse,
WorkspaceSummaryResponse,
)
from controllers.openapi.auth.role_gate import require_workspace_role
from controllers.openapi._models import WorkspaceDetailResponse, WorkspaceListResponse, WorkspaceSummaryResponse
from controllers.openapi.auth.surface_gate import accept_subjects
from extensions.ext_database import db
from libs.oauth_bearer import (
@@ -44,105 +23,8 @@ from libs.oauth_bearer import (
get_auth_ctx,
validate_bearer,
)
from models import Account, Tenant, TenantAccountJoin
from models.account import TenantAccountRole, TenantStatus
from services.account_service import AccountService, RegisterService, TenantService
from services.errors.account import (
AccountAlreadyInTenantError,
AccountNotLinkTenantError,
AccountRegisterError,
CannotOperateSelfError,
MemberNotInTenantError,
NoPermissionError,
RoleAlreadyAssignedError,
)
from services.feature_service import FeatureService
def _validate_body[M: BaseModel](model: type[M]) -> M:
"""Validate JSON body against ``model``. Validation errors → HTTP 400.
The workspace spec is explicit that bad email / unknown role payloads
are 400, not Pydantic's default 422 — handle uniformly here.
"""
body = request.get_json(silent=True) or {}
try:
return model.model_validate(body)
except ValidationError as exc:
raise BadRequest(str(exc))
def _member_response(account: Account) -> MemberResponse:
return MemberResponse(
id=str(account.id),
name=account.name,
email=account.email,
role=account.role.value if account.role else "",
status=account.status.value if account.status else "",
avatar=account.avatar,
)
def _load_tenant(workspace_id: str) -> Tenant:
tenant = TenantService.get_tenant_by_id(db.session, workspace_id)
if tenant is None or tenant.status != TenantStatus.NORMAL:
raise NotFound("workspace not found")
return tenant
def _load_account(account_id: object) -> Account:
"""Load the caller's Account. Missing == auth wiring bug, not user error."""
account = AccountService.get_account_by_id(db.session, str(account_id)) if account_id else None
if account is None:
raise RuntimeError("authenticated account_id has no Account row")
return account
def _quota_error(*, code: str, message: str, hint: str) -> Forbidden:
"""Build a 403 with envelope ``{code, message, hint}``.
CLI ``error-mapper`` reads ``message`` and ``hint`` off the wire body
verbatim — the structured envelope lets it surface remediation guidance
(e.g. "upgrade your plan") without the CLI needing to know edition
semantics.
"""
err = Forbidden(message)
err.response = make_response(
jsonify({"code": code, "message": message, "hint": hint}),
403,
)
return err
def _check_member_invite_quota(tenant_id: str) -> None:
"""Edition-aware member-count gate for invite.
Both branches self-disable on CE because ``FeatureService.get_features``
leaves ``billing.enabled`` and ``workspace_members.enabled`` False by
default; SaaS billing API and EE license activation are what flip them on.
Mirrors the two checks the console invite path performs (decorator at
``console/wraps.py:106`` for billing + inline at
``console/workspace/members.py:130`` for license).
"""
features = FeatureService.get_features(tenant_id)
if features.billing.enabled:
members = features.members
if 0 < members.limit <= members.size:
raise _quota_error(
code="members.limit_exceeded",
message="Subscription member limit reached.",
hint="Upgrade your plan to invite more members or remove an existing member first.",
)
if features.workspace_members.enabled:
if not features.workspace_members.is_available(1):
raise _quota_error(
code="workspace_members.license_exceeded",
message="Workspace member license capacity reached.",
hint="Contact your workspace administrator to expand the license seat count.",
)
from models import Tenant, TenantAccountJoin
from services.account_service import TenantService
@openapi_ns.route("/workspaces")
@@ -175,187 +57,6 @@ class WorkspaceByIdApi(Resource):
return _workspace_detail(tenant, membership).model_dump(mode="json"), 200
@openapi_ns.route("/workspaces/<string:workspace_id>/switch")
class WorkspaceSwitchApi(Resource):
"""Server-side switch — equivalent to the console's POST /workspaces/switch.
CLI `difyctl use workspace <id>` calls this; it does NOT mutate
``hosts.yml`` on its own. Failure here must abort the local write so
that ``hosts.yml`` never diverges from the server's ``current`` state.
"""
@openapi_ns.response(200, "Workspace detail", openapi_ns.models[WorkspaceDetailResponse.__name__])
@validate_bearer(accept=ACCEPT_USER_ANY)
@accept_subjects(SubjectType.ACCOUNT)
@require_workspace_role()
def post(self, workspace_id: str):
ctx = get_auth_ctx()
account = _load_account(ctx.account_id)
try:
TenantService.switch_tenant(account, workspace_id)
except AccountNotLinkTenantError:
# Membership existed at gate time but Tenant.status != NORMAL or
# the row was just removed — treat as not-found.
raise NotFound("workspace not found")
row = TenantService.find_workspace_for_account(db.session, str(ctx.account_id), workspace_id)
if row is None:
raise NotFound("workspace not found")
tenant, membership = row
return _workspace_detail(tenant, membership).model_dump(mode="json"), 200
@openapi_ns.route("/workspaces/<string:workspace_id>/members")
class WorkspaceMembersApi(Resource):
"""List + invite members.
GET is any-member. POST requires admin/owner — owner can never be
assigned through invite (ownership transfer is console-only).
"""
@openapi_ns.doc(params=query_params_from_model(MemberListQuery))
@openapi_ns.response(200, "Member list", openapi_ns.models[MemberListResponse.__name__])
@validate_bearer(accept=ACCEPT_USER_ANY)
@accept_subjects(SubjectType.ACCOUNT)
@require_workspace_role()
def get(self, workspace_id: str):
try:
query = MemberListQuery.model_validate(request.args.to_dict(flat=True))
except ValidationError as exc:
raise BadRequest(str(exc))
tenant = _load_tenant(workspace_id)
# Members per workspace are bounded by SaaS plan caps (≤50) or EE
# license seats (low thousands worst-case), so we materialize and
# slice in-memory rather than push pagination into the service —
# matches how the rest of the service exposes member lists.
members = TenantService.get_tenant_members(tenant)
total = len(members)
start = (query.page - 1) * query.limit
page_items = members[start : start + query.limit]
return MemberListResponse(
page=query.page,
limit=query.limit,
total=total,
has_more=query.page * query.limit < total,
data=[_member_response(m) for m in page_items],
).model_dump(mode="json"), 200
@openapi_ns.expect(openapi_ns.models[MemberInvitePayload.__name__])
@openapi_ns.response(201, "Member invited", openapi_ns.models[MemberInviteResponse.__name__])
@validate_bearer(accept=ACCEPT_USER_ANY)
@accept_subjects(SubjectType.ACCOUNT)
@require_workspace_role(TenantAccountRole.OWNER, TenantAccountRole.ADMIN)
def post(self, workspace_id: str):
payload = _validate_body(MemberInvitePayload)
ctx = get_auth_ctx()
inviter = _load_account(ctx.account_id)
tenant = _load_tenant(workspace_id)
_check_member_invite_quota(str(tenant.id))
try:
token = RegisterService.invite_new_member(
tenant=tenant,
email=payload.email,
language=None,
role=payload.role,
inviter=inviter,
)
except AccountAlreadyInTenantError as exc:
raise BadRequest(str(exc))
except NoPermissionError as exc:
raise BadRequest(str(exc))
except AccountRegisterError as exc:
raise BadRequest(str(exc))
normalized_email = payload.email.lower()
member = AccountService.get_account_by_email_with_case_fallback(normalized_email)
if member is None:
# invite_new_member just created or fetched this account.
raise RuntimeError("invited member missing from DB after invite")
encoded_email = parse.quote(normalized_email)
invite_url = f"{dify_config.CONSOLE_WEB_URL}/activate?email={encoded_email}&token={token}"
return MemberInviteResponse(
email=normalized_email,
role=payload.role,
member_id=str(member.id),
invite_url=invite_url,
tenant_id=str(tenant.id),
).model_dump(mode="json"), 201
@openapi_ns.route("/workspaces/<string:workspace_id>/members/<string:member_id>")
class WorkspaceMemberApi(Resource):
"""Remove a member.
Self-removal and owner-removal are explicitly rejected by the service
layer (CannotOperateSelfError, NoPermissionError) — both surface as
400 per the spec, with the service's message preserved.
"""
@openapi_ns.response(200, "Member removed", openapi_ns.models[MemberActionResponse.__name__])
@validate_bearer(accept=ACCEPT_USER_ANY)
@accept_subjects(SubjectType.ACCOUNT)
@require_workspace_role(TenantAccountRole.OWNER, TenantAccountRole.ADMIN)
def delete(self, workspace_id: str, member_id: str):
ctx = get_auth_ctx()
operator = _load_account(ctx.account_id)
tenant = _load_tenant(workspace_id)
member = AccountService.get_account_by_id(db.session, member_id)
if member is None:
raise NotFound("member not found")
try:
TenantService.remove_member_from_tenant(tenant, member, operator)
except CannotOperateSelfError as exc:
raise BadRequest(str(exc))
except NoPermissionError as exc:
raise BadRequest(str(exc))
except MemberNotInTenantError as exc:
raise NotFound(str(exc))
return MemberActionResponse().model_dump(mode="json"), 200
@openapi_ns.route("/workspaces/<string:workspace_id>/members/<string:member_id>/role")
class WorkspaceMemberRoleApi(Resource):
"""Change a member's role.
Owner cannot be assigned here (closed enum). Admin cannot demote the
standing owner (service NoPermissionError → 400, per spec).
"""
@openapi_ns.expect(openapi_ns.models[MemberRoleUpdatePayload.__name__])
@openapi_ns.response(200, "Role updated", openapi_ns.models[MemberActionResponse.__name__])
@validate_bearer(accept=ACCEPT_USER_ANY)
@accept_subjects(SubjectType.ACCOUNT)
@require_workspace_role(TenantAccountRole.OWNER, TenantAccountRole.ADMIN)
def put(self, workspace_id: str, member_id: str):
payload = _validate_body(MemberRoleUpdatePayload)
ctx = get_auth_ctx()
operator = _load_account(ctx.account_id)
tenant = _load_tenant(workspace_id)
member = AccountService.get_account_by_id(db.session, member_id)
if member is None:
raise NotFound("member not found")
try:
TenantService.update_member_role(tenant, member, payload.role, operator)
except CannotOperateSelfError as exc:
raise BadRequest(str(exc))
except NoPermissionError as exc:
raise BadRequest(str(exc))
except MemberNotInTenantError as exc:
raise NotFound(str(exc))
except RoleAlreadyAssignedError as exc:
raise BadRequest(str(exc))
return MemberActionResponse().model_dump(mode="json"), 200
def _workspace_summary(tenant: Tenant, membership: TenantAccountJoin) -> WorkspaceSummaryResponse:
return WorkspaceSummaryResponse(
id=str(tenant.id),
@@ -1,14 +1,11 @@
from uuid import UUID
from controllers.common.schema import register_response_schema_models, register_schema_models
from controllers.common.schema import register_schema_model
from controllers.console.datasets.hit_testing_base import DatasetsHitTestingBase, HitTestingPayload
from controllers.service_api import service_api_ns
from controllers.service_api.wraps import DatasetApiResource, cloud_edition_billing_rate_limit_check
from fields.hit_testing_fields import HitTestingResponse
from libs.helper import dump_response
register_schema_models(service_api_ns, HitTestingPayload)
register_response_schema_models(service_api_ns, HitTestingResponse)
register_schema_model(service_api_ns, HitTestingPayload)
@service_api_ns.route("/datasets/<uuid:dataset_id>/hit-testing", "/datasets/<uuid:dataset_id>/retrieve")
@@ -16,16 +13,16 @@ class HitTestingApi(DatasetApiResource, DatasetsHitTestingBase):
@service_api_ns.doc("dataset_hit_testing")
@service_api_ns.doc(description="Perform hit testing on a dataset")
@service_api_ns.doc(params={"dataset_id": "Dataset ID"})
@service_api_ns.response(
200,
"Hit testing results",
model=service_api_ns.models[HitTestingResponse.__name__],
@service_api_ns.doc(
responses={
200: "Hit testing results",
401: "Unauthorized - invalid API token",
404: "Dataset not found",
}
)
@service_api_ns.response(401, "Unauthorized - invalid API token")
@service_api_ns.response(404, "Dataset not found")
@service_api_ns.expect(service_api_ns.models[HitTestingPayload.__name__])
@cloud_edition_billing_rate_limit_check("knowledge", "dataset")
def post(self, tenant_id: str, dataset_id: UUID) -> dict[str, object]:
def post(self, tenant_id, dataset_id: UUID):
"""Perform hit testing on a dataset.
Tests retrieval performance for the specified dataset.
@@ -36,4 +33,4 @@ class HitTestingApi(DatasetApiResource, DatasetsHitTestingBase):
args = self.parse_args(service_api_ns.payload)
self.hit_testing_args_check(args)
return dump_response(HitTestingResponse, self.perform_hit_testing(dataset, args))
return self.perform_hit_testing(dataset, args)
+5 -12
View File
@@ -13,7 +13,6 @@ from pydantic import BaseModel
from sqlalchemy import select
from werkzeug.exceptions import Forbidden, NotFound, Unauthorized
from configs import dify_config
from enums.cloud_plan import CloudPlan
from extensions.ext_database import db
from extensions.ext_redis import redis_client
@@ -141,26 +140,20 @@ def cloud_edition_billing_resource_check[**P, R](
def interceptor(view: Callable[P, R]):
def decorated(*args: P.args, **kwargs: P.kwargs):
api_token = validate_and_get_api_token(api_token_type)
if resource == "vector_space":
if not dify_config.BILLING_ENABLED:
return view(*args, **kwargs)
vector_space = FeatureService.get_vector_space(api_token.tenant_id)
if 0 < vector_space.limit <= vector_space.size:
raise Forbidden("The capacity of the vector space has reached the limit of your subscription.")
return view(*args, **kwargs)
features = FeatureService.get_features(api_token.tenant_id, exclude_vector_space=True)
features = FeatureService.get_features(api_token.tenant_id)
if features.billing.enabled:
members = features.members
apps = features.apps
vector_space = features.vector_space
documents_upload_quota = features.documents_upload_quota
if resource == "members" and 0 < members.limit <= members.size:
raise Forbidden("The number of members has reached the limit of your subscription.")
elif resource == "apps" and 0 < apps.limit <= apps.size:
raise Forbidden("The number of apps has reached the limit of your subscription.")
elif resource == "vector_space" and 0 < vector_space.limit <= vector_space.size:
raise Forbidden("The capacity of the vector space has reached the limit of your subscription.")
elif resource == "documents" and 0 < documents_upload_quota.limit <= documents_upload_quota.size:
raise Forbidden("The number of documents has reached the limit of your subscription.")
else:
@@ -181,7 +174,7 @@ def cloud_edition_billing_knowledge_limit_check[**P, R](
@wraps(view)
def decorated(*args: P.args, **kwargs: P.kwargs):
api_token = validate_and_get_api_token(api_token_type)
features = FeatureService.get_features(api_token.tenant_id, exclude_vector_space=True)
features = FeatureService.get_features(api_token.tenant_id)
if features.billing.enabled:
if resource == "add_segment":
if features.billing.subscription.plan == CloudPlan.SANDBOX:
+3 -3
View File
@@ -12,7 +12,7 @@ from controllers.common.schema import register_response_schema_models, register_
from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
from libs.passport import PassportService
from libs.token import extract_webapp_passport
from models.model import App, AppMode, EndUser
from models.model import App, AppMode
from services.app_service import AppService
from services.enterprise.enterprise_service import EnterpriseService
from services.feature_service import FeatureService
@@ -56,7 +56,7 @@ class AppParameterApi(WebApiResource):
500: "Internal Server Error",
}
)
def get(self, app_model: App, end_user: EndUser):
def get(self, app_model: App, end_user):
"""Retrieve app parameters."""
if app_model.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
workflow = app_model.workflow
@@ -92,7 +92,7 @@ class AppMeta(WebApiResource):
500: "Internal Server Error",
}
)
def get(self, app_model: App, end_user: EndUser):
def get(self, app_model: App, end_user):
"""Get app meta"""
return AppService().get_app_meta(app_model)
+5 -5
View File
@@ -29,7 +29,7 @@ from core.errors.error import (
from graphon.model_runtime.errors.invoke import InvokeError
from libs import helper
from libs.helper import uuid_value
from models.model import App, AppMode, EndUser
from models.model import AppMode
from services.app_generate_service import AppGenerateService
from services.app_task_service import AppTaskService
from services.errors.llm import InvokeRateLimitError
@@ -86,7 +86,7 @@ class CompletionApi(WebApiResource):
500: "Internal Server Error",
}
)
def post(self, app_model: App, end_user: EndUser):
def post(self, app_model, end_user):
if app_model.mode != AppMode.COMPLETION:
raise NotCompletionAppError()
@@ -140,7 +140,7 @@ class CompletionStopApi(WebApiResource):
}
)
@web_ns.response(200, "Success", web_ns.models[SimpleResultResponse.__name__])
def post(self, app_model: App, end_user: EndUser, task_id: str):
def post(self, app_model, end_user, task_id: str):
if app_model.mode != AppMode.COMPLETION:
raise NotCompletionAppError()
@@ -169,7 +169,7 @@ class ChatApi(WebApiResource):
500: "Internal Server Error",
}
)
def post(self, app_model: App, end_user: EndUser):
def post(self, app_model, end_user):
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
@@ -226,7 +226,7 @@ class ChatStopApi(WebApiResource):
}
)
@web_ns.response(200, "Success", web_ns.models[SimpleResultResponse.__name__])
def post(self, app_model: App, end_user: EndUser, task_id: str):
def post(self, app_model, end_user, task_id: str):
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
+6 -6
View File
@@ -19,7 +19,7 @@ from fields.conversation_fields import (
SimpleConversation,
)
from libs.helper import uuid_value
from models.model import App, AppMode, EndUser
from models.model import AppMode
from services.conversation_service import ConversationService
from services.errors.conversation import ConversationNotExistsError, LastConversationNotExistsError
from services.web_conversation_service import WebConversationService
@@ -81,7 +81,7 @@ class ConversationListApi(WebApiResource):
500: "Internal Server Error",
}
)
def get(self, app_model: App, end_user: EndUser):
def get(self, app_model, end_user):
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
@@ -127,7 +127,7 @@ class ConversationApi(WebApiResource):
500: "Internal Server Error",
}
)
def delete(self, app_model: App, end_user: EndUser, c_id: UUID):
def delete(self, app_model, end_user, c_id: UUID):
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
@@ -166,7 +166,7 @@ class ConversationRenameApi(WebApiResource):
500: "Internal Server Error",
}
)
def post(self, app_model: App, end_user: EndUser, c_id: UUID):
def post(self, app_model, end_user, c_id: UUID):
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
@@ -204,7 +204,7 @@ class ConversationPinApi(WebApiResource):
}
)
@web_ns.response(200, "Conversation pinned successfully", web_ns.models[ResultResponse.__name__])
def patch(self, app_model: App, end_user: EndUser, c_id: UUID):
def patch(self, app_model, end_user, c_id: UUID):
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
@@ -235,7 +235,7 @@ class ConversationUnPinApi(WebApiResource):
}
)
@web_ns.response(200, "Conversation unpinned successfully", web_ns.models[ResultResponse.__name__])
def patch(self, app_model: App, end_user: EndUser, c_id: UUID):
def patch(self, app_model, end_user, c_id: UUID):
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
+1 -2
View File
@@ -13,7 +13,6 @@ from controllers.web import web_ns
from controllers.web.wraps import WebApiResource
from extensions.ext_database import db
from fields.file_fields import FileResponse
from models.model import App, EndUser
from services.file_service import FileService
register_schema_models(web_ns, FileResponse)
@@ -32,7 +31,7 @@ class FileApi(WebApiResource):
}
)
@web_ns.response(201, "File uploaded successfully", web_ns.models[FileResponse.__name__])
def post(self, app_model: App, end_user: EndUser):
def post(self, app_model, end_user):
"""Upload a file for use in web applications.
Accepts file uploads for use within web applications, supporting
+5 -5
View File
@@ -27,7 +27,7 @@ from fields.message_fields import SuggestedQuestionsResponse, WebMessageInfinite
from graphon.model_runtime.errors.invoke import InvokeError
from libs import helper
from models.enums import FeedbackRating
from models.model import App, AppMode, EndUser
from models.model import AppMode
from services.app_generate_service import AppGenerateService
from services.errors.app import MoreLikeThisDisabledError
from services.errors.conversation import ConversationNotExistsError
@@ -81,7 +81,7 @@ class MessageListApi(WebApiResource):
500: "Internal Server Error",
}
)
def get(self, app_model: App, end_user: EndUser):
def get(self, app_model, end_user):
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
@@ -133,7 +133,7 @@ class MessageFeedbackApi(WebApiResource):
}
)
@web_ns.response(200, "Feedback submitted successfully", web_ns.models[ResultResponse.__name__])
def post(self, app_model: App, end_user: EndUser, message_id: UUID):
def post(self, app_model, end_user, message_id: UUID):
message_id_str = str(message_id)
payload = MessageFeedbackPayload.model_validate(web_ns.payload or {})
@@ -167,7 +167,7 @@ class MessageMoreLikeThisApi(WebApiResource):
500: "Internal Server Error",
}
)
def get(self, app_model: App, end_user: EndUser, message_id: UUID):
def get(self, app_model, end_user, message_id: UUID):
if app_model.mode != "completion":
raise NotCompletionAppError()
@@ -223,7 +223,7 @@ class MessageSuggestedQuestionApi(WebApiResource):
500: "Internal Server Error",
}
)
def get(self, app_model: App, end_user: EndUser, message_id: UUID):
def get(self, app_model, end_user, message_id: UUID):
app_mode = AppMode.value_of(app_model.mode)
if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
raise NotChatAppError()
+2 -3
View File
@@ -13,7 +13,6 @@ from core.helper import ssrf_proxy
from extensions.ext_database import db
from fields.file_fields import FileWithSignedUrl, RemoteFileInfo
from graphon.file import helpers as file_helpers
from models.model import App, EndUser
from services.file_service import FileService
from ..common.schema import register_response_schema_models, register_schema_models
@@ -42,7 +41,7 @@ class RemoteFileInfoApi(WebApiResource):
}
)
@web_ns.response(200, "Remote file info", web_ns.models[RemoteFileInfo.__name__])
def get(self, app_model: App, end_user: EndUser, url: str):
def get(self, app_model, end_user, url):
"""Get information about a remote file.
Retrieves basic information about a file located at a remote URL,
@@ -86,7 +85,7 @@ class RemoteFileUploadApi(WebApiResource):
}
)
@web_ns.response(201, "Remote file uploaded", web_ns.models[FileWithSignedUrl.__name__])
def post(self, app_model: App, end_user: EndUser):
def post(self, app_model, end_user):
"""Upload a file from a remote URL.
Downloads a file from the provided remote URL and uploads it
+3 -4
View File
@@ -11,7 +11,6 @@ from controllers.web.error import NotCompletionAppError
from controllers.web.wraps import WebApiResource
from fields.conversation_fields import ResultResponse
from fields.message_fields import SavedMessageInfiniteScrollPagination, SavedMessageItem
from models.model import App, EndUser
from services.errors.message import MessageNotExistsError
from services.saved_message_service import SavedMessageService
@@ -44,7 +43,7 @@ class SavedMessageListApi(WebApiResource):
500: "Internal Server Error",
}
)
def get(self, app_model: App, end_user: EndUser):
def get(self, app_model, end_user):
if app_model.mode != "completion":
raise NotCompletionAppError()
@@ -78,7 +77,7 @@ class SavedMessageListApi(WebApiResource):
}
)
@web_ns.response(200, "Message saved successfully", web_ns.models[ResultResponse.__name__])
def post(self, app_model: App, end_user: EndUser):
def post(self, app_model, end_user):
if app_model.mode != "completion":
raise NotCompletionAppError()
@@ -107,7 +106,7 @@ class SavedMessageApi(WebApiResource):
500: "Internal Server Error",
}
)
def delete(self, app_model: App, end_user: EndUser, message_id: UUID):
def delete(self, app_model, end_user, message_id: UUID):
message_id_str = str(message_id)
if app_model.mode != "completion":
+5 -5
View File
@@ -10,7 +10,7 @@ from controllers.web.wraps import WebApiResource
from extensions.ext_database import db
from libs.helper import AppIconUrlField
from models.account import TenantStatus
from models.model import App, EndUser, Site
from models.model import App, Site
from services.feature_service import FeatureService
@@ -70,7 +70,7 @@ class AppSiteApi(WebApiResource):
}
)
@marshal_with(app_fields)
def get(self, app_model: App, end_user: EndUser):
def get(self, app_model, end_user):
"""Retrieve app site info."""
# get site
site = db.session.scalar(select(Site).where(Site.app_id == app_model.id).limit(1))
@@ -78,10 +78,10 @@ class AppSiteApi(WebApiResource):
if not site:
raise Forbidden()
if app_model.tenant and app_model.tenant.status == TenantStatus.ARCHIVE:
if app_model.tenant.status == TenantStatus.ARCHIVE:
raise Forbidden()
can_replace_logo = FeatureService.get_features(app_model.tenant_id, exclude_vector_space=True).can_replace_logo
can_replace_logo = FeatureService.get_features(app_model.tenant_id).can_replace_logo
return AppSiteInfo(app_model.tenant, app_model, site, end_user.id, can_replace_logo)
@@ -119,6 +119,6 @@ def serialize_site(site: Site) -> dict[str, Any]:
def serialize_app_site_payload(app_model: App, site: Site, end_user_id: str | None) -> dict[str, Any]:
can_replace_logo = FeatureService.get_features(app_model.tenant_id, exclude_vector_space=True).can_replace_logo
can_replace_logo = FeatureService.get_features(app_model.tenant_id).can_replace_logo
app_site_info = AppSiteInfo(app_model.tenant, app_model, site, end_user_id, can_replace_logo)
return cast(dict[str, Any], marshal(app_site_info, AppSiteApi.app_fields))
+73 -2
View File
@@ -22,6 +22,9 @@ from core.memory.token_buffer_memory import TokenBufferMemory
from core.model_manager import ModelInstance
from core.prompt.utils.extract_thread_messages import extract_thread_messages
from core.tools.__base.tool import Tool
from core.tools.entities.tool_entities import (
ToolParameter,
)
from core.tools.tool_manager import ToolManager
from core.tools.utils.dataset_retriever_tool import DatasetRetrieverTool
from extensions.ext_database import db
@@ -147,9 +150,44 @@ class BaseAgentRunner(AppRunner):
message_tool = PromptMessageTool(
name=tool.tool_name,
description=tool_entity.entity.description.llm,
parameters=tool_entity.get_llm_parameters_json_schema(),
parameters={
"type": "object",
"properties": {},
"required": [],
},
)
parameters = tool_entity.get_merged_runtime_parameters()
for parameter in parameters:
if parameter.form != ToolParameter.ToolParameterForm.LLM:
continue
parameter_type = parameter.type.as_normal_type()
if parameter.type in {
ToolParameter.ToolParameterType.SYSTEM_FILES,
ToolParameter.ToolParameterType.FILE,
ToolParameter.ToolParameterType.FILES,
}:
continue
enum = []
if parameter.type == ToolParameter.ToolParameterType.SELECT:
enum = [option.value for option in parameter.options] if parameter.options else []
message_tool.parameters["properties"][parameter.name] = (
{
"type": parameter_type,
"description": parameter.llm_description or "",
}
if parameter.input_schema is None
else parameter.input_schema
)
if len(enum) > 0:
message_tool.parameters["properties"][parameter.name]["enum"] = enum
if parameter.required:
message_tool.parameters["required"].append(parameter.name)
return message_tool, tool_entity
def _convert_dataset_retriever_tool_to_prompt_message_tool(self, tool: DatasetRetrieverTool) -> PromptMessageTool:
@@ -214,7 +252,40 @@ class BaseAgentRunner(AppRunner):
"""
update prompt message tool
"""
prompt_tool.parameters = tool.get_llm_parameters_json_schema()
# try to get tool runtime parameters
tool_runtime_parameters = tool.get_runtime_parameters()
for parameter in tool_runtime_parameters:
if parameter.form != ToolParameter.ToolParameterForm.LLM:
continue
parameter_type = parameter.type.as_normal_type()
if parameter.type in {
ToolParameter.ToolParameterType.SYSTEM_FILES,
ToolParameter.ToolParameterType.FILE,
ToolParameter.ToolParameterType.FILES,
}:
continue
enum = []
if parameter.type == ToolParameter.ToolParameterType.SELECT:
enum = [option.value for option in parameter.options] if parameter.options else []
prompt_tool.parameters["properties"][parameter.name] = (
{
"type": parameter_type,
"description": parameter.llm_description or "",
}
if parameter.input_schema is None
else parameter.input_schema
)
if len(enum) > 0:
prompt_tool.parameters["properties"][parameter.name]["enum"] = enum
if parameter.required:
if parameter.name not in prompt_tool.parameters["required"]:
prompt_tool.parameters["required"].append(parameter.name)
return prompt_tool
def create_agent_thought(
+35 -41
View File
@@ -7,7 +7,6 @@ from typing import TYPE_CHECKING, Any, Union
from core.app.app_config.entities import ExternalDataVariableEntity, PromptTemplateEntity
from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
from core.app.apps.exc import GenerateTaskStoppedError
from core.app.entities.app_invoke_entities import (
AppGenerateEntity,
EasyUIBasedAppGenerateEntity,
@@ -293,51 +292,46 @@ class AppRunner:
prompt_messages: list[PromptMessage] = []
text = ""
usage = None
try:
for result in invoke_result:
if not agent:
queue_manager.publish(QueueLLMChunkEvent(chunk=result), PublishFrom.APPLICATION_MANAGER)
else:
queue_manager.publish(QueueAgentMessageEvent(chunk=result), PublishFrom.APPLICATION_MANAGER)
for result in invoke_result:
if not agent:
queue_manager.publish(QueueLLMChunkEvent(chunk=result), PublishFrom.APPLICATION_MANAGER)
else:
queue_manager.publish(QueueAgentMessageEvent(chunk=result), PublishFrom.APPLICATION_MANAGER)
message = result.delta.message
if isinstance(message.content, str):
text += message.content
elif isinstance(message.content, list):
for content in message.content:
if isinstance(content, str):
text += content
elif isinstance(content, TextPromptMessageContent):
text += content.data
elif isinstance(content, ImagePromptMessageContent):
if message_id and user_id and tenant_id:
try:
self._handle_multimodal_image_content(
content=content,
message_id=message_id,
user_id=user_id,
tenant_id=tenant_id,
queue_manager=queue_manager,
)
except Exception:
_logger.exception("Failed to handle multimodal image output")
else:
_logger.warning("Received multimodal output but missing required parameters")
message = result.delta.message
if isinstance(message.content, str):
text += message.content
elif isinstance(message.content, list):
for content in message.content:
if isinstance(content, str):
text += content
elif isinstance(content, TextPromptMessageContent):
text += content.data
elif isinstance(content, ImagePromptMessageContent):
if message_id and user_id and tenant_id:
try:
self._handle_multimodal_image_content(
content=content,
message_id=message_id,
user_id=user_id,
tenant_id=tenant_id,
queue_manager=queue_manager,
)
except Exception:
_logger.exception("Failed to handle multimodal image output")
else:
text += content.data if hasattr(content, "data") else str(content)
_logger.warning("Received multimodal output but missing required parameters")
else:
text += content.data if hasattr(content, "data") else str(content)
if not model:
model = result.model
if not model:
model = result.model
if not prompt_messages:
prompt_messages = list(result.prompt_messages)
if not prompt_messages:
prompt_messages = list(result.prompt_messages)
if result.delta.usage:
usage = result.delta.usage
except GenerateTaskStoppedError:
# Explicitly close provider stream to stop in-flight token generation ASAP.
invoke_result.close()
raise
if result.delta.usage:
usage = result.delta.usage
if usage is None:
usage = LLMUsage.empty_usage()
@@ -47,12 +47,6 @@ from graphon.graph_events import (
)
from graphon.node_events import NodeRunResult
from libs.datetime_utils import naive_utc_now
from services.workflow.inspector_events import (
publish_node_changed as _inspector_publish_node_changed,
)
from services.workflow.inspector_events import (
publish_workflow_completed as _inspector_publish_workflow_completed,
)
@dataclass(slots=True)
@@ -169,7 +163,6 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
self._workflow_execution_repository.save(execution)
self._enqueue_trace_task(execution)
_inspector_publish_workflow_completed(workflow_run_id=execution.id_, status=str(execution.status.value))
def _handle_graph_run_partial_succeeded(self, event: GraphRunPartialSucceededEvent) -> None:
execution = self._get_workflow_execution()
@@ -180,7 +173,6 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
self._workflow_execution_repository.save(execution)
self._enqueue_trace_task(execution)
_inspector_publish_workflow_completed(workflow_run_id=execution.id_, status=str(execution.status.value))
def _handle_graph_run_failed(self, event: GraphRunFailedEvent) -> None:
execution = self._get_workflow_execution()
@@ -192,7 +184,6 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
self._fail_running_node_executions(error_message=event.error)
self._workflow_execution_repository.save(execution)
self._enqueue_trace_task(execution)
_inspector_publish_workflow_completed(workflow_run_id=execution.id_, status=str(execution.status.value))
def _handle_graph_run_aborted(self, event: GraphRunAbortedEvent) -> None:
execution = self._get_workflow_execution()
@@ -203,7 +194,6 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
self._fail_running_node_executions(error_message=execution.error_message or "")
self._workflow_execution_repository.save(execution)
self._enqueue_trace_task(execution)
_inspector_publish_workflow_completed(workflow_run_id=execution.id_, status=str(execution.status.value))
def _handle_graph_run_paused(self, event: GraphRunPausedEvent) -> None:
execution = self._get_workflow_execution()
@@ -251,7 +241,6 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
created_at=event.start_at,
)
self._node_snapshots[event.id] = snapshot
_inspector_publish_node_changed(workflow_run_id=execution.id_, node_id=event.node_id, status="running")
def _handle_node_retry(self, event: NodeRunRetryEvent) -> None:
domain_execution = self._get_node_execution(event.id)
@@ -259,11 +248,6 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
domain_execution.error = event.error
self._workflow_node_execution_repository.save(domain_execution)
self._workflow_node_execution_repository.save_execution_data(domain_execution)
_inspector_publish_node_changed(
workflow_run_id=self._get_workflow_execution().id_,
node_id=domain_execution.node_id,
status="retry",
)
def _handle_node_succeeded(self, event: NodeRunSucceededEvent) -> None:
domain_execution = self._get_node_execution(event.id)
@@ -273,11 +257,6 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
WorkflowNodeExecutionStatus.SUCCEEDED,
finished_at=event.finished_at,
)
_inspector_publish_node_changed(
workflow_run_id=self._get_workflow_execution().id_,
node_id=domain_execution.node_id,
status="succeeded",
)
def _handle_node_failed(self, event: NodeRunFailedEvent) -> None:
domain_execution = self._get_node_execution(event.id)
@@ -288,11 +267,6 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
error=event.error,
finished_at=event.finished_at,
)
_inspector_publish_node_changed(
workflow_run_id=self._get_workflow_execution().id_,
node_id=domain_execution.node_id,
status="failed",
)
def _handle_node_exception(self, event: NodeRunExceptionEvent) -> None:
domain_execution = self._get_node_execution(event.id)
@@ -303,11 +277,6 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
error=event.error,
finished_at=event.finished_at,
)
_inspector_publish_node_changed(
workflow_run_id=self._get_workflow_execution().id_,
node_id=domain_execution.node_id,
status="exception",
)
def _handle_node_pause_requested(self, event: NodeRunPauseRequestedEvent) -> None:
domain_execution = self._get_node_execution(event.id)
+1 -3
View File
@@ -534,9 +534,7 @@ class ProviderManager:
cache_key = f"tenant:{tenant_id}:model_load_balancing_enabled"
cache_result = redis_client.get(cache_key)
if cache_result is None:
model_load_balancing_enabled = FeatureService.get_features(
tenant_id, exclude_vector_space=True
).model_load_balancing_enabled
model_load_balancing_enabled = FeatureService.get_features(tenant_id).model_load_balancing_enabled
redis_client.setex(cache_key, 120, str(model_load_balancing_enabled))
else:
cache_result = cache_result.decode("utf-8")
+19 -74
View File
@@ -126,89 +126,34 @@ class Tool(ABC):
message_id: str | None = None,
) -> list[ToolParameter]:
"""
Get the effective parameter declarations for this tool.
Runtime parameters override declared parameters by name and append new
parameters, but the returned list is always detached from the tool's
cached declarations so callers can safely mutate it while building
downstream schemas.
get merged runtime parameters
:return: merged runtime parameters
"""
parameters = [deepcopy(parameter) for parameter in self.entity.parameters or []]
user_parameters = [
deepcopy(parameter)
for parameter in self.get_runtime_parameters(
conversation_id=conversation_id,
app_id=app_id,
message_id=message_id,
)
or []
]
parameter_indexes = {parameter.name: index for index, parameter in enumerate(parameters)}
parameters = self.entity.parameters
parameters = parameters.copy()
user_parameters = self.get_runtime_parameters() or []
user_parameters = user_parameters.copy()
# override parameters
for parameter in user_parameters:
existing_index = parameter_indexes.get(parameter.name)
if existing_index is None:
parameter_indexes[parameter.name] = len(parameters)
# check if parameter in tool parameters
for tool_parameter in parameters:
if tool_parameter.name == parameter.name:
# override parameter
tool_parameter.type = parameter.type
tool_parameter.form = parameter.form
tool_parameter.required = parameter.required
tool_parameter.default = parameter.default
tool_parameter.options = parameter.options
tool_parameter.llm_description = parameter.llm_description
break
else:
# add new parameter
parameters.append(parameter)
continue
parameters[existing_index] = parameter
return parameters
def get_llm_parameters_json_schema(
self,
conversation_id: str | None = None,
app_id: str | None = None,
message_id: str | None = None,
) -> dict[str, Any]:
"""Build the model-visible JSON schema from effective tool parameters.
Hidden/manual parameters stay available for invocation preparation on the
API side, but are intentionally omitted from the LLM-facing schema.
"""
schema: dict[str, Any] = {
"type": "object",
"properties": {},
"required": [],
}
for parameter in self.get_merged_runtime_parameters(
conversation_id=conversation_id,
app_id=app_id,
message_id=message_id,
):
if parameter.form != ToolParameter.ToolParameterForm.LLM:
continue
if parameter.type in {
ToolParameter.ToolParameterType.SYSTEM_FILES,
ToolParameter.ToolParameterType.FILE,
ToolParameter.ToolParameterType.FILES,
}:
continue
parameter_schema: dict[str, Any] = (
{
"type": parameter.type.as_normal_type(),
"description": parameter.llm_description or "",
}
if parameter.input_schema is None
else deepcopy(parameter.input_schema)
)
parameter_schema.setdefault("description", parameter.llm_description or "")
if parameter.type == ToolParameter.ToolParameterType.SELECT and parameter.options:
parameter_schema["enum"] = [option.value for option in parameter.options]
schema["properties"][parameter.name] = parameter_schema
if parameter.required:
schema["required"].append(parameter.name)
return schema
def create_image_message(
self,
image: str,
@@ -1,268 +0,0 @@
from __future__ import annotations
from collections.abc import Mapping
from typing import Any, Protocol, cast
from dify_agent.layers.dify_plugin import (
DifyPluginCredentialValue,
DifyPluginToolConfig,
DifyPluginToolCredentialType,
DifyPluginToolParameter,
DifyPluginToolParameterForm,
DifyPluginToolsLayerConfig,
)
from core.agent.entities import AgentToolEntity
from core.app.entities.app_invoke_entities import InvokeFrom
from core.tools.__base.tool import Tool
from core.tools.entities.tool_entities import ToolProviderType
from core.tools.errors import (
ToolProviderCredentialValidationError,
ToolProviderNotFoundError,
)
from core.tools.tool_manager import ToolManager
from models.agent_config_entities import AgentSoulDifyToolConfig, AgentSoulToolsConfig
from models.provider_ids import ToolProviderID
class WorkflowAgentPluginToolsBuildError(ValueError):
"""Raised when Agent Soul tools cannot be prepared for Agent backend."""
def __init__(self, error_code: str, message: str) -> None:
self.error_code = error_code
super().__init__(message)
class AgentToolRuntimeProvider(Protocol):
def get_agent_tool_runtime(
self,
tenant_id: str,
app_id: str,
agent_tool: AgentToolEntity,
user_id: str | None = None,
invoke_from: InvokeFrom = InvokeFrom.DEBUGGER,
variable_pool: Any | None = None,
) -> Tool: ...
class WorkflowAgentPluginToolsBuilder:
"""Prepare Agent Soul Dify Plugin Tools for the public Agent backend DTO."""
def __init__(self, *, tool_runtime_provider: AgentToolRuntimeProvider | None = None) -> None:
self._tool_runtime_provider = tool_runtime_provider or ToolManager
def build(
self,
*,
tenant_id: str,
app_id: str,
user_id: str | None,
tools: AgentSoulToolsConfig,
invoke_from: InvokeFrom,
) -> DifyPluginToolsLayerConfig | None:
"""Resolve user-selected Dify Plugin Tools into the Agent backend DTO.
``invoke_from`` is the *real* runtime caller category (DEBUGGER for a
Composer test run, SERVICE_API / WEB_APP for a published run). It must
be threaded through to :class:`ToolManager` so credential quotas, rate
limits, and audit tags match the actual call site.
"""
enabled_tools = [tool for tool in tools.dify_tools if tool.enabled]
if not enabled_tools:
return None
prepared: list[DifyPluginToolConfig] = []
seen_names: set[str] = set()
for tool_config in enabled_tools:
agent_tool = self._to_agent_tool_entity(tool_config)
tool_runtime = self._fetch_tool_runtime(
tenant_id=tenant_id,
app_id=app_id,
user_id=user_id,
agent_tool=agent_tool,
invoke_from=invoke_from,
tool_config=tool_config,
)
exposed_name = self._exposed_tool_name(tool_config)
if exposed_name in seen_names:
raise WorkflowAgentPluginToolsBuildError(
"agent_tool_name_duplicated",
f"Duplicate Dify Plugin Tool name {exposed_name!r}.",
)
seen_names.add(exposed_name)
prepared.append(self._to_backend_tool_config(tool_config, tool_runtime, exposed_name))
return DifyPluginToolsLayerConfig(tools=prepared)
def _fetch_tool_runtime(
self,
*,
tenant_id: str,
app_id: str,
user_id: str | None,
agent_tool: AgentToolEntity,
invoke_from: InvokeFrom,
tool_config: AgentSoulDifyToolConfig,
) -> Tool:
"""Resolve the API-side ``Tool`` runtime, mapping fetch errors to
Inspector-friendly error codes so callers can render distinct UX for
"tool definition gone" vs "credential failed".
"""
try:
return self._tool_runtime_provider.get_agent_tool_runtime(
tenant_id=tenant_id,
app_id=app_id,
agent_tool=agent_tool,
user_id=user_id,
invoke_from=invoke_from,
variable_pool=None,
)
except ToolProviderNotFoundError as exc:
raise WorkflowAgentPluginToolsBuildError(
"agent_tool_declaration_not_found",
f"Dify Plugin Tool {tool_config.tool_name!r} declaration not found: {exc}",
) from exc
except ToolProviderCredentialValidationError as exc:
raise WorkflowAgentPluginToolsBuildError(
"agent_tool_credential_invalid",
f"Dify Plugin Tool {tool_config.tool_name!r} credential validation failed: {exc}",
) from exc
except ValueError as exc:
# ToolManager raises bare ValueError when the agent tool's
# ``runtime`` / runtime parameters are missing. Surface it under a
# narrower error code than a generic "declaration not found" so
# frontend can render an actionable hint.
raise WorkflowAgentPluginToolsBuildError(
"agent_tool_config_invalid",
f"Dify Plugin Tool {tool_config.tool_name!r} runtime construction failed: {exc}",
) from exc
@staticmethod
def _to_agent_tool_entity(tool_config: AgentSoulDifyToolConfig) -> AgentToolEntity:
return AgentToolEntity(
provider_type=ToolProviderType.value_of(tool_config.provider_type),
provider_id=WorkflowAgentPluginToolsBuilder._provider_id(tool_config),
tool_name=tool_config.tool_name,
tool_parameters=dict(tool_config.runtime_parameters),
credential_id=tool_config.credential_ref.id if tool_config.credential_ref else None,
)
@staticmethod
def _provider_id(tool_config: AgentSoulDifyToolConfig) -> str:
if tool_config.provider_id:
return tool_config.provider_id
assert tool_config.plugin_id is not None
assert tool_config.provider is not None
return f"{tool_config.plugin_id}/{tool_config.provider}"
@staticmethod
def _exposed_tool_name(tool_config: AgentSoulDifyToolConfig) -> str:
# Stage 3.1 decision: no user rename yet. Keep the model-visible tool
# name aligned with the plugin declaration identity.
return tool_config.tool_name
def _to_backend_tool_config(
self,
tool_config: AgentSoulDifyToolConfig,
tool_runtime: Tool,
exposed_name: str,
) -> DifyPluginToolConfig:
runtime = tool_runtime.runtime
if runtime is None:
raise WorkflowAgentPluginToolsBuildError(
"agent_tool_config_invalid",
f"Dify Plugin Tool {tool_config.tool_name!r} has no runtime.",
)
provider_id = self._provider_id(tool_config)
plugin_id, provider = self._plugin_provider(tool_config, provider_id)
parameters = [
DifyPluginToolParameter.model_validate(parameter.model_dump(mode="json"))
for parameter in tool_runtime.get_merged_runtime_parameters()
]
runtime_parameters = self._runtime_parameters(tool_runtime, parameters)
description = tool_config.description
if description is None and tool_runtime.entity.description is not None:
description = tool_runtime.entity.description.llm
return DifyPluginToolConfig(
plugin_id=plugin_id,
provider=provider,
tool_name=tool_config.tool_name,
credential_type=self._credential_type(tool_config, runtime.credentials),
name=exposed_name,
description=description,
credentials=self._normalize_credentials(runtime.credentials, tool_name=tool_config.tool_name),
runtime_parameters=runtime_parameters,
parameters=parameters,
parameters_json_schema=cast(dict[str, Any], tool_runtime.get_llm_parameters_json_schema()),
)
@staticmethod
def _plugin_provider(tool_config: AgentSoulDifyToolConfig, provider_id: str) -> tuple[str, str]:
if tool_config.plugin_id and tool_config.provider:
return tool_config.plugin_id, tool_config.provider
provider_id_entity = ToolProviderID(provider_id)
return provider_id_entity.plugin_id, provider_id_entity.provider_name
@staticmethod
def _credential_type(
tool_config: AgentSoulDifyToolConfig,
credentials: Mapping[str, Any],
) -> DifyPluginToolCredentialType:
if not credentials and tool_config.credential_type == "unauthorized":
return "unauthorized"
return tool_config.credential_type
@staticmethod
def _runtime_parameters(
tool_runtime: Tool,
parameters: list[DifyPluginToolParameter],
) -> dict[str, Any]:
runtime = tool_runtime.runtime
runtime_parameters = dict(runtime.runtime_parameters if runtime is not None else {})
missing = [
parameter.name
for parameter in parameters
if parameter.form is not DifyPluginToolParameterForm.LLM
and parameter.required
and parameter.default is None
and parameter.name not in runtime_parameters
]
if missing:
names = ", ".join(sorted(missing))
raise WorkflowAgentPluginToolsBuildError(
"agent_tool_runtime_parameter_missing",
f"Dify Plugin Tool {tool_runtime.entity.identity.name!r} is missing runtime parameters: {names}.",
)
return runtime_parameters
@staticmethod
def _normalize_credentials(
credentials: Mapping[str, Any],
*,
tool_name: str,
) -> dict[str, DifyPluginCredentialValue]:
"""Forward only scalar credential values to the Agent backend.
``DifyPluginCredentialValue`` is ``str | int | float | bool | None``.
Refusing non-scalar values (lists, dicts, custom objects) is safer than
``str(value)`` stringifying a nested OAuth token blob produces a
Python ``repr`` that the plugin daemon cannot use, and we'd rather
surface a clear ``agent_tool_credential_shape_invalid`` than send junk.
"""
normalized: dict[str, DifyPluginCredentialValue] = {}
for key, value in credentials.items():
if isinstance(value, str | int | float | bool) or value is None:
normalized[key] = value
continue
raise WorkflowAgentPluginToolsBuildError(
"agent_tool_credential_shape_invalid",
(
f"Dify Plugin Tool {tool_name!r} credential {key!r} has a non-scalar value "
f"({type(value).__name__}); only str/int/float/bool/None are forwarded to the daemon."
),
)
return normalized
@@ -11,14 +11,13 @@ SUPPORTED_AGENT_BACKEND_FEATURES = frozenset(
"workflow_context",
"model",
"structured_output",
"tools.dify_tools",
}
)
RESERVED_AGENT_BACKEND_FEATURES = frozenset(
{
"skills_files",
"tools.cli_tools",
"tools",
"knowledge",
"human",
"env",
@@ -33,7 +32,7 @@ def build_runtime_feature_manifest(agent_soul: AgentSoulConfig) -> dict[str, Any
warnings: list[dict[str, str]] = []
soul_dump = agent_soul.model_dump(mode="json")
for section in sorted(RESERVED_AGENT_BACKEND_FEATURES):
value = _get_nested(soul_dump, section)
value = soul_dump.get(section)
has_value = bool(value)
if isinstance(value, dict):
has_value = any(bool(item) for item in value.values())
@@ -42,12 +41,11 @@ def build_runtime_feature_manifest(agent_soul: AgentSoulConfig) -> dict[str, Any
{
"section": f"agent_soul.{section}",
"code": "agent_backend_layer_not_available",
"message": f"{section} is saved in Agent Soul but is not executed by Agent backend.",
"message": f"{section} is saved in Agent Soul but is not executed by Agent backend in phase 3.",
}
)
reserved_status = dict.fromkeys(sorted(RESERVED_AGENT_BACKEND_FEATURES), "reserved_not_executed")
reserved_status["tools.dify_tools"] = "supported_when_config_valid"
return {
"supported": sorted(SUPPORTED_AGENT_BACKEND_FEATURES),
@@ -55,12 +53,3 @@ def build_runtime_feature_manifest(agent_soul: AgentSoulConfig) -> dict[str, Any
"reserved_status": reserved_status,
"unsupported_runtime_warnings": warnings,
}
def _get_nested(value: dict[str, Any], path: str) -> Any:
current: Any = value
for part in path.split("."):
if not isinstance(current, dict):
return None
current = current.get(part)
return current
@@ -4,8 +4,7 @@ from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import Any, Literal, Protocol, cast
from dify_agent.layers.execution_context import DifyExecutionContextLayerConfig
from dify_agent.protocol import CreateRunRequest
from dify_agent.protocol import CreateRunRequest, ExecutionContext
from clients.agent_backend import (
AgentBackendModelConfig,
@@ -30,7 +29,6 @@ from models.agent_config_entities import (
)
from .output_failure_orchestrator import retry_idempotency_key
from .plugin_tools_builder import WorkflowAgentPluginToolsBuilder, WorkflowAgentPluginToolsBuildError
from .runtime_feature_manifest import build_runtime_feature_manifest
@@ -85,11 +83,9 @@ class WorkflowAgentRuntimeRequestBuilder:
*,
credentials_provider: CredentialsProvider,
request_builder: AgentBackendRunRequestBuilder | None = None,
plugin_tools_builder: WorkflowAgentPluginToolsBuilder | None = None,
) -> None:
self._credentials_provider = credentials_provider
self._request_builder = request_builder or AgentBackendRunRequestBuilder()
self._plugin_tools_builder = plugin_tools_builder or WorkflowAgentPluginToolsBuilder()
def build(self, context: WorkflowAgentRuntimeBuildContext) -> WorkflowAgentRuntimeRequest:
agent_soul = AgentSoulConfig.model_validate(context.snapshot.config_snapshot_dict)
@@ -105,44 +101,20 @@ class WorkflowAgentRuntimeRequestBuilder:
workflow_job_prompt = node_job.workflow_prompt.strip() or "Run this workflow Agent Node for the current run."
user_prompt = workflow_context_prompt.strip() or "Use the current workflow context."
credentials = self._credentials_provider.fetch(agent_soul.model.model_provider, agent_soul.model.model)
try:
tools_layer = self._plugin_tools_builder.build(
tenant_id=context.dify_context.tenant_id,
app_id=context.dify_context.app_id,
user_id=context.dify_context.user_id,
tools=agent_soul.tools,
# Thread the *real* runtime invocation source through to
# ToolManager so credential quotas, rate limits, and audit
# trails match the actual call site (DEBUGGER for draft test
# run, SERVICE_API / WEB_APP for published run).
invoke_from=context.dify_context.invoke_from,
)
except WorkflowAgentPluginToolsBuildError as error:
raise WorkflowAgentRuntimeRequestBuildError(error.error_code, str(error)) from error
if tools_layer is not None:
metadata["agent_tools"] = {
"dify_tool_count": len(tools_layer.tools),
"dify_tool_names": [tool.name or tool.tool_name for tool in tools_layer.tools],
"cli_tool_count": len(agent_soul.tools.cli_tools),
}
request = self._request_builder.build_for_workflow_node(
AgentBackendWorkflowNodeRunInput(
model=AgentBackendModelConfig(
tenant_id=context.dify_context.tenant_id,
plugin_id=agent_soul.model.plugin_id,
model_provider=agent_soul.model.model_provider,
model=agent_soul.model.model,
user_id=context.dify_context.user_id,
credentials=self._normalize_credentials(credentials),
model_settings=cast(dict[str, Any], agent_soul.model.model_settings),
),
# The execution-context layer is now the only public protocol
# carrier for Dify tenant/user/run identifiers. ``user_id`` must
# be forwarded here because downstream plugin-daemon provider and
# tool clients read it from this layer rather than from any
# parallel top-level request field.
execution_context=DifyExecutionContextLayerConfig(
execution_context=ExecutionContext(
tenant_id=context.dify_context.tenant_id,
user_id=context.dify_context.user_id,
app_id=context.dify_context.app_id,
workflow_id=context.workflow_id,
workflow_run_id=context.workflow_run_id,
@@ -157,7 +129,6 @@ class WorkflowAgentRuntimeRequestBuilder:
workflow_node_job_prompt=workflow_job_prompt,
user_prompt=user_prompt,
output=self._build_output_config(node_job.declared_outputs),
tools=tools_layer,
idempotency_key=self._idempotency_key(context),
metadata=metadata,
)
@@ -126,7 +126,6 @@ class WorkflowAgentNodeValidator:
raise WorkflowAgentNodeValidationError(
f"Workflow Agent node {binding.node_id} requires Agent Soul model config."
)
cls._validate_agent_soul_tools(binding=binding, agent_soul=agent_soul)
node_job = WorkflowNodeJobConfig.model_validate(binding.node_job_config_dict)
cls.validate_node_job(session=session, binding=binding, node_job=node_job, topology=topology)
@@ -281,26 +280,6 @@ class WorkflowAgentNodeValidator:
f"Workflow Agent node {binding.node_id} references unsupported human contact channel {channel}."
)
@classmethod
def _validate_agent_soul_tools(
cls,
*,
binding: WorkflowAgentNodeBinding,
agent_soul: AgentSoulConfig,
) -> None:
exposed_names: set[str] = set()
for tool in agent_soul.tools.dify_tools:
if not tool.enabled:
continue
exposed_name = tool.tool_name
if exposed_name in exposed_names:
raise WorkflowAgentNodeValidationError(
f"Workflow Agent node {binding.node_id} has duplicate Dify Plugin Tool name {exposed_name}."
)
exposed_names.add(exposed_name)
# CLI tools remain saved-but-not-executed. They are allowed at publish
# time so existing Agent Soul drafts are not blocked by a reserved field.
@staticmethod
def _validate_file_ref(
*,
+56 -90
View File
@@ -1,96 +1,62 @@
from datetime import datetime
from typing import Any
from flask_restx import fields
from pydantic import field_validator
from libs.helper import TimestampField
from fields.base import ResponseModel
from libs.helper import to_timestamp
document_fields = {
"id": fields.String,
"data_source_type": fields.String,
"name": fields.String,
"doc_type": fields.String,
"doc_metadata": fields.Raw,
}
segment_fields = {
"id": fields.String,
"position": fields.Integer,
"document_id": fields.String,
"content": fields.String,
"sign_content": fields.String,
"answer": fields.String,
"word_count": fields.Integer,
"tokens": fields.Integer,
"keywords": fields.List(fields.String),
"index_node_id": fields.String,
"index_node_hash": fields.String,
"hit_count": fields.Integer,
"enabled": fields.Boolean,
"disabled_at": TimestampField,
"disabled_by": fields.String,
"status": fields.String,
"created_by": fields.String,
"created_at": TimestampField,
"indexing_at": TimestampField,
"completed_at": TimestampField,
"error": fields.String,
"stopped_at": TimestampField,
"document": fields.Nested(document_fields),
}
class HitTestingQuery(ResponseModel):
content: str
child_chunk_fields = {
"id": fields.String,
"content": fields.String,
"position": fields.Integer,
"score": fields.Float,
}
files_fields = {
"id": fields.String,
"name": fields.String,
"size": fields.Integer,
"extension": fields.String,
"mime_type": fields.String,
"source_url": fields.String,
}
class HitTestingDocument(ResponseModel):
id: str
data_source_type: str
name: str
doc_type: str | None
doc_metadata: Any | None
@field_validator("data_source_type", "doc_type", mode="before")
@classmethod
def _normalize_enum_fields(cls, value: Any) -> Any:
return _normalize_enum(value)
class HitTestingSegment(ResponseModel):
id: str
position: int
document_id: str
content: str
sign_content: str | None
answer: str | None
word_count: int
tokens: int
keywords: list[str]
index_node_id: str | None
index_node_hash: str | None
hit_count: int
enabled: bool
disabled_at: int | None
disabled_by: str | None
status: str
created_by: str
created_at: int
indexing_at: int | None
completed_at: int | None
error: str | None
stopped_at: int | None
document: HitTestingDocument
@field_validator("disabled_at", "created_at", "indexing_at", "completed_at", "stopped_at", mode="before")
@classmethod
def _normalize_timestamp(cls, value: datetime | int | None) -> int | None:
return to_timestamp(value)
@field_validator("status", mode="before")
@classmethod
def _normalize_enum_fields(cls, value: Any) -> Any:
return _normalize_enum(value)
class HitTestingChildChunk(ResponseModel):
id: str
content: str
position: int
score: float
class HitTestingFile(ResponseModel):
id: str
name: str
size: int
extension: str
mime_type: str
source_url: str
class HitTestingRecord(ResponseModel):
segment: HitTestingSegment
child_chunks: list[HitTestingChildChunk]
score: float | None
tsne_position: Any | None
files: list[HitTestingFile]
summary: str | None
class HitTestingResponse(ResponseModel):
query: HitTestingQuery
records: list[HitTestingRecord]
def _normalize_enum(value: Any) -> Any:
if isinstance(value, str) or value is None:
return value
return getattr(value, "value", value)
hit_testing_record_fields = {
"segment": fields.Nested(segment_fields),
"child_chunks": fields.List(fields.Nested(child_chunk_fields)),
"score": fields.Float,
"tsne_position": fields.Raw,
"files": fields.List(fields.Nested(files_fields)),
"summary": fields.String, # Summary content if retrieved via summary index
}
+1 -1
View File
@@ -58,7 +58,7 @@ def check_workspace_owner_transfer_permission(workspace_id: str) -> None:
Raises:
Forbidden: If either billing plan or workspace policy prohibits ownership transfer
"""
features = FeatureService.get_features(workspace_id, exclude_vector_space=True)
features = FeatureService.get_features(workspace_id)
if not features.is_allow_transfer_workspace:
raise Forbidden("Your current plan does not allow workspace ownership transfer")
+2 -84
View File
@@ -1,6 +1,6 @@
import re
from enum import StrEnum
from typing import Any, Final, Literal
from typing import Any, Final
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
@@ -50,90 +50,8 @@ class AgentSoulSkillsFilesConfig(BaseModel):
skills: list[dict[str, Any]] = Field(default_factory=list)
class AgentSoulDifyToolCredentialRef(BaseModel):
"""Reference to a stored Dify Plugin Tool credential.
Secret values are resolved only at runtime. The legacy ``credential_id``
field is accepted by :class:`AgentSoulDifyToolConfig` and normalized here so
old Agent tool payloads can be read while new payloads stay explicit.
"""
model_config = ConfigDict(extra="ignore")
type: Literal["provider", "tool"] = "tool"
id: str | None = Field(default=None, max_length=255)
provider: str | None = Field(default=None, max_length=255)
class AgentSoulDifyToolConfig(BaseModel):
"""One Dify Plugin Tool configured on Agent Soul.
The API backend prepares this persisted product shape into
``DifyPluginToolConfig`` before sending a run request to Agent backend.
``provider_id`` keeps compatibility with existing Agent tool config payloads;
new callers should send ``plugin_id`` + ``provider`` when available.
"""
# ``extra="ignore"`` (not ``"allow"``) so historical Agent Soul payloads
# with unknown fields still load — but the extra keys are dropped instead
# of silently riding along into ``model_dump``. New callers should send the
# explicit schema fields below.
model_config = ConfigDict(extra="ignore")
enabled: bool = True
# Dify Plugin Tools live behind the ``PLUGIN`` provider type. ``BUILT_IN`` /
# ``WORKFLOW`` / ``API`` providers are not exposed to the Agent backend in
# this layer — keep the default narrow so a missing field surfaces as
# ``agent_tool_declaration_not_found`` against the correct provider table.
provider_type: str = "plugin"
provider_id: str | None = Field(default=None, max_length=255)
plugin_id: str | None = Field(default=None, max_length=255)
provider: str | None = Field(default=None, max_length=255)
tool_name: str = Field(min_length=1, max_length=255)
credential_type: Literal["api-key", "oauth2", "unauthorized"] = "api-key"
credential_ref: AgentSoulDifyToolCredentialRef | None = None
# Reserved for a future user-rename UX. Accepted but currently rejected at
# validation time so frontend cannot silently believe a rename took effect
# (see :meth:`_validate_provider_and_credentials`).
name: str | None = Field(default=None, max_length=255)
description: str | None = None
runtime_parameters: dict[str, Any] = Field(default_factory=dict)
@model_validator(mode="before")
@classmethod
def _normalize_legacy_payload(cls, value: Any) -> Any:
if not isinstance(value, dict):
return value
normalized = dict(value)
if normalized.get("provider_id") is None and isinstance(normalized.get("provider_name"), str):
normalized["provider_id"] = normalized["provider_name"]
if normalized.get("runtime_parameters") is None and isinstance(normalized.get("tool_parameters"), dict):
normalized["runtime_parameters"] = normalized["tool_parameters"]
if normalized.get("credential_ref") is None and normalized.get("credential_id"):
normalized["credential_ref"] = {
"type": "tool",
"id": normalized.get("credential_id"),
"provider": normalized.get("provider_id") or normalized.get("provider"),
}
return normalized
@model_validator(mode="after")
def _validate_provider_and_credentials(self) -> "AgentSoulDifyToolConfig":
if not self.provider_id and not (self.plugin_id and self.provider):
raise ValueError("Dify tool requires provider_id or plugin_id + provider")
if self.credential_type != "unauthorized" and (self.credential_ref is None or not self.credential_ref.id):
raise ValueError("credential_ref.id is required for credentialed Dify tools")
# ``name`` is reserved for a future user-rename UX. Until that lands
# the model-visible name is forced to match ``tool_name``; reject
# explicit values so a frontend bug surfaces immediately instead of
# producing a silently-ignored override.
if self.name is not None and self.name != self.tool_name:
raise ValueError("name override is not yet supported; omit ``name`` or set it equal to ``tool_name``.")
return self
class AgentSoulToolsConfig(BaseModel):
dify_tools: list[AgentSoulDifyToolConfig] = Field(default_factory=list)
dify_tools: list[dict[str, Any]] = Field(default_factory=list)
cli_tools: list[dict[str, Any]] = Field(default_factory=list)
+47 -256
View File
@@ -3447,89 +3447,6 @@ Run draft workflow
| 200 | Draft workflow run started successfully |
| 403 | Permission denied |
### /apps/{app_id}/workflows/draft/runs/{run_id}/node-outputs
#### GET
##### Description
Snapshot of every node's declared outputs for a draft workflow run.
##### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| app_id | path | Application ID | Yes | string |
| run_id | path | Workflow run ID | Yes | string |
##### Responses
| Code | Description |
| ---- | ----------- |
| 404 | Workflow run not found |
### /apps/{app_id}/workflows/draft/runs/{run_id}/node-outputs/events
#### GET
##### Description
Server-Sent Events stream of inspector deltas for a draft workflow run.
##### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| app_id | path | Application ID | Yes | string |
| run_id | path | Workflow run ID | Yes | string |
##### Responses
| Code | Description |
| ---- | ----------- |
| 404 | Workflow run not found |
### /apps/{app_id}/workflows/draft/runs/{run_id}/node-outputs/{node_id}
#### GET
##### Description
One node's declared outputs for a draft workflow run.
##### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| app_id | path | Application ID | Yes | string |
| node_id | path | Node ID inside the workflow graph | Yes | string |
| run_id | path | Workflow run ID | Yes | string |
##### Responses
| Code | Description |
| ---- | ----------- |
| 404 | Workflow run / node not found |
### /apps/{app_id}/workflows/draft/runs/{run_id}/node-outputs/{node_id}/{output_name}/preview
#### GET
##### Description
Full value for one declared output, including signed download URL for files.
##### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| app_id | path | Application ID | Yes | string |
| node_id | path | Node ID inside the workflow graph | Yes | string |
| output_name | path | Declared output name as exposed by Composer | Yes | string |
| run_id | path | Workflow run ID | Yes | string |
##### Responses
| Code | Description |
| ---- | ----------- |
| 404 | Workflow run / node / output not found |
### /apps/{app_id}/workflows/draft/system-variables
#### GET
@@ -3767,89 +3684,6 @@ Publish workflow
| ---- | ----------- |
| 200 | Success |
### /apps/{app_id}/workflows/published/runs/{run_id}/node-outputs
#### GET
##### Description
Snapshot of every node's declared outputs for a published workflow run.
##### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| app_id | path | Application ID | Yes | string |
| run_id | path | Workflow run ID | Yes | string |
##### Responses
| Code | Description |
| ---- | ----------- |
| 404 | Workflow run not found |
### /apps/{app_id}/workflows/published/runs/{run_id}/node-outputs/events
#### GET
##### Description
Server-Sent Events stream of inspector deltas for a published workflow run.
##### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| app_id | path | Application ID | Yes | string |
| run_id | path | Workflow run ID | Yes | string |
##### Responses
| Code | Description |
| ---- | ----------- |
| 404 | Workflow run not found |
### /apps/{app_id}/workflows/published/runs/{run_id}/node-outputs/{node_id}
#### GET
##### Description
One node's declared outputs for a published workflow run.
##### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| app_id | path | Application ID | Yes | string |
| node_id | path | Node ID inside the workflow graph | Yes | string |
| run_id | path | Workflow run ID | Yes | string |
##### Responses
| Code | Description |
| ---- | ----------- |
| 404 | Workflow run / node not found |
### /apps/{app_id}/workflows/published/runs/{run_id}/node-outputs/{node_id}/{output_name}/preview
#### GET
##### Description
Full value for one declared output of a published run.
##### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| app_id | path | Application ID | Yes | string |
| node_id | path | Node ID inside the workflow graph | Yes | string |
| output_name | path | Declared output name as exposed by Composer | Yes | string |
| run_id | path | Workflow run ID | Yes | string |
##### Responses
| Code | Description |
| ---- | ----------- |
| 404 | Workflow run / node / output not found |
### /apps/{app_id}/workflows/triggers/webhook
#### GET
@@ -10705,43 +10539,6 @@ Supported icon storage formats for Agent roster entries.
| skills_files | [AgentSoulSkillsFilesConfig](#agentsoulskillsfilesconfig) | | No |
| tools | [AgentSoulToolsConfig](#agentsoultoolsconfig) | | No |
#### AgentSoulDifyToolConfig
One Dify Plugin Tool configured on Agent Soul.
The API backend prepares this persisted product shape into
``DifyPluginToolConfig`` before sending a run request to Agent backend.
``provider_id`` keeps compatibility with existing Agent tool config payloads;
new callers should send ``plugin_id`` + ``provider`` when available.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| credential_ref | [AgentSoulDifyToolCredentialRef](#agentsouldifytoolcredentialref) | | No |
| credential_type | string | *Enum:* `"api-key"`, `"oauth2"`, `"unauthorized"` | No |
| description | string | | No |
| enabled | boolean | | No |
| name | string | | No |
| plugin_id | string | | No |
| provider | string | | No |
| provider_id | string | | No |
| provider_type | string | | No |
| runtime_parameters | object | | No |
| tool_name | string | | Yes |
#### AgentSoulDifyToolCredentialRef
Reference to a stored Dify Plugin Tool credential.
Secret values are resolved only at runtime. The legacy ``credential_id``
field is accepted by :class:`AgentSoulDifyToolConfig` and normalized here so
old Agent tool payloads can be read while new payloads stay explicit.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| id | string | | No |
| provider | string | | No |
| type | string | *Enum:* `"provider"`, `"tool"` | No |
#### AgentSoulEnvConfig
| Name | Type | Description | Required |
@@ -10819,7 +10616,7 @@ Reference to model credentials resolved only at runtime.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| cli_tools | [ object ] | | No |
| dify_tools | [ [AgentSoulDifyToolConfig](#agentsouldifytoolconfig) ] | | No |
| dify_tools | [ object ] | | No |
#### AgentThought
@@ -13051,31 +12848,31 @@ Request payload for bulk downloading documents as a zip archive.
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| content | string | | Yes |
| id | string | | Yes |
| position | integer | | Yes |
| score | number | | Yes |
| content | string | | No |
| id | string | | No |
| position | integer | | No |
| score | number | | No |
#### HitTestingDocument
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| data_source_type | string | | Yes |
| doc_metadata | | | Yes |
| doc_type | string | | Yes |
| id | string | | Yes |
| name | string | | Yes |
| data_source_type | string | | No |
| doc_metadata | | | No |
| doc_type | string | | No |
| id | string | | No |
| name | string | | No |
#### HitTestingFile
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| extension | string | | Yes |
| id | string | | Yes |
| mime_type | string | | Yes |
| name | string | | Yes |
| size | integer | | Yes |
| source_url | string | | Yes |
| extension | string | | No |
| id | string | | No |
| mime_type | string | | No |
| name | string | | No |
| size | integer | | No |
| source_url | string | | No |
#### HitTestingPayload
@@ -13086,57 +12883,51 @@ Request payload for bulk downloading documents as a zip archive.
| query | string | | Yes |
| retrieval_model | [RetrievalModel](#retrievalmodel) | | No |
#### HitTestingQuery
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| content | string | | Yes |
#### HitTestingRecord
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| child_chunks | [ [HitTestingChildChunk](#hittestingchildchunk) ] | | Yes |
| files | [ [HitTestingFile](#hittestingfile) ] | | Yes |
| score | number | | Yes |
| segment | [HitTestingSegment](#hittestingsegment) | | Yes |
| summary | string | | Yes |
| tsne_position | | | Yes |
| child_chunks | [ [HitTestingChildChunk](#hittestingchildchunk) ] | | No |
| files | [ [HitTestingFile](#hittestingfile) ] | | No |
| score | number | | No |
| segment | [HitTestingSegment](#hittestingsegment) | | No |
| summary | string | | No |
| tsne_position | | | No |
#### HitTestingResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| query | [HitTestingQuery](#hittestingquery) | | Yes |
| records | [ [HitTestingRecord](#hittestingrecord) ] | | Yes |
| query | string | | Yes |
| records | [ [HitTestingRecord](#hittestingrecord) ] | | No |
#### HitTestingSegment
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| answer | string | | Yes |
| completed_at | integer | | Yes |
| content | string | | Yes |
| created_at | integer | | Yes |
| created_by | string | | Yes |
| disabled_at | integer | | Yes |
| disabled_by | string | | Yes |
| document | [HitTestingDocument](#hittestingdocument) | | Yes |
| document_id | string | | Yes |
| enabled | boolean | | Yes |
| error | string | | Yes |
| hit_count | integer | | Yes |
| id | string | | Yes |
| index_node_hash | string | | Yes |
| index_node_id | string | | Yes |
| indexing_at | integer | | Yes |
| keywords | [ string ] | | Yes |
| position | integer | | Yes |
| sign_content | string | | Yes |
| status | string | | Yes |
| stopped_at | integer | | Yes |
| tokens | integer | | Yes |
| word_count | integer | | Yes |
| answer | string | | No |
| completed_at | integer | | No |
| content | string | | No |
| created_at | integer | | No |
| created_by | string | | No |
| disabled_at | integer | | No |
| disabled_by | string | | No |
| document | [HitTestingDocument](#hittestingdocument) | | No |
| document_id | string | | No |
| enabled | boolean | | No |
| error | string | | No |
| hit_count | integer | | No |
| id | string | | No |
| index_node_hash | string | | No |
| index_node_id | string | | No |
| indexing_at | integer | | No |
| keywords | [ string ] | | No |
| position | integer | | No |
| sign_content | string | | No |
| status | string | | No |
| stopped_at | integer | | No |
| tokens | integer | | No |
| word_count | integer | | No |
#### HumanInputContent
-139
View File
@@ -323,85 +323,6 @@ Upload a file to use as an input variable when running the app
| ---- | ----------- | ------ |
| 200 | Workspace detail | [WorkspaceDetailResponse](#workspacedetailresponse) |
### /workspaces/{workspace_id}/members
#### GET
##### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| workspace_id | path | | Yes | string |
| limit | query | | No | integer |
| page | query | | No | integer |
##### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Member list | [MemberListResponse](#memberlistresponse) |
#### POST
##### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| workspace_id | path | | Yes | string |
| payload | body | | Yes | [MemberInvitePayload](#memberinvitepayload) |
##### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 201 | Member invited | [MemberInviteResponse](#memberinviteresponse) |
### /workspaces/{workspace_id}/members/{member_id}
#### DELETE
##### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| member_id | path | | Yes | string |
| workspace_id | path | | Yes | string |
##### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Member removed | [MemberActionResponse](#memberactionresponse) |
### /workspaces/{workspace_id}/members/{member_id}/role
#### PUT
##### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| member_id | path | | Yes | string |
| workspace_id | path | | Yes | string |
| payload | body | | Yes | [MemberRoleUpdatePayload](#memberroleupdatepayload) |
##### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Role updated | [MemberActionResponse](#memberactionresponse) |
### /workspaces/{workspace_id}/switch
#### POST
##### Parameters
| Name | Located in | Description | Required | Schema |
| ---- | ---------- | ----------- | -------- | ------ |
| workspace_id | path | | Yes | string |
##### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Workspace detail | [WorkspaceDetailResponse](#workspacedetailresponse) |
---
### Models
@@ -605,66 +526,6 @@ mode is a closed enum.
| ---- | ---- | ----------- | -------- |
| JsonValue | | | |
#### MemberActionResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| result | string | | No |
#### MemberInvitePayload
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| email | string | | Yes |
| role | string | *Enum:* `"admin"`, `"normal"` | Yes |
#### MemberInviteResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| email | string | | Yes |
| invite_url | string | | Yes |
| member_id | string | | Yes |
| result | string | | No |
| role | string | | Yes |
| tenant_id | string | | Yes |
#### MemberListQuery
Strict (extra='forbid').
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| limit | integer | | No |
| page | integer | | No |
#### MemberListResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| data | [ [MemberResponse](#memberresponse) ] | | Yes |
| has_more | boolean | | Yes |
| limit | integer | | Yes |
| page | integer | | Yes |
| total | integer | | Yes |
#### MemberResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| avatar | string | | No |
| email | string | | Yes |
| id | string | | Yes |
| name | string | | Yes |
| role | string | | Yes |
| status | string | | Yes |
#### MemberRoleUpdatePayload
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| role | string | *Enum:* `"admin"`, `"normal"` | Yes |
#### MessageMetadata
| Name | Type | Description | Required |
+10 -92
View File
@@ -1363,11 +1363,11 @@ Tests retrieval performance for the specified dataset.
##### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Hit testing results | [HitTestingResponse](#hittestingresponse) |
| 401 | Unauthorized - invalid API token | |
| 404 | Dataset not found | |
| Code | Description |
| ---- | ----------- |
| 200 | Hit testing results |
| 401 | Unauthorized - invalid API token |
| 404 | Dataset not found |
### /datasets/{dataset_id}/metadata
@@ -1614,11 +1614,11 @@ Tests retrieval performance for the specified dataset.
##### Responses
| Code | Description | Schema |
| ---- | ----------- | ------ |
| 200 | Hit testing results | [HitTestingResponse](#hittestingresponse) |
| 401 | Unauthorized - invalid API token | |
| 404 | Dataset not found | |
| Code | Description |
| ---- | ----------- |
| 200 | Hit testing results |
| 401 | Unauthorized - invalid API token |
| 404 | Dataset not found |
### /datasets/{dataset_id}/tags
@@ -2691,36 +2691,6 @@ Note: The SQLAlchemy model defines an `is_anonymous` property for Flask-Login se
| tenant_id | string | | No |
| user_id | string | | No |
#### HitTestingChildChunk
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| content | string | | Yes |
| id | string | | Yes |
| position | integer | | Yes |
| score | number | | Yes |
#### HitTestingDocument
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| data_source_type | string | | Yes |
| doc_metadata | | | Yes |
| doc_type | string | | Yes |
| id | string | | Yes |
| name | string | | Yes |
#### HitTestingFile
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| extension | string | | Yes |
| id | string | | Yes |
| mime_type | string | | Yes |
| name | string | | Yes |
| size | integer | | Yes |
| source_url | string | | Yes |
#### HitTestingPayload
| Name | Type | Description | Required |
@@ -2730,58 +2700,6 @@ Note: The SQLAlchemy model defines an `is_anonymous` property for Flask-Login se
| query | string | | Yes |
| retrieval_model | [RetrievalModel](#retrievalmodel) | | No |
#### HitTestingQuery
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| content | string | | Yes |
#### HitTestingRecord
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| child_chunks | [ [HitTestingChildChunk](#hittestingchildchunk) ] | | Yes |
| files | [ [HitTestingFile](#hittestingfile) ] | | Yes |
| score | number | | Yes |
| segment | [HitTestingSegment](#hittestingsegment) | | Yes |
| summary | string | | Yes |
| tsne_position | | | Yes |
#### HitTestingResponse
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| query | [HitTestingQuery](#hittestingquery) | | Yes |
| records | [ [HitTestingRecord](#hittestingrecord) ] | | Yes |
#### HitTestingSegment
| Name | Type | Description | Required |
| ---- | ---- | ----------- | -------- |
| answer | string | | Yes |
| completed_at | integer | | Yes |
| content | string | | Yes |
| created_at | integer | | Yes |
| created_by | string | | Yes |
| disabled_at | integer | | Yes |
| disabled_by | string | | Yes |
| document | [HitTestingDocument](#hittestingdocument) | | Yes |
| document_id | string | | Yes |
| enabled | boolean | | Yes |
| error | string | | Yes |
| hit_count | integer | | Yes |
| id | string | | Yes |
| index_node_hash | string | | Yes |
| index_node_id | string | | Yes |
| indexing_at | integer | | Yes |
| keywords | [ string ] | | Yes |
| position | integer | | Yes |
| sign_content | string | | Yes |
| status | string | | Yes |
| stopped_at | integer | | Yes |
| tokens | integer | | Yes |
| word_count | integer | | Yes |
#### HumanInputFormSubmitPayload
| Name | Type | Description | Required |
@@ -42,9 +42,6 @@ class MilvusConfig(BaseModel):
database: str = "default" # Database name
enable_hybrid_search: bool = False # Flag to enable hybrid search
analyzer_params: str | None = None # Analyzer params
secure: bool = False # Enable one-way TLS to Milvus
server_pem_path: str | None = None # Path to server certificate (PEM) for TLS verification
server_name: str | None = None # Server name to verify against the certificate (SNI / CN)
@model_validator(mode="before")
@classmethod
@@ -391,19 +388,16 @@ class MilvusVector(BaseVector):
"""
Initialize and return a Milvus client.
"""
kwargs: dict[str, Any] = {"uri": config.uri, "db_name": config.database}
if config.token:
kwargs["token"] = config.token
client = MilvusClient(uri=config.uri, token=config.token, db_name=config.database)
else:
kwargs["user"] = config.user or ""
kwargs["password"] = config.password or ""
if config.secure:
kwargs["secure"] = True
if config.server_pem_path:
kwargs["server_pem_path"] = config.server_pem_path
if config.server_name:
kwargs["server_name"] = config.server_name
return MilvusClient(**kwargs)
client = MilvusClient(
uri=config.uri,
user=config.user or "",
password=config.password or "",
db_name=config.database,
)
return client
class MilvusVectorFactory(AbstractVectorFactory):
@@ -433,8 +427,5 @@ class MilvusVectorFactory(AbstractVectorFactory):
database=dify_config.MILVUS_DATABASE or "",
enable_hybrid_search=dify_config.MILVUS_ENABLE_HYBRID_SEARCH or False,
analyzer_params=dify_config.MILVUS_ANALYZER_PARAMS or "",
secure=dify_config.MILVUS_SECURE,
server_pem_path=dify_config.MILVUS_SERVER_PEM_PATH,
server_name=dify_config.MILVUS_SERVER_NAME,
),
)
@@ -163,35 +163,6 @@ def test_init_client_supports_token_and_user_password(milvus_module):
assert user_client.init_kwargs["password"] == "Milvus"
def test_init_client_passes_tls_kwargs_when_secure(milvus_module):
vector = milvus_module.MilvusVector.__new__(milvus_module.MilvusVector)
client = vector._init_client(
milvus_module.MilvusConfig.model_validate(
{
"uri": "https://milvus.example.com:19530",
"token": "abc",
"database": "db",
"secure": True,
"server_pem_path": "/etc/milvus/certs/server.pem",
"server_name": "milvus.example.com",
}
)
)
assert client.init_kwargs["secure"] is True
assert client.init_kwargs["server_pem_path"] == "/etc/milvus/certs/server.pem"
assert client.init_kwargs["server_name"] == "milvus.example.com"
def test_init_client_omits_tls_kwargs_when_not_secure(milvus_module):
vector = milvus_module.MilvusVector.__new__(milvus_module.MilvusVector)
client = vector._init_client(
milvus_module.MilvusConfig.model_validate({"uri": "http://localhost:19530", "token": "abc", "database": "db"})
)
assert "secure" not in client.init_kwargs
assert "server_pem_path" not in client.init_kwargs
assert "server_name" not in client.init_kwargs
def test_init_loads_fields_when_collection_exists(milvus_module):
client = milvus_module.MilvusClient(uri="http://localhost:19530")
client.has_collection.return_value = True
+1 -1
View File
@@ -6,7 +6,7 @@ requires-python = "~=3.12.0"
dependencies = [
# Legacy: mature and widely deployed
"bleach>=6.3.0,<7.0.0",
"boto3>=1.43.14,<2.0.0",
"boto3>=1.43.10,<2.0.0",
"celery>=5.6.3,<6.0.0",
"croniter>=6.2.2,<7.0.0",
"dify-agent",
+1 -1
View File
@@ -112,7 +112,7 @@ def clean_unused_datasets_task():
features_cache_key = f"features:{dataset.tenant_id}"
plan_cache = redis_client.get(features_cache_key)
if plan_cache is None:
features = FeatureService.get_features(dataset.tenant_id, exclude_vector_space=True)
features = FeatureService.get_features(dataset.tenant_id)
redis_client.setex(features_cache_key, 600, features.billing.subscription.plan)
plan = features.billing.subscription.plan
else:
@@ -45,7 +45,7 @@ def mail_clean_document_notify_task():
dataset_auto_disable_logs_map[dataset_auto_disable_log.tenant_id].append(dataset_auto_disable_log)
url = f"{dify_config.CONSOLE_WEB_URL}/datasets"
for tenant_id, tenant_dataset_auto_disable_logs in dataset_auto_disable_logs_map.items():
features = FeatureService.get_features(tenant_id, exclude_vector_space=True)
features = FeatureService.get_features(tenant_id)
plan = features.billing.subscription.plan
if plan != CloudPlan.SANDBOX:
knowledge_details = []
-28
View File
@@ -1287,34 +1287,6 @@ class TenantService:
).scalar_one_or_none()
return row is not None
@staticmethod
def get_account_role_in_tenant(
session: Session | scoped_session,
account_id: uuid.UUID | str | None,
tenant_id: str,
) -> TenantAccountRole | None:
"""Return the caller's role in ``tenant_id``, or ``None`` if not a member.
Backs ``controllers.openapi.auth.role_gate.require_workspace_role``:
the gate maps ``None`` to 404 (non-member no cross-tenant ID leak)
and an out-of-set role to 403, so it never touches the ORM itself.
``None``/empty ``account_id`` short-circuits to ``None`` so SSO
bearers (no account) collapse to the non-member path. Mirrors the
session-injection style of :meth:`account_belongs_to_tenant` rather
than :meth:`get_user_role`, which loads full ``Account``/``Tenant``
objects against the Flask-scoped ``db.session``.
"""
if not account_id:
return None
role = session.execute(
select(TenantAccountJoin.role).where(
TenantAccountJoin.tenant_id == tenant_id,
TenantAccountJoin.account_id == account_id,
)
).scalar_one_or_none()
return TenantAccountRole(role) if role is not None else None
@staticmethod
def get_tenant_by_id(session: Session | scoped_session, tenant_id: str) -> Tenant | None:
"""Plain ``session.get(Tenant, tenant_id)`` — no status filter.
+1 -1
View File
@@ -521,7 +521,7 @@ class AppAnnotationService:
)
# Check annotation quota limit
features = FeatureService.get_features(current_tenant_id, exclude_vector_space=True)
features = FeatureService.get_features(current_tenant_id)
if features.billing.enabled:
annotation_quota_limit = features.annotation_quota_limit
if annotation_quota_limit.limit < len(result) + annotation_quota_limit.size:
-3
View File
@@ -192,9 +192,6 @@ class BillingService:
params["exclude_vector_space"] = "true"
billing_info = cls._send_request("GET", "/subscription/info", params=params)
if exclude_vector_space and billing_info.get("vector_space") is None:
# Unset proto message fields can be serialized as null; the light billing contract treats it as absent.
billing_info.pop("vector_space", None)
return _billing_info_adapter.validate_python(billing_info)
@classmethod
+3 -3
View File
@@ -1325,7 +1325,7 @@ class DatasetService:
def get_dataset_auto_disable_logs(dataset_id: str) -> AutoDisableLogsDict:
assert isinstance(current_user, Account)
assert current_user.current_tenant_id is not None
features = FeatureService.get_features(current_user.current_tenant_id, exclude_vector_space=True)
features = FeatureService.get_features(current_user.current_tenant_id)
if not features.billing.enabled or features.billing.subscription.plan == CloudPlan.SANDBOX:
return {
"document_ids": [],
@@ -2007,7 +2007,7 @@ class DocumentService:
assert isinstance(current_user, Account)
assert current_user.current_tenant_id is not None
features = FeatureService.get_features(current_user.current_tenant_id, exclude_vector_space=True)
features = FeatureService.get_features(current_user.current_tenant_id)
if features.billing.enabled:
if not knowledge_config.original_document_id:
@@ -2798,7 +2798,7 @@ class DocumentService:
assert current_user.current_tenant_id is not None
assert knowledge_config.data_source
features = FeatureService.get_features(current_user.current_tenant_id, exclude_vector_space=True)
features = FeatureService.get_features(current_user.current_tenant_id)
if features.billing.enabled:
count = 0
+1 -1
View File
@@ -41,7 +41,7 @@ class DocumentTaskProxyBase(ABC):
@cached_property
def features(self):
return FeatureService.get_features(self._tenant_id, exclude_vector_space=True)
return FeatureService.get_features(self._tenant_id)
@abstractmethod
def _send_to_direct_queue(self, task_func: Callable[..., Any]):
+1 -4
View File
@@ -130,7 +130,7 @@ class FeatureModel(FeatureResponseModel):
education: EducationModel = EducationModel()
members: LimitationModel = LimitationModel(size=0, limit=1)
apps: LimitationModel = LimitationModel(size=0, limit=10)
vector_space: LimitationModel | None = LimitationModel(size=0, limit=5)
vector_space: LimitationModel = LimitationModel(size=0, limit=5)
knowledge_rate_limit: int = 10
annotation_quota_limit: LimitationModel = LimitationModel(size=0, limit=10)
documents_upload_quota: LimitationModel = LimitationModel(size=0, limit=50)
@@ -188,8 +188,6 @@ class FeatureService:
@classmethod
def get_features(cls, tenant_id: str, exclude_vector_space: bool = False) -> FeatureModel:
features = FeatureModel()
if exclude_vector_space:
features.vector_space = None
cls._fulfill_params_from_env(features)
@@ -349,7 +347,6 @@ class FeatureService:
features.apps.limit = billing_info["apps"]["limit"]
if not exclude_vector_space:
assert features.vector_space is not None
cls._fulfill_vector_space_from_billing_info(features.vector_space, billing_info)
if "documents_upload_quota" in billing_info:
@@ -136,7 +136,7 @@ class EmailDeliveryTestHandler:
) -> DeliveryTestResult:
if not isinstance(method, EmailDeliveryMethod):
raise DeliveryTestUnsupportedError("Delivery method does not support test send.")
features = FeatureService.get_features(context.tenant_id, exclude_vector_space=True)
features = FeatureService.get_features(context.tenant_id)
if not features.human_input_email_delivery_enabled:
raise DeliveryTestError("Email delivery is not available for current plan.")
if not mail.is_inited():
@@ -29,7 +29,7 @@ class RagPipelineTaskProxy:
@cached_property
def features(self):
return FeatureService.get_features(self._dataset_tenant_id, exclude_vector_space=True)
return FeatureService.get_features(self._dataset_tenant_id)
def _upload_invoke_entities(self) -> str:
text = [item.model_dump() for item in self._rag_pipeline_invoke_entities]
-194
View File
@@ -1,194 +0,0 @@
"""Inspector pub/sub fanout for live workflow run updates (Stage 4 §8.5).
The Node Output Inspector exposes a Server-Sent Events stream alongside its
three REST endpoints so the frontend can render per-output progress without
DB polling. This module owns the redis pub/sub channel that connects the two
sides:
* :func:`publish_node_changed` / :func:`publish_workflow_completed`
invoked by :class:`core.app.workflow.layers.persistence.WorkflowPersistenceLayer`
at the very end of each handler, after the DB write has already
succeeded. Publish failures are swallowed so the engine never trips on a
flaky redis connection.
* :func:`subscribe` async iterator the SSE endpoint consumes.
Channel layout
--------------
``dify:inspector:workflow_run:{workflow_run_id}``
One channel per workflow run; the SSE endpoint subscribes for the lifetime of
the run and unsubscribes on the terminal event. Multiple clients can attach
to the same run safely redis pub/sub fans every message out to every
listener.
The message envelope intentionally carries only the *delta* needed to invalidate
a slice of the inspector view; the SSE handler re-reads the canonical
``WorkflowNodeExecutionModel`` row from the DB so we never serialize stale
state across the wire. This means messages stay tiny (~150 bytes) and the
inspector view stays consistent even if a publisher races persistence.
Decision D-5: the on-wire SSE envelope ``{event, data, id}`` is shared with
the babysit chat stream; this module only emits the *internal* pub/sub
message the SSE controller turns it into the public envelope.
"""
from __future__ import annotations
import json
import logging
from collections.abc import Iterator
from dataclasses import asdict, dataclass
from typing import Final, Literal
from extensions.ext_redis import redis_client
logger = logging.getLogger(__name__)
# ──────────────────────────────────────────────────────────────────────────────
# Channel naming
# ──────────────────────────────────────────────────────────────────────────────
_CHANNEL_PREFIX: Final = "dify:inspector:workflow_run"
def channel_for(workflow_run_id: str) -> str:
"""Return the pub/sub channel name for ``workflow_run_id``.
Kept as a module-level helper so tests can pin the channel without
reaching into the publish/subscribe code paths.
"""
return f"{_CHANNEL_PREFIX}:{workflow_run_id}"
# ──────────────────────────────────────────────────────────────────────────────
# Message envelope
# ──────────────────────────────────────────────────────────────────────────────
#: Tags discriminating the wire-level message kinds. Kept narrow so the SSE
#: controller can pattern-match exhaustively.
InspectorMessageKind = Literal["node_changed", "workflow_completed"]
@dataclass(frozen=True, slots=True)
class InspectorMessage:
"""Minimal delta carried across the pub/sub channel.
``node_id`` is set only for ``node_changed`` messages; ``status`` is the
coarse string status straight from the persistence layer (``"running"`` /
``"succeeded"`` / ``"failed"`` for nodes, plus ``"succeeded"`` /
``"failed"`` / ``"partial_succeeded"`` / ``"stopped"`` for workflow runs).
"""
kind: InspectorMessageKind
workflow_run_id: str
node_id: str | None = None
status: str | None = None
def to_json(self) -> str:
return json.dumps(asdict(self), ensure_ascii=False)
@classmethod
def from_json(cls, blob: str) -> InspectorMessage | None:
"""Decode a payload, returning ``None`` for any shape we can't trust."""
try:
decoded = json.loads(blob)
except (json.JSONDecodeError, TypeError):
return None
if not isinstance(decoded, dict):
return None
kind = decoded.get("kind")
if kind not in ("node_changed", "workflow_completed"):
return None
workflow_run_id = decoded.get("workflow_run_id")
if not isinstance(workflow_run_id, str) or not workflow_run_id:
return None
node_id = decoded.get("node_id")
if node_id is not None and not isinstance(node_id, str):
return None
status = decoded.get("status")
if status is not None and not isinstance(status, str):
return None
return cls(kind=kind, workflow_run_id=workflow_run_id, node_id=node_id, status=status)
# ──────────────────────────────────────────────────────────────────────────────
# Publisher (called from the persistence layer)
# ──────────────────────────────────────────────────────────────────────────────
def _publish(message: InspectorMessage) -> None:
"""Best-effort fire-and-forget publish.
Persistence runs inside the workflow engine thread; we never want a redis
glitch to break the workflow. Any exception is logged at debug level so
operators still see them when they grep, but the engine keeps running.
"""
try:
redis_client.publish(channel_for(message.workflow_run_id), message.to_json())
except Exception:
logger.debug("InspectorEventPublisher: publish failed for %s", message.workflow_run_id, exc_info=True)
def publish_node_changed(*, workflow_run_id: str, node_id: str, status: str) -> None:
"""Announce that one node's execution row just changed.
The SSE handler will recompute the node slice from the DB on receipt.
"""
_publish(InspectorMessage(kind="node_changed", workflow_run_id=workflow_run_id, node_id=node_id, status=status))
def publish_workflow_completed(*, workflow_run_id: str, status: str) -> None:
"""Announce that the workflow run reached a terminal state.
The SSE handler emits one last envelope and disconnects.
"""
_publish(InspectorMessage(kind="workflow_completed", workflow_run_id=workflow_run_id, status=status))
# ──────────────────────────────────────────────────────────────────────────────
# Subscriber (consumed by the SSE controller)
# ──────────────────────────────────────────────────────────────────────────────
def subscribe(workflow_run_id: str, *, timeout_seconds: float = 1.0) -> Iterator[InspectorMessage]:
"""Yield ``InspectorMessage`` instances until the consumer abandons us.
The loop polls redis with ``timeout_seconds`` so the SSE handler can
interleave keepalive heartbeats. Yields ``None`` on timeout so the caller
can decide whether to keep blocking; malformed payloads are silently
skipped.
The pub/sub connection is closed when the iterator is garbage-collected
(the wrapping ``finally`` releases it as soon as the SSE handler exits).
"""
pubsub = redis_client.pubsub()
pubsub.subscribe(channel_for(workflow_run_id))
try:
while True:
raw = pubsub.get_message(ignore_subscribe_messages=True, timeout=timeout_seconds)
if raw is None:
# Surface a heartbeat tick — caller can keep-alive or check
# disconnection without blocking redis any longer.
yield InspectorMessage(kind="node_changed", workflow_run_id=workflow_run_id, node_id=None, status=None)
continue
data = raw.get("data") if isinstance(raw, dict) else None
if isinstance(data, bytes):
data = data.decode("utf-8", errors="replace")
if not isinstance(data, str):
continue
message = InspectorMessage.from_json(data)
if message is None:
continue
yield message
finally:
try:
pubsub.unsubscribe(channel_for(workflow_run_id))
pubsub.close()
except Exception:
logger.debug(
"InspectorEventPublisher: pubsub teardown failed for %s",
workflow_run_id,
exc_info=True,
)
@@ -1,712 +0,0 @@
"""Node Output Inspector service (Stage 4 §8).
PRD §Node Output Inspector renames every workflow "Variable" to a "Node Output"
and re-organizes the panel by **producer node** rather than consumer node. This
service backs the new REST surface
``/apps/{app_id}/workflows/draft/runs/{run_id}/node-outputs[/...]`` with three
read-only views:
* :meth:`snapshot_workflow_run` every node + its declared outputs + per-output
status, for one debug workflow run.
* :meth:`node_detail` the same shape filtered down to one node.
* :meth:`output_preview` full payload for one output, with signed download
URL when the output references an upload file.
Design constraints baked into this version:
1. **No new tables** (§8.1). Topology comes from ``WorkflowRun.graph`` (the
graph snapshot taken at execution time so the view stays consistent even
if the draft was edited mid-run). Execution facts come from
``WorkflowNodeExecutionModel`` rows already produced by the workflow
runtime.
2. **Draft + published runs** (decision D-1 lifted 2026-05-26). The Inspector
accepts ``WorkflowRunTriggeredFrom.DEBUGGING`` (draft test runs) as well as
``APP_RUN`` / ``WEBHOOK`` / ``SCHEDULE`` / ``PLUGIN`` / ``RAG_PIPELINE_*``
triggers; the underlying graph + executions are uniform across all of them.
Cross-tenant / cross-app rows still 404 via the standard tenant/app scope.
3. **Declared outputs by node kind**:
* Agent v2 nodes resolve their declared list via
:class:`WorkflowAgentBindingResolver` (the binding owns the canonical
``DeclaredOutputConfig`` list and falls back to PRD defaults when empty).
* Other node kinds don't have a declared-output schema yet; we surface the
keys present in the execution payload as a best-effort list typed
``unknown`` so the panel can still render them.
4. **Per-output status** is derived from the metadata the agent_v2 stack
already records (``output_type_check`` and ``output_check`` blobs); falling
back to the node's overall status when those signals aren't present.
5. **SSE stream** (design §8.5) lives in
:mod:`controllers.console.app.workflow_node_output_inspector` alongside the
REST endpoints. The Inspector and the babysit chat SSE share the
``{event, data, id}`` envelope per decision D-5.
"""
from __future__ import annotations
import json
import logging
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime
from enum import StrEnum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import select
from core.db.session_factory import session_factory
from core.workflow.nodes.agent_v2.binding_resolver import (
WorkflowAgentBindingError,
WorkflowAgentBindingResolver,
)
from core.workflow.nodes.agent_v2.runtime_request_builder import (
WorkflowAgentRuntimeRequestBuilder,
)
from graphon.enums import (
BuiltinNodeTypes,
WorkflowExecutionStatus,
WorkflowNodeExecutionStatus,
)
from graphon.file import helpers as file_helpers
from models import App
from models.agent_config_entities import DeclaredOutputConfig, DeclaredOutputType
from models.workflow import WorkflowNodeExecutionModel, WorkflowRun
logger = logging.getLogger(__name__)
# ──────────────────────────────────────────────────────────────────────────────
# Public dataclasses / enums (Pydantic — these go straight on the wire)
# ──────────────────────────────────────────────────────────────────────────────
class NodeOutputStatus(StrEnum):
"""Lifecycle status of a single declared output within a run."""
PENDING = "pending" # node not started
RUNNING = "running" # node running, output not ready yet
READY = "ready"
TYPE_CHECK_FAILED = "type_check_failed"
OUTPUT_CHECK_FAILED = "output_check_failed"
NOT_PRODUCED = "not_produced" # node succeeded but did not produce this declared output
FAILED = "failed" # node itself failed/exception/stopped
class NodeStatus(StrEnum):
"""Coarse node-level status used by Inspector to pick a banner."""
IDLE = "idle"
RUNNING = "running"
READY = "ready"
FAILED = "failed"
class CheckResultView(BaseModel):
"""``type_check`` / ``output_check`` per-output summary block."""
model_config = ConfigDict(extra="forbid")
passed: bool
reason: str | None = None
class NodeOutputView(BaseModel):
model_config = ConfigDict(extra="forbid")
name: str
type: DeclaredOutputType | None = None
status: NodeOutputStatus
value_preview: Any = None
type_check: CheckResultView | None = None
output_check: CheckResultView | None = None
retried: int = 0
class NodeOutputsView(BaseModel):
model_config = ConfigDict(extra="forbid")
node_id: str
node_kind: str
node_display_name: str
node_status: NodeStatus
node_started_at: datetime | None = None
node_completed_at: datetime | None = None
outputs: list[NodeOutputView] = Field(default_factory=list)
class WorkflowRunSnapshotView(BaseModel):
model_config = ConfigDict(extra="forbid")
workflow_run_id: str
workflow_run_status: WorkflowExecutionStatus
node_outputs: list[NodeOutputsView] = Field(default_factory=list)
class OutputPreviewView(BaseModel):
model_config = ConfigDict(extra="forbid")
node_id: str
output_name: str
type: DeclaredOutputType | None = None
status: NodeOutputStatus
value: Any = None # full value (with signed URL for file refs)
class NodeOutputInspectorError(Exception):
"""Raised when a request cannot be served (404-level conditions)."""
def __init__(self, code: str, message: str) -> None:
super().__init__(message)
self.code = code
# ──────────────────────────────────────────────────────────────────────────────
# Internal helpers — declared outputs per node
# ──────────────────────────────────────────────────────────────────────────────
@dataclass(frozen=True, slots=True)
class _ResolvedDeclaration:
"""Declared output the Inspector should expose, with a normalized type.
``inferred`` is ``True`` when the node kind has no declared-output schema
(we derived the list from the execution payload). The frontend can use
this to dim the type column.
"""
name: str
declared_type: DeclaredOutputType | None
inferred: bool
def _is_agent_v2_node(node: Mapping[str, Any]) -> bool:
"""A graph node is Agent v2 iff its ``data.type`` is the AGENT builtin
AND its ``data.version`` is ``"2"``.
``BuiltinNodeTypes.AGENT`` is a ``ClassVar[NodeType]`` (plain string), not
a StrEnum, so we compare against it directly without ``.value``.
"""
data = node.get("data") or {}
if not isinstance(data, Mapping):
return False
if data.get("type") != BuiltinNodeTypes.AGENT:
return False
return str(data.get("version", "")) == "2"
def _graph_nodes(workflow_run: WorkflowRun) -> list[Mapping[str, Any]]:
"""Parse ``WorkflowRun.graph`` (LongText JSON) into a node list.
The graph blob may be missing / unparseable for very old runs; we treat
that as "no nodes" rather than failing the Inspector, so the panel still
renders an empty state.
"""
if not workflow_run.graph:
return []
try:
parsed = json.loads(workflow_run.graph)
except (json.JSONDecodeError, TypeError):
logger.warning("NodeOutputInspector: workflow_run %s has unparseable graph blob", workflow_run.id)
return []
if not isinstance(parsed, Mapping):
return []
nodes = parsed.get("nodes")
if not isinstance(nodes, list):
return []
return [n for n in nodes if isinstance(n, Mapping) and "id" in n]
# ──────────────────────────────────────────────────────────────────────────────
# Internal helpers — per-output status derivation
# ──────────────────────────────────────────────────────────────────────────────
def _decode_json_blob(blob: str | None) -> Mapping[str, Any] | None:
if not blob:
return None
try:
decoded = json.loads(blob)
except (json.JSONDecodeError, TypeError):
return None
if not isinstance(decoded, Mapping):
return None
return decoded
def _node_status_for(execution: WorkflowNodeExecutionModel | None) -> NodeStatus:
if execution is None:
return NodeStatus.IDLE
if execution.status == WorkflowNodeExecutionStatus.RUNNING:
return NodeStatus.RUNNING
if execution.status == WorkflowNodeExecutionStatus.SUCCEEDED:
return NodeStatus.READY
return NodeStatus.FAILED
def _type_check_by_name(metadata: Mapping[str, Any] | None) -> dict[str, Mapping[str, Any]]:
if not metadata:
return {}
block = metadata.get("output_type_check")
if not isinstance(block, Mapping):
return {}
results = block.get("results") or []
if not isinstance(results, list):
return {}
indexed: dict[str, Mapping[str, Any]] = {}
for r in results:
if isinstance(r, Mapping) and isinstance(r.get("name"), str):
indexed[r["name"]] = r
return indexed
def _output_check_by_name(metadata: Mapping[str, Any] | None) -> dict[str, Mapping[str, Any]]:
if not metadata:
return {}
block = metadata.get("output_check")
if not isinstance(block, Mapping):
return {}
results = block.get("results") or []
if not isinstance(results, list):
return {}
indexed: dict[str, Mapping[str, Any]] = {}
for r in results:
if isinstance(r, Mapping) and isinstance(r.get("name"), str):
indexed[r["name"]] = r
return indexed
def _retried_attempt_count(metadata: Mapping[str, Any] | None) -> int:
"""The agent_v2 runtime records the final attempt index in metadata.
``attempt`` is 0-indexed so a single execution with no retry has
``attempt == 0`` and a Inspector-friendly ``retried == 0``.
"""
if not metadata:
return 0
attempt = metadata.get("attempt")
if isinstance(attempt, int) and attempt > 0:
return attempt
return 0
# ──────────────────────────────────────────────────────────────────────────────
# Value preview (file refs get signed URLs)
# ──────────────────────────────────────────────────────────────────────────────
_PREVIEW_TEXT_LIMIT = 500
_FILE_ID_KEYS: tuple[str, ...] = ("file_id", "upload_file_id", "tool_file_id")
def _looks_like_file_ref(value: Any) -> str | None:
"""Return the resolved ``file_id`` when ``value`` is a file-shaped dict."""
if not isinstance(value, Mapping):
return None
for key in _FILE_ID_KEYS:
candidate = value.get(key)
if isinstance(candidate, str) and candidate:
return candidate
return None
def _value_preview(value: Any) -> Any:
"""Compact preview suitable for the snapshot endpoint.
File refs are augmented with a signed download URL so the panel can render
a thumbnail / link without a second round-trip; long strings are truncated;
other scalar / dict / list shapes are returned as-is (the Pydantic layer
enforces JSON-safety on serialization).
"""
file_id = _looks_like_file_ref(value)
if file_id:
assert isinstance(value, Mapping)
try:
preview_url = file_helpers.get_signed_file_url(upload_file_id=file_id)
except Exception:
logger.warning("NodeOutputInspector: signed URL failed for file_id=%s", file_id, exc_info=True)
preview_url = None
return {**dict(value), "preview_url": preview_url}
if isinstance(value, str) and len(value) > _PREVIEW_TEXT_LIMIT:
return value[:_PREVIEW_TEXT_LIMIT] + ""
return value
def _full_value(value: Any) -> Any:
"""Same shape as :func:`_value_preview` minus the truncation."""
file_id = _looks_like_file_ref(value)
if file_id:
assert isinstance(value, Mapping)
try:
preview_url = file_helpers.get_signed_file_url(upload_file_id=file_id)
except Exception:
logger.warning("NodeOutputInspector: signed URL failed for file_id=%s", file_id, exc_info=True)
preview_url = None
return {**dict(value), "preview_url": preview_url}
return value
# ──────────────────────────────────────────────────────────────────────────────
# Service
# ──────────────────────────────────────────────────────────────────────────────
class NodeOutputInspectorService:
"""Read-only Inspector for draft + published workflow runs.
The service is dependency-light: it holds a single
:class:`WorkflowAgentBindingResolver` so agent v2 nodes can map to their
declared outputs without re-implementing binding lookup. All other I/O
uses the global session factory so workflow runs / executions stay on the
repo-default code path.
Tenancy is enforced via ``app_model.tenant_id`` + ``app_model.id`` on
every load the same scope guard regardless of trigger source.
"""
def __init__(self, binding_resolver: WorkflowAgentBindingResolver | None = None) -> None:
self._binding_resolver = binding_resolver or WorkflowAgentBindingResolver()
# ── public API ────────────────────────────────────────────────────────
def snapshot_workflow_run(self, *, app_model: App, workflow_run_id: str) -> WorkflowRunSnapshotView:
"""Build the per-node snapshot for one debug workflow run."""
workflow_run, executions = self._load_run_and_executions(app_model=app_model, workflow_run_id=workflow_run_id)
executions_by_node = self._index_executions_by_node(executions)
graph_nodes = _graph_nodes(workflow_run)
node_views: list[NodeOutputsView] = []
for raw_node in graph_nodes:
node_id = str(raw_node["id"])
execution = executions_by_node.get(node_id)
view = self._build_node_view(
tenant_id=app_model.tenant_id,
app_id=app_model.id,
workflow_id=workflow_run.workflow_id,
raw_node=raw_node,
execution=execution,
)
node_views.append(view)
return WorkflowRunSnapshotView(
workflow_run_id=workflow_run.id,
workflow_run_status=workflow_run.status,
node_outputs=node_views,
)
def node_detail(self, *, app_model: App, workflow_run_id: str, node_id: str) -> NodeOutputsView:
"""Per-node Inspector entry — returns one ``NodeOutputsView``."""
workflow_run, executions = self._load_run_and_executions(app_model=app_model, workflow_run_id=workflow_run_id)
graph_nodes = _graph_nodes(workflow_run)
raw_node = next((n for n in graph_nodes if str(n.get("id")) == node_id), None)
if raw_node is None:
raise NodeOutputInspectorError(
"node_not_in_workflow_run",
f"Node {node_id!r} does not appear in workflow run {workflow_run_id!r}.",
)
execution = self._index_executions_by_node(executions).get(node_id)
return self._build_node_view(
tenant_id=app_model.tenant_id,
app_id=app_model.id,
workflow_id=workflow_run.workflow_id,
raw_node=raw_node,
execution=execution,
)
def output_preview(
self,
*,
app_model: App,
workflow_run_id: str,
node_id: str,
output_name: str,
) -> OutputPreviewView:
"""Full payload for one declared output (with signed file URL)."""
detail = self.node_detail(
app_model=app_model,
workflow_run_id=workflow_run_id,
node_id=node_id,
)
view = next((o for o in detail.outputs if o.name == output_name), None)
if view is None:
raise NodeOutputInspectorError(
"node_output_not_declared",
f"Output {output_name!r} is not declared on node {node_id!r}.",
)
# ``node_detail`` already produced a truncated value_preview; reload
# the raw value from the execution payload so the preview endpoint can
# return the full thing (still wrapped through ``_full_value`` for
# signed file URLs).
execution = self._index_executions_by_node(
self._load_run_and_executions(app_model=app_model, workflow_run_id=workflow_run_id)[1]
).get(node_id)
full_value: Any = None
if execution is not None:
outputs = _decode_json_blob(execution.outputs) or {}
if output_name in outputs:
full_value = _full_value(outputs[output_name])
return OutputPreviewView(
node_id=node_id,
output_name=output_name,
type=view.type,
status=view.status,
value=full_value,
)
# ── DB loading ────────────────────────────────────────────────────────
def _load_run_and_executions(
self, *, app_model: App, workflow_run_id: str
) -> tuple[WorkflowRun, Sequence[WorkflowNodeExecutionModel]]:
"""Fetch the ``WorkflowRun`` row + every execution that belongs to it.
Enforces:
* row exists,
* row belongs to the app's tenant + app.
The trigger source (DEBUGGING vs. APP_RUN / WEBHOOK / SCHEDULE / ...) is
deliberately not checked here D-1 was lifted 2026-05-26 and the
Inspector now serves both draft and published runs.
"""
with session_factory.create_session() as session:
workflow_run = session.scalar(
select(WorkflowRun).where(
WorkflowRun.id == workflow_run_id,
WorkflowRun.app_id == app_model.id,
WorkflowRun.tenant_id == app_model.tenant_id,
)
)
if workflow_run is None:
raise NodeOutputInspectorError("workflow_run_not_found", "Workflow run not found.")
executions = session.scalars(
select(WorkflowNodeExecutionModel).where(
WorkflowNodeExecutionModel.workflow_run_id == workflow_run_id,
WorkflowNodeExecutionModel.tenant_id == app_model.tenant_id,
WorkflowNodeExecutionModel.app_id == app_model.id,
)
).all()
return workflow_run, executions
@staticmethod
def _index_executions_by_node(
executions: Sequence[WorkflowNodeExecutionModel],
) -> dict[str, WorkflowNodeExecutionModel]:
"""Keep the latest execution per ``node_id``.
A given node may have multiple rows when retries or iterations occur;
``index`` is the per-run sequence counter, so we keep the one with
the highest index as the canonical "current" view.
"""
latest: dict[str, WorkflowNodeExecutionModel] = {}
for execution in executions:
existing = latest.get(execution.node_id)
if existing is None or execution.index > existing.index:
latest[execution.node_id] = execution
return latest
# ── Per-node view construction ────────────────────────────────────────
def _build_node_view(
self,
*,
tenant_id: str,
app_id: str,
workflow_id: str,
raw_node: Mapping[str, Any],
execution: WorkflowNodeExecutionModel | None,
) -> NodeOutputsView:
node_id = str(raw_node["id"])
data = raw_node.get("data") or {}
if not isinstance(data, Mapping):
data = {}
node_kind = str(data.get("type") or (execution.node_type if execution else "") or "unknown")
display_name = str(data.get("title") or (execution.title if execution else node_id))
node_status = _node_status_for(execution)
declarations = self._resolve_declared_outputs(
tenant_id=tenant_id,
app_id=app_id,
workflow_id=workflow_id,
node_id=node_id,
raw_node=raw_node,
execution=execution,
)
outputs_dict = _decode_json_blob(execution.outputs) if execution else None
metadata_dict = _decode_json_blob(execution.execution_metadata) if execution else None
type_check_by_name = _type_check_by_name(metadata_dict)
output_check_by_name = _output_check_by_name(metadata_dict)
retried = _retried_attempt_count(metadata_dict)
output_views: list[NodeOutputView] = []
for declaration in declarations:
output_views.append(
self._build_output_view(
declaration=declaration,
node_status=node_status,
outputs_dict=outputs_dict,
type_check_by_name=type_check_by_name,
output_check_by_name=output_check_by_name,
retried=retried,
)
)
return NodeOutputsView(
node_id=node_id,
node_kind=node_kind,
node_display_name=display_name,
node_status=node_status,
node_started_at=execution.created_at if execution else None,
node_completed_at=execution.finished_at if execution else None,
outputs=output_views,
)
def _build_output_view(
self,
*,
declaration: _ResolvedDeclaration,
node_status: NodeStatus,
outputs_dict: Mapping[str, Any] | None,
type_check_by_name: Mapping[str, Mapping[str, Any]],
output_check_by_name: Mapping[str, Mapping[str, Any]],
retried: int,
) -> NodeOutputView:
name = declaration.name
declared_type = declaration.declared_type
if node_status == NodeStatus.IDLE:
return NodeOutputView(
name=name,
type=declared_type,
status=NodeOutputStatus.PENDING,
retried=retried,
)
if node_status == NodeStatus.RUNNING:
return NodeOutputView(
name=name,
type=declared_type,
status=NodeOutputStatus.RUNNING,
retried=retried,
)
if node_status == NodeStatus.FAILED:
return NodeOutputView(
name=name,
type=declared_type,
status=NodeOutputStatus.FAILED,
retried=retried,
)
# ── node succeeded ────────────────────────────────────────────
type_check_result = type_check_by_name.get(name)
output_check_result = output_check_by_name.get(name)
type_check_view = self._coerce_check_view(type_check_result)
output_check_view = self._coerce_check_view(output_check_result)
# type check loses first; output check next; otherwise ready.
status: NodeOutputStatus
if type_check_result and not _is_passing(type_check_result):
status = NodeOutputStatus.TYPE_CHECK_FAILED
elif output_check_result and not _is_passing(output_check_result):
status = NodeOutputStatus.OUTPUT_CHECK_FAILED
elif outputs_dict is not None and name not in outputs_dict:
status = NodeOutputStatus.NOT_PRODUCED
else:
status = NodeOutputStatus.READY
value_preview = _value_preview(outputs_dict.get(name)) if outputs_dict and name in outputs_dict else None
return NodeOutputView(
name=name,
type=declared_type,
status=status,
value_preview=value_preview,
type_check=type_check_view,
output_check=output_check_view,
retried=retried,
)
@staticmethod
def _coerce_check_view(result: Mapping[str, Any] | None) -> CheckResultView | None:
if not result:
return None
# type_check rows use ``status``; output_check rows use ``status`` too —
# both record per-output state. We treat ``status == "ready"``/"passed"
# as passing and everything else as failing, so the view stays
# stable regardless of which producer wrote the metadata.
return CheckResultView(passed=_is_passing(result), reason=result.get("reason"))
# ── Declared-output resolution ────────────────────────────────────────
def _resolve_declared_outputs(
self,
*,
tenant_id: str,
app_id: str,
workflow_id: str,
node_id: str,
raw_node: Mapping[str, Any],
execution: WorkflowNodeExecutionModel | None,
) -> list[_ResolvedDeclaration]:
if _is_agent_v2_node(raw_node):
agent_decl = self._declared_outputs_for_agent_v2(
tenant_id=tenant_id,
app_id=app_id,
workflow_id=workflow_id,
node_id=node_id,
)
if agent_decl is not None:
return [_ResolvedDeclaration(name=o.name, declared_type=o.type, inferred=False) for o in agent_decl]
# Non-agent (or agent-binding-missing) fall back to inferring from the
# produced payload so the Inspector still has something to show.
return self._infer_outputs_from_payload(execution=execution)
def _declared_outputs_for_agent_v2(
self,
*,
tenant_id: str,
app_id: str,
workflow_id: str,
node_id: str,
) -> list[DeclaredOutputConfig] | None:
try:
bundle = self._binding_resolver.resolve(
tenant_id=tenant_id,
app_id=app_id,
workflow_id=workflow_id,
node_id=node_id,
)
except WorkflowAgentBindingError:
return None
try:
from models.agent_config_entities import WorkflowNodeJobConfig
node_job = WorkflowNodeJobConfig.model_validate(bundle.binding.node_job_config_dict)
except Exception:
logger.warning(
"NodeOutputInspector: malformed node_job_config for binding %s", bundle.binding.id, exc_info=True
)
return None
return list(WorkflowAgentRuntimeRequestBuilder.effective_declared_outputs(list(node_job.declared_outputs)))
@staticmethod
def _infer_outputs_from_payload(*, execution: WorkflowNodeExecutionModel | None) -> list[_ResolvedDeclaration]:
if execution is None:
return []
outputs = _decode_json_blob(execution.outputs)
if not outputs:
return []
return [_ResolvedDeclaration(name=name, declared_type=None, inferred=True) for name in outputs]
def _is_passing(result: Mapping[str, Any]) -> bool:
"""A check-result row is "passing" when its ``status`` is the ready/passed
sentinel emitted by the type-checker / output-check executor."""
status = result.get("status")
if status in {"ready", "passed", "not_produced"}:
return True
return False
+1 -1
View File
@@ -33,7 +33,7 @@ class WorkspaceService:
assert tenant_account_join is not None, "TenantAccountJoin not found"
tenant_info["role"] = tenant_account_join.role
feature = FeatureService.get_features(tenant.id, exclude_vector_space=True)
feature = FeatureService.get_features(tenant.id)
can_replace_logo = feature.can_replace_logo
if can_replace_logo and TenantService.has_roles(tenant, [TenantAccountRole.OWNER, TenantAccountRole.ADMIN]):
+36 -38
View File
@@ -102,13 +102,12 @@ class AppExecutionParams(BaseModel):
workflow_run_id: str | None = None,
):
user_params: _Account | _EndUser
match user:
case Account():
user_params = _Account(user_id=user.id)
case EndUser():
user_params = _EndUser(end_user_id=user.id)
case _:
raise AssertionError("this statement should be unreachable.")
if isinstance(user, Account):
user_params = _Account(user_id=user.id)
elif isinstance(user, EndUser):
user_params = _EndUser(end_user_id=user.id)
else:
raise AssertionError("this statement should be unreachable.")
return cls(
app_id=app_model.id,
workflow_id=workflow.id,
@@ -366,37 +365,36 @@ def _resume_app_execution(payload: dict[str, Any]) -> None:
state_owner_user_id=workflow.created_by,
)
match generate_entity:
case AdvancedChatAppGenerateEntity():
assert conversation is not None
assert message is not None
_resume_advanced_chat(
app_model=app_model,
workflow=workflow,
user=user,
conversation=conversation,
message=message,
generate_entity=generate_entity,
graph_runtime_state=graph_runtime_state,
session_factory=session_factory,
pause_state_config=pause_config,
workflow_run_id=workflow_run_id,
workflow_run=workflow_run,
)
case WorkflowAppGenerateEntity():
_resume_workflow(
app_model=app_model,
workflow=workflow,
user=user,
generate_entity=generate_entity,
graph_runtime_state=graph_runtime_state,
session_factory=session_factory,
pause_state_config=pause_config,
workflow_run_id=workflow_run_id,
workflow_run=workflow_run,
workflow_run_repo=workflow_run_repo,
pause_entity=pause_entity,
)
if isinstance(generate_entity, AdvancedChatAppGenerateEntity):
assert conversation is not None
assert message is not None
_resume_advanced_chat(
app_model=app_model,
workflow=workflow,
user=user,
conversation=conversation,
message=message,
generate_entity=generate_entity,
graph_runtime_state=graph_runtime_state,
session_factory=session_factory,
pause_state_config=pause_config,
workflow_run_id=workflow_run_id,
workflow_run=workflow_run,
)
elif isinstance(generate_entity, WorkflowAppGenerateEntity):
_resume_workflow(
app_model=app_model,
workflow=workflow,
user=user,
generate_entity=generate_entity,
graph_runtime_state=graph_runtime_state,
session_factory=session_factory,
pause_state_config=pause_config,
workflow_run_id=workflow_run_id,
workflow_run=workflow_run,
workflow_run_repo=workflow_run_repo,
pause_entity=pause_entity,
)
def _resume_advanced_chat(
-1
View File
@@ -66,7 +66,6 @@ def _document_indexing(dataset_id: str, document_ids: Sequence[str]):
try:
if features.billing.enabled:
vector_space = features.vector_space
assert vector_space is not None
count = len(document_ids)
batch_upload_limit = int(dify_config.BATCH_UPLOAD_LIMIT)
if features.billing.subscription.plan == CloudPlan.SANDBOX and count > 1:
@@ -92,7 +92,6 @@ def _duplicate_document_indexing_task(dataset_id: str, document_ids: Sequence[st
try:
if features.billing.enabled:
vector_space = features.vector_space
assert vector_space is not None
count = len(document_ids)
if features.billing.subscription.plan == CloudPlan.SANDBOX and count > 1:
raise ValueError("Your current plan does not support batch upload, please upgrade your plan.")
+1 -1
View File
@@ -157,7 +157,7 @@ def dispatch_human_input_email_task(form_id: str, node_title: str | None = None,
if form is None:
logger.warning("Human input form not found, form_id=%s", form_id)
return
features = FeatureService.get_features(form.tenant_id, exclude_vector_space=True)
features = FeatureService.get_features(form.tenant_id)
if not features.human_input_email_delivery_enabled:
logger.info(
"Human input email delivery is not available for tenant=%s, form_id=%s",
@@ -52,7 +52,6 @@ def retry_document_indexing_task(dataset_id: str, document_ids: list[str], user_
try:
if features.billing.enabled:
vector_space = features.vector_space
assert vector_space is not None
if 0 < vector_space.limit <= vector_space.size:
raise ValueError(
"Your total number of documents plus the number of uploads have over the limit of "
@@ -39,7 +39,6 @@ def sync_website_document_indexing_task(dataset_id: str, document_id: str):
try:
if features.billing.enabled:
vector_space = features.vector_space
assert vector_space is not None
if 0 < vector_space.limit <= vector_space.size:
raise ValueError(
"Your total number of documents plus the number of uploads have over the limit of "
@@ -1,475 +0,0 @@
"""End-to-end tests for ``NodeOutputInspectorService`` (Stage 4 §8 / ENG-373).
These integration tests exercise the service against a real Postgres
(``dify-db-1``) same pattern as :mod:`test_remove_app_and_related_data_task`:
seed rows via ``session_factory.create_session()`` with explicit commits,
exercise the service, clean up by ID at teardown.
Coverage:
1. Snapshot for a draft run with one agent v2 node + one tool node
2. Type-check failure visible in snapshot
3. Output-check failure visible in snapshot
4. Published run returns ``published_run_inspector_not_implemented``
5. Cross-tenant access returns ``workflow_run_not_found``
6. File output preview endpoint returns full value with signed URL
7. ``node_detail`` path serves a single node view
"""
from __future__ import annotations
import json
import uuid
from collections.abc import Generator
from datetime import UTC, datetime
from types import SimpleNamespace
from typing import Any
from unittest.mock import patch
import pytest
from sqlalchemy import delete
from core.db.session_factory import session_factory
from graphon.enums import WorkflowExecutionStatus, WorkflowNodeExecutionStatus
from models.enums import CreatorUserRole, WorkflowRunTriggeredFrom
from models.workflow import (
WorkflowNodeExecutionModel,
WorkflowNodeExecutionTriggeredFrom,
WorkflowRun,
WorkflowType,
)
from services.workflow.node_output_inspector_service import (
NodeOutputInspectorError,
NodeOutputInspectorService,
NodeOutputStatus,
NodeStatus,
)
@pytest.fixture
def fake_app_model() -> SimpleNamespace:
"""Lightweight stand-in for the ``App`` model that the service consumes.
``App`` is only read for ``id`` and ``tenant_id``; the service does not
poke at any ORM relationship so a SimpleNamespace is enough and it
keeps us free of needing the ``apps`` row to actually exist (which would
drag in Account / Tenant setup).
"""
return SimpleNamespace(
id=str(uuid.uuid4()),
tenant_id=str(uuid.uuid4()),
)
def _make_workflow_run(
*,
app_id: str,
tenant_id: str,
triggered_from: WorkflowRunTriggeredFrom = WorkflowRunTriggeredFrom.DEBUGGING,
status: WorkflowExecutionStatus = WorkflowExecutionStatus.RUNNING,
graph: dict[str, Any] | None = None,
) -> WorkflowRun:
"""Build a ``WorkflowRun`` row with all required fields populated."""
return WorkflowRun(
id=str(uuid.uuid4()),
tenant_id=tenant_id,
app_id=app_id,
workflow_id=str(uuid.uuid4()),
type=WorkflowType.WORKFLOW,
triggered_from=triggered_from,
version="draft",
graph=json.dumps(graph or {"nodes": []}),
status=status,
created_by_role=CreatorUserRole.ACCOUNT,
created_by=str(uuid.uuid4()),
)
def _make_execution(
*,
app_id: str,
tenant_id: str,
workflow_id: str,
workflow_run_id: str,
node_id: str,
node_type: str = "agent",
title: str = "",
status: WorkflowNodeExecutionStatus = WorkflowNodeExecutionStatus.SUCCEEDED,
outputs: dict[str, Any] | None = None,
execution_metadata: dict[str, Any] | None = None,
index: int = 1,
) -> WorkflowNodeExecutionModel:
"""Build a ``WorkflowNodeExecutionModel`` row with all required fields."""
return WorkflowNodeExecutionModel(
id=str(uuid.uuid4()),
tenant_id=tenant_id,
app_id=app_id,
workflow_id=workflow_id,
triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
workflow_run_id=workflow_run_id,
index=index,
node_id=node_id,
node_type=node_type,
title=title or node_id,
status=status,
outputs=json.dumps(outputs) if outputs is not None else None,
execution_metadata=json.dumps(execution_metadata) if execution_metadata is not None else None,
created_by_role=CreatorUserRole.ACCOUNT,
created_by=str(uuid.uuid4()),
created_at=datetime.now(UTC),
finished_at=datetime.now(UTC),
)
@pytest.fixture
def seeded_run(
flask_req_ctx, fake_app_model: SimpleNamespace
) -> Generator[tuple[SimpleNamespace, WorkflowRun, list[WorkflowNodeExecutionModel]], None, None]:
"""Seed one debug ``WorkflowRun`` + 2 node executions in real Postgres.
Yields ``(app_model, workflow_run, executions)``. Cleans both rows up at
teardown via direct ``DELETE`` so a failed test never leaves debris.
"""
graph = {
"nodes": [
{
"id": "agent-node-1",
"data": {"type": "agent", "version": "2", "title": "My Agent"},
},
{
"id": "tool-node-1",
"data": {"type": "tool", "title": "Slack"},
},
]
}
workflow_run = _make_workflow_run(
app_id=fake_app_model.id,
tenant_id=fake_app_model.tenant_id,
graph=graph,
)
agent_execution = _make_execution(
app_id=fake_app_model.id,
tenant_id=fake_app_model.tenant_id,
workflow_id=workflow_run.workflow_id,
workflow_run_id=workflow_run.id,
node_id="agent-node-1",
node_type="agent",
outputs={"text": "hello world"},
execution_metadata={
"output_type_check": {
"passed": True,
"results": [{"name": "text", "type": "string", "status": "ready"}],
},
"attempt": 0,
},
index=1,
)
tool_execution = _make_execution(
app_id=fake_app_model.id,
tenant_id=fake_app_model.tenant_id,
workflow_id=workflow_run.workflow_id,
workflow_run_id=workflow_run.id,
node_id="tool-node-1",
node_type="tool",
outputs={"message": "sent", "ok": True},
index=2,
)
with session_factory.create_session() as session:
session.add(workflow_run)
session.add(agent_execution)
session.add(tool_execution)
session.commit()
run_id = workflow_run.id
execution_ids = [agent_execution.id, tool_execution.id]
try:
yield fake_app_model, workflow_run, [agent_execution, tool_execution]
finally:
with session_factory.create_session() as session:
session.execute(delete(WorkflowNodeExecutionModel).where(WorkflowNodeExecutionModel.id.in_(execution_ids)))
session.execute(delete(WorkflowRun).where(WorkflowRun.id == run_id))
session.commit()
# ──────────────────────────────────────────────────────────────────────────────
# Stub binding resolver — declared outputs for the agent v2 node
# ──────────────────────────────────────────────────────────────────────────────
def _stub_resolver(declared_outputs_payload: list[dict[str, Any]]):
"""Return a stand-in binding resolver whose ``.resolve`` always returns
one bundle with the supplied declared_outputs.
The real resolver hits ``workflow_agent_node_bindings``; we skip that
table here so the Inspector can be tested without binding-row setup.
"""
binding = SimpleNamespace(
id="binding-1",
node_job_config_dict={
"workflow_prompt": "stub",
"declared_outputs": declared_outputs_payload,
},
)
bundle = SimpleNamespace(binding=binding, agent=None, snapshot=None)
class _Resolver:
def resolve(self, **_: Any):
return bundle
return _Resolver()
# ──────────────────────────────────────────────────────────────────────────────
# Tests
# ──────────────────────────────────────────────────────────────────────────────
def test_snapshot_returns_agent_v2_declared_outputs_with_status_ready(seeded_run):
"""Happy path: agent v2 node + tool node both render, statuses come from
real ``WorkflowRun`` + ``WorkflowNodeExecutionModel`` rows."""
app_model, workflow_run, _ = seeded_run
service = NodeOutputInspectorService(binding_resolver=_stub_resolver([{"name": "text", "type": "string"}]))
snapshot = service.snapshot_workflow_run(
app_model=app_model,
workflow_run_id=workflow_run.id,
)
assert snapshot.workflow_run_id == workflow_run.id
assert snapshot.workflow_run_status == WorkflowExecutionStatus.RUNNING
by_node = {n.node_id: n for n in snapshot.node_outputs}
agent_view = by_node["agent-node-1"]
assert agent_view.node_status == NodeStatus.READY
assert agent_view.outputs[0].name == "text"
assert agent_view.outputs[0].status == NodeOutputStatus.READY
assert agent_view.outputs[0].value_preview == "hello world"
tool_view = by_node["tool-node-1"]
# Tool node's declared outputs are *inferred* from the produced payload.
output_names = sorted(o.name for o in tool_view.outputs)
assert output_names == ["message", "ok"]
assert all(o.type is None for o in tool_view.outputs)
def test_snapshot_404s_for_missing_run(fake_app_model):
"""Service raises ``workflow_run_not_found`` when the row doesn't exist."""
service = NodeOutputInspectorService(binding_resolver=_stub_resolver([]))
with pytest.raises(NodeOutputInspectorError) as exc:
service.snapshot_workflow_run(app_model=fake_app_model, workflow_run_id=str(uuid.uuid4()))
assert exc.value.code == "workflow_run_not_found"
def test_snapshot_404s_for_cross_tenant_access(seeded_run):
"""A wrong-tenant app_model must not see another tenant's run."""
_, workflow_run, _ = seeded_run
intruder = SimpleNamespace(id=str(uuid.uuid4()), tenant_id=str(uuid.uuid4()))
service = NodeOutputInspectorService(binding_resolver=_stub_resolver([]))
with pytest.raises(NodeOutputInspectorError) as exc:
service.snapshot_workflow_run(app_model=intruder, workflow_run_id=workflow_run.id)
assert exc.value.code == "workflow_run_not_found"
def test_snapshot_404s_for_published_run_per_decision_d1(flask_req_ctx, fake_app_model):
"""Decision D-1: published / app-run Inspector deferred to stage 4.1."""
workflow_run = _make_workflow_run(
app_id=fake_app_model.id,
tenant_id=fake_app_model.tenant_id,
triggered_from=WorkflowRunTriggeredFrom.APP_RUN,
graph={"nodes": []},
)
with session_factory.create_session() as session:
session.add(workflow_run)
session.commit()
run_id = workflow_run.id
try:
service = NodeOutputInspectorService(binding_resolver=_stub_resolver([]))
with pytest.raises(NodeOutputInspectorError) as exc:
service.snapshot_workflow_run(app_model=fake_app_model, workflow_run_id=run_id)
assert exc.value.code == "published_run_inspector_not_implemented"
finally:
with session_factory.create_session() as session:
session.execute(delete(WorkflowRun).where(WorkflowRun.id == run_id))
session.commit()
def test_snapshot_surfaces_type_check_failure_from_metadata(flask_req_ctx, fake_app_model):
"""Per-output ``TYPE_CHECK_FAILED`` derived from the metadata blob the
Stage 4 §5 stack records on the execution row."""
graph = {"nodes": [{"id": "agent-1", "data": {"type": "agent", "version": "2"}}]}
workflow_run = _make_workflow_run(app_id=fake_app_model.id, tenant_id=fake_app_model.tenant_id, graph=graph)
execution = _make_execution(
app_id=fake_app_model.id,
tenant_id=fake_app_model.tenant_id,
workflow_id=workflow_run.workflow_id,
workflow_run_id=workflow_run.id,
node_id="agent-1",
outputs={"summary": 123}, # int despite declared string
execution_metadata={
"output_type_check": {
"passed": False,
"results": [
{
"name": "summary",
"type": "string",
"status": "type_check_failed",
"reason": "expected string, got int",
}
],
}
},
)
with session_factory.create_session() as session:
session.add(workflow_run)
session.add(execution)
session.commit()
run_id, execution_id = workflow_run.id, execution.id
try:
service = NodeOutputInspectorService(binding_resolver=_stub_resolver([{"name": "summary", "type": "string"}]))
snapshot = service.snapshot_workflow_run(app_model=fake_app_model, workflow_run_id=run_id)
output = snapshot.node_outputs[0].outputs[0]
assert output.status == NodeOutputStatus.TYPE_CHECK_FAILED
assert output.type_check is not None
assert output.type_check.passed is False
assert output.type_check.reason == "expected string, got int"
finally:
with session_factory.create_session() as session:
session.execute(delete(WorkflowNodeExecutionModel).where(WorkflowNodeExecutionModel.id == execution_id))
session.execute(delete(WorkflowRun).where(WorkflowRun.id == run_id))
session.commit()
def test_snapshot_surfaces_output_check_failure_from_metadata(flask_req_ctx, fake_app_model):
"""When ``output_type_check.passed`` but ``output_check.passed=False``, the
output is flagged ``OUTPUT_CHECK_FAILED``."""
graph = {"nodes": [{"id": "agent-1", "data": {"type": "agent", "version": "2"}}]}
workflow_run = _make_workflow_run(app_id=fake_app_model.id, tenant_id=fake_app_model.tenant_id, graph=graph)
execution = _make_execution(
app_id=fake_app_model.id,
tenant_id=fake_app_model.tenant_id,
workflow_id=workflow_run.workflow_id,
workflow_run_id=workflow_run.id,
node_id="agent-1",
outputs={"report": {"file_id": "550e8400-e29b-41d4-a716-446655440000"}},
execution_metadata={
"output_type_check": {"passed": True, "results": [{"name": "report", "status": "ready"}]},
"output_check": {
"passed": False,
"results": [{"name": "report", "status": "failed", "reason": "benchmark mismatch"}],
},
},
)
with session_factory.create_session() as session:
session.add(workflow_run)
session.add(execution)
session.commit()
run_id, execution_id = workflow_run.id, execution.id
try:
service = NodeOutputInspectorService(binding_resolver=_stub_resolver([{"name": "report", "type": "file"}]))
# Stub signed-URL so we don't depend on the workflow file runtime being
# bound (it isn't, in this minimal flask_req_ctx).
with patch(
"services.workflow.node_output_inspector_service.file_helpers.get_signed_file_url",
return_value="https://signed.example/report",
):
snapshot = service.snapshot_workflow_run(app_model=fake_app_model, workflow_run_id=run_id)
output = snapshot.node_outputs[0].outputs[0]
assert output.status == NodeOutputStatus.OUTPUT_CHECK_FAILED
assert output.output_check is not None
assert output.output_check.passed is False
assert output.output_check.reason == "benchmark mismatch"
finally:
with session_factory.create_session() as session:
session.execute(delete(WorkflowNodeExecutionModel).where(WorkflowNodeExecutionModel.id == execution_id))
session.execute(delete(WorkflowRun).where(WorkflowRun.id == run_id))
session.commit()
def test_node_detail_serves_one_node(seeded_run):
app_model, workflow_run, _ = seeded_run
service = NodeOutputInspectorService(binding_resolver=_stub_resolver([{"name": "text", "type": "string"}]))
view = service.node_detail(
app_model=app_model,
workflow_run_id=workflow_run.id,
node_id="agent-node-1",
)
assert view.node_id == "agent-node-1"
assert view.outputs[0].name == "text"
def test_output_preview_for_file_renders_signed_url(seeded_run, fake_app_model):
"""``preview`` returns the full value with signed_url for file refs."""
# Replace the seeded agent execution's output with a file ref.
_, workflow_run, executions = seeded_run
agent_execution = executions[0]
with session_factory.create_session() as session:
# Re-bind the persisted row so we can mutate + commit.
from sqlalchemy import select
row = session.scalar(
select(WorkflowNodeExecutionModel).where(WorkflowNodeExecutionModel.id == agent_execution.id)
)
assert row is not None
row.outputs = json.dumps({"text": {"file_id": "550e8400-e29b-41d4-a716-446655440000", "filename": "x.pdf"}})
session.commit()
service = NodeOutputInspectorService(binding_resolver=_stub_resolver([{"name": "text", "type": "file"}]))
with patch(
"services.workflow.node_output_inspector_service.file_helpers.get_signed_file_url",
return_value="https://signed.example/x.pdf",
):
preview = service.output_preview(
app_model=fake_app_model,
workflow_run_id=workflow_run.id,
node_id="agent-node-1",
output_name="text",
)
assert preview.output_name == "text"
assert isinstance(preview.value, dict)
assert preview.value["preview_url"] == "https://signed.example/x.pdf"
assert preview.value["filename"] == "x.pdf"
def test_keeps_latest_execution_per_node_by_index(flask_req_ctx, fake_app_model):
"""Multiple executions for the same node_id → service keeps the highest
``index`` (matches the agent_v2 retry pattern that re-emits node
executions)."""
graph = {"nodes": [{"id": "agent-1", "data": {"type": "agent", "version": "2"}}]}
workflow_run = _make_workflow_run(app_id=fake_app_model.id, tenant_id=fake_app_model.tenant_id, graph=graph)
older = _make_execution(
app_id=fake_app_model.id,
tenant_id=fake_app_model.tenant_id,
workflow_id=workflow_run.workflow_id,
workflow_run_id=workflow_run.id,
node_id="agent-1",
outputs={"text": "first attempt"},
index=1,
)
newer = _make_execution(
app_id=fake_app_model.id,
tenant_id=fake_app_model.tenant_id,
workflow_id=workflow_run.workflow_id,
workflow_run_id=workflow_run.id,
node_id="agent-1",
outputs={"text": "second attempt"},
index=5,
)
with session_factory.create_session() as session:
session.add(workflow_run)
session.add(older)
session.add(newer)
session.commit()
run_id, ex_ids = workflow_run.id, [older.id, newer.id]
try:
service = NodeOutputInspectorService(binding_resolver=_stub_resolver([{"name": "text", "type": "string"}]))
snapshot = service.snapshot_workflow_run(app_model=fake_app_model, workflow_run_id=run_id)
assert snapshot.node_outputs[0].outputs[0].value_preview == "second attempt"
finally:
with session_factory.create_session() as session:
session.execute(delete(WorkflowNodeExecutionModel).where(WorkflowNodeExecutionModel.id.in_(ex_ids)))
session.execute(delete(WorkflowRun).where(WorkflowRun.id == run_id))
session.commit()
@@ -19,7 +19,6 @@ def test_get_api_key_auth_data_source(
test_client_with_containers: FlaskClient,
) -> None:
account, tenant = create_console_account_and_tenant(db_session_with_containers)
foreign_account, foreign_tenant = create_console_account_and_tenant(db_session_with_containers)
binding = DataSourceApiKeyAuthBinding(
tenant_id=tenant.id,
category="api_key",
@@ -27,16 +26,8 @@ def test_get_api_key_auth_data_source(
credentials=json.dumps({"auth_type": "api_key", "config": {"api_key": "encrypted"}}),
disabled=False,
)
foreign_binding = DataSourceApiKeyAuthBinding(
tenant_id=foreign_tenant.id,
category="api_key",
provider="foreign_provider",
credentials=json.dumps({"auth_type": "api_key", "config": {"api_key": "encrypted"}}),
disabled=False,
)
db_session_with_containers.add_all([binding, foreign_binding])
db_session_with_containers.add(binding)
db_session_with_containers.commit()
authenticate_console_client(test_client_with_containers, foreign_account)
response = test_client_with_containers.get(
"/console/api/api-key-auth/data-source",
@@ -69,23 +60,20 @@ def test_create_binding_successful(
db_session_with_containers: Session,
test_client_with_containers: FlaskClient,
) -> None:
account, tenant = create_console_account_and_tenant(db_session_with_containers)
tenant_id = tenant.id
payload = {"category": "api_key", "provider": "custom", "credentials": {"key": "value"}}
account, _tenant = create_console_account_and_tenant(db_session_with_containers)
with (
patch("controllers.console.auth.data_source_bearer_auth.ApiKeyAuthService.validate_api_key_auth_args"),
patch("controllers.console.auth.data_source_bearer_auth.ApiKeyAuthService.create_provider_auth") as create_auth,
patch("controllers.console.auth.data_source_bearer_auth.ApiKeyAuthService.create_provider_auth"),
):
response = test_client_with_containers.post(
"/console/api/api-key-auth/data-source/binding",
json=payload,
json={"category": "api_key", "provider": "custom", "credentials": {"key": "value"}},
headers=authenticate_console_client(test_client_with_containers, account),
)
assert response.status_code == 200
assert response.get_json() == {"result": "success"}
create_auth.assert_called_once_with(tenant_id, payload)
def test_create_binding_failure(
@@ -141,35 +129,3 @@ def test_delete_binding_successful(
)
is None
)
def test_delete_binding_scopes_to_authenticated_tenant(
db_session_with_containers: Session,
test_client_with_containers: FlaskClient,
) -> None:
account, _tenant = create_console_account_and_tenant(db_session_with_containers)
foreign_account, foreign_tenant = create_console_account_and_tenant(db_session_with_containers)
foreign_binding = DataSourceApiKeyAuthBinding(
tenant_id=foreign_tenant.id,
category="api_key",
provider="custom_provider",
credentials=json.dumps({"auth_type": "api_key", "config": {"api_key": "encrypted"}}),
disabled=False,
)
db_session_with_containers.add(foreign_binding)
db_session_with_containers.commit()
foreign_binding_id = foreign_binding.id
authenticate_console_client(test_client_with_containers, foreign_account)
response = test_client_with_containers.delete(
f"/console/api/api-key-auth/data-source/{foreign_binding_id}",
headers=authenticate_console_client(test_client_with_containers, account),
)
assert response.status_code == 204
assert (
db_session_with_containers.scalar(
select(DataSourceApiKeyAuthBinding).where(DataSourceApiKeyAuthBinding.id == foreign_binding_id)
)
is not None
)
@@ -1,116 +0,0 @@
"""Integration tests for console external knowledge API endpoints."""
from __future__ import annotations
import json
from flask.testing import FlaskClient
from sqlalchemy.orm import Session
from models.dataset import ExternalKnowledgeApis
from tests.test_containers_integration_tests.controllers.console.helpers import (
authenticate_console_client,
create_console_account_and_tenant,
)
def _create_external_api(
db_session: Session,
*,
tenant_id: str,
account_id: str,
name: str,
) -> ExternalKnowledgeApis:
external_api = ExternalKnowledgeApis(
tenant_id=tenant_id,
created_by=account_id,
updated_by=account_id,
name=name,
description=f"{name} description",
settings=json.dumps(
{
"endpoint": "https://example.com",
"api_key": "test-api-key",
}
),
)
db_session.add(external_api)
db_session.commit()
return external_api
def test_external_api_template_list_filters_paginates_and_scopes_to_authenticated_tenant(
db_session_with_containers: Session,
test_client_with_containers: FlaskClient,
) -> None:
"""Exercise the real list route, including query parsing, DB lookup, and tenant isolation."""
account, tenant = create_console_account_and_tenant(db_session_with_containers)
foreign_account, foreign_tenant = create_console_account_and_tenant(db_session_with_containers)
account_id = account.id
tenant_id = tenant.id
foreign_account_id = foreign_account.id
foreign_tenant_id = foreign_tenant.id
headers = authenticate_console_client(test_client_with_containers, account)
_create_external_api(
db_session_with_containers,
tenant_id=tenant_id,
account_id=account_id,
name="Alpha Primary",
)
_create_external_api(
db_session_with_containers,
tenant_id=tenant_id,
account_id=account_id,
name="Alpha Secondary",
)
_create_external_api(
db_session_with_containers,
tenant_id=tenant_id,
account_id=account_id,
name="Beta Unmatched",
)
_create_external_api(
db_session_with_containers,
tenant_id=foreign_tenant_id,
account_id=foreign_account_id,
name="Alpha Foreign",
)
response = test_client_with_containers.get(
"/console/api/datasets/external-knowledge-api?page=1&limit=1&keyword=Alpha",
headers=headers,
)
assert response.status_code == 200
assert response.json is not None
assert response.json["page"] == 1
assert response.json["limit"] == 1
assert response.json["total"] == 2
assert response.json["has_more"] is True
assert len(response.json["data"]) == 1
first_page_item = response.json["data"][0]
assert first_page_item["tenant_id"] == tenant_id
assert first_page_item["name"] in {"Alpha Primary", "Alpha Secondary"}
assert first_page_item["settings"] == {
"endpoint": "https://example.com",
"api_key": "test-api-key",
}
assert first_page_item["dataset_bindings"] == []
second_response = test_client_with_containers.get(
"/console/api/datasets/external-knowledge-api?page=2&limit=1&keyword=Alpha",
headers=headers,
)
assert second_response.status_code == 200
assert second_response.json is not None
assert second_response.json["page"] == 2
assert second_response.json["limit"] == 1
assert second_response.json["total"] == 2
assert len(second_response.json["data"]) == 1
second_page_item = second_response.json["data"][0]
assert second_page_item["name"] in {"Alpha Primary", "Alpha Secondary"}
assert second_response.json["data"][0]["tenant_id"] == tenant_id
@@ -6,13 +6,11 @@ from unittest.mock import patch
import pytest
from flask.testing import FlaskClient
from sqlalchemy import select
from sqlalchemy.orm import Session
from constants import HIDDEN_VALUE
from libs.rsa import generate_key_pair
from models.api_based_extension import APIBasedExtension
from services.api_based_extension_service import APIBasedExtensionService
from models import Tenant
from tests.test_containers_integration_tests.controllers.console.helpers import (
authenticate_console_client,
create_console_account_and_tenant,
@@ -29,14 +27,13 @@ def _masked_api_key(api_key: str) -> str:
def api_extension_client(
db_session_with_containers: Session,
test_client_with_containers: FlaskClient,
) -> tuple[FlaskClient, dict[str, str], str]:
) -> tuple[FlaskClient, dict[str, str], Tenant]:
account, tenant = create_console_account_and_tenant(db_session_with_containers)
tenant_id = tenant.id
tenant.encrypt_public_key = generate_key_pair(tenant.id)
db_session_with_containers.commit()
headers = authenticate_console_client(test_client_with_containers, account)
return test_client_with_containers, headers, tenant_id
return test_client_with_containers, headers, tenant
@pytest.fixture(autouse=True)
@@ -47,10 +44,9 @@ def mock_api_based_extension_ping():
def test_create_response_masks_plaintext_api_key(
api_extension_client: tuple[FlaskClient, dict[str, str], str],
db_session_with_containers: Session,
api_extension_client: tuple[FlaskClient, dict[str, str], Tenant],
) -> None:
client, headers, tenant_id = api_extension_client
client, headers, _ = api_extension_client
api_key = "plain-secret-12345"
response = client.post(
@@ -66,57 +62,10 @@ def test_create_response_masks_plaintext_api_key(
assert response.status_code == 201
assert response.json is not None
assert response.json["api_key"] == _masked_api_key(api_key)
extension = db_session_with_containers.scalar(
select(APIBasedExtension).where(APIBasedExtension.id == response.json["id"]).limit(1)
)
assert extension is not None
assert extension.tenant_id == tenant_id
def test_list_scopes_api_based_extensions_to_authenticated_tenant(
db_session_with_containers: Session,
test_client_with_containers: FlaskClient,
) -> None:
account, tenant = create_console_account_and_tenant(db_session_with_containers)
account_headers = authenticate_console_client(test_client_with_containers, account)
tenant.encrypt_public_key = generate_key_pair(tenant.id)
_foreign_account, foreign_tenant = create_console_account_and_tenant(db_session_with_containers)
foreign_tenant_id = foreign_tenant.id
foreign_tenant.encrypt_public_key = generate_key_pair(foreign_tenant.id)
db_session_with_containers.commit()
account_create_response = test_client_with_containers.post(
"/console/api/api-based-extension",
headers=account_headers,
json={
"name": "Tenant API",
"api_endpoint": "https://tenant.example.com/hook",
"api_key": "tenant-secret-12345",
},
)
assert account_create_response.status_code == 201
APIBasedExtensionService.save(
APIBasedExtension(
tenant_id=foreign_tenant_id,
name="Foreign API",
api_endpoint="https://foreign.example.com/hook",
api_key="foreign-secret-12345",
)
)
response = test_client_with_containers.get(
"/console/api/api-based-extension",
headers=account_headers,
)
assert response.status_code == 200
assert response.json is not None
assert [item["name"] for item in response.json] == ["Tenant API"]
def test_update_response_masks_new_plaintext_api_key(
api_extension_client: tuple[FlaskClient, dict[str, str], str],
api_extension_client: tuple[FlaskClient, dict[str, str], Tenant],
) -> None:
client, headers, _ = api_extension_client
new_api_key = "new-secret-67890"
@@ -147,7 +96,7 @@ def test_update_response_masks_new_plaintext_api_key(
def test_update_response_masks_existing_plaintext_api_key_when_hidden_value_is_submitted(
api_extension_client: tuple[FlaskClient, dict[str, str], str],
api_extension_client: tuple[FlaskClient, dict[str, str], Tenant],
) -> None:
client, headers, _ = api_extension_client
existing_api_key = "old-secret-12345"
@@ -7,11 +7,9 @@ from unittest.mock import MagicMock, patch
import pytest
from flask import Flask
from flask.testing import FlaskClient
from sqlalchemy import delete, select
from sqlalchemy import delete
from sqlalchemy.orm import Session
from models import Account
from models.account import AccountStatus, TenantAccountRole
from models.enums import ApiTokenType
from models.model import ApiToken, App, AppMode
from tests.test_containers_integration_tests.controllers.console.helpers import (
@@ -60,24 +58,6 @@ class TestAppApiKeyListResource:
assert data["token"].startswith("app-")
assert data["id"] is not None
def test_create_api_key_persists_authenticated_tenant(
self,
setup_app: tuple[FlaskClient, dict[str, str], App],
db_session_with_containers: Session,
) -> None:
client, headers, app = setup_app
tenant_id = app.tenant_id
resp = client.post(f"/console/api/apps/{app.id}/api-keys", headers=headers)
assert resp.status_code == 201
assert resp.json is not None
api_token = db_session_with_containers.scalar(select(ApiToken).where(ApiToken.id == resp.json["id"]))
assert api_token is not None
assert api_token.tenant_id == tenant_id
assert api_token.app_id == app.id
assert api_token.type == ApiTokenType.APP
def test_get_keys_after_create(self, setup_app: tuple[FlaskClient, dict[str, str], App]) -> None:
client, headers, app = setup_app
client.post(f"/console/api/apps/{app.id}/api-keys", headers=headers)
@@ -113,21 +93,6 @@ class TestAppApiKeyListResource:
)
assert resp.status_code == 404
def test_get_foreign_app_keys_not_found(
self,
setup_app: tuple[FlaskClient, dict[str, str], App],
db_session_with_containers: Session,
) -> None:
client, headers, _ = setup_app
foreign_account, foreign_tenant = create_console_account_and_tenant(db_session_with_containers)
foreign_app = create_console_app(
db_session_with_containers, foreign_tenant.id, foreign_account.id, AppMode.CHAT
)
resp = client.get(f"/console/api/apps/{foreign_app.id}/api-keys", headers=headers)
assert resp.status_code == 404
class TestAppApiKeyResource:
"""Tests for DELETE /apps/<resource_id>/api-keys/<api_key_id>."""
@@ -174,13 +139,16 @@ class TestAppApiKeyResource:
resource.resource_model = MagicMock()
resource.resource_id_field = "app_id"
non_admin = Account(name="Normal User", email="normal@example.com", status=AccountStatus.ACTIVE)
non_admin.id = "normal-user"
non_admin.role = TenantAccountRole.NORMAL
non_admin = MagicMock()
non_admin.is_admin_or_owner = False
with (
flask_app_with_containers.test_request_context("/"),
patch(
"controllers.console.apikey.current_account_with_tenant",
return_value=(non_admin, "tenant-id"),
),
patch("controllers.console.apikey._get_resource"),
):
with pytest.raises(Forbidden):
BaseApiKeyResource.delete(resource, "rid", "kid", "tenant-id", non_admin)
BaseApiKeyResource.delete(resource, "rid", "kid")
@@ -1,65 +0,0 @@
"""Integration tests for console feature endpoints."""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import patch
from flask.testing import FlaskClient
from sqlalchemy.orm import Session
from services.feature_service import FeatureModel, FeatureService, LimitationModel
from tests.test_containers_integration_tests.controllers.console.helpers import (
authenticate_console_client,
create_console_account_and_tenant,
)
def test_feature_list_returns_current_tenant_configuration_without_vector_space(
db_session_with_containers: Session,
test_client_with_containers: FlaskClient,
) -> None:
"""Exercise auth, tenant injection, and the feature response shaping contract."""
account, tenant = create_console_account_and_tenant(db_session_with_containers)
tenant_id = tenant.id
headers = authenticate_console_client(test_client_with_containers, account)
feature_model = FeatureModel(
members=LimitationModel(size=1, limit=2),
apps=LimitationModel(size=3, limit=4),
vector_space=LimitationModel(size=5, limit=6),
)
with patch.object(FeatureService, "get_features", return_value=feature_model) as get_features:
response = test_client_with_containers.get(
"/console/api/features",
headers=headers,
)
assert response.status_code == 200
assert response.json is not None
assert response.json["members"] == {"size": 1, "limit": 2}
assert response.json["apps"] == {"size": 3, "limit": 4}
assert "vector_space" not in response.json
get_features.assert_called_once_with(tenant_id, exclude_vector_space=True)
def test_feature_vector_space_returns_current_tenant_usage(
db_session_with_containers: Session,
test_client_with_containers: FlaskClient,
) -> None:
"""Exercise tenant injection and vector-space response serialization through the registered route."""
account, tenant = create_console_account_and_tenant(db_session_with_containers)
tenant_id = tenant.id
headers = authenticate_console_client(test_client_with_containers, account)
vector_space = SimpleNamespace(model_dump=lambda: {"size": 0, "limit": 100})
with patch.object(FeatureService, "get_vector_space", return_value=vector_space) as get_vector_space:
response = test_client_with_containers.get(
"/console/api/features/vector-space",
headers=headers,
)
assert response.status_code == 200
assert response.json == {"size": 0, "limit": 100}
get_vector_space.assert_called_once_with(tenant_id)
@@ -1,101 +0,0 @@
"""Integration tests for console file endpoints."""
from __future__ import annotations
from io import BytesIO
from flask.testing import FlaskClient
from sqlalchemy import select
from sqlalchemy.orm import Session
from configs import dify_config
from models.model import UploadFile
from tests.test_containers_integration_tests.controllers.console.helpers import (
authenticate_console_client,
create_console_account_and_tenant,
)
def test_file_upload_config_returns_console_limits(
db_session_with_containers: Session,
test_client_with_containers: FlaskClient,
) -> None:
"""Exercise the authenticated upload-config route and response contract."""
account, _tenant = create_console_account_and_tenant(db_session_with_containers)
headers = authenticate_console_client(test_client_with_containers, account)
response = test_client_with_containers.get(
"/console/api/files/upload",
headers=headers,
)
assert response.status_code == 200
assert response.json == {
"file_size_limit": dify_config.UPLOAD_FILE_SIZE_LIMIT,
"batch_count_limit": dify_config.UPLOAD_FILE_BATCH_LIMIT,
"file_upload_limit": dify_config.BATCH_UPLOAD_LIMIT,
"image_file_size_limit": dify_config.UPLOAD_IMAGE_FILE_SIZE_LIMIT,
"video_file_size_limit": dify_config.UPLOAD_VIDEO_FILE_SIZE_LIMIT,
"audio_file_size_limit": dify_config.UPLOAD_AUDIO_FILE_SIZE_LIMIT,
"workflow_file_upload_limit": dify_config.WORKFLOW_FILE_UPLOAD_LIMIT,
"image_file_batch_limit": dify_config.IMAGE_FILE_BATCH_LIMIT,
"single_chunk_attachment_limit": dify_config.SINGLE_CHUNK_ATTACHMENT_LIMIT,
"attachment_image_file_size_limit": dify_config.ATTACHMENT_IMAGE_FILE_SIZE_LIMIT,
}
def test_file_upload_persists_file_for_authenticated_current_user(
db_session_with_containers: Session,
test_client_with_containers: FlaskClient,
) -> None:
"""Exercise real upload behavior plus current-user and tenant propagation."""
account, tenant = create_console_account_and_tenant(db_session_with_containers)
account_id = account.id
tenant_id = tenant.id
headers = authenticate_console_client(test_client_with_containers, account)
content = b"hello from console integration"
response = test_client_with_containers.post(
"/console/api/files/upload",
headers=headers,
data={"file": (BytesIO(content), "tenant-owned.txt")},
content_type="multipart/form-data",
)
assert response.status_code == 201
assert response.json is not None
assert response.json["name"] == "tenant-owned.txt"
assert response.json["size"] == len(content)
assert response.json["extension"] == "txt"
assert response.json["mime_type"] == "text/plain"
assert response.json["created_by"] == account_id
upload_file = db_session_with_containers.scalar(
select(UploadFile).where(UploadFile.id == response.json["id"]).limit(1)
)
assert upload_file is not None
assert upload_file.tenant_id == tenant_id
assert upload_file.created_by == account_id
assert upload_file.name == "tenant-owned.txt"
assert upload_file.size == len(content)
assert f"/{tenant_id}/" in upload_file.key
def test_file_upload_rejects_missing_file_after_authentication(
db_session_with_containers: Session,
test_client_with_containers: FlaskClient,
) -> None:
"""Exercise the route's validation path with a real authenticated account."""
account, _tenant = create_console_account_and_tenant(db_session_with_containers)
headers = authenticate_console_client(test_client_with_containers, account)
response = test_client_with_containers.post(
"/console/api/files/upload",
headers=headers,
data={},
content_type="multipart/form-data",
)
assert response.status_code == 400
assert response.json is not None
assert response.json["code"] == "no_file_uploaded"
@@ -127,7 +127,7 @@ class TestEmailDeliveryTestHandler:
monkeypatch.setattr(
service_module.FeatureService,
"get_features",
lambda _tenant_id, **_kwargs: SimpleNamespace(human_input_email_delivery_enabled=False),
lambda _tenant_id: SimpleNamespace(human_input_email_delivery_enabled=False),
)
handler = EmailDeliveryTestHandler(session_factory=MagicMock())
context = DeliveryTestContext(
@@ -142,7 +142,7 @@ class TestEmailDeliveryTestHandler:
monkeypatch.setattr(
service_module.FeatureService,
"get_features",
lambda _id, **_kwargs: SimpleNamespace(human_input_email_delivery_enabled=True),
lambda _id: SimpleNamespace(human_input_email_delivery_enabled=True),
)
monkeypatch.setattr(service_module.mail, "is_inited", lambda: False)
@@ -159,7 +159,7 @@ class TestEmailDeliveryTestHandler:
monkeypatch.setattr(
service_module.FeatureService,
"get_features",
lambda _id, **_kwargs: SimpleNamespace(human_input_email_delivery_enabled=True),
lambda _id: SimpleNamespace(human_input_email_delivery_enabled=True),
)
monkeypatch.setattr(service_module.mail, "is_inited", lambda: True)
@@ -178,7 +178,7 @@ class TestEmailDeliveryTestHandler:
monkeypatch.setattr(
service_module.FeatureService,
"get_features",
lambda _id, **_kwargs: SimpleNamespace(human_input_email_delivery_enabled=True),
lambda _id: SimpleNamespace(human_input_email_delivery_enabled=True),
)
monkeypatch.setattr(service_module.mail, "is_inited", lambda: True)
mock_mail_send = MagicMock()
@@ -214,7 +214,7 @@ class TestEmailDeliveryTestHandler:
monkeypatch.setattr(
service_module.FeatureService,
"get_features",
lambda _id, **_kwargs: SimpleNamespace(human_input_email_delivery_enabled=True),
lambda _id: SimpleNamespace(human_input_email_delivery_enabled=True),
)
monkeypatch.setattr(service_module.mail, "is_inited", lambda: True)
mock_mail_send = MagicMock()

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