refactor: enhance webhook trigger db session management

.github/workflows/pyrefly-diff-comment.yml

@@ -76,11 +76,13 @@ jobs:
               diff += '\\n\\n... (truncated) ...';
             }
 
-            if (diff.trim()) {
-              await github.rest.issues.createComment({
-                issue_number: prNumber,
-                owner: context.repo.owner,
-                repo: context.repo.repo,
-                body: '### Pyrefly Diff\n<details>\n<summary>base → PR</summary>\n\n```diff\n' + diff + '\n```\n</details>',
-              });
-            }
+            const body = diff.trim()
+              ? '### Pyrefly Diff\n<details>\n<summary>base → PR</summary>\n\n```diff\n' + diff + '\n```\n</details>'
+              : '### Pyrefly Diff\nNo changes detected.';
+
+            await github.rest.issues.createComment({
+              issue_number: prNumber,
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              body,
+            });

.github/workflows/web-tests.yml

@@ -89,37 +89,3 @@ jobs:
           flags: web
         env:
           CODECOV_TOKEN: ${{ env.CODECOV_TOKEN }}
-
-  dify-ui-test:
-    name: dify-ui Tests
-    runs-on: ubuntu-latest
-    env:
-      CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
-    defaults:
-      run:
-        shell: bash
-        working-directory: ./packages/dify-ui
-
-    steps:
-      - name: Checkout code
-        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
-        with:
-          persist-credentials: false
-
-      - name: Setup web environment
-        uses: ./.github/actions/setup-web
-
-      - name: Install Chromium for Browser Mode
-        run: vp exec playwright install --with-deps chromium
-
-      - name: Run dify-ui tests
-        run: vp test run --coverage --silent=passed-only
-
-      - name: Report coverage
-        if: ${{ env.CODECOV_TOKEN != '' }}
-        uses: codecov/codecov-action@57e3a136b779b570ffcdbf80b3bdc90e7fab3de2 # v6.0.0
-        with:
-          directory: packages/dify-ui/coverage
-          flags: dify-ui
-        env:
-          CODECOV_TOKEN: ${{ env.CODECOV_TOKEN }}

api/app_factory.py

@@ -159,7 +159,6 @@ def initialize_extensions(app: DifyApp):
         ext_logstore,
         ext_mail,
         ext_migrate,
-        ext_oauth_bearer,
         ext_orjson,
         ext_otel,
         ext_proxy_fix,
@@ -204,7 +203,6 @@ def initialize_extensions(app: DifyApp):
         ext_enterprise_telemetry,
         ext_request_logging,
         ext_session_factory,
-        ext_oauth_bearer,
     ]
     for ext in extensions:
         short_name = ext.__name__.split(".")[-1]

api/commands/account.py

@@ -2,7 +2,6 @@ import base64
 import secrets
 
 import click
-from sqlalchemy.orm import Session
 
 from constants.languages import languages
 from extensions.ext_database import db
@@ -44,11 +43,10 @@ def reset_password(email, new_password, password_confirm):
     # encrypt password with salt
     password_hashed = hash_password(new_password, salt)
     base64_password_hashed = base64.b64encode(password_hashed).decode()
-    with Session(db.engine) as session:
-        account = session.merge(account)
-        account.password = base64_password_hashed
-        account.password_salt = base64_salt
-        session.commit()
+    account = db.session.merge(account)
+    account.password = base64_password_hashed
+    account.password_salt = base64_salt
+    db.session.commit()
     AccountService.reset_login_error_rate_limit(normalized_email)
     click.echo(click.style("Password reset successfully.", fg="green"))
 
@@ -79,10 +77,9 @@ def reset_email(email, new_email, email_confirm):
         click.echo(click.style(f"Invalid email: {new_email}", fg="red"))
         return
 
-    with Session(db.engine) as session:
-        account = session.merge(account)
-        account.email = normalized_new_email
-        session.commit()
+    account = db.session.merge(account)
+    account.email = normalized_new_email
+    db.session.commit()
     click.echo(click.style("Email updated successfully.", fg="green"))
 
 

api/configs/feature/__init__.py

@@ -499,35 +499,6 @@ class HttpConfig(BaseSettings):
     def WEB_API_CORS_ALLOW_ORIGINS(self) -> list[str]:
         return self.inner_WEB_API_CORS_ALLOW_ORIGINS.split(",")
 
-    inner_OPENAPI_CORS_ALLOW_ORIGINS: str = Field(
-        description=(
-            "Comma-separated allowlist for /openapi/v1/* CORS. "
-            "Default empty = same-origin only. Browser-cookie routes within "
-            "the group reject cross-origin OPTIONS regardless of this list."
-        ),
-        validation_alias=AliasChoices("OPENAPI_CORS_ALLOW_ORIGINS"),
-        default="",
-    )
-
-    @computed_field
-    def OPENAPI_CORS_ALLOW_ORIGINS(self) -> list[str]:
-        return [o for o in self.inner_OPENAPI_CORS_ALLOW_ORIGINS.split(",") if o]
-
-    inner_OPENAPI_KNOWN_CLIENT_IDS: str = Field(
-        description=(
-            "Comma-separated client_id values accepted at "
-            "POST /openapi/v1/oauth/device/code. New CLIs / SDKs added here "
-            "without code changes. Unknown client_id returns 400 unsupported_client."
-        ),
-        validation_alias=AliasChoices("OPENAPI_KNOWN_CLIENT_IDS"),
-        default="difyctl",
-    )
-
-    @computed_field  # type: ignore[misc]
-    @property
-    def OPENAPI_KNOWN_CLIENT_IDS(self) -> frozenset[str]:
-        return frozenset(c for c in self.inner_OPENAPI_KNOWN_CLIENT_IDS.split(",") if c)
-
     HTTP_REQUEST_MAX_CONNECT_TIMEOUT: int = Field(
         ge=1, description="Maximum connection timeout in seconds for HTTP requests", default=10
     )
@@ -903,17 +874,6 @@ class AuthConfig(BaseSettings):
         default=86400,
     )
 
-    ENABLE_OAUTH_BEARER: bool = Field(
-        description="Enable OAuth bearer authentication (device-flow + Service API /v1/* bearer middleware).",
-        default=True,
-    )
-
-    OPENAPI_RATE_LIMIT_PER_TOKEN: PositiveInt = Field(
-        description="Per-token rate limit on /openapi/v1/* (requests per minute). "
-        "Bucket keyed on sha256(token), shared across api replicas via Redis.",
-        default=60,
-    )
-
 
 class ModerationConfig(BaseSettings):
     """
@@ -1188,14 +1148,6 @@ class CeleryScheduleTasksConfig(BaseSettings):
         description="Enable scheduled workflow run cleanup task",
         default=False,
     )
-    ENABLE_CLEAN_OAUTH_ACCESS_TOKENS_TASK: bool = Field(
-        description="Enable scheduled cleanup of revoked/expired OAuth access-token rows past retention.",
-        default=True,
-    )
-    OAUTH_ACCESS_TOKEN_RETENTION_DAYS: PositiveInt = Field(
-        description="Days to retain revoked OAuth access-token rows before deletion.",
-        default=30,
-    )
     ENABLE_MAIL_CLEAN_DOCUMENT_NOTIFY_TASK: bool = Field(
         description="Enable mail clean document notify task",
         default=False,

api/constants/dsl_version.py

@@ -1 +0,0 @@
-CURRENT_APP_DSL_VERSION = "0.6.0"

api/controllers/console/app/app.py

@@ -129,7 +129,6 @@ class AppNamePayload(BaseModel):
 
 class AppIconPayload(BaseModel):
     icon: str | None = Field(default=None, description="Icon data")
-    icon_type: IconType | None = Field(default=None, description="Icon type")
     icon_background: str | None = Field(default=None, description="Icon background color")
 
 
@@ -730,12 +729,7 @@ class AppIconApi(Resource):
         args = AppIconPayload.model_validate(console_ns.payload or {})
 
         app_service = AppService()
-        app_model = app_service.update_app_icon(
-            app_model,
-            args.icon or "",
-            args.icon_background or "",
-            args.icon_type,
-        )
+        app_model = app_service.update_app_icon(app_model, args.icon or "", args.icon_background or "")
         response_model = AppDetail.model_validate(app_model, from_attributes=True)
         return response_model.model_dump(mode="json")
 

api/controllers/console/app/conversation_variables.py

@@ -45,7 +45,7 @@ class ConversationVariableResponse(ResponseModel):
     def _normalize_value_type(cls, value: Any) -> str:
         exposed_type = getattr(value, "exposed_type", None)
         if callable(exposed_type):
-            return str(exposed_type())
+            return str(exposed_type().value)
         if isinstance(value, str):
             return value
         try:

api/controllers/console/app/workflow_draft_variable.py

@@ -102,7 +102,7 @@ def _serialize_var_value(variable: WorkflowDraftVariable):
 
 def _serialize_variable_type(workflow_draft_var: WorkflowDraftVariable) -> str:
     value_type = workflow_draft_var.value_type
-    return str(value_type.exposed_type())
+    return value_type.exposed_type().value
 
 
 class FullContentDict(TypedDict):
@@ -122,7 +122,7 @@ def _serialize_full_content(variable: WorkflowDraftVariable) -> FullContentDict
 
     result: FullContentDict = {
         "size_bytes": variable_file.size,
-        "value_type": str(variable_file.value_type.exposed_type()),
+        "value_type": variable_file.value_type.exposed_type().value,
         "length": variable_file.length,
         "download_url": file_helpers.get_signed_file_url(variable_file.upload_file_id, as_attachment=True),
     }
@@ -598,7 +598,7 @@ class EnvironmentVariableCollectionApi(Resource):
                     "name": v.name,
                     "description": v.description,
                     "selector": v.selector,
-                    "value_type": str(v.value_type.exposed_type()),
+                    "value_type": v.value_type.exposed_type().value,
                     "value": v.value,
                     # Do not track edited for env vars.
                     "edited": False,

api/controllers/console/datasets/datasets.py

@@ -50,7 +50,6 @@ from fields.dataset_fields import (
 from fields.document_fields import document_status_fields
 from graphon.model_runtime.entities.model_entities import ModelType
 from libs.login import current_account_with_tenant, login_required
-from libs.url_utils import normalize_api_base_url
 from models import ApiToken, Dataset, Document, DocumentSegment, UploadFile
 from models.dataset import DatasetPermission, DatasetPermissionEnum
 from models.enums import ApiTokenType, SegmentStatus
@@ -890,8 +889,7 @@ class DatasetApiBaseUrlApi(Resource):
     @login_required
     @account_initialization_required
     def get(self):
-        base = dify_config.SERVICE_API_URL or request.host_url.rstrip("/")
-        return {"api_base_url": normalize_api_base_url(base)}
+        return {"api_base_url": (dify_config.SERVICE_API_URL or request.host_url.rstrip("/")) + "/v1"}
 
 
 @console_ns.route("/datasets/retrieval-setting")

api/controllers/console/workspace/tool_providers.py

@@ -1131,14 +1131,6 @@ class ToolMCPAuthApi(Resource):
             with sessionmaker(db.engine).begin() as session:
                 service = MCPToolManageService(session=session)
                 service.clear_provider_credentials(provider_id=provider_id, tenant_id=tenant_id)
-            parsed = urlparse(server_url)
-            sanitized_url = f"{parsed.scheme}://{parsed.hostname}{parsed.path}"
-            logger.warning(
-                "MCP authorization failed for provider %s (url=%s)",
-                provider_id,
-                sanitized_url,
-                exc_info=True,
-            )
             raise ValueError(f"Failed to connect to MCP server: {e}") from e
 
 

api/controllers/inner_api/plugin/wraps.py

@@ -20,13 +20,10 @@ class TenantUserPayload(BaseModel):
 
 def get_user(tenant_id: str, user_id: str | None) -> EndUser:
     """
-    Get current user.
+    Get current user
 
     NOTE: user_id is not trusted, it could be maliciously set to any value.
-    As a result, it could only be considered as an end user id. Even when a
-    concrete end-user ID is supplied, lookups must stay tenant-scoped so one
-    tenant cannot bind another tenant's user record into the plugin request
-    context.
+    As a result, it could only be considered as an end user id.
     """
     if not user_id:
         user_id = DefaultEndUserSessionID.DEFAULT_SESSION_ID
@@ -45,14 +42,7 @@ def get_user(tenant_id: str, user_id: str | None) -> EndUser:
                     .limit(1)
                 )
             else:
-                user_model = session.scalar(
-                    select(EndUser)
-                    .where(
-                        EndUser.id == user_id,
-                        EndUser.tenant_id == tenant_id,
-                    )
-                    .limit(1)
-                )
+                user_model = session.get(EndUser, user_id)
 
             if not user_model:
                 user_model = EndUser(

api/controllers/openapi/__init__.py

@@ -1,41 +0,0 @@
-from flask import Blueprint
-from flask_restx import Namespace
-
-from libs.device_flow_security import attach_anti_framing
-from libs.external_api import ExternalApi
-
-bp = Blueprint("openapi", __name__, url_prefix="/openapi/v1")
-attach_anti_framing(bp)
-
-api = ExternalApi(
-    bp,
-    version="1.0",
-    title="OpenAPI",
-    description="User-scoped programmatic API (bearer auth)",
-)
-
-openapi_ns = Namespace("openapi", description="User-scoped operations", path="/")
-
-from . import (
-    account,
-    app_run,
-    apps,
-    apps_permitted,
-    index,
-    oauth_device,
-    oauth_device_sso,
-    workspaces,
-)
-
-__all__ = [
-    "account",
-    "app_run",
-    "apps",
-    "apps_permitted",
-    "index",
-    "oauth_device",
-    "oauth_device_sso",
-    "workspaces",
-]
-
-api.add_namespace(openapi_ns)

api/controllers/openapi/_audit.py

@@ -1,33 +0,0 @@
-"""Audit emission for openapi app-run endpoints.
-
-Pattern: logger.info with extra={"audit": True, "event": "app.run.openapi", ...}
-matches the existing oauth_device convention. The EE OTel exporter consults
-its own allowlist to decide whether to ship the line.
-"""
-
-from __future__ import annotations
-
-import logging
-
-logger = logging.getLogger(__name__)
-
-EVENT_APP_RUN_OPENAPI = "app.run.openapi"
-
-
-def emit_app_run(*, app_id: str, tenant_id: str, caller_kind: str, mode: str) -> None:
-    logger.info(
-        "audit: %s app_id=%s tenant_id=%s caller_kind=%s mode=%s",
-        EVENT_APP_RUN_OPENAPI,
-        app_id,
-        tenant_id,
-        caller_kind,
-        mode,
-        extra={
-            "audit": True,
-            "event": EVENT_APP_RUN_OPENAPI,
-            "app_id": app_id,
-            "tenant_id": tenant_id,
-            "caller_kind": caller_kind,
-            "mode": mode,
-        },
-    )

api/controllers/openapi/_input_schema.py

@@ -1,143 +0,0 @@
-"""Server-side JSON Schema derivation from Dify `user_input_form`."""
-
-from __future__ import annotations
-
-from typing import Any, cast
-
-from controllers.service_api.app.error import AppUnavailableError
-from models import App
-from models.model import AppMode
-
-JSON_SCHEMA_DRAFT = "https://json-schema.org/draft/2020-12/schema"
-
-EMPTY_INPUT_SCHEMA: dict[str, Any] = {
-    "$schema": JSON_SCHEMA_DRAFT,
-    "type": "object",
-    "properties": {},
-    "required": [],
-}
-
-_CHAT_FAMILY = frozenset({AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT})
-
-
-def _file_object_shape() -> dict[str, Any]:
-    """Single-file value shape. Forward-compat placeholder; refine when file-API contract pins."""
-    return {
-        "type": "object",
-        "properties": {
-            "type": {"type": "string"},
-            "transfer_method": {"type": "string"},
-            "url": {"type": "string"},
-            "upload_file_id": {"type": "string"},
-        },
-        "additionalProperties": True,
-    }
-
-
-def _row_to_schema(row_type: str, row: dict[str, Any]) -> dict[str, Any] | None:
-    label = row.get("label") or row.get("variable", "")
-    base: dict[str, Any] = {"title": label} if label else {}
-
-    if row_type in ("text-input", "paragraph"):
-        out = {"type": "string"} | base
-        max_length = row.get("max_length")
-        if isinstance(max_length, int) and max_length > 0:
-            out["maxLength"] = max_length
-        return out
-
-    if row_type == "select":
-        return {"type": "string"} | base | {"enum": list(row.get("options") or [])}
-
-    if row_type == "number":
-        return {"type": "number"} | base
-
-    if row_type == "file":
-        return _file_object_shape() | base
-
-    if row_type == "file-list":
-        return {
-            "type": "array",
-            "items": _file_object_shape(),
-        } | base
-
-    return None
-
-
-def _form_to_jsonschema(form: list[dict[str, Any]]) -> tuple[dict[str, Any], list[str]]:
-    """Translate a user_input_form row list into (properties, required-list).
-
-    Each row is a single-key dict: `{"text-input": {variable, label, required, ...}}`.
-    Unknown variable types are skipped (forward-compat).
-    """
-    properties: dict[str, Any] = {}
-    required: list[str] = []
-    for row in form:
-        if not isinstance(row, dict) or len(row) != 1:
-            continue
-        ((row_type, row_body),) = row.items()
-        if not isinstance(row_body, dict):
-            continue
-        variable = row_body.get("variable")
-        if not variable:
-            continue
-        schema = _row_to_schema(row_type, row_body)
-        if schema is None:
-            continue
-        properties[variable] = schema
-        if row_body.get("required"):
-            required.append(variable)
-    return properties, required
-
-
-def resolve_app_config(app: App) -> tuple[dict[str, Any], list[dict[str, Any]]]:
-    """Resolve `(features_dict, user_input_form)` for parameters / schema derivation.
-
-    Raises `AppUnavailableError` on misconfigured apps.
-    """
-    if app.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
-        workflow = app.workflow
-        if workflow is None:
-            raise AppUnavailableError()
-        return (
-            workflow.features_dict,
-            cast(list[dict[str, Any]], workflow.user_input_form(to_old_structure=True)),
-        )
-
-    app_model_config = app.app_model_config
-    if app_model_config is None:
-        raise AppUnavailableError()
-    features_dict = cast(dict[str, Any], app_model_config.to_dict())
-    return features_dict, cast(list[dict[str, Any]], features_dict.get("user_input_form", []))
-
-
-def build_input_schema(app: App) -> dict[str, Any]:
-    """Derive Draft 2020-12 JSON Schema from `user_input_form` + app mode.
-
-    chat / agent-chat / advanced-chat: top-level `query` (required, minLength=1) + `inputs` object.
-    completion / workflow: `inputs` object only.
-    Raises `AppUnavailableError` on misconfigured apps.
-    """
-    _, user_input_form = resolve_app_config(app)
-    inputs_props, inputs_required = _form_to_jsonschema(user_input_form)
-
-    properties: dict[str, Any] = {}
-    required: list[str] = []
-
-    if app.mode in _CHAT_FAMILY:
-        properties["query"] = {"type": "string", "minLength": 1}
-        required.append("query")
-
-    properties["inputs"] = {
-        "type": "object",
-        "properties": inputs_props,
-        "required": inputs_required,
-        "additionalProperties": False,
-    }
-    required.append("inputs")
-
-    return {
-        "$schema": JSON_SCHEMA_DRAFT,
-        "type": "object",
-        "properties": properties,
-        "required": required,
-    }

api/controllers/openapi/_models.py

@@ -1,112 +0,0 @@
-"""Shared response substructures for openapi endpoints."""
-
-from __future__ import annotations
-
-from typing import Any, Literal
-
-from pydantic import BaseModel, Field
-
-# Server-side cap on `limit` query param for any /openapi/v1/* list endpoint.
-# Sibling endpoints (`/apps`, `/account/sessions`, future routes) all clamp to
-# this; do not introduce per-endpoint caps without raising the constant.
-MAX_PAGE_LIMIT = 200
-
-
-class UsageInfo(BaseModel):
-    prompt_tokens: int = 0
-    completion_tokens: int = 0
-    total_tokens: int = 0
-
-
-class MessageMetadata(BaseModel):
-    usage: UsageInfo | None = None
-    retriever_resources: list[dict[str, Any]] = []
-
-
-class PaginationEnvelope[T](BaseModel):
-    """Canonical pagination envelope for `/openapi/v1/*` list endpoints."""
-
-    page: int
-    limit: int
-    total: int
-    has_more: bool
-    data: list[T]
-
-    @classmethod
-    def build(cls, *, page: int, limit: int, total: int, items: list[T]) -> PaginationEnvelope[T]:
-        return cls(page=page, limit=limit, total=total, has_more=page * limit < total, data=items)
-
-
-class AppListRow(BaseModel):
-    id: str
-    name: str
-    description: str | None = None
-    mode: str
-    tags: list[dict[str, str]] = []
-    updated_at: str | None = None
-    created_by_name: str | None = None
-    workspace_id: str | None = None
-    workspace_name: str | None = None
-
-
-class AppInfoResponse(BaseModel):
-    id: str
-    name: str
-    description: str | None = None
-    mode: str
-    author: str | None = None
-    tags: list[dict[str, str]] = []
-
-
-class AppDescribeInfo(AppInfoResponse):
-    updated_at: str | None = None
-    service_api_enabled: bool
-
-
-class AppDescribeResponse(BaseModel):
-    info: AppDescribeInfo | None = None
-    parameters: dict[str, Any] | None = None
-    input_schema: dict[str, Any] | None = None
-
-
-class ChatMessageResponse(BaseModel):
-    event: str
-    task_id: str
-    id: str
-    message_id: str
-    conversation_id: str
-    mode: str
-    answer: str
-    metadata: MessageMetadata = Field(default_factory=MessageMetadata)
-    created_at: int
-
-
-class CompletionMessageResponse(BaseModel):
-    event: str
-    task_id: str
-    id: str
-    message_id: str
-    mode: str
-    answer: str
-    metadata: MessageMetadata = Field(default_factory=MessageMetadata)
-    created_at: int
-
-
-class WorkflowRunData(BaseModel):
-    id: str
-    workflow_id: str
-    status: str
-    outputs: dict[str, Any] = Field(default_factory=dict)
-    error: str | None = None
-    elapsed_time: float | None = None
-    total_tokens: int | None = None
-    total_steps: int | None = None
-    created_at: int | None = None
-    finished_at: int | None = None
-
-
-class WorkflowRunResponse(BaseModel):
-    workflow_run_id: str
-    task_id: str
-    mode: Literal["workflow"] = "workflow"
-    data: WorkflowRunData

api/controllers/openapi/account.py

@@ -1,236 +0,0 @@
-"""User-scoped account endpoints. /account is the bearer-authed
-identity read; /account/sessions and /account/sessions/<id> manage
-the user's active OAuth tokens.
-"""
-
-from __future__ import annotations
-
-from datetime import UTC, datetime
-
-from flask import g, request
-from flask_restx import Resource
-from sqlalchemy import and_, select, update
-from werkzeug.exceptions import BadRequest, NotFound
-
-from controllers.openapi import openapi_ns
-from controllers.openapi._models import MAX_PAGE_LIMIT, PaginationEnvelope
-from extensions.ext_database import db
-from extensions.ext_redis import redis_client
-from libs.oauth_bearer import (
-    ACCEPT_USER_ANY,
-    TOKEN_CACHE_KEY_FMT,
-    AuthContext,
-    SubjectType,
-    validate_bearer,
-)
-from libs.rate_limit import (
-    LIMIT_ME_PER_ACCOUNT,
-    LIMIT_ME_PER_EMAIL,
-    enforce,
-)
-from models import Account, OAuthAccessToken, Tenant, TenantAccountJoin
-
-
-@openapi_ns.route("/account")
-class AccountApi(Resource):
-    @validate_bearer(accept=ACCEPT_USER_ANY)
-    def get(self):
-        ctx = g.auth_ctx
-
-        if ctx.subject_type == SubjectType.EXTERNAL_SSO:
-            enforce(LIMIT_ME_PER_EMAIL, key=f"subject:{ctx.subject_email}")
-        else:
-            enforce(LIMIT_ME_PER_ACCOUNT, key=f"account:{ctx.account_id}")
-
-        if ctx.subject_type == SubjectType.EXTERNAL_SSO:
-            return {
-                "subject_type": ctx.subject_type,
-                "subject_email": ctx.subject_email,
-                "subject_issuer": ctx.subject_issuer,
-                "account": None,
-                "workspaces": [],
-                "default_workspace_id": None,
-            }
-
-        account = (
-            db.session.query(Account).filter(Account.id == ctx.account_id).one_or_none() if ctx.account_id else None
-        )
-        memberships = _load_memberships(ctx.account_id) if ctx.account_id else []
-        default_ws_id = _pick_default_workspace(memberships)
-
-        return {
-            "subject_type": ctx.subject_type,
-            "subject_email": ctx.subject_email or (account.email if account else None),
-            "account": _account_payload(account) if account else None,
-            "workspaces": [_workspace_payload(m) for m in memberships],
-            "default_workspace_id": default_ws_id,
-        }
-
-
-@openapi_ns.route("/account/sessions/self")
-class AccountSessionsSelfApi(Resource):
-    @validate_bearer(accept=ACCEPT_USER_ANY)
-    def delete(self):
-        ctx = g.auth_ctx
-        _require_oauth_subject(ctx)
-        _revoke_token_by_id(str(ctx.token_id))
-        return {"status": "revoked"}, 200
-
-
-@openapi_ns.route("/account/sessions")
-class AccountSessionsApi(Resource):
-    @validate_bearer(accept=ACCEPT_USER_ANY)
-    def get(self):
-        ctx = g.auth_ctx
-        now = datetime.now(UTC)
-        page = int(request.args.get("page", "1"))
-        limit = min(int(request.args.get("limit", "100")), MAX_PAGE_LIMIT)
-
-        all_rows = db.session.execute(
-            select(
-                OAuthAccessToken.id,
-                OAuthAccessToken.prefix,
-                OAuthAccessToken.client_id,
-                OAuthAccessToken.device_label,
-                OAuthAccessToken.created_at,
-                OAuthAccessToken.last_used_at,
-                OAuthAccessToken.expires_at,
-            )
-            .where(
-                and_(
-                    *_subject_match(ctx),
-                    OAuthAccessToken.revoked_at.is_(None),
-                    OAuthAccessToken.token_hash.is_not(None),
-                    OAuthAccessToken.expires_at > now,
-                )
-            )
-            .order_by(OAuthAccessToken.created_at.desc())
-        ).all()
-
-        total = len(all_rows)
-        sliced = all_rows[(page - 1) * limit : page * limit]
-
-        items = [
-            {
-                "id": str(r.id),
-                "prefix": r.prefix,
-                "client_id": r.client_id,
-                "device_label": r.device_label,
-                "created_at": _iso(r.created_at),
-                "last_used_at": _iso(r.last_used_at),
-                "expires_at": _iso(r.expires_at),
-            }
-            for r in sliced
-        ]
-
-        return (
-            PaginationEnvelope.build(page=page, limit=limit, total=total, items=items).model_dump(mode="json"),
-            200,
-        )
-
-
-@openapi_ns.route("/account/sessions/<string:session_id>")
-class AccountSessionByIdApi(Resource):
-    @validate_bearer(accept=ACCEPT_USER_ANY)
-    def delete(self, session_id: str):
-        ctx = g.auth_ctx
-        _require_oauth_subject(ctx)
-
-        # Subject-match guard. 404 (not 403) on cross-subject so the
-        # endpoint doesn't leak token IDs that belong to other subjects.
-        owns = db.session.execute(
-            select(OAuthAccessToken.id).where(
-                and_(
-                    OAuthAccessToken.id == session_id,
-                    *_subject_match(ctx),
-                )
-            )
-        ).first()
-        if owns is None:
-            raise NotFound("session not found")
-
-        _revoke_token_by_id(session_id)
-        return {"status": "revoked"}, 200
-
-
-def _subject_match(ctx: AuthContext) -> tuple:
-    """Where-clauses that scope a query to the bearer's subject. Works
-    for both account (account_id) and external_sso (email + issuer).
-    """
-    if ctx.subject_type == SubjectType.ACCOUNT:
-        return (OAuthAccessToken.account_id == str(ctx.account_id),)
-    return (
-        OAuthAccessToken.subject_email == ctx.subject_email,
-        OAuthAccessToken.subject_issuer == ctx.subject_issuer,
-        OAuthAccessToken.account_id.is_(None),
-    )
-
-
-def _require_oauth_subject(ctx: AuthContext) -> None:
-    if not ctx.source.startswith("oauth"):
-        raise BadRequest(
-            "this endpoint revokes OAuth bearer tokens; use /openapi/v1/personal-access-tokens/self for PATs"
-        )
-
-
-def _revoke_token_by_id(token_id: str) -> None:
-    # Snapshot pre-revoke hash for cache invalidation; UPDATE WHERE
-    # makes double-revoke idempotent.
-    row = (
-        db.session.query(OAuthAccessToken.token_hash)
-        .filter(
-            OAuthAccessToken.id == token_id,
-            OAuthAccessToken.revoked_at.is_(None),
-        )
-        .one_or_none()
-    )
-    pre_revoke_hash = row[0] if row else None
-
-    stmt = (
-        update(OAuthAccessToken)
-        .where(
-            OAuthAccessToken.id == token_id,
-            OAuthAccessToken.revoked_at.is_(None),
-        )
-        .values(revoked_at=datetime.now(UTC), token_hash=None)
-    )
-    db.session.execute(stmt)
-    db.session.commit()
-
-    if pre_revoke_hash:
-        redis_client.delete(TOKEN_CACHE_KEY_FMT.format(hash=pre_revoke_hash))
-
-
-def _iso(dt: datetime | None) -> str | None:
-    if dt is None:
-        return None
-    if dt.tzinfo is None:
-        dt = dt.replace(tzinfo=UTC)
-    return dt.isoformat().replace("+00:00", "Z")
-
-
-def _load_memberships(account_id):
-    return (
-        db.session.query(TenantAccountJoin, Tenant)
-        .join(Tenant, Tenant.id == TenantAccountJoin.tenant_id)
-        .filter(TenantAccountJoin.account_id == account_id)
-        .all()
-    )
-
-
-def _pick_default_workspace(memberships) -> str | None:
-    if not memberships:
-        return None
-    for join, tenant in memberships:
-        if getattr(join, "current", False):
-            return str(tenant.id)
-    return str(memberships[0][1].id)
-
-
-def _workspace_payload(row) -> dict:
-    join, tenant = row
-    return {"id": str(tenant.id), "name": tenant.name, "role": getattr(join, "role", "")}
-
-
-def _account_payload(account) -> dict:
-    return {"id": str(account.id), "email": account.email, "name": account.name}

api/controllers/openapi/app_run.py

@@ -1,198 +0,0 @@
-"""POST /openapi/v1/apps/<app_id>/run — mode-agnostic runner."""
-
-from __future__ import annotations
-
-import logging
-from collections.abc import Callable, Iterator, Mapping
-from contextlib import contextmanager
-from typing import Any, Literal
-from uuid import UUID
-
-from flask import request
-from flask_restx import Resource
-from pydantic import BaseModel, ValidationError, field_validator
-from werkzeug.exceptions import BadRequest, HTTPException, InternalServerError, NotFound, UnprocessableEntity
-
-import services
-from controllers.openapi import openapi_ns
-from controllers.openapi._audit import emit_app_run
-from controllers.openapi._models import (
-    ChatMessageResponse,
-    CompletionMessageResponse,
-    WorkflowRunResponse,
-)
-from controllers.openapi.auth.composition import OAUTH_BEARER_PIPELINE
-from controllers.service_api.app.error import (
-    AppUnavailableError,
-    CompletionRequestError,
-    ConversationCompletedError,
-    ProviderModelCurrentlyNotSupportError,
-    ProviderNotInitializeError,
-    ProviderQuotaExceededError,
-)
-from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
-from core.app.entities.app_invoke_entities import InvokeFrom
-from core.errors.error import (
-    ModelCurrentlyNotSupportError,
-    ProviderTokenNotInitError,
-    QuotaExceededError,
-)
-from graphon.model_runtime.errors.invoke import InvokeError
-from libs import helper
-from libs.helper import UUIDStrOrEmpty
-from libs.oauth_bearer import Scope
-from models.model import App, AppMode
-from services.app_generate_service import AppGenerateService
-from services.errors.app import (
-    IsDraftWorkflowError,
-    WorkflowIdFormatError,
-    WorkflowNotFoundError,
-)
-from services.errors.llm import InvokeRateLimitError
-
-logger = logging.getLogger(__name__)
-
-
-class AppRunRequest(BaseModel):
-    inputs: dict[str, Any]
-    query: str | None = None
-    files: list[dict[str, Any]] | None = None
-    response_mode: Literal["blocking", "streaming"] | None = None
-    conversation_id: UUIDStrOrEmpty | None = None
-    auto_generate_name: bool = True
-    workflow_id: str | None = None
-
-    @field_validator("conversation_id", mode="before")
-    @classmethod
-    def _normalize_conv(cls, value: str | UUID | None) -> str | None:
-        if isinstance(value, str):
-            value = value.strip()
-        if not value:
-            return None
-        try:
-            return helper.uuid_value(value)
-        except ValueError as exc:
-            raise ValueError("conversation_id must be a valid UUID") from exc
-
-
-@contextmanager
-def _translate_service_errors() -> Iterator[None]:
-    try:
-        yield
-    except WorkflowNotFoundError as ex:
-        raise NotFound(str(ex))
-    except (IsDraftWorkflowError, WorkflowIdFormatError) as ex:
-        raise BadRequest(str(ex))
-    except services.errors.conversation.ConversationNotExistsError:
-        raise NotFound("Conversation Not Exists.")
-    except services.errors.conversation.ConversationCompletedError:
-        raise ConversationCompletedError()
-    except services.errors.app_model_config.AppModelConfigBrokenError:
-        logger.exception("App model config broken.")
-        raise AppUnavailableError()
-    except ProviderTokenNotInitError as ex:
-        raise ProviderNotInitializeError(ex.description)
-    except QuotaExceededError:
-        raise ProviderQuotaExceededError()
-    except ModelCurrentlyNotSupportError:
-        raise ProviderModelCurrentlyNotSupportError()
-    except InvokeRateLimitError as ex:
-        raise InvokeRateLimitHttpError(ex.description)
-    except InvokeError as e:
-        raise CompletionRequestError(e.description)
-
-
-def _unpack_blocking(response: Any) -> Mapping[str, Any]:
-    if isinstance(response, tuple):
-        response = response[0]
-    if not isinstance(response, Mapping):
-        raise InternalServerError("blocking generate returned non-mapping response")
-    return response
-
-
-def _generate(app: App, caller: Any, args: dict[str, Any], streaming: bool):
-    return AppGenerateService.generate(
-        app_model=app,
-        user=caller,
-        args=args,
-        invoke_from=InvokeFrom.OPENAPI,
-        streaming=streaming,
-    )
-
-
-def _run_chat(app: App, caller: Any, payload: AppRunRequest, streaming: bool):
-    if not payload.query or not payload.query.strip():
-        raise UnprocessableEntity("query_required_for_chat")
-    args = payload.model_dump(exclude_none=True)
-    with _translate_service_errors():
-        response = _generate(app, caller, args, streaming)
-    if streaming:
-        return response, None
-    return None, ChatMessageResponse.model_validate(_unpack_blocking(response)).model_dump(mode="json")
-
-
-def _run_completion(app: App, caller: Any, payload: AppRunRequest, streaming: bool):
-    args = payload.model_dump(exclude_none=True)
-    args["auto_generate_name"] = False
-    args.setdefault("query", "")
-    with _translate_service_errors():
-        response = _generate(app, caller, args, streaming)
-    if streaming:
-        return response, None
-    return None, CompletionMessageResponse.model_validate(_unpack_blocking(response)).model_dump(mode="json")
-
-
-def _run_workflow(app: App, caller: Any, payload: AppRunRequest, streaming: bool):
-    if payload.query is not None:
-        raise UnprocessableEntity("query_not_supported_for_workflow")
-    args = payload.model_dump(exclude={"query", "conversation_id", "auto_generate_name"}, exclude_none=True)
-    with _translate_service_errors():
-        response = _generate(app, caller, args, streaming)
-    if streaming:
-        return response, None
-    return None, WorkflowRunResponse.model_validate(_unpack_blocking(response)).model_dump(mode="json")
-
-
-_DISPATCH: dict[AppMode, Callable[[App, Any, AppRunRequest, bool], tuple[Any, dict[str, Any] | None]]] = {
-    AppMode.CHAT: _run_chat,
-    AppMode.AGENT_CHAT: _run_chat,
-    AppMode.ADVANCED_CHAT: _run_chat,
-    AppMode.COMPLETION: _run_completion,
-    AppMode.WORKFLOW: _run_workflow,
-}
-
-
-@openapi_ns.route("/apps/<string:app_id>/run")
-class AppRunApi(Resource):
-    @OAUTH_BEARER_PIPELINE.guard(scope=Scope.APPS_RUN)
-    def post(self, app_id: str, app_model: App, caller, caller_kind: str):
-        body = request.get_json(silent=True) or {}
-        body.pop("user", None)
-        try:
-            payload = AppRunRequest.model_validate(body)
-        except ValidationError as exc:
-            raise UnprocessableEntity(exc.json())
-
-        handler = _DISPATCH.get(app_model.mode)
-        if handler is None:
-            raise UnprocessableEntity("mode_not_runnable")
-
-        streaming = payload.response_mode == "streaming"
-        try:
-            stream_obj, blocking_body = handler(app_model, caller, payload, streaming)
-        except HTTPException:
-            raise
-        except Exception:
-            logger.exception("internal server error.")
-            raise InternalServerError()
-
-        emit_app_run(
-            app_id=app_model.id,
-            tenant_id=app_model.tenant_id,
-            caller_kind=caller_kind,
-            mode=str(app_model.mode),
-        )
-
-        if streaming:
-            return helper.compact_generate_response(stream_obj)
-        return blocking_body, 200

api/controllers/openapi/apps.py

@@ -1,315 +0,0 @@
-"""GET /openapi/v1/apps and per-app reads.
-
-Decorator order: `method_decorators` is innermost-first. `validate_bearer`
-is last → outermost → sets `g.auth_ctx` before `require_scope` reads it.
-"""
-
-from __future__ import annotations
-
-import uuid as _uuid
-from typing import Any
-
-import sqlalchemy as sa
-from flask import g, request
-from flask_restx import Resource
-from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator
-from werkzeug.exceptions import Conflict, NotFound, UnprocessableEntity
-
-from controllers.common.fields import Parameters
-from controllers.openapi import openapi_ns
-from controllers.openapi._input_schema import EMPTY_INPUT_SCHEMA, build_input_schema, resolve_app_config
-from controllers.openapi._models import (
-    MAX_PAGE_LIMIT,
-    AppDescribeInfo,
-    AppDescribeResponse,
-    AppListRow,
-    PaginationEnvelope,
-)
-from controllers.service_api.app.error import AppUnavailableError
-from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
-from extensions.ext_database import db
-from libs.oauth_bearer import (
-    ACCEPT_USER_ANY,
-    AuthContext,
-    Scope,
-    SubjectType,
-    require_scope,
-    require_workspace_member,
-    validate_bearer,
-)
-from models import App, Tenant
-from models.model import AppMode
-from services.app_service import AppService
-from services.tag_service import TagService
-
-_APPS_READ_DECORATORS = [
-    require_scope(Scope.APPS_READ),
-    validate_bearer(accept=ACCEPT_USER_ANY),
-]
-
-_ALLOWED_DESCRIBE_FIELDS: frozenset[str] = frozenset({"info", "parameters", "input_schema"})
-
-
-class AppDescribeQuery(BaseModel):
-    """`?fields=` allow-list for GET /apps/<id>/describe.
-
-    Empty / omitted → all blocks. Unknown member → ValidationError → 422.
-    """
-
-    model_config = ConfigDict(extra="forbid")
-
-    fields: set[str] | None = None
-    workspace_id: str | None = None
-
-    @field_validator("workspace_id", mode="before")
-    @classmethod
-    def _validate_workspace_id(cls, v: object) -> str | None:
-        if v is None or v == "":
-            return None
-        if not isinstance(v, str):
-            raise ValueError("workspace_id must be a string")
-        try:
-            _uuid.UUID(v)
-        except ValueError:
-            raise ValueError("workspace_id must be a valid UUID")
-        return v
-
-    @field_validator("fields", mode="before")
-    @classmethod
-    def _parse_fields(cls, v: object) -> set[str] | None:
-        if v is None or v == "":
-            return None
-        if not isinstance(v, str):
-            raise ValueError("fields must be a comma-separated string")
-        members = {m.strip() for m in v.split(",") if m.strip()}
-        unknown = members - _ALLOWED_DESCRIBE_FIELDS
-        if unknown:
-            raise ValueError(f"unknown field(s): {sorted(unknown)}")
-        return members
-
-
-_EMPTY_PARAMETERS: dict[str, Any] = {
-    "opening_statement": None,
-    "suggested_questions": [],
-    "user_input_form": [],
-    "file_upload": None,
-    "system_parameters": {},
-}
-
-
-class AppReadResource(Resource):
-    """Base for per-app read endpoints; subclasses call `_load()` for SSO/membership/exists checks."""
-
-    method_decorators = _APPS_READ_DECORATORS
-
-    def _load(self, app_id: str, workspace_id: str | None = None) -> tuple[App, AuthContext]:
-        ctx = g.auth_ctx
-        if ctx.subject_type != SubjectType.ACCOUNT or ctx.account_id is None:
-            raise NotFound("app not found")
-
-        try:
-            parsed_uuid = _uuid.UUID(app_id)
-            is_uuid = True
-        except ValueError:
-            parsed_uuid = None
-            is_uuid = False
-
-        if is_uuid:
-            app = db.session.get(App, str(parsed_uuid))  # normalised dashed form
-            if not app or app.status != "normal":
-                raise NotFound("app not found")
-        else:
-            if not workspace_id:
-                raise UnprocessableEntity("workspace_id is required for name-based lookup")
-            matches = list(
-                db.session.execute(
-                    sa.select(App).where(
-                        App.name == app_id,
-                        App.tenant_id == workspace_id,
-                        App.status == "normal",
-                    )
-                ).scalars()
-            )
-            if len(matches) == 0:
-                raise NotFound("app not found")
-            if len(matches) > 1:
-                lines = [f"app name {app_id!r} is ambiguous — re-run with a UUID:\n\n"]
-                lines.append(f"  {'ID':<36}  {'MODE':<12}  NAME\n")
-                for m in matches:
-                    lines.append(f"  {str(m.id):<36}  {str(m.mode.value):<12}  {m.name}\n")
-                raise Conflict("".join(lines))
-            app = matches[0]
-
-        require_workspace_member(ctx, str(app.tenant_id))
-        return app, ctx
-
-
-def parameters_payload(app: App) -> dict:
-    """Mirrors service_api/app/app.py::AppParameterApi response body."""
-    features_dict, user_input_form = resolve_app_config(app)
-    parameters = get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
-    return Parameters.model_validate(parameters).model_dump(mode="json")
-
-
-@openapi_ns.route("/apps/<string:app_id>/describe")
-class AppDescribeApi(AppReadResource):
-    def get(self, app_id: str):
-        try:
-            query = AppDescribeQuery.model_validate(request.args.to_dict(flat=True))
-        except ValidationError as exc:
-            raise UnprocessableEntity(exc.json())
-
-        app, _ = self._load(app_id, workspace_id=query.workspace_id)
-
-        requested = query.fields
-        want_info = requested is None or "info" in requested
-        want_params = requested is None or "parameters" in requested
-        want_schema = requested is None or "input_schema" in requested
-
-        info = (
-            AppDescribeInfo(
-                id=str(app.id),
-                name=app.name,
-                mode=app.mode,
-                description=app.description,
-                tags=[{"name": t.name} for t in app.tags],
-                author=app.author_name,
-                updated_at=app.updated_at.isoformat() if app.updated_at else None,
-                service_api_enabled=bool(app.enable_api),
-            )
-            if want_info
-            else None
-        )
-
-        parameters: dict[str, Any] | None = None
-        input_schema: dict[str, Any] | None = None
-        if want_params:
-            try:
-                parameters = parameters_payload(app)
-            except AppUnavailableError:
-                parameters = dict(_EMPTY_PARAMETERS)
-        if want_schema:
-            try:
-                input_schema = build_input_schema(app)
-            except AppUnavailableError:
-                input_schema = dict(EMPTY_INPUT_SCHEMA)
-
-        return (
-            AppDescribeResponse(
-                info=info,
-                parameters=parameters,
-                input_schema=input_schema,
-            ).model_dump(mode="json", exclude_none=False),
-            200,
-        )
-
-
-class AppListQuery(BaseModel):
-    """`mode` is a closed enum — unknown values 422 instead of silently-empty data."""
-
-    workspace_id: str
-    page: int = Field(1, ge=1)
-    limit: int = Field(20, ge=1, le=MAX_PAGE_LIMIT)
-    mode: AppMode | None = None
-    name: str | None = Field(None, max_length=200)
-    tag: str | None = Field(None, max_length=100)
-
-
-@openapi_ns.route("/apps")
-class AppListApi(Resource):
-    method_decorators = _APPS_READ_DECORATORS
-
-    def get(self):
-        ctx = g.auth_ctx
-        if ctx.subject_type != SubjectType.ACCOUNT or ctx.account_id is None:
-            return PaginationEnvelope[AppListRow].build(page=1, limit=0, total=0, items=[]).model_dump(mode="json"), 200
-
-        try:
-            query = AppListQuery.model_validate(request.args.to_dict(flat=True))
-        except ValidationError as exc:
-            raise UnprocessableEntity(exc.json())
-
-        workspace_id = query.workspace_id
-        require_workspace_member(ctx, workspace_id)
-
-        empty = (
-            PaginationEnvelope[AppListRow]
-            .build(page=query.page, limit=query.limit, total=0, items=[])
-            .model_dump(mode="json"),
-            200,
-        )
-
-        if query.name:
-            try:
-                parsed_uuid = _uuid.UUID(query.name)
-            except ValueError:
-                parsed_uuid = None
-        else:
-            parsed_uuid = None
-
-        if parsed_uuid is not None:
-            app = db.session.get(App, str(parsed_uuid))
-            if not app or app.status != "normal" or str(app.tenant_id) != workspace_id:
-                return empty
-            tenant_name = db.session.execute(
-                sa.select(Tenant.name).where(Tenant.id == workspace_id)
-            ).scalar_one_or_none()
-            item = AppListRow(
-                id=str(app.id),
-                name=app.name,
-                description=app.description,
-                mode=app.mode,
-                tags=[{"name": t.name} for t in app.tags],
-                updated_at=app.updated_at.isoformat() if app.updated_at else None,
-                created_by_name=getattr(app, "author_name", None),
-                workspace_id=str(workspace_id),
-                workspace_name=tenant_name,
-            )
-            env = PaginationEnvelope[AppListRow].build(page=1, limit=1, total=1, items=[item])
-            return env.model_dump(mode="json"), 200
-
-        tag_ids: list[str] | None = None
-        if query.tag:
-            tags = TagService.get_tag_by_tag_name("app", workspace_id, query.tag)
-            if not tags:
-                return empty
-            tag_ids = [tag.id for tag in tags]
-
-        args: dict[str, Any] = {
-            "page": query.page,
-            "limit": query.limit,
-            "mode": query.mode.value if query.mode else "",
-            "name": query.name,
-            "status": "normal",
-        }
-        if tag_ids:
-            args["tag_ids"] = tag_ids
-
-        pagination = AppService().get_paginate_apps(ctx.account_id, workspace_id, args)
-        if pagination is None:
-            return empty
-
-        tenant_name: str | None = None
-        if pagination.items:
-            tenant_name = db.session.execute(
-                sa.select(Tenant.name).where(Tenant.id == workspace_id)
-            ).scalar_one_or_none()
-
-        items = [
-            AppListRow(
-                id=str(r.id),
-                name=r.name,
-                description=r.description,
-                mode=r.mode,
-                tags=[{"name": t.name} for t in r.tags],
-                updated_at=r.updated_at.isoformat() if r.updated_at else None,
-                created_by_name=getattr(r, "author_name", None),
-                workspace_id=str(workspace_id),
-                workspace_name=tenant_name,
-            )
-            for r in pagination.items
-        ]
-        env = PaginationEnvelope[AppListRow].build(
-            page=query.page, limit=query.limit, total=int(pagination.total), items=items
-        )
-        return env.model_dump(mode="json"), 200

api/controllers/openapi/apps_permitted.py

@@ -1,101 +0,0 @@
-"""GET /openapi/v1/apps/permitted — external-subject app discovery (EE only)."""
-
-from __future__ import annotations
-
-import sqlalchemy as sa
-from flask import request
-from flask_restx import Resource
-from pydantic import BaseModel, ConfigDict, Field, ValidationError
-from werkzeug.exceptions import UnprocessableEntity
-
-from controllers.openapi import openapi_ns
-from controllers.openapi._models import (
-    MAX_PAGE_LIMIT,
-    AppListRow,
-    PaginationEnvelope,
-)
-from extensions.ext_database import db
-from libs.device_flow_security import enterprise_only
-from libs.oauth_bearer import (
-    ACCEPT_USER_EXT_SSO,
-    Scope,
-    require_scope,
-    validate_bearer,
-)
-from models import App, Tenant
-from models.model import AppMode
-from services.enterprise.app_permitted_service import list_permitted_apps
-
-
-class AppPermittedListQuery(BaseModel):
-    """Strict (`extra='forbid'`) — rejects `workspace_id`/`tag`/etc. that are valid on /apps but not here."""
-
-    model_config = ConfigDict(extra="forbid")
-
-    page: int = Field(1, ge=1)
-    limit: int = Field(20, ge=1, le=MAX_PAGE_LIMIT)
-    mode: AppMode | None = None
-    name: str | None = Field(None, max_length=200)
-
-
-@openapi_ns.route("/apps/permitted")
-class AppPermittedListApi(Resource):
-    method_decorators = [
-        require_scope(Scope.APPS_READ_PERMITTED),
-        validate_bearer(accept=ACCEPT_USER_EXT_SSO),
-        enterprise_only,
-    ]
-
-    def get(self):
-        try:
-            query = AppPermittedListQuery.model_validate(request.args.to_dict(flat=True))
-        except ValidationError as exc:
-            raise UnprocessableEntity(exc.json())
-
-        page_result = list_permitted_apps(
-            page=query.page,
-            limit=query.limit,
-            mode=query.mode.value if query.mode else None,
-            name=query.name,
-        )
-
-        if not page_result.app_ids:
-            env = PaginationEnvelope[AppListRow].build(
-                page=query.page, limit=query.limit, total=page_result.total, items=[]
-            )
-            return env.model_dump(mode="json"), 200
-
-        apps_by_id = {
-            str(a.id): a
-            for a in db.session.execute(sa.select(App).where(App.id.in_(page_result.app_ids))).scalars().all()
-        }
-        tenant_ids = list({a.tenant_id for a in apps_by_id.values()})
-        tenants_by_id = {
-            str(t.id): t for t in db.session.execute(sa.select(Tenant).where(Tenant.id.in_(tenant_ids))).scalars().all()
-        }
-
-        items: list[AppListRow] = []
-        for app_id in page_result.app_ids:
-            app = apps_by_id.get(app_id)
-            if not app or app.status != "normal":
-                continue
-            tenant = tenants_by_id.get(str(app.tenant_id))
-            items.append(
-                AppListRow(
-                    id=str(app.id),
-                    name=app.name,
-                    description=app.description,
-                    mode=app.mode,
-                    tags=[],  # tenant-scoped; not surfaced cross-tenant
-                    updated_at=app.updated_at.isoformat() if app.updated_at else None,
-                    created_by_name=None,  # cross-tenant author leak prevention
-                    workspace_id=str(app.tenant_id),
-                    workspace_name=tenant.name if tenant else None,
-                )
-            )
-
-        # total/has_more reflect the EE-side allow-list; len(items) may be < limit when local rows are dropped.
-        env = PaginationEnvelope[AppListRow].build(
-            page=query.page, limit=query.limit, total=page_result.total, items=items
-        )
-        return env.model_dump(mode="json"), 200

api/controllers/openapi/auth/__init__.py

@@ -1,3 +0,0 @@
-from controllers.openapi.auth.composition import OAUTH_BEARER_PIPELINE
-
-__all__ = ["OAUTH_BEARER_PIPELINE"]

api/controllers/openapi/auth/composition.py

@@ -1,43 +0,0 @@
-"""`OAUTH_BEARER_PIPELINE` — the auth scheme for openapi `/run` endpoints.
-
-Endpoints attach via `@OAUTH_BEARER_PIPELINE.guard(scope=…)`. No alternative
-paths. Read endpoints (`/apps`, `/info`, `/parameters`, `/describe`) skip
-the pipeline and use `validate_bearer + require_scope + require_workspace_member`
-inline — they don't need `AppAuthzCheck`/`CallerMount`.
-"""
-
-from __future__ import annotations
-
-from controllers.openapi.auth.pipeline import Pipeline
-from controllers.openapi.auth.steps import (
-    AppAuthzCheck,
-    AppResolver,
-    BearerCheck,
-    CallerMount,
-    ScopeCheck,
-    WorkspaceMembershipCheck,
-)
-from controllers.openapi.auth.strategies import (
-    AccountMounter,
-    AclStrategy,
-    AppAuthzStrategy,
-    EndUserMounter,
-    MembershipStrategy,
-)
-from services.feature_service import FeatureService
-
-
-def _resolve_app_authz_strategy() -> AppAuthzStrategy:
-    if FeatureService.get_system_features().webapp_auth.enabled:
-        return AclStrategy()
-    return MembershipStrategy()
-
-
-OAUTH_BEARER_PIPELINE = Pipeline(
-    BearerCheck(),
-    ScopeCheck(),
-    AppResolver(),
-    WorkspaceMembershipCheck(),
-    AppAuthzCheck(_resolve_app_authz_strategy),
-    CallerMount(AccountMounter(), EndUserMounter()),
-)

api/controllers/openapi/auth/context.py

@@ -1,46 +0,0 @@
-"""Mutable per-request context for the openapi auth pipeline.
-
-Every field starts None / empty and is filled in by a step. The pipeline
-is the only thing that should construct or mutate Context — handlers
-read populated values via the decorator's kwargs unpacking.
-"""
-
-from __future__ import annotations
-
-import uuid
-from dataclasses import dataclass, field
-from datetime import datetime
-from typing import TYPE_CHECKING, Literal, Protocol
-
-from flask import Request
-
-from libs.oauth_bearer import Scope, SubjectType
-
-if TYPE_CHECKING:
-    from models import App, Tenant
-
-
-@dataclass
-class Context:
-    request: Request
-    required_scope: Scope
-    subject_type: SubjectType | None = None
-    subject_email: str | None = None
-    subject_issuer: str | None = None
-    account_id: uuid.UUID | None = None
-    scopes: frozenset[Scope] = field(default_factory=frozenset)
-    token_id: uuid.UUID | None = None
-    token_hash: str | None = None
-    cached_verified_tenants: dict[str, bool] | None = None
-    source: str | None = None
-    expires_at: datetime | None = None
-    app: App | None = None
-    tenant: Tenant | None = None
-    caller: object | None = None
-    caller_kind: Literal["account", "end_user"] | None = None
-
-
-class Step(Protocol):
-    """One responsibility. Mutate ctx or raise to short-circuit."""
-
-    def __call__(self, ctx: Context) -> None: ...

api/controllers/openapi/auth/pipeline.py

@@ -1,41 +0,0 @@
-"""Pipeline IS the auth scheme.
-
-`Pipeline.guard(scope=…)` is the only attachment point for endpoints —
-that is the design lock-in: forgetting an auth layer is structurally
-impossible because there is no "sometimes wrap, sometimes don't" choice.
-"""
-
-from __future__ import annotations
-
-from functools import wraps
-
-from flask import request
-
-from controllers.openapi.auth.context import Context, Step
-from libs.oauth_bearer import Scope
-
-
-class Pipeline:
-    def __init__(self, *steps: Step) -> None:
-        self._steps = steps
-
-    def run(self, ctx: Context) -> None:
-        for step in self._steps:
-            step(ctx)
-
-    def guard(self, *, scope: Scope):
-        def decorator(view):
-            @wraps(view)
-            def decorated(*args, **kwargs):
-                ctx = Context(request=request, required_scope=scope)
-                self.run(ctx)
-                kwargs.update(
-                    app_model=ctx.app,
-                    caller=ctx.caller,
-                    caller_kind=ctx.caller_kind,
-                )
-                return view(*args, **kwargs)
-
-            return decorated
-
-        return decorator

api/controllers/openapi/auth/steps.py

@@ -1,131 +0,0 @@
-"""Pipeline steps. Each is one responsibility.
-
-`BearerCheck` is the only step that touches the token registry; downstream
-steps see only the populated `Context`.
-"""
-
-from __future__ import annotations
-
-from collections.abc import Callable
-
-from werkzeug.exceptions import BadRequest, Forbidden, NotFound, Unauthorized
-
-from configs import dify_config
-from controllers.openapi.auth.context import Context
-from controllers.openapi.auth.strategies import AppAuthzStrategy, CallerMounter
-from extensions.ext_database import db
-from libs.oauth_bearer import (
-    InvalidBearerError,
-    Scope,
-    SubjectType,
-    _extract_bearer,  # type: ignore[attr-defined]
-    check_workspace_membership,
-    get_authenticator,
-)
-from models import App, Tenant, TenantStatus
-
-
-class BearerCheck:
-    """Resolve bearer → populate identity fields. Rate-limit is enforced
-    inside `BearerAuthenticator.authenticate`, so no separate step here."""
-
-    def __call__(self, ctx: Context) -> None:
-        token = _extract_bearer(ctx.request)
-        if not token:
-            raise Unauthorized("bearer required")
-
-        try:
-            authn = get_authenticator().authenticate(token)
-        except InvalidBearerError as e:
-            raise Unauthorized(str(e))
-
-        ctx.subject_type = authn.subject_type
-        ctx.subject_email = authn.subject_email
-        ctx.subject_issuer = authn.subject_issuer
-        ctx.account_id = authn.account_id
-        ctx.scopes = frozenset(authn.scopes)
-        ctx.source = authn.source
-        ctx.token_id = authn.token_id
-        ctx.expires_at = authn.expires_at
-        ctx.token_hash = authn.token_hash
-        ctx.cached_verified_tenants = dict(authn.verified_tenants)
-
-
-class ScopeCheck:
-    """Verify ctx.scopes (already populated by BearerCheck) covers required."""
-
-    def __call__(self, ctx: Context) -> None:
-        if Scope.FULL in ctx.scopes or ctx.required_scope in ctx.scopes:
-            return
-        raise Forbidden("insufficient_scope")
-
-
-class AppResolver:
-    """Read app_id from request.view_args, populate ctx.app + ctx.tenant.
-
-    Every endpoint using the OAuth bearer pipeline must declare
-    ``<string:app_id>`` in its route — that is the design lock-in (no body /
-    header coupling).
-    """
-
-    def __call__(self, ctx: Context) -> None:
-        app_id = (ctx.request.view_args or {}).get("app_id")
-        if not app_id:
-            raise BadRequest("app_id is required in path")
-        app = db.session.get(App, app_id)
-        if not app or app.status != "normal":
-            raise NotFound("app not found")
-        if not app.enable_api:
-            raise Forbidden("service_api_disabled")
-        tenant = db.session.get(Tenant, app.tenant_id)
-        if tenant is None or tenant.status == TenantStatus.ARCHIVE:
-            raise Forbidden("workspace unavailable")
-        ctx.app, ctx.tenant = app, tenant
-
-
-class WorkspaceMembershipCheck:
-    """Layer 0 — workspace membership gate.
-
-    CE-only (skipped when ENTERPRISE_ENABLED). Account-subject bearers
-    (dfoa_) only — SSO subjects skip.
-    """
-
-    def __call__(self, ctx: Context) -> None:
-        if dify_config.ENTERPRISE_ENABLED:
-            return
-        if ctx.subject_type != SubjectType.ACCOUNT:
-            return
-        if ctx.account_id is None or ctx.tenant is None:
-            raise Unauthorized("account_id or tenant unset — BearerCheck or AppResolver did not run")
-        if ctx.token_hash is None:
-            raise Unauthorized("token_hash unset — BearerCheck did not run")
-
-        check_workspace_membership(
-            account_id=ctx.account_id,
-            tenant_id=ctx.tenant.id,
-            token_hash=ctx.token_hash,
-            cached_verdicts=ctx.cached_verified_tenants or {},
-        )
-
-
-class AppAuthzCheck:
-    def __init__(self, resolve_strategy: Callable[[], AppAuthzStrategy]) -> None:
-        self._resolve = resolve_strategy
-
-    def __call__(self, ctx: Context) -> None:
-        if not self._resolve().authorize(ctx):
-            raise Forbidden("subject_no_app_access")
-
-
-class CallerMount:
-    def __init__(self, *mounters: CallerMounter) -> None:
-        self._mounters = mounters
-
-    def __call__(self, ctx: Context) -> None:
-        if ctx.subject_type is None:
-            raise Unauthorized("subject_type unset — BearerCheck did not run")
-        for m in self._mounters:
-            if m.applies_to(ctx.subject_type):
-                m.mount(ctx)
-                return
-        raise Unauthorized("no caller mounter for subject type")

api/controllers/openapi/auth/strategies.py

@@ -1,115 +0,0 @@
-"""Strategy classes for the openapi auth pipeline.
-
-App authorization (Acl/Membership) and caller mounting (Account/EndUser)
-vary along independent axes; each strategy is one class so the pipeline
-composition stays a flat list.
-"""
-
-from __future__ import annotations
-
-import uuid
-from typing import Protocol
-
-from flask import current_app
-from flask_login import user_logged_in
-from sqlalchemy import select
-
-from controllers.openapi.auth.context import Context
-from core.app.entities.app_invoke_entities import InvokeFrom
-from extensions.ext_database import db
-from libs.oauth_bearer import SubjectType
-from models import Account, TenantAccountJoin
-from services.end_user_service import EndUserService
-from services.enterprise.enterprise_service import EnterpriseService
-
-
-class AppAuthzStrategy(Protocol):
-    def authorize(self, ctx: Context) -> bool: ...
-
-
-class AclStrategy:
-    """Per-app ACL via the workspace-auth inner API.
-
-    Used when webapp-auth is enabled (EE deployment). The inner-API
-    allowlist is the source of truth.
-    """
-
-    def authorize(self, ctx: Context) -> bool:
-        if ctx.subject_email is None or ctx.app is None:
-            return False
-        return EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(
-            user_id=ctx.subject_email,
-            app_id=ctx.app.id,
-        )
-
-
-class MembershipStrategy:
-    """Tenant-membership fallback.
-
-    Used when webapp-auth is disabled (CE deployment). Account-bearing
-    subjects pass if they have a TenantAccountJoin row; EXTERNAL_SSO is
-    denied (it requires the webapp-auth surface).
-    """
-
-    def authorize(self, ctx: Context) -> bool:
-        if ctx.subject_type == SubjectType.EXTERNAL_SSO:
-            return False
-        if ctx.tenant is None:
-            return False
-        return _has_tenant_membership(ctx.account_id, ctx.tenant.id)
-
-
-def _has_tenant_membership(account_id: uuid.UUID | str | None, tenant_id: str) -> bool:
-    if not account_id:
-        return False
-    row = db.session.execute(
-        select(TenantAccountJoin.id).where(
-            TenantAccountJoin.tenant_id == tenant_id,
-            TenantAccountJoin.account_id == account_id,
-        )
-    ).scalar_one_or_none()
-    return row is not None
-
-
-def _login_as(user) -> None:
-    """Set Flask-Login request user so downstream services see the caller."""
-    current_app.login_manager._update_request_context_with_user(user)
-    user_logged_in.send(current_app._get_current_object(), user=user)
-
-
-class CallerMounter(Protocol):
-    def applies_to(self, subject_type: SubjectType) -> bool: ...
-
-    def mount(self, ctx: Context) -> None: ...
-
-
-class AccountMounter:
-    def applies_to(self, subject_type: SubjectType) -> bool:
-        return subject_type == SubjectType.ACCOUNT
-
-    def mount(self, ctx: Context) -> None:
-        if ctx.account_id is None:
-            raise RuntimeError("AccountMounter: account_id unset — BearerCheck did not run")
-        account = db.session.get(Account, ctx.account_id)
-        if account is None:
-            raise RuntimeError("AccountMounter: account row missing for resolved bearer")
-        account.current_tenant = ctx.tenant
-        _login_as(account)
-        ctx.caller, ctx.caller_kind = account, "account"
-
-
-class EndUserMounter:
-    def applies_to(self, subject_type: SubjectType) -> bool:
-        return subject_type == SubjectType.EXTERNAL_SSO
-
-    def mount(self, ctx: Context) -> None:
-        if ctx.tenant is None or ctx.app is None or ctx.subject_email is None:
-            raise RuntimeError("EndUserMounter: tenant/app/subject_email unset — earlier steps did not run")
-        end_user = EndUserService.get_or_create_end_user_by_type(
-            InvokeFrom.OPENAPI,
-            tenant_id=ctx.tenant.id,
-            app_id=ctx.app.id,
-            user_id=ctx.subject_email,
-        )
-        _login_as(end_user)
-        ctx.caller, ctx.caller_kind = end_user, "end_user"

api/controllers/openapi/index.py

@@ -1,9 +0,0 @@
-from flask_restx import Resource
-
-from controllers.openapi import openapi_ns
-
-
-@openapi_ns.route("/_health")
-class HealthApi(Resource):
-    def get(self):
-        return {"ok": True}

api/controllers/openapi/oauth_device.py

@@ -1,392 +0,0 @@
-"""Device-flow endpoints under /openapi/v1/oauth/device/*. Two
-sub-groups in one module:
-
-  Protocol (RFC 8628, public + rate-limited):
-    POST /oauth/device/code
-    POST /oauth/device/token
-    GET  /oauth/device/lookup
-
-  Approval (account branch, console-cookie authed):
-    POST /oauth/device/approve
-    POST /oauth/device/deny
-
-SSO branch lives in oauth_device_sso.py.
-"""
-
-from __future__ import annotations
-
-import logging
-
-from flask import request
-from flask_login import login_required
-from flask_restx import Resource
-from pydantic import BaseModel, ValidationError
-from werkzeug.exceptions import BadRequest
-
-from configs import dify_config
-from controllers.console.wraps import account_initialization_required, setup_required
-from controllers.openapi import openapi_ns
-from extensions.ext_database import db
-from extensions.ext_redis import redis_client
-from libs.helper import extract_remote_ip
-from libs.login import current_account_with_tenant
-from libs.oauth_bearer import SubjectType, bearer_feature_required
-from libs.rate_limit import (
-    LIMIT_APPROVE_CONSOLE,
-    LIMIT_DEVICE_CODE_PER_IP,
-    LIMIT_LOOKUP_PUBLIC,
-    rate_limit,
-)
-from services.oauth_device_flow import (
-    ACCOUNT_ISSUER_SENTINEL,
-    DEFAULT_POLL_INTERVAL_SECONDS,
-    DEVICE_FLOW_TTL_SECONDS,
-    PREFIX_OAUTH_ACCOUNT,
-    DeviceFlowRedis,
-    DeviceFlowStatus,
-    InvalidTransitionError,
-    SlowDownDecision,
-    StateNotFoundError,
-    mint_oauth_token,
-    oauth_ttl_days,
-)
-
-logger = logging.getLogger(__name__)
-
-
-# =========================================================================
-# Request / query schemas
-# =========================================================================
-
-
-class DeviceCodeRequest(BaseModel):
-    client_id: str
-    device_label: str
-
-
-class DevicePollRequest(BaseModel):
-    device_code: str
-    client_id: str
-
-
-class DeviceLookupQuery(BaseModel):
-    user_code: str
-
-
-class DeviceMutateRequest(BaseModel):
-    user_code: str
-
-
-def _validate_json[M: BaseModel](model: type[M]) -> M:
-    body = request.get_json(silent=True) or {}
-    try:
-        return model.model_validate(body)
-    except ValidationError as exc:
-        raise BadRequest(str(exc))
-
-
-def _validate_query[M: BaseModel](model: type[M]) -> M:
-    try:
-        return model.model_validate(request.args.to_dict(flat=True))
-    except ValidationError as exc:
-        raise BadRequest(str(exc))
-
-
-# =========================================================================
-# Protocol endpoints — RFC 8628 (public + per-IP rate limit)
-# =========================================================================
-
-
-@openapi_ns.route("/oauth/device/code")
-class OAuthDeviceCodeApi(Resource):
-    @rate_limit(LIMIT_DEVICE_CODE_PER_IP)
-    def post(self):
-        payload = _validate_json(DeviceCodeRequest)
-        client_id = payload.client_id
-        device_label = payload.device_label
-
-        if client_id not in dify_config.OPENAPI_KNOWN_CLIENT_IDS:
-            return {"error": "unsupported_client"}, 400
-
-        store = DeviceFlowRedis(redis_client)
-        ip = extract_remote_ip(request)
-        device_code, user_code, expires_in = store.start(client_id, device_label, created_ip=ip)
-
-        return {
-            "device_code": device_code,
-            "user_code": user_code,
-            "verification_uri": _verification_uri(),
-            "expires_in": expires_in,
-            "interval": DEFAULT_POLL_INTERVAL_SECONDS,
-        }, 200
-
-
-@openapi_ns.route("/oauth/device/token")
-class OAuthDeviceTokenApi(Resource):
-    """RFC 8628 poll."""
-
-    def post(self):
-        payload = _validate_json(DevicePollRequest)
-        device_code = payload.device_code
-
-        store = DeviceFlowRedis(redis_client)
-
-        # slow_down beats every other branch — polling-too-fast clients
-        # see only that response regardless of underlying state.
-        if store.record_poll(device_code, DEFAULT_POLL_INTERVAL_SECONDS) is SlowDownDecision.SLOW_DOWN:
-            return {"error": "slow_down"}, 400
-
-        state = store.load_by_device_code(device_code)
-        if state is None:
-            return {"error": "expired_token"}, 400
-
-        if state.status is DeviceFlowStatus.PENDING:
-            return {"error": "authorization_pending"}, 400
-
-        terminal = store.consume_on_poll(device_code)
-        if terminal is None:
-            return {"error": "expired_token"}, 400
-
-        if terminal.status is DeviceFlowStatus.DENIED:
-            return {"error": "access_denied"}, 400
-
-        poll_payload = terminal.poll_payload or {}
-        if "token" not in poll_payload:
-            logger.error("device_flow: approved state missing poll_payload for %s", device_code)
-            return {"error": "expired_token"}, 400
-
-        _audit_cross_ip_if_needed(state)
-        return poll_payload, 200
-
-
-@openapi_ns.route("/oauth/device/lookup")
-class OAuthDeviceLookupApi(Resource):
-    """Read-only — public for pre-validate before login. user_code is
-    high-entropy + short-TTL; per-IP rate limit blocks enumeration.
-    """
-
-    @rate_limit(LIMIT_LOOKUP_PUBLIC)
-    def get(self):
-        payload = _validate_query(DeviceLookupQuery)
-        user_code = payload.user_code.strip().upper()
-
-        store = DeviceFlowRedis(redis_client)
-        found = store.load_by_user_code(user_code)
-        if found is None:
-            return {"valid": False, "expires_in_remaining": 0, "client_id": None}, 200
-
-        _device_code, state = found
-        if state.status is not DeviceFlowStatus.PENDING:
-            return {"valid": False, "expires_in_remaining": 0, "client_id": state.client_id}, 200
-
-        return {
-            "valid": True,
-            "expires_in_remaining": DEVICE_FLOW_TTL_SECONDS,
-            "client_id": state.client_id,
-        }, 200
-
-
-# =========================================================================
-# Approval endpoints — account branch (cookie-authed)
-# =========================================================================
-
-
-_APPROVE_GUARD_KEY_FMT = "device_code:{code}:approving"
-_APPROVE_GUARD_TTL_SECONDS = 10
-
-
-@openapi_ns.route("/oauth/device/approve")
-class DeviceApproveApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @bearer_feature_required
-    @rate_limit(LIMIT_APPROVE_CONSOLE)
-    def post(self):
-        payload = _validate_json(DeviceMutateRequest)
-        user_code = payload.user_code.strip().upper()
-
-        account, tenant = current_account_with_tenant()
-        store = DeviceFlowRedis(redis_client)
-
-        found = store.load_by_user_code(user_code)
-        if found is None:
-            return {"error": "expired_or_unknown"}, 404
-        device_code, state = found
-        if state.status is not DeviceFlowStatus.PENDING:
-            return {"error": "already_resolved"}, 409
-
-        # SET NX guard — without it, two in-flight approves both pass
-        # PENDING, both mint, and the second upsert silently rotates the
-        # first caller into an already-revoked token.
-        guard_key = _APPROVE_GUARD_KEY_FMT.format(code=device_code)
-        if not redis_client.set(guard_key, "1", nx=True, ex=_APPROVE_GUARD_TTL_SECONDS):
-            return {"error": "approve_in_progress"}, 409
-
-        try:
-            ttl_days = oauth_ttl_days(tenant_id=tenant)
-            mint = mint_oauth_token(
-                db.session,
-                redis_client,
-                subject_email=account.email,
-                subject_issuer=ACCOUNT_ISSUER_SENTINEL,
-                account_id=str(account.id),
-                client_id=state.client_id,
-                device_label=state.device_label,
-                prefix=PREFIX_OAUTH_ACCOUNT,
-                ttl_days=ttl_days,
-            )
-
-            poll_payload = _build_account_poll_payload(account, tenant, mint)
-            try:
-                store.approve(
-                    device_code,
-                    subject_email=account.email,
-                    account_id=str(account.id),
-                    subject_issuer=ACCOUNT_ISSUER_SENTINEL,
-                    minted_token=mint.token,
-                    token_id=str(mint.token_id),
-                    poll_payload=poll_payload,
-                )
-            except (StateNotFoundError, InvalidTransitionError):
-                # Row minted but state vanished — roll forward; the orphan
-                # token is revocable via auth devices list / Authorized Apps.
-                logger.exception("device_flow: approve raced on %s", device_code)
-                return {"error": "state_lost"}, 409
-        finally:
-            redis_client.delete(guard_key)
-
-        _emit_approve_audit(state, account, tenant, mint)
-        return {"status": "approved"}, 200
-
-
-@openapi_ns.route("/oauth/device/deny")
-class DeviceDenyApi(Resource):
-    @setup_required
-    @login_required
-    @account_initialization_required
-    @bearer_feature_required
-    @rate_limit(LIMIT_APPROVE_CONSOLE)
-    def post(self):
-        payload = _validate_json(DeviceMutateRequest)
-        user_code = payload.user_code.strip().upper()
-
-        store = DeviceFlowRedis(redis_client)
-        found = store.load_by_user_code(user_code)
-        if found is None:
-            return {"error": "expired_or_unknown"}, 404
-        device_code, state = found
-        if state.status is not DeviceFlowStatus.PENDING:
-            return {"error": "already_resolved"}, 409
-
-        try:
-            store.deny(device_code)
-        except (StateNotFoundError, InvalidTransitionError):
-            logger.exception("device_flow: deny raced on %s", device_code)
-            return {"error": "state_lost"}, 409
-
-        _emit_deny_audit(state)
-        return {"status": "denied"}, 200
-
-
-# =========================================================================
-# Helpers
-# =========================================================================
-
-
-def _verification_uri() -> str:
-    base = getattr(dify_config, "CONSOLE_WEB_URL", None)
-    if base:
-        return f"{base.rstrip('/')}/device"
-    return f"{request.host_url.rstrip('/')}/device"
-
-
-def _audit_cross_ip_if_needed(state) -> None:
-    poll_ip = extract_remote_ip(request)
-    if state.created_ip and poll_ip and poll_ip != state.created_ip:
-        logger.warning(
-            "audit: oauth.device_code_cross_ip_poll token_id=%s creation_ip=%s poll_ip=%s",
-            state.token_id,
-            state.created_ip,
-            poll_ip,
-            extra={
-                "audit": True,
-                "token_id": state.token_id,
-                "creation_ip": state.created_ip,
-                "poll_ip": poll_ip,
-            },
-        )
-
-
-def _build_account_poll_payload(account, tenant, mint) -> dict:
-    """Pre-render the poll-response body so the unauthenticated poll
-    handler doesn't re-query accounts/tenants for authz data.
-    """
-    from models import Tenant, TenantAccountJoin
-
-    rows = (
-        db.session.query(Tenant, TenantAccountJoin)
-        .join(TenantAccountJoin, TenantAccountJoin.tenant_id == Tenant.id)
-        .filter(TenantAccountJoin.account_id == account.id)
-        .all()
-    )
-    workspaces = [{"id": str(t.id), "name": t.name, "role": getattr(m, "role", "")} for t, m in rows]
-    # Prefer active session tenant → DB-flagged current join → first membership.
-    default_ws_id = None
-    if tenant and any(w["id"] == str(tenant) for w in workspaces):
-        default_ws_id = str(tenant)
-    if default_ws_id is None:
-        for _t, m in rows:
-            if getattr(m, "current", False):
-                default_ws_id = str(m.tenant_id)
-                break
-    if default_ws_id is None and workspaces:
-        default_ws_id = workspaces[0]["id"]
-
-    return {
-        "token": mint.token,
-        "expires_at": mint.expires_at.isoformat(),
-        "subject_type": SubjectType.ACCOUNT,
-        "account": {"id": str(account.id), "email": account.email, "name": account.name},
-        "workspaces": workspaces,
-        "default_workspace_id": default_ws_id,
-        "token_id": str(mint.token_id),
-    }
-
-
-def _emit_approve_audit(state, account, tenant, mint) -> None:
-    logger.warning(
-        "audit: oauth.device_flow_approved token_id=%s subject=%s client_id=%s device_label=%s rotated=? expires_at=%s",
-        mint.token_id,
-        account.email,
-        state.client_id,
-        state.device_label,
-        mint.expires_at,
-        extra={
-            "audit": True,
-            "event": "oauth.device_flow_approved",
-            "token_id": str(mint.token_id),
-            "subject_type": SubjectType.ACCOUNT,
-            "subject_email": account.email,
-            "account_id": str(account.id),
-            "tenant_id": tenant,
-            "client_id": state.client_id,
-            "device_label": state.device_label,
-            "scopes": ["full"],
-            "expires_at": mint.expires_at.isoformat(),
-        },
-    )
-
-
-def _emit_deny_audit(state) -> None:
-    logger.warning(
-        "audit: oauth.device_flow_denied client_id=%s device_label=%s",
-        state.client_id,
-        state.device_label,
-        extra={
-            "audit": True,
-            "event": "oauth.device_flow_denied",
-            "client_id": state.client_id,
-            "device_label": state.device_label,
-        },
-    )

api/controllers/openapi/oauth_device_sso.py

@@ -1,287 +0,0 @@
-"""SSO-branch device-flow endpoints under /openapi/v1/oauth/device/*.
-EE-only. Browser flow:
-
-  GET  /oauth/device/sso-initiate     → 302 to IdP authorize URL
-  GET  /oauth/device/sso-complete     → ACS callback, sets approval-grant cookie
-  GET  /oauth/device/approval-context → SPA reads cookie claims (idempotent)
-  POST /oauth/device/approve-external → mints dfoe_ token + clears cookie
-
-Function-based (raw @bp.route) rather than Resource classes because the
-handlers do redirects + cookie kwargs that don't fit the Resource shape.
-"""
-
-from __future__ import annotations
-
-import logging
-import secrets
-
-from flask import jsonify, make_response, redirect, request
-from werkzeug.exceptions import (
-    BadGateway,
-    BadRequest,
-    Conflict,
-    Forbidden,
-    NotFound,
-    Unauthorized,
-)
-
-from controllers.openapi import bp
-from extensions.ext_database import db
-from extensions.ext_redis import redis_client
-from libs import jws
-from libs.device_flow_security import (
-    APPROVAL_GRANT_COOKIE_NAME,
-    ApprovalGrantClaims,
-    approval_grant_cleared_cookie_kwargs,
-    approval_grant_cookie_kwargs,
-    consume_approval_grant_nonce,
-    consume_sso_assertion_nonce,
-    enterprise_only,
-    mint_approval_grant,
-    verify_approval_grant,
-)
-from libs.oauth_bearer import SubjectType
-from libs.rate_limit import (
-    LIMIT_APPROVE_EXT_PER_EMAIL,
-    LIMIT_SSO_INITIATE_PER_IP,
-    enforce,
-    rate_limit,
-)
-from services.enterprise.enterprise_service import EnterpriseService
-from services.oauth_device_flow import (
-    PREFIX_OAUTH_EXTERNAL_SSO,
-    DeviceFlowRedis,
-    DeviceFlowStatus,
-    InvalidTransitionError,
-    StateNotFoundError,
-    mint_oauth_token,
-    oauth_ttl_days,
-)
-
-logger = logging.getLogger(__name__)
-
-
-# Matches DEVICE_FLOW_TTL_SECONDS so the signed state can't outlive the
-# device_code it references.
-STATE_ENVELOPE_TTL_SECONDS = 15 * 60
-
-# Canonical sso-complete path. IdP-side ACS callback URL must point here.
-_SSO_COMPLETE_PATH = "/openapi/v1/oauth/device/sso-complete"
-
-
-@bp.route("/oauth/device/sso-initiate", methods=["GET"])
-@enterprise_only
-@rate_limit(LIMIT_SSO_INITIATE_PER_IP)
-def sso_initiate():
-    user_code = (request.args.get("user_code") or "").strip().upper()
-    if not user_code:
-        raise BadRequest("user_code required")
-
-    store = DeviceFlowRedis(redis_client)
-    found = store.load_by_user_code(user_code)
-    if found is None:
-        raise BadRequest("invalid_user_code")
-    _, state = found
-    if state.status is not DeviceFlowStatus.PENDING:
-        raise BadRequest("invalid_user_code")
-
-    keyset = jws.KeySet.from_shared_secret()
-    signed_state = jws.sign(
-        keyset,
-        payload={
-            "redirect_url": "",
-            "app_code": "",
-            "intent": "device_flow",
-            "user_code": user_code,
-            "nonce": secrets.token_urlsafe(16),
-            "return_to": "",
-            "idp_callback_url": f"{request.host_url.rstrip('/')}{_SSO_COMPLETE_PATH}",
-        },
-        aud=jws.AUD_STATE_ENVELOPE,
-        ttl_seconds=STATE_ENVELOPE_TTL_SECONDS,
-    )
-
-    try:
-        reply = EnterpriseService.initiate_device_flow_sso(signed_state)
-    except Exception as e:
-        logger.warning("sso-initiate: enterprise call failed: %s", e)
-        raise BadGateway("sso_initiate_failed") from e
-
-    url = (reply or {}).get("url")
-    if not url:
-        raise BadGateway("sso_initiate_missing_url")
-
-    # Clear stale approval-grant — defends against cross-tab/back-button mixing.
-    resp = redirect(url, code=302)
-    resp.set_cookie(**approval_grant_cleared_cookie_kwargs())
-    return resp
-
-
-@bp.route("/oauth/device/sso-complete", methods=["GET"])
-@enterprise_only
-def sso_complete():
-    blob = request.args.get("sso_assertion")
-    if not blob:
-        raise BadRequest("sso_assertion required")
-
-    keyset = jws.KeySet.from_shared_secret()
-
-    try:
-        claims = jws.verify(keyset, blob, expected_aud=jws.AUD_EXT_SUBJECT_ASSERTION)
-    except jws.VerifyError as e:
-        logger.warning("sso-complete: rejected assertion: %s", e)
-        raise BadRequest("invalid_sso_assertion") from e
-
-    if not consume_sso_assertion_nonce(redis_client, claims.get("nonce", "")):
-        raise BadRequest("invalid_sso_assertion")
-
-    user_code = (claims.get("user_code") or "").strip().upper()
-    store = DeviceFlowRedis(redis_client)
-    found = store.load_by_user_code(user_code)
-    if found is None:
-        raise Conflict("user_code_not_pending")
-    _, state = found
-    if state.status is not DeviceFlowStatus.PENDING:
-        raise Conflict("user_code_not_pending")
-
-    iss = request.host_url.rstrip("/")
-    cookie_value, _ = mint_approval_grant(
-        keyset=keyset,
-        iss=iss,
-        subject_email=claims["email"],
-        subject_issuer=claims["issuer"],
-        user_code=user_code,
-    )
-
-    resp = redirect("/device?sso_verified=1", code=302)
-    resp.set_cookie(**approval_grant_cookie_kwargs(cookie_value))
-    return resp
-
-
-@bp.route("/oauth/device/approval-context", methods=["GET"])
-@enterprise_only
-def approval_context():
-    token = request.cookies.get(APPROVAL_GRANT_COOKIE_NAME)
-    if not token:
-        raise Unauthorized("no_session")
-
-    keyset = jws.KeySet.from_shared_secret()
-    try:
-        claims = verify_approval_grant(keyset, token)
-    except jws.VerifyError as e:
-        logger.warning("approval-context: bad cookie: %s", e)
-        raise Unauthorized("no_session") from e
-
-    return jsonify(
-        {
-            "subject_email": claims.subject_email,
-            "subject_issuer": claims.subject_issuer,
-            "user_code": claims.user_code,
-            "csrf_token": claims.csrf_token,
-            "expires_at": claims.expires_at.isoformat(),
-        }
-    ), 200
-
-
-@bp.route("/oauth/device/approve-external", methods=["POST"])
-@enterprise_only
-def approve_external():
-    token = request.cookies.get(APPROVAL_GRANT_COOKIE_NAME)
-    if not token:
-        raise Unauthorized("invalid_session")
-
-    keyset = jws.KeySet.from_shared_secret()
-    try:
-        claims: ApprovalGrantClaims = verify_approval_grant(keyset, token)
-    except jws.VerifyError as e:
-        logger.warning("approve-external: bad cookie: %s", e)
-        raise Unauthorized("invalid_session") from e
-
-    enforce(LIMIT_APPROVE_EXT_PER_EMAIL, key=f"subject:{claims.subject_email}")
-
-    csrf_header = request.headers.get("X-CSRF-Token", "")
-    if not csrf_header or csrf_header != claims.csrf_token:
-        raise Forbidden("csrf_mismatch")
-
-    data = request.get_json(silent=True) or {}
-    body_user_code = (data.get("user_code") or "").strip().upper()
-    if body_user_code != claims.user_code:
-        raise BadRequest("user_code_mismatch")
-
-    store = DeviceFlowRedis(redis_client)
-    found = store.load_by_user_code(claims.user_code)
-    if found is None:
-        raise NotFound("user_code_not_pending")
-    device_code, state = found
-    if state.status is not DeviceFlowStatus.PENDING:
-        raise Conflict("user_code_not_pending")
-
-    if not consume_approval_grant_nonce(redis_client, claims.nonce):
-        raise Unauthorized("session_already_consumed")
-
-    ttl_days = oauth_ttl_days(tenant_id=None)
-    mint = mint_oauth_token(
-        db.session,
-        redis_client,
-        subject_email=claims.subject_email,
-        subject_issuer=claims.subject_issuer,
-        account_id=None,
-        client_id=state.client_id,
-        device_label=state.device_label,
-        prefix=PREFIX_OAUTH_EXTERNAL_SSO,
-        ttl_days=ttl_days,
-    )
-
-    poll_payload = {
-        "token": mint.token,
-        "expires_at": mint.expires_at.isoformat(),
-        "subject_type": SubjectType.EXTERNAL_SSO,
-        "subject_email": claims.subject_email,
-        "subject_issuer": claims.subject_issuer,
-        "account": None,
-        "workspaces": [],
-        "default_workspace_id": None,
-        "token_id": str(mint.token_id),
-    }
-
-    try:
-        store.approve(
-            device_code,
-            subject_email=claims.subject_email,
-            account_id=None,
-            subject_issuer=claims.subject_issuer,
-            minted_token=mint.token,
-            token_id=str(mint.token_id),
-            poll_payload=poll_payload,
-        )
-    except (StateNotFoundError, InvalidTransitionError) as e:
-        logger.exception("approve-external: state transition raced")
-        raise Conflict("state_lost") from e
-
-    _emit_approve_external_audit(state, claims, mint)
-
-    resp = make_response(jsonify({"status": "approved"}), 200)
-    resp.set_cookie(**approval_grant_cleared_cookie_kwargs())
-    return resp
-
-
-def _emit_approve_external_audit(state, claims, mint) -> None:
-    logger.warning(
-        "audit: oauth.device_flow_approved subject_type=%s subject_email=%s subject_issuer=%s token_id=%s",
-        SubjectType.EXTERNAL_SSO,
-        claims.subject_email,
-        claims.subject_issuer,
-        mint.token_id,
-        extra={
-            "audit": True,
-            "event": "oauth.device_flow_approved",
-            "subject_type": SubjectType.EXTERNAL_SSO,
-            "subject_email": claims.subject_email,
-            "subject_issuer": claims.subject_issuer,
-            "token_id": str(mint.token_id),
-            "client_id": state.client_id,
-            "device_label": state.device_label,
-            "scopes": ["apps:run"],
-            "expires_at": mint.expires_at.isoformat(),
-        },
-    )

api/controllers/openapi/workspaces.py

@@ -1,89 +0,0 @@
-"""User-scoped workspace reads under /openapi/v1/workspaces. 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.
-"""
-
-from __future__ import annotations
-
-from itertools import starmap
-
-from flask import g
-from flask_restx import Resource
-from sqlalchemy import select
-from werkzeug.exceptions import NotFound
-
-from controllers.openapi import openapi_ns
-from extensions.ext_database import db
-from libs.oauth_bearer import (
-    ACCEPT_USER_ANY,
-    SubjectType,
-    validate_bearer,
-)
-from models import Tenant, TenantAccountJoin
-
-
-@openapi_ns.route("/workspaces")
-class WorkspacesApi(Resource):
-    @validate_bearer(accept=ACCEPT_USER_ANY)
-    def get(self):
-        ctx = g.auth_ctx
-        if ctx.subject_type != SubjectType.ACCOUNT or not ctx.account_id:
-            return {"workspaces": []}, 200
-
-        rows = db.session.execute(
-            select(Tenant, TenantAccountJoin)
-            .join(TenantAccountJoin, TenantAccountJoin.tenant_id == Tenant.id)
-            .where(TenantAccountJoin.account_id == str(ctx.account_id))
-            .order_by(Tenant.created_at.asc())
-        ).all()
-
-        return {"workspaces": list(starmap(_workspace_summary, rows))}, 200
-
-
-@openapi_ns.route("/workspaces/<string:workspace_id>")
-class WorkspaceByIdApi(Resource):
-    @validate_bearer(accept=ACCEPT_USER_ANY)
-    def get(self, workspace_id: str):
-        ctx = g.auth_ctx
-        # External SSO + missing account → never a member of anything; 404.
-        if ctx.subject_type != SubjectType.ACCOUNT or not ctx.account_id:
-            raise NotFound("workspace not found")
-
-        row = db.session.execute(
-            select(Tenant, TenantAccountJoin)
-            .join(TenantAccountJoin, TenantAccountJoin.tenant_id == Tenant.id)
-            .where(
-                Tenant.id == workspace_id,
-                TenantAccountJoin.account_id == str(ctx.account_id),
-            )
-        ).first()
-        # 404 (not 403) on non-member so workspace IDs don't leak across tenants.
-        if row is None:
-            raise NotFound("workspace not found")
-
-        tenant, membership = row
-        return _workspace_detail(tenant, membership), 200
-
-
-def _workspace_summary(tenant: Tenant, membership: TenantAccountJoin) -> dict:
-    return {
-        "id": str(tenant.id),
-        "name": tenant.name,
-        "role": getattr(membership, "role", ""),
-        "status": tenant.status,
-        "current": getattr(membership, "current", False),
-    }
-
-
-def _workspace_detail(tenant: Tenant, membership: TenantAccountJoin) -> dict:
-    return {
-        "id": str(tenant.id),
-        "name": tenant.name,
-        "role": getattr(membership, "role", ""),
-        "status": tenant.status,
-        "current": getattr(membership, "current", False),
-        "created_at": tenant.created_at.isoformat() if tenant.created_at else None,
-    }

api/controllers/service_api/app/conversation.py

@@ -84,10 +84,10 @@ class ConversationVariableResponse(ResponseModel):
     def normalize_value_type(cls, value: Any) -> str:
         exposed_type = getattr(value, "exposed_type", None)
         if callable(exposed_type):
-            return str(exposed_type())
+            return str(exposed_type().value)
         if isinstance(value, str):
             try:
-                return str(SegmentType(value).exposed_type())
+                return str(SegmentType(value).exposed_type().value)
             except ValueError:
                 return value
         try:

api/core/agent/base_agent_runner.py

@@ -42,7 +42,7 @@ from graphon.model_runtime.entities import (
 )
 from graphon.model_runtime.entities.message_entities import ImagePromptMessageContent, PromptMessageContentUnionTypes
 from graphon.model_runtime.entities.model_entities import ModelFeature
-from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
+from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from models.enums import CreatorUserRole
 from models.model import Conversation, Message, MessageAgentThought, MessageFile
 

api/core/agent/fc_agent_runner.py

@@ -299,9 +299,7 @@ class FunctionCallAgentRunner(BaseAgentRunner):
 
             # update prompt tool
             for prompt_tool in prompt_messages_tools:
-                tool_instance = tool_instances.get(prompt_tool.name)
-                if tool_instance:
-                    self.update_prompt_message_tool(tool_instance, prompt_tool)
+                self.update_prompt_message_tool(tool_instances[prompt_tool.name], prompt_tool)
 
             iteration_step += 1
 

api/core/app/app_config/easy_ui_based_app/model_config/converter.py

@@ -7,7 +7,7 @@ from core.errors.error import ModelCurrentlyNotSupportError, ProviderTokenNotIni
 from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager
 from graphon.model_runtime.entities.llm_entities import LLMMode
 from graphon.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
-from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
+from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 
 
 class ModelConfigConverter:

api/core/app/apps/agent_chat/app_runner.py

@@ -18,7 +18,7 @@ from core.moderation.base import ModerationError
 from extensions.ext_database import db
 from graphon.model_runtime.entities.llm_entities import LLMMode
 from graphon.model_runtime.entities.model_entities import ModelFeature, ModelPropertyKey
-from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
+from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from models.model import App, Conversation, Message
 
 logger = logging.getLogger(__name__)

api/core/app/apps/workflow/generate_task_pipeline.py

@@ -685,8 +685,6 @@ class WorkflowAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         match invoke_from:
             case InvokeFrom.SERVICE_API:
                 created_from = WorkflowAppLogCreatedFrom.SERVICE_API
-            case InvokeFrom.OPENAPI:
-                created_from = WorkflowAppLogCreatedFrom.OPENAPI
             case InvokeFrom.EXPLORE:
                 created_from = WorkflowAppLogCreatedFrom.INSTALLED_APP
             case InvokeFrom.WEB_APP:

api/core/app/entities/app_invoke_entities.py

@@ -24,7 +24,6 @@ class UserFrom(StrEnum):
 
 class InvokeFrom(StrEnum):
     SERVICE_API = "service-api"
-    OPENAPI = "openapi"
     WEB_APP = "web-app"
     TRIGGER = "trigger"
     EXPLORE = "explore"

api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py

@@ -59,7 +59,7 @@ from graphon.model_runtime.entities.message_entities import (
     AssistantPromptMessage,
     TextPromptMessageContent,
 )
-from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
+from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from libs.datetime_utils import naive_utc_now
 from models.model import AppMode, Conversation, Message, MessageAgentThought, MessageFile, UploadFile
 

api/core/app/workflow/file_runtime.py

@@ -12,14 +12,13 @@ from typing import TYPE_CHECKING, Literal
 from configs import dify_config
 from core.app.file_access import DatabaseFileAccessController, FileAccessControllerProtocol
 from core.db.session_factory import session_factory
-from core.helper.ssrf_proxy import graphon_ssrf_proxy
+from core.helper.ssrf_proxy import ssrf_proxy
 from core.tools.signature import sign_tool_file
 from core.workflow.file_reference import parse_file_reference
 from extensions.ext_storage import storage
 from graphon.file import FileTransferMethod
-from graphon.file.protocols import WorkflowFileRuntimeProtocol
+from graphon.file.protocols import HttpResponseProtocol, WorkflowFileRuntimeProtocol
 from graphon.file.runtime import set_workflow_file_runtime
-from graphon.http.protocols import HttpResponseProtocol
 
 if TYPE_CHECKING:
     from graphon.file import File
@@ -44,7 +43,7 @@ class DifyWorkflowFileRuntime(WorkflowFileRuntimeProtocol):
         return dify_config.MULTIMODAL_SEND_FORMAT
 
     def http_get(self, url: str, *, follow_redirects: bool = True) -> HttpResponseProtocol:
-        return graphon_ssrf_proxy.get(url, follow_redirects=follow_redirects)
+        return ssrf_proxy.get(url, follow_redirects=follow_redirects)
 
     def storage_load(self, path: str, *, stream: bool = False) -> bytes | Generator:
         return storage.load(path, stream=stream)

api/core/app/workflow/layers/persistence.py

@@ -349,7 +349,7 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
         execution.total_tokens = runtime_state.total_tokens
         execution.total_steps = runtime_state.node_run_steps
         execution.outputs = execution.outputs or runtime_state.outputs
-        execution.exceptions_count = max(execution.exceptions_count, runtime_state.exceptions_count)
+        execution.exceptions_count = runtime_state.exceptions_count
 
     def _update_node_execution(
         self,

api/core/datasource/datasource_manager.py

@@ -352,11 +352,11 @@ class DatasourceManager:
                 raise ValueError(f"UploadFile not found for file_id={file_id}, tenant_id={tenant_id}")
 
         file_info = File(
-            file_id=upload_file.id,
+            id=upload_file.id,
             filename=upload_file.name,
             extension="." + upload_file.extension,
             mime_type=upload_file.mime_type,
-            file_type=FileType.CUSTOM,
+            type=FileType.CUSTOM,
             transfer_method=FileTransferMethod.LOCAL_FILE,
             remote_url=upload_file.source_url,
             reference=build_file_reference(record_id=str(upload_file.id)),

api/core/entities/provider_configuration.py

@@ -31,7 +31,7 @@ from graphon.model_runtime.entities.provider_entities import (
     FormType,
     ProviderEntity,
 )
-from graphon.model_runtime.model_providers.base.ai_model import AIModel
+from graphon.model_runtime.model_providers.__base.ai_model import AIModel
 from graphon.model_runtime.model_providers.model_provider_factory import ModelProviderFactory
 from graphon.model_runtime.runtime import ModelRuntime
 from libs.datetime_utils import naive_utc_now
@@ -318,28 +318,34 @@ class ProviderConfiguration(BaseModel):
             else [],
         )
 
-    def validate_provider_credentials(self, credentials: dict[str, Any], credential_id: str = ""):
+    def validate_provider_credentials(
+        self, credentials: dict[str, Any], credential_id: str = "", session: Session | None = None
+    ):
         """
         Validate custom credentials.
         :param credentials: provider credentials
         :param credential_id: (Optional)If provided, can use existing credential's hidden api key to validate
+        :param session: optional database session
         :return:
         """
-        provider_credential_secret_variables = self.extract_secret_variables(
-            self.provider.provider_credential_schema.credential_form_schemas
-            if self.provider.provider_credential_schema
-            else []
-        )
 
-        if credential_id:
-            with Session(db.engine) as session:
+        def _validate(s: Session):
+            # Get provider credential secret variables
+            provider_credential_secret_variables = self.extract_secret_variables(
+                self.provider.provider_credential_schema.credential_form_schemas
+                if self.provider.provider_credential_schema
+                else []
+            )
+
+            if credential_id:
                 try:
                     stmt = select(ProviderCredential).where(
                         ProviderCredential.tenant_id == self.tenant_id,
                         ProviderCredential.provider_name.in_(self._get_provider_names()),
                         ProviderCredential.id == credential_id,
                     )
-                    credential_record = session.execute(stmt).scalar_one_or_none()
+                    credential_record = s.execute(stmt).scalar_one_or_none()
+                    # fix origin data
                     if credential_record and credential_record.encrypted_config:
                         if not credential_record.encrypted_config.startswith("{"):
                             original_credentials = {"openai_api_key": credential_record.encrypted_config}
@@ -350,23 +356,31 @@ class ProviderConfiguration(BaseModel):
                 except JSONDecodeError:
                     original_credentials = {}
 
-            for key, value in credentials.items():
+                # encrypt credentials
+                for key, value in credentials.items():
+                    if key in provider_credential_secret_variables:
+                        # if send [__HIDDEN__] in secret input, it will be same as original value
+                        if value == HIDDEN_VALUE and key in original_credentials:
+                            credentials[key] = encrypter.decrypt_token(
+                                tenant_id=self.tenant_id, token=original_credentials[key]
+                            )
+
+            model_provider_factory = self.get_model_provider_factory()
+            validated_credentials = model_provider_factory.provider_credentials_validate(
+                provider=self.provider.provider, credentials=credentials
+            )
+
+            for key, value in validated_credentials.items():
                 if key in provider_credential_secret_variables:
-                    if value == HIDDEN_VALUE and key in original_credentials:
-                        credentials[key] = encrypter.decrypt_token(
-                            tenant_id=self.tenant_id, token=original_credentials[key]
-                        )
+                    validated_credentials[key] = encrypter.encrypt_token(self.tenant_id, value)
 
-        model_provider_factory = self.get_model_provider_factory()
-        validated_credentials = model_provider_factory.provider_credentials_validate(
-            provider=self.provider.provider, credentials=credentials
-        )
+            return validated_credentials
 
-        for key, value in validated_credentials.items():
-            if key in provider_credential_secret_variables and isinstance(value, str):
-                validated_credentials[key] = encrypter.encrypt_token(self.tenant_id, value)
-
-        return validated_credentials
+        if session:
+            return _validate(session)
+        else:
+            with Session(db.engine) as new_session:
+                return _validate(new_session)
 
     def _generate_provider_credential_name(self, session) -> str:
         """
@@ -443,16 +457,14 @@ class ProviderConfiguration(BaseModel):
         :param credential_name: credential name
         :return:
         """
-        with Session(db.engine) as pre_session:
+        with Session(db.engine) as session:
             if credential_name:
-                if self._check_provider_credential_name_exists(credential_name=credential_name, session=pre_session):
+                if self._check_provider_credential_name_exists(credential_name=credential_name, session=session):
                     raise ValueError(f"Credential with name '{credential_name}' already exists.")
             else:
-                credential_name = self._generate_provider_credential_name(pre_session)
+                credential_name = self._generate_provider_credential_name(session)
 
-        credentials = self.validate_provider_credentials(credentials=credentials)
-
-        with Session(db.engine) as session:
+            credentials = self.validate_provider_credentials(credentials=credentials, session=session)
             provider_record = self._get_provider_record(session)
             try:
                 new_record = ProviderCredential(
@@ -465,6 +477,7 @@ class ProviderConfiguration(BaseModel):
                 session.flush()
 
                 if not provider_record:
+                    # If provider record does not exist, create it
                     provider_record = Provider(
                         tenant_id=self.tenant_id,
                         provider_name=self.provider.provider,
@@ -517,15 +530,15 @@ class ProviderConfiguration(BaseModel):
         :param credential_name: credential name
         :return:
         """
-        with Session(db.engine) as pre_session:
+        with Session(db.engine) as session:
             if credential_name and self._check_provider_credential_name_exists(
-                credential_name=credential_name, session=pre_session, exclude_id=credential_id
+                credential_name=credential_name, session=session, exclude_id=credential_id
             ):
                 raise ValueError(f"Credential with name '{credential_name}' already exists.")
 
-        credentials = self.validate_provider_credentials(credentials=credentials, credential_id=credential_id)
-
-        with Session(db.engine) as session:
+            credentials = self.validate_provider_credentials(
+                credentials=credentials, credential_id=credential_id, session=session
+            )
             provider_record = self._get_provider_record(session)
             stmt = select(ProviderCredential).where(
                 ProviderCredential.id == credential_id,
@@ -533,10 +546,12 @@ class ProviderConfiguration(BaseModel):
                 ProviderCredential.provider_name.in_(self._get_provider_names()),
             )
 
+            # Get the credential record to update
             credential_record = session.execute(stmt).scalar_one_or_none()
             if not credential_record:
                 raise ValueError("Credential record not found.")
             try:
+                # Update credential
                 credential_record.encrypted_config = json.dumps(credentials)
                 credential_record.updated_at = naive_utc_now()
                 if credential_name:
@@ -864,6 +879,7 @@ class ProviderConfiguration(BaseModel):
         model: str,
         credentials: dict[str, Any],
         credential_id: str = "",
+        session: Session | None = None,
     ):
         """
         Validate custom model credentials.
@@ -874,14 +890,16 @@ class ProviderConfiguration(BaseModel):
         :param credential_id: (Optional)If provided, can use existing credential's hidden api key to validate
         :return:
         """
-        provider_credential_secret_variables = self.extract_secret_variables(
-            self.provider.model_credential_schema.credential_form_schemas
-            if self.provider.model_credential_schema
-            else []
-        )
 
-        if credential_id:
-            with Session(db.engine) as session:
+        def _validate(s: Session):
+            # Get provider credential secret variables
+            provider_credential_secret_variables = self.extract_secret_variables(
+                self.provider.model_credential_schema.credential_form_schemas
+                if self.provider.model_credential_schema
+                else []
+            )
+
+            if credential_id:
                 try:
                     stmt = select(ProviderModelCredential).where(
                         ProviderModelCredential.id == credential_id,
@@ -890,7 +908,7 @@ class ProviderConfiguration(BaseModel):
                         ProviderModelCredential.model_name == model,
                         ProviderModelCredential.model_type == model_type,
                     )
-                    credential_record = session.execute(stmt).scalar_one_or_none()
+                    credential_record = s.execute(stmt).scalar_one_or_none()
                     original_credentials = (
                         json.loads(credential_record.encrypted_config)
                         if credential_record and credential_record.encrypted_config
@@ -899,23 +917,31 @@ class ProviderConfiguration(BaseModel):
                 except JSONDecodeError:
                     original_credentials = {}
 
-            for key, value in credentials.items():
+                # decrypt credentials
+                for key, value in credentials.items():
+                    if key in provider_credential_secret_variables:
+                        # if send [__HIDDEN__] in secret input, it will be same as original value
+                        if value == HIDDEN_VALUE and key in original_credentials:
+                            credentials[key] = encrypter.decrypt_token(
+                                tenant_id=self.tenant_id, token=original_credentials[key]
+                            )
+
+            model_provider_factory = self.get_model_provider_factory()
+            validated_credentials = model_provider_factory.model_credentials_validate(
+                provider=self.provider.provider, model_type=model_type, model=model, credentials=credentials
+            )
+
+            for key, value in validated_credentials.items():
                 if key in provider_credential_secret_variables:
-                    if value == HIDDEN_VALUE and key in original_credentials:
-                        credentials[key] = encrypter.decrypt_token(
-                            tenant_id=self.tenant_id, token=original_credentials[key]
-                        )
+                    validated_credentials[key] = encrypter.encrypt_token(self.tenant_id, value)
 
-        model_provider_factory = self.get_model_provider_factory()
-        validated_credentials = model_provider_factory.model_credentials_validate(
-            provider=self.provider.provider, model_type=model_type, model=model, credentials=credentials
-        )
+            return validated_credentials
 
-        for key, value in validated_credentials.items():
-            if key in provider_credential_secret_variables and isinstance(value, str):
-                validated_credentials[key] = encrypter.encrypt_token(self.tenant_id, value)
-
-        return validated_credentials
+        if session:
+            return _validate(session)
+        else:
+            with Session(db.engine) as new_session:
+                return _validate(new_session)
 
     def create_custom_model_credential(
         self, model_type: ModelType, model: str, credentials: dict[str, Any], credential_name: str | None
@@ -928,22 +954,20 @@ class ProviderConfiguration(BaseModel):
         :param credentials: model credentials dict
         :return:
         """
-        with Session(db.engine) as pre_session:
+        with Session(db.engine) as session:
             if credential_name:
                 if self._check_custom_model_credential_name_exists(
-                    model=model, model_type=model_type, credential_name=credential_name, session=pre_session
+                    model=model, model_type=model_type, credential_name=credential_name, session=session
                 ):
                     raise ValueError(f"Model credential with name '{credential_name}' already exists for {model}.")
             else:
                 credential_name = self._generate_custom_model_credential_name(
-                    model=model, model_type=model_type, session=pre_session
+                    model=model, model_type=model_type, session=session
                 )
-
-        credentials = self.validate_custom_model_credentials(
-            model_type=model_type, model=model, credentials=credentials
-        )
-
-        with Session(db.engine) as session:
+            # validate custom model config
+            credentials = self.validate_custom_model_credentials(
+                model_type=model_type, model=model, credentials=credentials, session=session
+            )
             provider_model_record = self._get_custom_model_record(model_type=model_type, model=model, session=session)
 
             try:
@@ -958,6 +982,7 @@ class ProviderConfiguration(BaseModel):
                 session.add(credential)
                 session.flush()
 
+                # save provider model
                 if not provider_model_record:
                     provider_model_record = ProviderModel(
                         tenant_id=self.tenant_id,
@@ -999,24 +1024,23 @@ class ProviderConfiguration(BaseModel):
         :param credential_id: credential id
         :return:
         """
-        with Session(db.engine) as pre_session:
+        with Session(db.engine) as session:
             if credential_name and self._check_custom_model_credential_name_exists(
                 model=model,
                 model_type=model_type,
                 credential_name=credential_name,
-                session=pre_session,
+                session=session,
                 exclude_id=credential_id,
             ):
                 raise ValueError(f"Model credential with name '{credential_name}' already exists for {model}.")
-
-        credentials = self.validate_custom_model_credentials(
-            model_type=model_type,
-            model=model,
-            credentials=credentials,
-            credential_id=credential_id,
-        )
-
-        with Session(db.engine) as session:
+            # validate custom model config
+            credentials = self.validate_custom_model_credentials(
+                model_type=model_type,
+                model=model,
+                credentials=credentials,
+                credential_id=credential_id,
+                session=session,
+            )
             provider_model_record = self._get_custom_model_record(model_type=model_type, model=model, session=session)
 
             stmt = select(ProviderModelCredential).where(
@@ -1031,6 +1055,7 @@ class ProviderConfiguration(BaseModel):
                 raise ValueError("Credential record not found.")
 
             try:
+                # Update credential
                 credential_record.encrypted_config = json.dumps(credentials)
                 credential_record.updated_at = naive_utc_now()
                 if credential_name:

api/core/helper/code_executor/template_transformer.py

@@ -102,7 +102,7 @@ class TemplateTransformer(ABC):
 
     @classmethod
     def serialize_inputs(cls, inputs: Mapping[str, Any]) -> str:
-        inputs_json_str = dumps_with_segments(inputs).encode()
+        inputs_json_str = dumps_with_segments(inputs, ensure_ascii=False).encode()
         input_base64_encoded = b64encode(inputs_json_str).decode("utf-8")
         return input_base64_encoded
 

api/core/helper/moderation.py

@@ -8,7 +8,7 @@ from core.plugin.impl.model_runtime_factory import create_plugin_model_provider_
 from extensions.ext_hosting_provider import hosting_configuration
 from graphon.model_runtime.entities.model_entities import ModelType
 from graphon.model_runtime.errors.invoke import InvokeBadRequestError
-from graphon.model_runtime.model_providers.base.moderation_model import ModerationModel
+from graphon.model_runtime.model_providers.__base.moderation_model import ModerationModel
 from models.provider import ProviderType
 
 logger = logging.getLogger(__name__)

api/core/helper/ssrf_proxy.py

@@ -12,7 +12,6 @@ from pydantic import TypeAdapter, ValidationError
 from configs import dify_config
 from core.helper.http_client_pooling import get_pooled_http_client
 from core.tools.errors import ToolSSRFError
-from graphon.http.response import HttpResponse
 
 logger = logging.getLogger(__name__)
 
@@ -268,47 +267,4 @@ class SSRFProxy:
         return patch(url=url, max_retries=max_retries, **kwargs)
 
 
-def _to_graphon_http_response(response: httpx.Response) -> HttpResponse:
-    """Convert an ``httpx`` response into Graphon's transport-agnostic wrapper."""
-    return HttpResponse(
-        status_code=response.status_code,
-        headers=dict(response.headers),
-        content=response.content,
-        url=str(response.url) if response.url else None,
-        reason_phrase=response.reason_phrase,
-        fallback_text=response.text,
-    )
-
-
-class GraphonSSRFProxy:
-    """Adapter exposing SSRF helpers behind Graphon's ``HttpClientProtocol``."""
-
-    @property
-    def max_retries_exceeded_error(self) -> type[Exception]:
-        return max_retries_exceeded_error
-
-    @property
-    def request_error(self) -> type[Exception]:
-        return request_error
-
-    def get(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> HttpResponse:
-        return _to_graphon_http_response(get(url=url, max_retries=max_retries, **kwargs))
-
-    def head(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> HttpResponse:
-        return _to_graphon_http_response(head(url=url, max_retries=max_retries, **kwargs))
-
-    def post(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> HttpResponse:
-        return _to_graphon_http_response(post(url=url, max_retries=max_retries, **kwargs))
-
-    def put(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> HttpResponse:
-        return _to_graphon_http_response(put(url=url, max_retries=max_retries, **kwargs))
-
-    def delete(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> HttpResponse:
-        return _to_graphon_http_response(delete(url=url, max_retries=max_retries, **kwargs))
-
-    def patch(self, url: str, max_retries: int = SSRF_DEFAULT_MAX_RETRIES, **kwargs: Any) -> HttpResponse:
-        return _to_graphon_http_response(patch(url=url, max_retries=max_retries, **kwargs))
-
-
 ssrf_proxy = SSRFProxy()
-graphon_ssrf_proxy = GraphonSSRFProxy()

api/core/mcp/client/streamable_client.py

@@ -303,16 +303,9 @@ class StreamableHTTPTransport:
 
             if response.status_code == 404:
                 if isinstance(message.root, JSONRPCRequest):
-                    error_msg = (
-                        f"MCP server URL returned 404 Not Found: {self.url} "
-                        "— verify the server URL is correct and the server is running"
-                        if is_initialization
-                        else "Session terminated by server"
-                    )
                     self._send_session_terminated_error(
                         ctx.server_to_client_queue,
                         message.root.id,
-                        message=error_msg,
                     )
                 return
 
@@ -388,13 +381,12 @@ class StreamableHTTPTransport:
         self,
         server_to_client_queue: ServerToClientQueue,
         request_id: RequestId,
-        message: str = "Session terminated by server",
     ):
         """Send a session terminated error response."""
         jsonrpc_error = JSONRPCError(
             jsonrpc="2.0",
             id=request_id,
-            error=ErrorData(code=32600, message=message),
+            error=ErrorData(code=32600, message="Session terminated by server"),
         )
         session_message = SessionMessage(JSONRPCMessage(jsonrpc_error))
         server_to_client_queue.put(session_message)

api/core/model_manager.py

@@ -1,6 +1,6 @@
 import logging
 from collections.abc import Callable, Generator, Iterable, Mapping, Sequence
-from typing import IO, Any, Literal, Optional, ParamSpec, TypeVar, Union, cast, overload
+from typing import IO, Any, Literal, Optional, Union, cast, overload
 
 from configs import dify_config
 from core.entities import PluginCredentialType
@@ -18,17 +18,15 @@ from graphon.model_runtime.entities.model_entities import AIModelEntity, ModelFe
 from graphon.model_runtime.entities.rerank_entities import MultimodalRerankInput, RerankResult
 from graphon.model_runtime.entities.text_embedding_entities import EmbeddingResult
 from graphon.model_runtime.errors.invoke import InvokeAuthorizationError, InvokeConnectionError, InvokeRateLimitError
-from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
-from graphon.model_runtime.model_providers.base.moderation_model import ModerationModel
-from graphon.model_runtime.model_providers.base.rerank_model import RerankModel
-from graphon.model_runtime.model_providers.base.speech2text_model import Speech2TextModel
-from graphon.model_runtime.model_providers.base.text_embedding_model import TextEmbeddingModel
-from graphon.model_runtime.model_providers.base.tts_model import TTSModel
+from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
+from graphon.model_runtime.model_providers.__base.moderation_model import ModerationModel
+from graphon.model_runtime.model_providers.__base.rerank_model import RerankModel
+from graphon.model_runtime.model_providers.__base.speech2text_model import Speech2TextModel
+from graphon.model_runtime.model_providers.__base.text_embedding_model import TextEmbeddingModel
+from graphon.model_runtime.model_providers.__base.tts_model import TTSModel
 from models.provider import ProviderType
 
 logger = logging.getLogger(__name__)
-P = ParamSpec("P")
-R = TypeVar("R")
 
 
 class ModelInstance:
@@ -170,7 +168,7 @@ class ModelInstance:
         return cast(
             Union[LLMResult, Generator],
             self._round_robin_invoke(
-                self.model_type_instance.invoke,
+                function=self.model_type_instance.invoke,
                 model=self.model_name,
                 credentials=self.credentials,
                 prompt_messages=list(prompt_messages),
@@ -195,7 +193,7 @@ class ModelInstance:
         if not isinstance(self.model_type_instance, LargeLanguageModel):
             raise Exception("Model type instance is not LargeLanguageModel")
         return self._round_robin_invoke(
-            self.model_type_instance.get_num_tokens,
+            function=self.model_type_instance.get_num_tokens,
             model=self.model_name,
             credentials=self.credentials,
             prompt_messages=list(prompt_messages),
@@ -215,7 +213,7 @@ class ModelInstance:
         if not isinstance(self.model_type_instance, TextEmbeddingModel):
             raise Exception("Model type instance is not TextEmbeddingModel")
         return self._round_robin_invoke(
-            self.model_type_instance.invoke,
+            function=self.model_type_instance.invoke,
             model=self.model_name,
             credentials=self.credentials,
             texts=texts,
@@ -237,7 +235,7 @@ class ModelInstance:
         if not isinstance(self.model_type_instance, TextEmbeddingModel):
             raise Exception("Model type instance is not TextEmbeddingModel")
         return self._round_robin_invoke(
-            self.model_type_instance.invoke,
+            function=self.model_type_instance.invoke,
             model=self.model_name,
             credentials=self.credentials,
             multimodel_documents=multimodel_documents,
@@ -254,7 +252,7 @@ class ModelInstance:
         if not isinstance(self.model_type_instance, TextEmbeddingModel):
             raise Exception("Model type instance is not TextEmbeddingModel")
         return self._round_robin_invoke(
-            self.model_type_instance.get_num_tokens,
+            function=self.model_type_instance.get_num_tokens,
             model=self.model_name,
             credentials=self.credentials,
             texts=texts,
@@ -279,7 +277,7 @@ class ModelInstance:
         if not isinstance(self.model_type_instance, RerankModel):
             raise Exception("Model type instance is not RerankModel")
         return self._round_robin_invoke(
-            self.model_type_instance.invoke,
+            function=self.model_type_instance.invoke,
             model=self.model_name,
             credentials=self.credentials,
             query=query,
@@ -307,7 +305,7 @@ class ModelInstance:
         if not isinstance(self.model_type_instance, RerankModel):
             raise Exception("Model type instance is not RerankModel")
         return self._round_robin_invoke(
-            self.model_type_instance.invoke_multimodal_rerank,
+            function=self.model_type_instance.invoke_multimodal_rerank,
             model=self.model_name,
             credentials=self.credentials,
             query=query,
@@ -326,7 +324,7 @@ class ModelInstance:
         if not isinstance(self.model_type_instance, ModerationModel):
             raise Exception("Model type instance is not ModerationModel")
         return self._round_robin_invoke(
-            self.model_type_instance.invoke,
+            function=self.model_type_instance.invoke,
             model=self.model_name,
             credentials=self.credentials,
             text=text,
@@ -342,7 +340,7 @@ class ModelInstance:
         if not isinstance(self.model_type_instance, Speech2TextModel):
             raise Exception("Model type instance is not Speech2TextModel")
         return self._round_robin_invoke(
-            self.model_type_instance.invoke,
+            function=self.model_type_instance.invoke,
             model=self.model_name,
             credentials=self.credentials,
             file=file,
@@ -359,14 +357,14 @@ class ModelInstance:
         if not isinstance(self.model_type_instance, TTSModel):
             raise Exception("Model type instance is not TTSModel")
         return self._round_robin_invoke(
-            self.model_type_instance.invoke,
+            function=self.model_type_instance.invoke,
             model=self.model_name,
             credentials=self.credentials,
             content_text=content_text,
             voice=voice,
         )
 
-    def _round_robin_invoke(self, function: Callable[P, R], *args: P.args, **kwargs: P.kwargs) -> R:
+    def _round_robin_invoke[**P, R](self, function: Callable[P, R], *args: P.args, **kwargs: P.kwargs) -> R:
         """
         Round-robin invoke
         :param function: function to invoke

api/core/ops/aliyun_trace/aliyun_trace.py

@@ -4,20 +4,7 @@ from collections.abc import Sequence
 from opentelemetry.trace import SpanKind
 from sqlalchemy.orm import sessionmaker
 
-from core.ops.base_trace_instance import BaseTraceInstance
-from core.ops.entities.trace_entity import (
-    BaseTraceInfo,
-    DatasetRetrievalTraceInfo,
-    GenerateNameTraceInfo,
-    MessageTraceInfo,
-    ModerationTraceInfo,
-    SuggestedQuestionTraceInfo,
-    ToolTraceInfo,
-    WorkflowTraceInfo,
-)
-from core.repositories import DifyCoreRepositoryFactory
-from dify_trace_aliyun.config import AliyunConfig
-from dify_trace_aliyun.data_exporter.traceclient import (
+from core.ops.aliyun_trace.data_exporter.traceclient import (
     TraceClient,
     build_endpoint,
     convert_datetime_to_nanoseconds,
@@ -25,8 +12,8 @@ from dify_trace_aliyun.data_exporter.traceclient import (
     convert_to_trace_id,
     generate_span_id,
 )
-from dify_trace_aliyun.entities.aliyun_trace_entity import SpanData, TraceMetadata
-from dify_trace_aliyun.entities.semconv import (
+from core.ops.aliyun_trace.entities.aliyun_trace_entity import SpanData, TraceMetadata
+from core.ops.aliyun_trace.entities.semconv import (
     DIFY_APP_ID,
     GEN_AI_COMPLETION,
     GEN_AI_INPUT_MESSAGE,
@@ -45,7 +32,7 @@ from dify_trace_aliyun.entities.semconv import (
     TOOL_PARAMETERS,
     GenAISpanKind,
 )
-from dify_trace_aliyun.utils import (
+from core.ops.aliyun_trace.utils import (
     create_common_span_attributes,
     create_links_from_trace_id,
     create_status_from_error,
@@ -57,6 +44,19 @@ from dify_trace_aliyun.utils import (
     get_workflow_node_status,
     serialize_json_data,
 )
+from core.ops.base_trace_instance import BaseTraceInstance
+from core.ops.entities.config_entity import AliyunConfig
+from core.ops.entities.trace_entity import (
+    BaseTraceInfo,
+    DatasetRetrievalTraceInfo,
+    GenerateNameTraceInfo,
+    MessageTraceInfo,
+    ModerationTraceInfo,
+    SuggestedQuestionTraceInfo,
+    ToolTraceInfo,
+    WorkflowTraceInfo,
+)
+from core.repositories import DifyCoreRepositoryFactory
 from extensions.ext_database import db
 from graphon.entities import WorkflowNodeExecution
 from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey

api/core/ops/aliyun_trace/data_exporter/traceclient.py

@@ -26,8 +26,8 @@ from opentelemetry.semconv.attributes import service_attributes
 from opentelemetry.trace import Link, SpanContext, TraceFlags
 
 from configs import dify_config
-from dify_trace_aliyun.entities.aliyun_trace_entity import SpanData
-from dify_trace_aliyun.entities.semconv import ACS_ARMS_SERVICE_FEATURE
+from core.ops.aliyun_trace.entities.aliyun_trace_entity import SpanData
+from core.ops.aliyun_trace.entities.semconv import ACS_ARMS_SERVICE_FEATURE
 
 INVALID_SPAN_ID: Final[int] = 0x0000000000000000
 INVALID_TRACE_ID: Final[int] = 0x00000000000000000000000000000000

api/core/ops/aliyun_trace/utils.py

@@ -4,8 +4,7 @@ from typing import Any, TypedDict
 
 from opentelemetry.trace import Link, Status, StatusCode
 
-from core.rag.models.document import Document
-from dify_trace_aliyun.entities.semconv import (
+from core.ops.aliyun_trace.entities.semconv import (
     GEN_AI_FRAMEWORK,
     GEN_AI_SESSION_ID,
     GEN_AI_SPAN_KIND,
@@ -14,6 +13,7 @@ from dify_trace_aliyun.entities.semconv import (
     OUTPUT_VALUE,
     GenAISpanKind,
 )
+from core.rag.models.document import Document
 from extensions.ext_database import db
 from graphon.entities import WorkflowNodeExecution
 from graphon.enums import WorkflowNodeExecutionStatus
@@ -48,7 +48,7 @@ def get_workflow_node_status(node_execution: WorkflowNodeExecution) -> Status:
 
 
 def create_links_from_trace_id(trace_id: str | None) -> list[Link]:
-    from dify_trace_aliyun.data_exporter.traceclient import create_link
+    from core.ops.aliyun_trace.data_exporter.traceclient import create_link
 
     links = []
     if trace_id:

api/core/ops/arize_phoenix_trace/arize_phoenix_trace.py

@@ -25,6 +25,7 @@ from opentelemetry.util.types import AttributeValue
 from sqlalchemy.orm import sessionmaker
 
 from core.ops.base_trace_instance import BaseTraceInstance
+from core.ops.entities.config_entity import ArizeConfig, PhoenixConfig
 from core.ops.entities.trace_entity import (
     BaseTraceInfo,
     DatasetRetrievalTraceInfo,
@@ -38,7 +39,6 @@ from core.ops.entities.trace_entity import (
 )
 from core.ops.utils import JSON_DICT_ADAPTER
 from core.repositories import DifyCoreRepositoryFactory
-from dify_trace_arize_phoenix.config import ArizeConfig, PhoenixConfig
 from extensions.ext_database import db
 from graphon.enums import WorkflowNodeExecutionStatus
 from models.model import EndUser, MessageFile

api/core/ops/entities/config_entity.py

@@ -1,8 +1,8 @@
 from enum import StrEnum
 
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationInfo, field_validator
 
-from core.ops.utils import validate_project_name, validate_url
+from core.ops.utils import validate_integer_id, validate_project_name, validate_url, validate_url_with_path
 
 
 class TracingProviderEnum(StrEnum):
@@ -52,5 +52,220 @@ class BaseTracingConfig(BaseModel):
         return validate_project_name(v, default_name)
 
 
+class ArizeConfig(BaseTracingConfig):
+    """
+    Model class for Arize tracing config.
+    """
+
+    api_key: str | None = None
+    space_id: str | None = None
+    project: str | None = None
+    endpoint: str = "https://otlp.arize.com"
+
+    @field_validator("project")
+    @classmethod
+    def project_validator(cls, v, info: ValidationInfo):
+        return cls.validate_project_field(v, "default")
+
+    @field_validator("endpoint")
+    @classmethod
+    def endpoint_validator(cls, v, info: ValidationInfo):
+        return cls.validate_endpoint_url(v, "https://otlp.arize.com")
+
+
+class PhoenixConfig(BaseTracingConfig):
+    """
+    Model class for Phoenix tracing config.
+    """
+
+    api_key: str | None = None
+    project: str | None = None
+    endpoint: str = "https://app.phoenix.arize.com"
+
+    @field_validator("project")
+    @classmethod
+    def project_validator(cls, v, info: ValidationInfo):
+        return cls.validate_project_field(v, "default")
+
+    @field_validator("endpoint")
+    @classmethod
+    def endpoint_validator(cls, v, info: ValidationInfo):
+        return validate_url_with_path(v, "https://app.phoenix.arize.com")
+
+
+class LangfuseConfig(BaseTracingConfig):
+    """
+    Model class for Langfuse tracing config.
+    """
+
+    public_key: str
+    secret_key: str
+    host: str = "https://api.langfuse.com"
+
+    @field_validator("host")
+    @classmethod
+    def host_validator(cls, v, info: ValidationInfo):
+        return validate_url_with_path(v, "https://api.langfuse.com")
+
+
+class LangSmithConfig(BaseTracingConfig):
+    """
+    Model class for Langsmith tracing config.
+    """
+
+    api_key: str
+    project: str
+    endpoint: str = "https://api.smith.langchain.com"
+
+    @field_validator("endpoint")
+    @classmethod
+    def endpoint_validator(cls, v, info: ValidationInfo):
+        # LangSmith only allows HTTPS
+        return validate_url(v, "https://api.smith.langchain.com", allowed_schemes=("https",))
+
+
+class OpikConfig(BaseTracingConfig):
+    """
+    Model class for Opik tracing config.
+    """
+
+    api_key: str | None = None
+    project: str | None = None
+    workspace: str | None = None
+    url: str = "https://www.comet.com/opik/api/"
+
+    @field_validator("project")
+    @classmethod
+    def project_validator(cls, v, info: ValidationInfo):
+        return cls.validate_project_field(v, "Default Project")
+
+    @field_validator("url")
+    @classmethod
+    def url_validator(cls, v, info: ValidationInfo):
+        return validate_url_with_path(v, "https://www.comet.com/opik/api/", required_suffix="/api/")
+
+
+class WeaveConfig(BaseTracingConfig):
+    """
+    Model class for Weave tracing config.
+    """
+
+    api_key: str
+    entity: str | None = None
+    project: str
+    endpoint: str = "https://trace.wandb.ai"
+    host: str | None = None
+
+    @field_validator("endpoint")
+    @classmethod
+    def endpoint_validator(cls, v, info: ValidationInfo):
+        # Weave only allows HTTPS for endpoint
+        return validate_url(v, "https://trace.wandb.ai", allowed_schemes=("https",))
+
+    @field_validator("host")
+    @classmethod
+    def host_validator(cls, v, info: ValidationInfo):
+        if v is not None and v.strip() != "":
+            return validate_url(v, v, allowed_schemes=("https", "http"))
+        return v
+
+
+class AliyunConfig(BaseTracingConfig):
+    """
+    Model class for Aliyun tracing config.
+    """
+
+    app_name: str = "dify_app"
+    license_key: str
+    endpoint: str
+
+    @field_validator("app_name")
+    @classmethod
+    def app_name_validator(cls, v, info: ValidationInfo):
+        return cls.validate_project_field(v, "dify_app")
+
+    @field_validator("license_key")
+    @classmethod
+    def license_key_validator(cls, v, info: ValidationInfo):
+        if not v or v.strip() == "":
+            raise ValueError("License key cannot be empty")
+        return v
+
+    @field_validator("endpoint")
+    @classmethod
+    def endpoint_validator(cls, v, info: ValidationInfo):
+        # aliyun uses two URL formats, which may include a URL path
+        return validate_url_with_path(v, "https://tracing-analysis-dc-hz.aliyuncs.com")
+
+
+class TencentConfig(BaseTracingConfig):
+    """
+    Tencent APM tracing config
+    """
+
+    token: str
+    endpoint: str
+    service_name: str
+
+    @field_validator("token")
+    @classmethod
+    def token_validator(cls, v, info: ValidationInfo):
+        if not v or v.strip() == "":
+            raise ValueError("Token cannot be empty")
+        return v
+
+    @field_validator("endpoint")
+    @classmethod
+    def endpoint_validator(cls, v, info: ValidationInfo):
+        return cls.validate_endpoint_url(v, "https://apm.tencentcloudapi.com")
+
+    @field_validator("service_name")
+    @classmethod
+    def service_name_validator(cls, v, info: ValidationInfo):
+        return cls.validate_project_field(v, "dify_app")
+
+
+class MLflowConfig(BaseTracingConfig):
+    """
+    Model class for MLflow tracing config.
+    """
+
+    tracking_uri: str = "http://localhost:5000"
+    experiment_id: str = "0"  # Default experiment id in MLflow is 0
+    username: str | None = None
+    password: str | None = None
+
+    @field_validator("tracking_uri")
+    @classmethod
+    def tracking_uri_validator(cls, v, info: ValidationInfo):
+        if isinstance(v, str) and v.startswith("databricks"):
+            raise ValueError(
+                "Please use Databricks tracing config below to record traces to Databricks-managed MLflow instances."
+            )
+        return validate_url_with_path(v, "http://localhost:5000")
+
+    @field_validator("experiment_id")
+    @classmethod
+    def experiment_id_validator(cls, v, info: ValidationInfo):
+        return validate_integer_id(v)
+
+
+class DatabricksConfig(BaseTracingConfig):
+    """
+    Model class for Databricks (Databricks-managed MLflow) tracing config.
+    """
+
+    experiment_id: str
+    host: str
+    client_id: str | None = None
+    client_secret: str | None = None
+    personal_access_token: str | None = None
+
+    @field_validator("experiment_id")
+    @classmethod
+    def experiment_id_validator(cls, v, info: ValidationInfo):
+        return validate_integer_id(v)
+
+
 OPS_FILE_PATH = "ops_trace/"
 OPS_TRACE_FAILED_KEY = "FAILED_OPS_TRACE"

api/core/ops/langfuse_trace/langfuse_trace.py

@@ -16,6 +16,7 @@ from langfuse.api.commons.types.usage import Usage
 from sqlalchemy.orm import sessionmaker
 
 from core.ops.base_trace_instance import BaseTraceInstance
+from core.ops.entities.config_entity import LangfuseConfig
 from core.ops.entities.trace_entity import (
     BaseTraceInfo,
     DatasetRetrievalTraceInfo,
@@ -27,10 +28,7 @@ from core.ops.entities.trace_entity import (
     TraceTaskName,
     WorkflowTraceInfo,
 )
-from core.ops.utils import filter_none_values
-from core.repositories import DifyCoreRepositoryFactory
-from dify_trace_langfuse.config import LangfuseConfig
-from dify_trace_langfuse.entities.langfuse_trace_entity import (
+from core.ops.langfuse_trace.entities.langfuse_trace_entity import (
     GenerationUsage,
     LangfuseGeneration,
     LangfuseSpan,
@@ -38,6 +36,8 @@ from dify_trace_langfuse.entities.langfuse_trace_entity import (
     LevelEnum,
     UnitEnum,
 )
+from core.ops.utils import filter_none_values
+from core.repositories import DifyCoreRepositoryFactory
 from extensions.ext_database import db
 from graphon.enums import BuiltinNodeTypes
 from models import EndUser, WorkflowNodeExecutionTriggeredFrom

api/core/ops/langsmith_trace/langsmith_trace.py

@@ -9,6 +9,7 @@ from langsmith.schemas import RunBase
 from sqlalchemy.orm import sessionmaker
 
 from core.ops.base_trace_instance import BaseTraceInstance
+from core.ops.entities.config_entity import LangSmithConfig
 from core.ops.entities.trace_entity import (
     BaseTraceInfo,
     DatasetRetrievalTraceInfo,
@@ -20,14 +21,13 @@ from core.ops.entities.trace_entity import (
     TraceTaskName,
     WorkflowTraceInfo,
 )
-from core.ops.utils import filter_none_values, generate_dotted_order
-from core.repositories import DifyCoreRepositoryFactory
-from dify_trace_langsmith.config import LangSmithConfig
-from dify_trace_langsmith.entities.langsmith_trace_entity import (
+from core.ops.langsmith_trace.entities.langsmith_trace_entity import (
     LangSmithRunModel,
     LangSmithRunType,
     LangSmithRunUpdateModel,
 )
+from core.ops.utils import filter_none_values, generate_dotted_order
+from core.repositories import DifyCoreRepositoryFactory
 from extensions.ext_database import db
 from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey
 from models import EndUser, MessageFile, WorkflowNodeExecutionTriggeredFrom

api/core/ops/mlflow_trace/mlflow_trace.py

@@ -11,6 +11,7 @@ from mlflow.tracing.provider import detach_span_from_context, set_span_in_contex
 from sqlalchemy import select
 
 from core.ops.base_trace_instance import BaseTraceInstance
+from core.ops.entities.config_entity import DatabricksConfig, MLflowConfig
 from core.ops.entities.trace_entity import (
     BaseTraceInfo,
     DatasetRetrievalTraceInfo,
@@ -23,7 +24,6 @@ from core.ops.entities.trace_entity import (
     WorkflowTraceInfo,
 )
 from core.ops.utils import JSON_DICT_ADAPTER
-from dify_trace_mlflow.config import DatabricksConfig, MLflowConfig
 from extensions.ext_database import db
 from graphon.enums import BuiltinNodeTypes
 from models import EndUser

api/core/ops/opik_trace/opik_trace.py

@@ -10,6 +10,7 @@ from opik.id_helpers import uuid4_to_uuid7
 from sqlalchemy.orm import sessionmaker
 
 from core.ops.base_trace_instance import BaseTraceInstance
+from core.ops.entities.config_entity import OpikConfig
 from core.ops.entities.trace_entity import (
     BaseTraceInfo,
     DatasetRetrievalTraceInfo,
@@ -22,7 +23,6 @@ from core.ops.entities.trace_entity import (
     WorkflowTraceInfo,
 )
 from core.repositories import DifyCoreRepositoryFactory
-from dify_trace_opik.config import OpikConfig
 from extensions.ext_database import db
 from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey
 from models import EndUser, MessageFile, WorkflowNodeExecutionTriggeredFrom

api/core/ops/ops_trace_manager.py

@@ -204,117 +204,114 @@ class TracingProviderConfigEntry(TypedDict):
 
 class OpsTraceProviderConfigMap(collections.UserDict[str, TracingProviderConfigEntry]):
     def __getitem__(self, provider: str) -> TracingProviderConfigEntry:
-        try:
-            match provider:
-                case TracingProviderEnum.LANGFUSE:
-                    from dify_trace_langfuse.config import LangfuseConfig
-                    from dify_trace_langfuse.langfuse_trace import LangFuseDataTrace
+        match provider:
+            case TracingProviderEnum.LANGFUSE:
+                from core.ops.entities.config_entity import LangfuseConfig
+                from core.ops.langfuse_trace.langfuse_trace import LangFuseDataTrace
 
-                    return {
-                        "config_class": LangfuseConfig,
-                        "secret_keys": ["public_key", "secret_key"],
-                        "other_keys": ["host", "project_key"],
-                        "trace_instance": LangFuseDataTrace,
-                    }
+                return {
+                    "config_class": LangfuseConfig,
+                    "secret_keys": ["public_key", "secret_key"],
+                    "other_keys": ["host", "project_key"],
+                    "trace_instance": LangFuseDataTrace,
+                }
 
-                case TracingProviderEnum.LANGSMITH:
-                    from dify_trace_langsmith.config import LangSmithConfig
-                    from dify_trace_langsmith.langsmith_trace import LangSmithDataTrace
+            case TracingProviderEnum.LANGSMITH:
+                from core.ops.entities.config_entity import LangSmithConfig
+                from core.ops.langsmith_trace.langsmith_trace import LangSmithDataTrace
 
-                    return {
-                        "config_class": LangSmithConfig,
-                        "secret_keys": ["api_key"],
-                        "other_keys": ["project", "endpoint"],
-                        "trace_instance": LangSmithDataTrace,
-                    }
+                return {
+                    "config_class": LangSmithConfig,
+                    "secret_keys": ["api_key"],
+                    "other_keys": ["project", "endpoint"],
+                    "trace_instance": LangSmithDataTrace,
+                }
 
-                case TracingProviderEnum.OPIK:
-                    from dify_trace_opik.config import OpikConfig
-                    from dify_trace_opik.opik_trace import OpikDataTrace
+            case TracingProviderEnum.OPIK:
+                from core.ops.entities.config_entity import OpikConfig
+                from core.ops.opik_trace.opik_trace import OpikDataTrace
 
-                    return {
-                        "config_class": OpikConfig,
-                        "secret_keys": ["api_key"],
-                        "other_keys": ["project", "url", "workspace"],
-                        "trace_instance": OpikDataTrace,
-                    }
+                return {
+                    "config_class": OpikConfig,
+                    "secret_keys": ["api_key"],
+                    "other_keys": ["project", "url", "workspace"],
+                    "trace_instance": OpikDataTrace,
+                }
 
-                case TracingProviderEnum.WEAVE:
-                    from dify_trace_weave.config import WeaveConfig
-                    from dify_trace_weave.weave_trace import WeaveDataTrace
+            case TracingProviderEnum.WEAVE:
+                from core.ops.entities.config_entity import WeaveConfig
+                from core.ops.weave_trace.weave_trace import WeaveDataTrace
 
-                    return {
-                        "config_class": WeaveConfig,
-                        "secret_keys": ["api_key"],
-                        "other_keys": ["project", "entity", "endpoint", "host"],
-                        "trace_instance": WeaveDataTrace,
-                    }
-                case TracingProviderEnum.ARIZE:
-                    from dify_trace_arize_phoenix.arize_phoenix_trace import ArizePhoenixDataTrace
-                    from dify_trace_arize_phoenix.config import ArizeConfig
+                return {
+                    "config_class": WeaveConfig,
+                    "secret_keys": ["api_key"],
+                    "other_keys": ["project", "entity", "endpoint", "host"],
+                    "trace_instance": WeaveDataTrace,
+                }
+            case TracingProviderEnum.ARIZE:
+                from core.ops.arize_phoenix_trace.arize_phoenix_trace import ArizePhoenixDataTrace
+                from core.ops.entities.config_entity import ArizeConfig
 
-                    return {
-                        "config_class": ArizeConfig,
-                        "secret_keys": ["api_key", "space_id"],
-                        "other_keys": ["project", "endpoint"],
-                        "trace_instance": ArizePhoenixDataTrace,
-                    }
-                case TracingProviderEnum.PHOENIX:
-                    from dify_trace_arize_phoenix.arize_phoenix_trace import ArizePhoenixDataTrace
-                    from dify_trace_arize_phoenix.config import PhoenixConfig
+                return {
+                    "config_class": ArizeConfig,
+                    "secret_keys": ["api_key", "space_id"],
+                    "other_keys": ["project", "endpoint"],
+                    "trace_instance": ArizePhoenixDataTrace,
+                }
+            case TracingProviderEnum.PHOENIX:
+                from core.ops.arize_phoenix_trace.arize_phoenix_trace import ArizePhoenixDataTrace
+                from core.ops.entities.config_entity import PhoenixConfig
 
-                    return {
-                        "config_class": PhoenixConfig,
-                        "secret_keys": ["api_key"],
-                        "other_keys": ["project", "endpoint"],
-                        "trace_instance": ArizePhoenixDataTrace,
-                    }
-                case TracingProviderEnum.ALIYUN:
-                    from dify_trace_aliyun.aliyun_trace import AliyunDataTrace
-                    from dify_trace_aliyun.config import AliyunConfig
+                return {
+                    "config_class": PhoenixConfig,
+                    "secret_keys": ["api_key"],
+                    "other_keys": ["project", "endpoint"],
+                    "trace_instance": ArizePhoenixDataTrace,
+                }
+            case TracingProviderEnum.ALIYUN:
+                from core.ops.aliyun_trace.aliyun_trace import AliyunDataTrace
+                from core.ops.entities.config_entity import AliyunConfig
 
-                    return {
-                        "config_class": AliyunConfig,
-                        "secret_keys": ["license_key"],
-                        "other_keys": ["endpoint", "app_name"],
-                        "trace_instance": AliyunDataTrace,
-                    }
-                case TracingProviderEnum.MLFLOW:
-                    from dify_trace_mlflow.config import MLflowConfig
-                    from dify_trace_mlflow.mlflow_trace import MLflowDataTrace
+                return {
+                    "config_class": AliyunConfig,
+                    "secret_keys": ["license_key"],
+                    "other_keys": ["endpoint", "app_name"],
+                    "trace_instance": AliyunDataTrace,
+                }
+            case TracingProviderEnum.MLFLOW:
+                from core.ops.entities.config_entity import MLflowConfig
+                from core.ops.mlflow_trace.mlflow_trace import MLflowDataTrace
 
-                    return {
-                        "config_class": MLflowConfig,
-                        "secret_keys": ["password"],
-                        "other_keys": ["tracking_uri", "experiment_id", "username"],
-                        "trace_instance": MLflowDataTrace,
-                    }
-                case TracingProviderEnum.DATABRICKS:
-                    from dify_trace_mlflow.config import DatabricksConfig
-                    from dify_trace_mlflow.mlflow_trace import MLflowDataTrace
+                return {
+                    "config_class": MLflowConfig,
+                    "secret_keys": ["password"],
+                    "other_keys": ["tracking_uri", "experiment_id", "username"],
+                    "trace_instance": MLflowDataTrace,
+                }
+            case TracingProviderEnum.DATABRICKS:
+                from core.ops.entities.config_entity import DatabricksConfig
+                from core.ops.mlflow_trace.mlflow_trace import MLflowDataTrace
 
-                    return {
-                        "config_class": DatabricksConfig,
-                        "secret_keys": ["personal_access_token", "client_secret"],
-                        "other_keys": ["host", "client_id", "experiment_id"],
-                        "trace_instance": MLflowDataTrace,
-                    }
+                return {
+                    "config_class": DatabricksConfig,
+                    "secret_keys": ["personal_access_token", "client_secret"],
+                    "other_keys": ["host", "client_id", "experiment_id"],
+                    "trace_instance": MLflowDataTrace,
+                }
 
-                case TracingProviderEnum.TENCENT:
-                    from dify_trace_tencent.config import TencentConfig
-                    from dify_trace_tencent.tencent_trace import TencentDataTrace
+            case TracingProviderEnum.TENCENT:
+                from core.ops.entities.config_entity import TencentConfig
+                from core.ops.tencent_trace.tencent_trace import TencentDataTrace
 
-                    return {
-                        "config_class": TencentConfig,
-                        "secret_keys": ["token"],
-                        "other_keys": ["endpoint", "service_name"],
-                        "trace_instance": TencentDataTrace,
-                    }
+                return {
+                    "config_class": TencentConfig,
+                    "secret_keys": ["token"],
+                    "other_keys": ["endpoint", "service_name"],
+                    "trace_instance": TencentDataTrace,
+                }
 
-                case _:
-                    raise KeyError(f"Unsupported tracing provider: {provider}")
-        except ImportError:
-            raise ImportError(f"Provider {provider} is not installed.")
+            case _:
+                raise KeyError(f"Unsupported tracing provider: {provider}")
 
 
 provider_config_map = OpsTraceProviderConfigMap()

api/core/ops/tencent_trace/span_builder.py

@@ -14,8 +14,7 @@ from core.ops.entities.trace_entity import (
     ToolTraceInfo,
     WorkflowTraceInfo,
 )
-from core.rag.models.document import Document
-from dify_trace_tencent.entities.semconv import (
+from core.ops.tencent_trace.entities.semconv import (
     GEN_AI_COMPLETION,
     GEN_AI_FRAMEWORK,
     GEN_AI_IS_ENTRY,
@@ -39,8 +38,9 @@ from dify_trace_tencent.entities.semconv import (
     TOOL_PARAMETERS,
     GenAISpanKind,
 )
-from dify_trace_tencent.entities.tencent_trace_entity import SpanData
-from dify_trace_tencent.utils import TencentTraceUtils
+from core.ops.tencent_trace.entities.tencent_trace_entity import SpanData
+from core.ops.tencent_trace.utils import TencentTraceUtils
+from core.rag.models.document import Document
 from graphon.entities import WorkflowNodeExecution
 from graphon.enums import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 

api/core/ops/tencent_trace/tencent_trace.py

@@ -1,12 +1,14 @@
-"""Tencent APM tracing with idempotent client cleanup."""
+"""
+Tencent APM tracing implementation with separated concerns
+"""
 
-import inspect
 import logging
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session, sessionmaker
 
 from core.ops.base_trace_instance import BaseTraceInstance
+from core.ops.entities.config_entity import TencentConfig
 from core.ops.entities.trace_entity import (
     BaseTraceInfo,
     DatasetRetrievalTraceInfo,
@@ -17,12 +19,11 @@ from core.ops.entities.trace_entity import (
     ToolTraceInfo,
     WorkflowTraceInfo,
 )
+from core.ops.tencent_trace.client import TencentTraceClient
+from core.ops.tencent_trace.entities.tencent_trace_entity import SpanData
+from core.ops.tencent_trace.span_builder import TencentSpanBuilder
+from core.ops.tencent_trace.utils import TencentTraceUtils
 from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
-from dify_trace_tencent.client import TencentTraceClient
-from dify_trace_tencent.config import TencentConfig
-from dify_trace_tencent.entities.tencent_trace_entity import SpanData
-from dify_trace_tencent.span_builder import TencentSpanBuilder
-from dify_trace_tencent.utils import TencentTraceUtils
 from extensions.ext_database import db
 from graphon.entities.workflow_node_execution import (
     WorkflowNodeExecution,
@@ -37,18 +38,10 @@ class TencentDataTrace(BaseTraceInstance):
     """
     Tencent APM trace implementation with single responsibility principle.
     Acts as a coordinator that delegates specific tasks to specialized classes.
-
-    The instance owns a long-lived ``TencentTraceClient``. Cleanup may happen
-    explicitly in tests or implicitly during garbage collection, so shutdown
-    must be safe to call multiple times.
     """
 
-    trace_client: TencentTraceClient
-    _closed: bool
-
     def __init__(self, tencent_config: TencentConfig):
         super().__init__(tencent_config)
-        self._closed = False
         self.trace_client = TencentTraceClient(
             service_name=tencent_config.service_name,
             endpoint=tencent_config.endpoint,
@@ -520,25 +513,10 @@ class TencentDataTrace(BaseTraceInstance):
         except Exception:
             logger.debug("[Tencent APM] Failed to record message trace duration")
 
-    def close(self) -> None:
-        """Synchronously and idempotently shutdown the underlying trace client."""
-        if getattr(self, "_closed", False):
-            return
-
-        self._closed = True
-        trace_client = getattr(self, "trace_client", None)
-        if trace_client is None:
-            return
-
+    def __del__(self):
+        """Ensure proper cleanup on garbage collection."""
         try:
-            shutdown_result = trace_client.shutdown()
-            if inspect.isawaitable(shutdown_result):
-                close_awaitable = getattr(shutdown_result, "close", None)
-                if callable(close_awaitable):
-                    close_awaitable()
+            if hasattr(self, "trace_client"):
+                self.trace_client.shutdown()
         except Exception:
             logger.exception("[Tencent APM] Failed to shutdown trace client during cleanup")
-
-    def __del__(self):
-        """Ensure best-effort cleanup on garbage collection without retrying shutdown."""
-        self.close()

api/core/ops/weave_trace/weave_trace.py

@@ -17,6 +17,7 @@ from weave.trace_server.trace_server_interface import (
 )
 
 from core.ops.base_trace_instance import BaseTraceInstance
+from core.ops.entities.config_entity import WeaveConfig
 from core.ops.entities.trace_entity import (
     BaseTraceInfo,
     DatasetRetrievalTraceInfo,
@@ -28,9 +29,8 @@ from core.ops.entities.trace_entity import (
     TraceTaskName,
     WorkflowTraceInfo,
 )
+from core.ops.weave_trace.entities.weave_trace_entity import WeaveTraceModel
 from core.repositories import DifyCoreRepositoryFactory
-from dify_trace_weave.config import WeaveConfig
-from dify_trace_weave.entities.weave_trace_entity import WeaveTraceModel
 from extensions.ext_database import db
 from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey
 from models import EndUser, MessageFile, WorkflowNodeExecutionTriggeredFrom

api/core/plugin/impl/model_runtime.py

@@ -66,15 +66,15 @@ class PluginModelRuntime(ModelRuntime):
             if not provider_schema.icon_small:
                 raise ValueError(f"Provider {provider} does not have small icon.")
             file_name = (
-                provider_schema.icon_small.zh_hans if lang.lower() == "zh_hans" else provider_schema.icon_small.en_us
+                provider_schema.icon_small.zh_Hans if lang.lower() == "zh_hans" else provider_schema.icon_small.en_US
             )
         elif icon_type.lower() == "icon_small_dark":
             if not provider_schema.icon_small_dark:
                 raise ValueError(f"Provider {provider} does not have small dark icon.")
             file_name = (
-                provider_schema.icon_small_dark.zh_hans
+                provider_schema.icon_small_dark.zh_Hans
                 if lang.lower() == "zh_hans"
-                else provider_schema.icon_small_dark.en_us
+                else provider_schema.icon_small_dark.en_US
             )
         else:
             raise ValueError(f"Unsupported icon type: {icon_type}.")

api/core/prompt/agent_history_prompt_transform.py

@@ -10,7 +10,7 @@ from graphon.model_runtime.entities.message_entities import (
     SystemPromptMessage,
     UserPromptMessage,
 )
-from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
+from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 
 
 class AgentHistoryPromptTransform(PromptTransform):

api/core/rag/embedding/cached_embedding.py

@@ -14,7 +14,7 @@ from core.rag.embedding.embedding_base import Embeddings
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from graphon.model_runtime.entities.model_entities import ModelPropertyKey
-from graphon.model_runtime.model_providers.base.text_embedding_model import TextEmbeddingModel
+from graphon.model_runtime.model_providers.__base.text_embedding_model import TextEmbeddingModel
 from libs import helper
 from models.dataset import Embedding
 

api/core/rag/extractor/word_extractor.py

@@ -3,7 +3,6 @@
 Supports local file paths and remote URLs (downloaded via `core.helper.ssrf_proxy`).
 """
 
-import inspect
 import logging
 import mimetypes
 import os
@@ -37,11 +36,8 @@ class WordExtractor(BaseExtractor):
         file_path: Path to the file to load.
     """
 
-    _closed: bool
-
     def __init__(self, file_path: str, tenant_id: str, user_id: str):
         """Initialize with file path."""
-        self._closed = False
         self.file_path = file_path
         self.tenant_id = tenant_id
         self.user_id = user_id
@@ -69,27 +65,9 @@ class WordExtractor(BaseExtractor):
         elif not os.path.isfile(self.file_path):
             raise ValueError(f"File path {self.file_path} is not a valid file or url")
 
-    def close(self) -> None:
-        """Best-effort cleanup for downloaded temporary files."""
-        if getattr(self, "_closed", False):
-            return
-
-        self._closed = True
-        temp_file = getattr(self, "temp_file", None)
-        if temp_file is None:
-            return
-
-        try:
-            close_result = temp_file.close()
-            if inspect.isawaitable(close_result):
-                close_awaitable = getattr(close_result, "close", None)
-                if callable(close_awaitable):
-                    close_awaitable()
-        except Exception:
-            logger.debug("Failed to cleanup downloaded word temp file", exc_info=True)
-
     def __del__(self):
-        self.close()
+        if hasattr(self, "temp_file"):
+            self.temp_file.close()
 
     def extract(self) -> list[Document]:
         """Load given path as single page."""

api/core/rag/index_processor/processor/paragraph_index_processor.py

@@ -609,11 +609,11 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             try:
                 # Create File object directly (similar to DatasetRetrieval)
                 file_obj = File(
-                    file_id=upload_file.id,
+                    id=upload_file.id,
                     filename=upload_file.name,
                     extension="." + upload_file.extension,
                     mime_type=upload_file.mime_type,
-                    file_type=FileType.IMAGE,
+                    type=FileType.IMAGE,
                     transfer_method=FileTransferMethod.LOCAL_FILE,
                     remote_url=upload_file.source_url,
                     reference=build_file_reference(

api/core/rag/retrieval/dataset_retrieval.py

@@ -68,7 +68,7 @@ from graphon.file import File, FileTransferMethod, FileType
 from graphon.model_runtime.entities.llm_entities import LLMMode, LLMResult, LLMUsage
 from graphon.model_runtime.entities.message_entities import PromptMessage, PromptMessageRole, PromptMessageTool
 from graphon.model_runtime.entities.model_entities import ModelFeature, ModelType
-from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
+from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from libs.helper import parse_uuid_str_or_none
 from libs.json_in_md_parser import parse_and_check_json_markdown
 from models import UploadFile
@@ -517,11 +517,11 @@ class DatasetRetrieval:
                         if attachments_with_bindings:
                             for _, upload_file in attachments_with_bindings:
                                 attachment_info = File(
-                                    file_id=upload_file.id,
+                                    id=upload_file.id,
                                     filename=upload_file.name,
                                     extension="." + upload_file.extension,
                                     mime_type=upload_file.mime_type,
-                                    file_type=FileType.IMAGE,
+                                    type=FileType.IMAGE,
                                     transfer_method=FileTransferMethod.LOCAL_FILE,
                                     remote_url=upload_file.source_url,
                                     reference=build_file_reference(

api/core/rag/splitter/fixed_text_splitter.py

@@ -9,7 +9,7 @@ from typing import Any, Literal
 
 from core.model_manager import ModelInstance
 from core.rag.splitter.text_splitter import RecursiveCharacterTextSplitter
-from graphon.model_runtime.model_providers.base.tokenizers.gpt2_tokenizer import GPT2Tokenizer
+from graphon.model_runtime.model_providers.__base.tokenizers.gpt2_tokenizer import GPT2Tokenizer
 
 
 class EnhanceRecursiveCharacterTextSplitter(RecursiveCharacterTextSplitter):

api/core/repositories/human_input_repository.py

@@ -8,7 +8,7 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session, selectinload
 
 from core.db.session_factory import session_factory
-from core.workflow.human_input_adapter import (
+from core.workflow.human_input_compat import (
     BoundRecipient,
     DeliveryChannelConfig,
     EmailDeliveryMethod,

api/core/tools/tool_file_manager.py

@@ -28,7 +28,7 @@ class ToolFileManager:
     def _build_graph_file_reference(tool_file: ToolFile) -> File:
         extension = guess_extension(tool_file.mimetype) or ".bin"
         return File(
-            file_type=get_file_type_by_mime_type(tool_file.mimetype),
+            type=get_file_type_by_mime_type(tool_file.mimetype),
             transfer_method=FileTransferMethod.TOOL_FILE,
             remote_url=tool_file.original_url,
             reference=build_file_reference(record_id=str(tool_file.id)),

api/core/tools/tool_manager.py

@@ -1082,12 +1082,7 @@ class ToolManager:
                         continue
                     tool_input = ToolNodeData.ToolInput.model_validate(tool_configurations.get(parameter.name, {}))
                     if tool_input.type == "variable":
-                        variable_selector = tool_input.value
-                        if not isinstance(variable_selector, list) or not all(
-                            isinstance(selector_part, str) for selector_part in variable_selector
-                        ):
-                            raise ToolParameterError("Variable tool input must be a variable selector")
-                        variable = variable_pool.get(variable_selector)
+                        variable = variable_pool.get(tool_input.value)
                         if variable is None:
                             raise ToolParameterError(f"Variable {tool_input.value} does not exist")
                         parameter_value = variable.value

api/core/tools/utils/model_invocation_utils.py

@@ -21,7 +21,7 @@ from graphon.model_runtime.errors.invoke import (
     InvokeRateLimitError,
     InvokeServerUnavailableError,
 )
-from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
+from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from graphon.model_runtime.utils.encoders import jsonable_encoder
 from models.tools import ToolModelInvoke
 

api/core/tools/workflow_as_tool/tool.py

@@ -357,10 +357,7 @@ class WorkflowTool(Tool):
 
     def _update_file_mapping(self, file_dict: dict[str, Any]) -> dict[str, Any]:
         file_id = resolve_file_record_id(file_dict.get("reference") or file_dict.get("related_id"))
-        transfer_method_value = file_dict.get("transfer_method")
-        if not isinstance(transfer_method_value, str):
-            raise ValueError("Workflow file mapping is missing a valid transfer_method")
-        transfer_method = FileTransferMethod.value_of(transfer_method_value)
+        transfer_method = FileTransferMethod.value_of(file_dict.get("transfer_method"))
         match transfer_method:
             case FileTransferMethod.TOOL_FILE:
                 file_dict["tool_file_id"] = file_id

api/core/workflow/human_input_compat.py

@@ -1,8 +1,8 @@
-"""Workflow-to-Graphon adapters for persisted node payloads.
+"""Workflow-layer adapters for legacy human-input payload keys.
 
-Stored workflow graphs and editor payloads still contain a small set of
-Dify-owned field spellings and value shapes. Adapt them here before handing the
-payload to Graphon so Graphon-owned models only see current contracts.
+Stored workflow graphs and editor payloads may still use Dify-specific human
+input recipient keys. Normalize them here before handing configs to
+`graphon` so graph-owned models only see graph-neutral field names.
 """
 
 from __future__ import annotations
@@ -185,7 +185,7 @@ def _copy_mapping(value: object) -> dict[str, Any] | None:
     return None
 
 
-def adapt_human_input_node_data_for_graph(node_data: Mapping[str, Any] | BaseModel) -> dict[str, Any]:
+def normalize_human_input_node_data_for_graph(node_data: Mapping[str, Any] | BaseModel) -> dict[str, Any]:
     normalized = _copy_mapping(node_data)
     if normalized is None:
         raise TypeError(f"human-input node data must be a mapping, got {type(node_data).__name__}")
@@ -215,7 +215,7 @@ def adapt_human_input_node_data_for_graph(node_data: Mapping[str, Any] | BaseMod
 
 
 def parse_human_input_delivery_methods(node_data: Mapping[str, Any] | BaseModel) -> list[DeliveryChannelConfig]:
-    normalized = adapt_human_input_node_data_for_graph(node_data)
+    normalized = normalize_human_input_node_data_for_graph(node_data)
     raw_delivery_methods = normalized.get("delivery_methods")
     if not isinstance(raw_delivery_methods, list):
         return []
@@ -229,20 +229,17 @@ def is_human_input_webapp_enabled(node_data: Mapping[str, Any] | BaseModel) -> b
     return False
 
 
-def adapt_node_data_for_graph(node_data: Mapping[str, Any] | BaseModel) -> dict[str, Any]:
+def normalize_node_data_for_graph(node_data: Mapping[str, Any] | BaseModel) -> dict[str, Any]:
     normalized = _copy_mapping(node_data)
     if normalized is None:
         raise TypeError(f"node data must be a mapping, got {type(node_data).__name__}")
 
-    node_type = normalized.get("type")
-    if node_type == BuiltinNodeTypes.HUMAN_INPUT:
-        return adapt_human_input_node_data_for_graph(normalized)
-    if node_type == BuiltinNodeTypes.TOOL:
-        return _adapt_tool_node_data_for_graph(normalized)
-    return normalized
+    if normalized.get("type") != BuiltinNodeTypes.HUMAN_INPUT:
+        return normalized
+    return normalize_human_input_node_data_for_graph(normalized)
 
 
-def adapt_node_config_for_graph(node_config: Mapping[str, Any] | BaseModel) -> dict[str, Any]:
+def normalize_node_config_for_graph(node_config: Mapping[str, Any] | BaseModel) -> dict[str, Any]:
     normalized = _copy_mapping(node_config)
     if normalized is None:
         raise TypeError(f"node config must be a mapping, got {type(node_config).__name__}")
@@ -251,65 +248,10 @@ def adapt_node_config_for_graph(node_config: Mapping[str, Any] | BaseModel) -> d
     if data_mapping is None:
         return normalized
 
-    normalized["data"] = adapt_node_data_for_graph(data_mapping)
+    normalized["data"] = normalize_node_data_for_graph(data_mapping)
     return normalized
 
 
-def _adapt_tool_node_data_for_graph(node_data: Mapping[str, Any]) -> dict[str, Any]:
-    normalized = dict(node_data)
-
-    raw_tool_configurations = normalized.get("tool_configurations")
-    if not isinstance(raw_tool_configurations, Mapping):
-        return normalized
-
-    existing_tool_parameters = normalized.get("tool_parameters")
-    normalized_tool_parameters = dict(existing_tool_parameters) if isinstance(existing_tool_parameters, Mapping) else {}
-    normalized_tool_configurations: dict[str, Any] = {}
-    found_legacy_tool_inputs = False
-
-    for name, value in raw_tool_configurations.items():
-        if not isinstance(value, Mapping):
-            normalized_tool_configurations[name] = value
-            continue
-
-        input_type = value.get("type")
-        input_value = value.get("value")
-        if input_type not in {"mixed", "variable", "constant"}:
-            normalized_tool_configurations[name] = value
-            continue
-
-        found_legacy_tool_inputs = True
-        normalized_tool_parameters.setdefault(name, dict(value))
-
-        flattened_value = _flatten_legacy_tool_configuration_value(
-            input_type=input_type,
-            input_value=input_value,
-        )
-        if flattened_value is not None:
-            normalized_tool_configurations[name] = flattened_value
-
-    if not found_legacy_tool_inputs:
-        return normalized
-
-    normalized["tool_parameters"] = normalized_tool_parameters
-    normalized["tool_configurations"] = normalized_tool_configurations
-    return normalized
-
-
-def _flatten_legacy_tool_configuration_value(*, input_type: Any, input_value: Any) -> str | int | float | bool | None:
-    if input_type in {"mixed", "constant"} and isinstance(input_value, str | int | float | bool):
-        return input_value
-
-    if (
-        input_type == "variable"
-        and isinstance(input_value, list)
-        and all(isinstance(item, str) for item in input_value)
-    ):
-        return "{{#" + ".".join(input_value) + "#}}"
-
-    return None
-
-
 def _normalize_email_recipients(recipients: Mapping[str, Any]) -> dict[str, Any]:
     normalized = dict(recipients)
 
@@ -349,9 +291,9 @@ __all__ = [
     "MemberRecipient",
     "WebAppDeliveryMethod",
     "_WebAppDeliveryConfig",
-    "adapt_human_input_node_data_for_graph",
-    "adapt_node_config_for_graph",
-    "adapt_node_data_for_graph",
     "is_human_input_webapp_enabled",
+    "normalize_human_input_node_data_for_graph",
+    "normalize_node_config_for_graph",
+    "normalize_node_data_for_graph",
     "parse_human_input_delivery_methods",
 ]

api/core/workflow/node_factory.py

@@ -15,12 +15,12 @@ from core.helper.code_executor.code_executor import (
     CodeExecutionError,
     CodeExecutor,
 )
-from core.helper.ssrf_proxy import graphon_ssrf_proxy
+from core.helper.ssrf_proxy import ssrf_proxy
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance
 from core.prompt.entities.advanced_prompt_entities import MemoryConfig
 from core.trigger.constants import TRIGGER_NODE_TYPES
-from core.workflow.human_input_adapter import adapt_node_config_for_graph
+from core.workflow.human_input_compat import normalize_node_config_for_graph
 from core.workflow.node_runtime import (
     DifyFileReferenceFactory,
     DifyHumanInputNodeRuntime,
@@ -46,7 +46,7 @@ from graphon.enums import BuiltinNodeTypes, NodeType
 from graphon.file.file_manager import file_manager
 from graphon.graph.graph import NodeFactory
 from graphon.model_runtime.memory import PromptMessageMemory
-from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
+from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from graphon.nodes.base.node import Node
 from graphon.nodes.code.code_node import WorkflowCodeExecutor
 from graphon.nodes.code.entities import CodeLanguage
@@ -121,7 +121,6 @@ def get_node_type_classes_mapping() -> Mapping[NodeType, Mapping[str, type[Node]
 
 
 def resolve_workflow_node_class(*, node_type: NodeType, node_version: str) -> type[Node]:
-    """Resolve the production node class for the requested type/version."""
     node_mapping = get_node_type_classes_mapping().get(node_type)
     if not node_mapping:
         raise ValueError(f"No class mapping found for node type: {node_type}")
@@ -298,7 +297,7 @@ class DifyNodeFactory(NodeFactory):
         )
         self._jinja2_template_renderer = CodeExecutorJinja2TemplateRenderer()
         self._template_transform_max_output_length = dify_config.TEMPLATE_TRANSFORM_MAX_LENGTH
-        self._http_request_http_client = graphon_ssrf_proxy
+        self._http_request_http_client = ssrf_proxy
         self._bound_tool_file_manager_factory = lambda: DifyToolFileManager(
             self._dify_context,
             conversation_id_getter=self._conversation_id,
@@ -365,14 +364,10 @@ class DifyNodeFactory(NodeFactory):
             (including pydantic ValidationError, which subclasses ValueError),
             if node type is unknown, or if no implementation exists for the resolved version
         """
-        typed_node_config = NodeConfigDictAdapter.validate_python(adapt_node_config_for_graph(node_config))
+        typed_node_config = NodeConfigDictAdapter.validate_python(normalize_node_config_for_graph(node_config))
         node_id = typed_node_config["id"]
         node_data = typed_node_config["data"]
         node_class = self._resolve_node_class(node_type=node_data.type, node_version=str(node_data.version))
-        # Graph configs are initially validated against permissive shared node data.
-        # Re-validate using the resolved node class so workflow-local node schemas
-        # stay explicit and constructors receive the concrete typed payload.
-        resolved_node_data = self._validate_resolved_node_data(node_class, node_data)
         node_type = node_data.type
         node_init_kwargs_factories: Mapping[NodeType, Callable[[], dict[str, object]]] = {
             BuiltinNodeTypes.CODE: lambda: {
@@ -396,7 +391,7 @@ class DifyNodeFactory(NodeFactory):
             },
             BuiltinNodeTypes.LLM: lambda: self._build_llm_compatible_node_init_kwargs(
                 node_class=node_class,
-                node_data=resolved_node_data,
+                node_data=node_data,
                 wrap_model_instance=True,
                 include_http_client=True,
                 include_llm_file_saver=True,
@@ -410,7 +405,7 @@ class DifyNodeFactory(NodeFactory):
             },
             BuiltinNodeTypes.QUESTION_CLASSIFIER: lambda: self._build_llm_compatible_node_init_kwargs(
                 node_class=node_class,
-                node_data=resolved_node_data,
+                node_data=node_data,
                 wrap_model_instance=True,
                 include_http_client=True,
                 include_llm_file_saver=True,
@@ -420,7 +415,7 @@ class DifyNodeFactory(NodeFactory):
             ),
             BuiltinNodeTypes.PARAMETER_EXTRACTOR: lambda: self._build_llm_compatible_node_init_kwargs(
                 node_class=node_class,
-                node_data=resolved_node_data,
+                node_data=node_data,
                 wrap_model_instance=True,
                 include_http_client=False,
                 include_llm_file_saver=False,
@@ -441,8 +436,8 @@ class DifyNodeFactory(NodeFactory):
         }
         node_init_kwargs = node_init_kwargs_factories.get(node_type, lambda: {})()
         return node_class(
-            node_id=node_id,
-            config=resolved_node_data,
+            id=node_id,
+            config=typed_node_config,
             graph_init_params=self.graph_init_params,
             graph_runtime_state=self.graph_runtime_state,
             **node_init_kwargs,
@@ -453,10 +448,7 @@ class DifyNodeFactory(NodeFactory):
         """
         Re-validate the permissive graph payload with the concrete NodeData model declared by the resolved node class.
         """
-        validate_node_data = getattr(node_class, "validate_node_data", None)
-        if callable(validate_node_data):
-            return cast("BaseNodeData", validate_node_data(node_data))
-        return node_data
+        return node_class.validate_node_data(node_data)
 
     @staticmethod
     def _resolve_node_class(*, node_type: NodeType, node_version: str) -> type[Node]:

api/core/workflow/node_runtime.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Callable, Generator, Mapping, Sequence
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any, Literal, cast, overload
+from typing import TYPE_CHECKING, Any, cast
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -41,7 +41,7 @@ from graphon.model_runtime.entities.llm_entities import (
 )
 from graphon.model_runtime.entities.message_entities import PromptMessage, PromptMessageTool
 from graphon.model_runtime.entities.model_entities import AIModelEntity
-from graphon.model_runtime.model_providers.base.large_language_model import LargeLanguageModel
+from graphon.model_runtime.model_providers.__base.large_language_model import LargeLanguageModel
 from graphon.nodes.human_input.entities import HumanInputNodeData
 from graphon.nodes.llm.runtime_protocols import (
     PreparedLLMProtocol,
@@ -64,7 +64,7 @@ from models.dataset import SegmentAttachmentBinding
 from models.model import UploadFile
 from services.tools.builtin_tools_manage_service import BuiltinToolManageService
 
-from .human_input_adapter import (
+from .human_input_compat import (
     BoundRecipient,
     DeliveryChannelConfig,
     DeliveryMethodType,
@@ -173,28 +173,6 @@ class DifyPreparedLLM(PreparedLLMProtocol):
     def get_llm_num_tokens(self, prompt_messages: Sequence[PromptMessage]) -> int:
         return self._model_instance.get_llm_num_tokens(prompt_messages)
 
-    @overload
-    def invoke_llm(
-        self,
-        *,
-        prompt_messages: Sequence[PromptMessage],
-        model_parameters: Mapping[str, Any],
-        tools: Sequence[PromptMessageTool] | None,
-        stop: Sequence[str] | None,
-        stream: Literal[False],
-    ) -> LLMResult: ...
-
-    @overload
-    def invoke_llm(
-        self,
-        *,
-        prompt_messages: Sequence[PromptMessage],
-        model_parameters: Mapping[str, Any],
-        tools: Sequence[PromptMessageTool] | None,
-        stop: Sequence[str] | None,
-        stream: Literal[True],
-    ) -> Generator[LLMResultChunk, None, None]: ...
-
     def invoke_llm(
         self,
         *,
@@ -212,28 +190,6 @@ class DifyPreparedLLM(PreparedLLMProtocol):
             stream=stream,
         )
 
-    @overload
-    def invoke_llm_with_structured_output(
-        self,
-        *,
-        prompt_messages: Sequence[PromptMessage],
-        json_schema: Mapping[str, Any],
-        model_parameters: Mapping[str, Any],
-        stop: Sequence[str] | None,
-        stream: Literal[False],
-    ) -> LLMResultWithStructuredOutput: ...
-
-    @overload
-    def invoke_llm_with_structured_output(
-        self,
-        *,
-        prompt_messages: Sequence[PromptMessage],
-        json_schema: Mapping[str, Any],
-        model_parameters: Mapping[str, Any],
-        stop: Sequence[str] | None,
-        stream: Literal[True],
-    ) -> Generator[LLMResultChunkWithStructuredOutput, None, None]: ...
-
     def invoke_llm_with_structured_output(
         self,
         *,

api/core/workflow/nodes/agent/agent_node.py

@@ -5,6 +5,7 @@ from typing import TYPE_CHECKING, Any
 
 from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
 from core.workflow.system_variables import SystemVariableKey, get_system_text
+from graphon.entities.graph_config import NodeConfigDict
 from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionStatus
 from graphon.node_events import NodeEventBase, NodeRunResult, StreamCompletedEvent
 from graphon.nodes.base.node import Node
@@ -34,18 +35,18 @@ class AgentNode(Node[AgentNodeData]):
 
     def __init__(
         self,
-        node_id: str,
-        config: AgentNodeData,
-        *,
+        id: str,
+        config: NodeConfigDict,
         graph_init_params: GraphInitParams,
         graph_runtime_state: GraphRuntimeState,
+        *,
         strategy_resolver: AgentStrategyResolver,
         presentation_provider: AgentStrategyPresentationProvider,
         runtime_support: AgentRuntimeSupport,
         message_transformer: AgentMessageTransformer,
     ) -> None:
         super().__init__(
-            node_id=node_id,
+            id=id,
             config=config,
             graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,

api/core/workflow/nodes/datasource/datasource_node.py

@@ -7,6 +7,7 @@ from core.datasource.entities.datasource_entities import DatasourceProviderType
 from core.plugin.impl.exc import PluginDaemonClientSideError
 from core.workflow.file_reference import resolve_file_record_id
 from core.workflow.system_variables import SystemVariableKey, get_system_segment
+from graphon.entities.graph_config import NodeConfigDict
 from graphon.enums import (
     BuiltinNodeTypes,
     NodeExecutionType,
@@ -35,14 +36,13 @@ class DatasourceNode(Node[DatasourceNodeData]):
 
     def __init__(
         self,
-        node_id: str,
-        config: DatasourceNodeData,
-        *,
+        id: str,
+        config: NodeConfigDict,
         graph_init_params: "GraphInitParams",
         graph_runtime_state: "GraphRuntimeState",
-    ) -> None:
+    ):
         super().__init__(
-            node_id=node_id,
+            id=id,
             config=config,
             graph_init_params=graph_init_params,
             graph_runtime_state=graph_runtime_state,