[autofix.ci] apply automated fixes

(cherry picked from commit 9231ec7c54)

api/configs/enterprise/__init__.py

@@ -16,7 +16,7 @@ class EnterpriseFeatureConfig(BaseSettings):
 
     CAN_REPLACE_LOGO: bool = Field(
         description="Allow customization of the enterprise logo.",
-        default=False,
+        default=True,
     )
 
     ENTERPRISE_REQUEST_TIMEOUT: int = Field(

api/controllers/console/app/conversation.py

@@ -137,7 +137,7 @@ class CompletionConversationApi(Resource):
                 .join(  # type: ignore
                     MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
                 )
-                .distinct()
+                .group_by(Conversation.id)
             )
         elif args.annotation_status == "not_annotated":
             query = (
@@ -275,7 +275,7 @@ class ChatConversationApi(Resource):
                     .join(  # type: ignore
                         MessageAnnotation, MessageAnnotation.conversation_id == Conversation.id
                     )
-                    .distinct()
+                    .group_by(Conversation.id)
                 )
             case "not_annotated":
                 query = (

api/controllers/console/datasets/datasets_segments.py

@@ -3,7 +3,7 @@ import uuid
 from flask import request
 from flask_restx import Resource, marshal
 from pydantic import BaseModel, Field
-from sqlalchemy import String, cast, func, or_, select
+from sqlalchemy import String, case, cast, func, literal, or_, select
 from sqlalchemy.dialects.postgresql import JSONB
 from werkzeug.exceptions import Forbidden, NotFound
 
@@ -159,9 +159,17 @@ class DatasetDocumentSegmentListApi(Resource):
             # Use database-specific methods for JSON array search
             if dify_config.SQLALCHEMY_DATABASE_URI_SCHEME == "postgresql":
                 # PostgreSQL: Use jsonb_array_elements_text to properly handle Unicode/Chinese text
+                # Feed the set-returning function a JSON array in every row. Filtering in
+                # the subquery is not enough because PostgreSQL can still evaluate the
+                # SRF on scalar JSON before applying the predicate.
+                keywords_jsonb = cast(DocumentSegment.keywords, JSONB)
+                keywords_array = case(
+                    (func.jsonb_typeof(keywords_jsonb) == "array", keywords_jsonb),
+                    else_=cast(literal("[]"), JSONB),
+                )
                 keywords_condition = func.array_to_string(
                     func.array(
-                        select(func.jsonb_array_elements_text(cast(DocumentSegment.keywords, JSONB)))
+                        select(func.jsonb_array_elements_text(keywords_array))
                         .correlate(DocumentSegment)
                         .scalar_subquery()
                     ),

api/core/app/layers/trigger_post_layer.py

@@ -7,7 +7,13 @@ from pydantic import TypeAdapter
 from core.db.session_factory import session_factory
 from core.workflow.system_variables import SystemVariableKey, get_system_text
 from graphon.graph_engine.layers import GraphEngineLayer
-from graphon.graph_events import GraphEngineEvent, GraphRunFailedEvent, GraphRunPausedEvent, GraphRunSucceededEvent
+from graphon.graph_events import (
+    GraphEngineEvent,
+    GraphRunAbortedEvent,
+    GraphRunFailedEvent,
+    GraphRunPausedEvent,
+    GraphRunSucceededEvent,
+)
 from models.enums import WorkflowTriggerStatus
 from repositories.sqlalchemy_workflow_trigger_log_repository import SQLAlchemyWorkflowTriggerLogRepository
 from tasks.workflow_cfs_scheduler.cfs_scheduler import AsyncWorkflowCFSPlanEntity
@@ -23,6 +29,7 @@ class TriggerPostLayer(GraphEngineLayer):
     _STATUS_MAP: ClassVar[dict[type[GraphEngineEvent], WorkflowTriggerStatus]] = {
         GraphRunSucceededEvent: WorkflowTriggerStatus.SUCCEEDED,
         GraphRunFailedEvent: WorkflowTriggerStatus.FAILED,
+        GraphRunAbortedEvent: WorkflowTriggerStatus.FAILED,
         GraphRunPausedEvent: WorkflowTriggerStatus.PAUSED,
     }
 
@@ -71,6 +78,8 @@ class TriggerPostLayer(GraphEngineLayer):
                 trigger_log.status = self._STATUS_MAP[type(event)]
                 trigger_log.workflow_run_id = workflow_run_id
                 trigger_log.outputs = TypeAdapter(dict[str, Any]).dump_json(outputs).decode()
+                if isinstance(event, GraphRunAbortedEvent):
+                    trigger_log.error = event.reason or "Workflow execution aborted"
 
                 if trigger_log.elapsed_time is None:
                     trigger_log.elapsed_time = elapsed_time

api/core/hosting_configuration.py

@@ -240,7 +240,8 @@ class HostingConfiguration:
         if len(quotas) > 0:
             credentials = {
                 "dashscope_api_key": dify_config.HOSTED_TONGYI_API_KEY,
-                "use_international_endpoint": dify_config.HOSTED_TONGYI_USE_INTERNATIONAL_ENDPOINT,
+                # SNP-494: keep temporary compatibility with tongyi plugin string credential checks.
+                "use_international_endpoint": str(dify_config.HOSTED_TONGYI_USE_INTERNATIONAL_ENDPOINT).lower(),
             }
 
             return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)

api/core/plugin/impl/base.py

@@ -3,6 +3,7 @@ import json
 import logging
 from collections.abc import Callable, Generator
 from typing import Any, cast
+from urllib.parse import unquote
 
 import httpx
 from pydantic import BaseModel
@@ -53,6 +54,9 @@ else:
 
 logger = logging.getLogger(__name__)
 
+PLUGIN_DAEMON_MAX_PATH_LENGTH = 4096
+PLUGIN_DAEMON_MAX_PATH_DECODE_DEPTH = 8
+
 _httpx_client: httpx.Client = get_pooled_http_client(
     "plugin_daemon",
     lambda: httpx.Client(limits=httpx.Limits(max_keepalive_connections=50, max_connections=100), trust_env=False),
@@ -103,6 +107,20 @@ class BasePluginClient:
         params: dict[str, Any] | None,
         files: dict[str, Any] | None,
     ) -> tuple[str, dict[str, str], bytes | dict[str, Any] | str | None, dict[str, Any] | None, dict[str, Any] | None]:
+        if len(path) > PLUGIN_DAEMON_MAX_PATH_LENGTH:
+            raise ValueError(f"Invalid plugin daemon path: path length exceeds {PLUGIN_DAEMON_MAX_PATH_LENGTH}")
+
+        decoded_path = path
+        for _ in range(PLUGIN_DAEMON_MAX_PATH_DECODE_DEPTH):
+            next_decoded_path = unquote(decoded_path)
+            if next_decoded_path == decoded_path:
+                break
+            decoded_path = next_decoded_path
+        else:
+            raise ValueError("Invalid plugin daemon path: path is too deeply encoded")
+
+        if any(seg == ".." for seg in decoded_path.split("/")):
+            raise ValueError(f"Invalid plugin daemon path: traversal sequence detected in {path!r}")
         url = plugin_daemon_inner_api_baseurl / path
         prepared_headers = dict(headers or {})
         prepared_headers["X-Api-Key"] = dify_config.PLUGIN_DAEMON_KEY

api/services/account_service.py

@@ -110,6 +110,8 @@ class TokenPair(BaseModel):
 REFRESH_TOKEN_PREFIX = "refresh_token:"
 ACCOUNT_REFRESH_TOKEN_PREFIX = "account_refresh_token:"
 REFRESH_TOKEN_EXPIRY = timedelta(days=dify_config.REFRESH_TOKEN_EXPIRE_DAYS)
+ACCOUNT_LAST_ACTIVE_REFRESH_PREFIX = "account_last_active_refresh:"
+ACCOUNT_LAST_ACTIVE_REFRESH_INTERVAL = timedelta(minutes=10)
 
 
 class AccountService:
@@ -146,6 +148,40 @@ class AccountService:
     def _get_account_refresh_token_key(account_id: str) -> str:
         return f"{ACCOUNT_REFRESH_TOKEN_PREFIX}{account_id}"
 
+    @staticmethod
+    def _get_account_last_active_refresh_key(account_id: str) -> str:
+        return f"{ACCOUNT_LAST_ACTIVE_REFRESH_PREFIX}{account_id}"
+
+    @staticmethod
+    @redis_fallback(default_return=True)
+    def _should_refresh_account_last_active(account_id: str) -> bool:
+        return bool(
+            redis_client.set(
+                AccountService._get_account_last_active_refresh_key(account_id),
+                1,
+                ex=int(ACCOUNT_LAST_ACTIVE_REFRESH_INTERVAL.total_seconds()),
+                nx=True,
+            )
+        )
+
+    @staticmethod
+    def _refresh_account_last_active(account: Account) -> None:
+        now = naive_utc_now()
+        refresh_before = now - ACCOUNT_LAST_ACTIVE_REFRESH_INTERVAL
+
+        if account.last_active_at >= refresh_before:
+            return
+
+        if not AccountService._should_refresh_account_last_active(account.id):
+            return
+
+        db.session.execute(
+            update(Account)
+            .where(Account.id == account.id, Account.last_active_at < refresh_before)
+            .values(last_active_at=now, updated_at=func.current_timestamp())
+        )
+        db.session.commit()
+
     @staticmethod
     def _store_refresh_token(refresh_token: str, account_id: str):
         redis_client.setex(AccountService._get_refresh_token_key(refresh_token), REFRESH_TOKEN_EXPIRY, account_id)
@@ -188,9 +224,7 @@ class AccountService:
             available_ta.current = True
             db.session.commit()
 
-        if naive_utc_now() - account.last_active_at > timedelta(minutes=10):
-            account.last_active_at = naive_utc_now()
-            db.session.commit()
+        AccountService._refresh_account_last_active(account)
         # NOTE: make sure account is accessible outside of a db session
         # This ensures that it will work correctly after upgrading to Flask version 3.1.2
         db.session.refresh(account)

api/services/billing_service.py

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

api/services/credit_pool_service.py

@@ -1,4 +1,14 @@
+"""Tenant credit pool accounting.
+
+Credit deductions are guarded by a tenant-level Redis lock before the database
+row lock is acquired. This keeps concurrent usage accounting for one tenant
+from piling up database transactions while preserving cross-tenant concurrency.
+"""
+
 import logging
+from collections.abc import Callable
+from dataclasses import dataclass
+from uuid import uuid4
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
@@ -7,13 +17,70 @@ from configs import dify_config
 from core.db.session_factory import session_factory
 from core.errors.error import QuotaExceededError
 from extensions.ext_database import db
+from extensions.ext_redis import redis_client
 from models import TenantCreditPool
 from models.enums import ProviderQuotaType
 
 logger = logging.getLogger(__name__)
 
+FEATURE_KEY_CREDIT_POOL = "credit_pool"
+CREDIT_POOL_TENANT_LOCK_TIMEOUT_SECONDS = 10
+CREDIT_POOL_TENANT_LOCK_BLOCKING_TIMEOUT_SECONDS = 5
+
+
+@dataclass(frozen=True)
+class CreditPoolBalance:
+    tenant_id: str
+    pool_type: str
+    quota_limit: int
+    quota_used: int
+
+    @property
+    def remaining_credits(self) -> int:
+        if self.quota_limit == -1:
+            return -1
+        return max(0, self.quota_limit - self.quota_used)
+
+    def has_sufficient_credits(self, required_credits: int) -> bool:
+        return self.quota_limit == -1 or self.remaining_credits >= required_credits
+
 
 class CreditPoolService:
+    @staticmethod
+    def _normalize_pool_type(pool_type: str | ProviderQuotaType) -> str:
+        return pool_type.value if isinstance(pool_type, ProviderQuotaType) else str(pool_type)
+
+    @staticmethod
+    def _use_billing_quota() -> bool:
+        return bool(dify_config.BILLING_ENABLED)
+
+    @staticmethod
+    def _get_tenant_lock_key(tenant_id: str) -> str:
+        return f"credit_pool:tenant:{tenant_id}:deduct_lock"
+
+    @classmethod
+    def _deduct_with_tenant_lock(cls, tenant_id: str, deduct: Callable[[], int]) -> int:
+        lock_key = cls._get_tenant_lock_key(tenant_id)
+        lock = redis_client.lock(
+            lock_key,
+            timeout=CREDIT_POOL_TENANT_LOCK_TIMEOUT_SECONDS,
+            blocking_timeout=CREDIT_POOL_TENANT_LOCK_BLOCKING_TIMEOUT_SECONDS,
+        )
+        lock_acquired = False
+
+        try:
+            lock_acquired = lock.acquire(blocking=True)
+            if not lock_acquired:
+                raise QuotaExceededError("Failed to acquire credit pool lock")
+
+            return deduct()
+        finally:
+            if lock_acquired:
+                try:
+                    lock.release()
+                except Exception:
+                    logger.warning("Failed to release credit pool lock, tenant_id=%s", tenant_id, exc_info=True)
+
     @staticmethod
     def _get_locked_pool(session: Session, tenant_id: str, pool_type: str) -> TenantCreditPool | None:
         return session.scalar(
@@ -40,14 +107,32 @@ class CreditPoolService:
         return credit_pool
 
     @classmethod
-    def get_pool(cls, tenant_id: str, pool_type: str = "trial") -> TenantCreditPool | None:
+    def get_pool(
+        cls, tenant_id: str, pool_type: str | ProviderQuotaType = "trial"
+    ) -> TenantCreditPool | CreditPoolBalance | None:
         """get tenant credit pool"""
+        normalized_pool_type = cls._normalize_pool_type(pool_type)
+        if cls._use_billing_quota():
+            from services.billing_service import BillingService
+
+            balance = BillingService.quota_get_balance(
+                tenant_id=tenant_id,
+                feature_key=FEATURE_KEY_CREDIT_POOL,
+                bucket=normalized_pool_type,
+            )
+            return CreditPoolBalance(
+                tenant_id=tenant_id,
+                pool_type=normalized_pool_type,
+                quota_limit=balance["quota"],
+                quota_used=balance["usage"],
+            )
+
         with session_factory.get_session_maker().begin() as session:
             return session.scalar(
                 select(TenantCreditPool)
                 .where(
                     TenantCreditPool.tenant_id == tenant_id,
-                    TenantCreditPool.pool_type == pool_type,
+                    TenantCreditPool.pool_type == normalized_pool_type,
                 )
                 .limit(1)
             )
@@ -63,7 +148,7 @@ class CreditPoolService:
         pool = cls.get_pool(tenant_id, pool_type)
         if not pool:
             return False
-        return pool.remaining_credits >= credits_required
+        return pool.has_sufficient_credits(credits_required)
 
     @classmethod
     def check_and_deduct_credits(
@@ -75,10 +160,54 @@ class CreditPoolService:
         """Deduct exactly the requested credits or raise without mutating the pool."""
         if credits_required <= 0:
             return 0
+        normalized_pool_type = cls._normalize_pool_type(pool_type)
 
-        try:
+        if cls._use_billing_quota():
+            from services.billing_service import BillingService
+
+            request_id = str(uuid4())
+            result = BillingService.quota_reserve(
+                tenant_id=tenant_id,
+                feature_key=FEATURE_KEY_CREDIT_POOL,
+                bucket=normalized_pool_type,
+                request_id=request_id,
+                amount=credits_required,
+                meta={"source": "credit_pool.check_and_deduct"},
+            )
+            reservation_id = result.get("reservation_id", "")
+            if not reservation_id:
+                raise QuotaExceededError("Insufficient credits remaining")
+            try:
+                BillingService.quota_commit(
+                    tenant_id=tenant_id,
+                    feature_key=FEATURE_KEY_CREDIT_POOL,
+                    bucket=normalized_pool_type,
+                    reservation_id=reservation_id,
+                    actual_amount=credits_required,
+                    meta={"source": "credit_pool.check_and_deduct"},
+                )
+            except Exception:
+                try:
+                    BillingService.quota_release(
+                        tenant_id=tenant_id,
+                        feature_key=FEATURE_KEY_CREDIT_POOL,
+                        bucket=normalized_pool_type,
+                        reservation_id=reservation_id,
+                    )
+                except Exception:
+                    logger.warning(
+                        "Failed to release reserved credit pool quota, tenant_id=%s, pool_type=%s, reservation_id=%s",
+                        tenant_id,
+                        normalized_pool_type,
+                        reservation_id,
+                        exc_info=True,
+                    )
+                raise
+            return credits_required
+
+        def deduct() -> int:
             with session_factory.get_session_maker().begin() as session:
-                pool = cls._get_locked_pool(session=session, tenant_id=tenant_id, pool_type=pool_type)
+                pool = cls._get_locked_pool(session=session, tenant_id=tenant_id, pool_type=normalized_pool_type)
                 if not pool:
                     raise QuotaExceededError("Credit pool not found")
 
@@ -89,14 +218,16 @@ class CreditPoolService:
                     raise QuotaExceededError("Insufficient credits remaining")
 
                 pool.quota_used += credits_required
+                return credits_required
+
+        try:
+            return cls._deduct_with_tenant_lock(tenant_id, deduct)
         except QuotaExceededError:
             raise
         except Exception:
             logger.exception("Failed to deduct credits for tenant %s", tenant_id)
             raise QuotaExceededError("Failed to deduct credits")
 
-        return credits_required
-
     @classmethod
     def deduct_credits_capped(
         cls,
@@ -107,12 +238,26 @@ class CreditPoolService:
         """Deduct up to the available balance and return the actual deducted credits."""
         if credits_required <= 0:
             return 0
+        normalized_pool_type = cls._normalize_pool_type(pool_type)
 
-        try:
+        if cls._use_billing_quota():
+            from services.billing_service import BillingService
+
+            result = BillingService.quota_consume_capped(
+                tenant_id=tenant_id,
+                feature_key=FEATURE_KEY_CREDIT_POOL,
+                bucket=normalized_pool_type,
+                request_id=str(uuid4()),
+                amount=credits_required,
+                meta={"source": "credit_pool.deduct_capped"},
+            )
+            return result["deducted"]
+
+        def deduct() -> int:
             with session_factory.get_session_maker().begin() as session:
-                pool = cls._get_locked_pool(session=session, tenant_id=tenant_id, pool_type=pool_type)
+                pool = cls._get_locked_pool(session=session, tenant_id=tenant_id, pool_type=normalized_pool_type)
                 if not pool:
-                    logger.warning("Credit pool not found, tenant_id=%s, pool_type=%s", tenant_id, pool_type)
+                    logger.warning("Credit pool not found, tenant_id=%s, pool_type=%s", tenant_id, normalized_pool_type)
                     return 0
 
                 deducted_credits = min(credits_required, pool.remaining_credits)
@@ -121,6 +266,9 @@ class CreditPoolService:
 
                 pool.quota_used += deducted_credits
                 return deducted_credits
+
+        try:
+            return cls._deduct_with_tenant_lock(tenant_id, deduct)
         except QuotaExceededError:
             raise
         except Exception:

api/services/dataset_service.py

@@ -170,6 +170,16 @@ class _AutomaticProcessRule(BaseModel):
     mode: Literal[ProcessRuleMode.AUTOMATIC]
     summary_index_setting: _SummaryIndexSetting | None = None
 
+    @field_validator("summary_index_setting", mode="before")
+    @classmethod
+    def _normalize_summary_index_setting(cls, v: Any) -> Any:
+        """Treat dicts with enable=None (or missing enable) as None (#36602)."""
+        if v is None:
+            return None
+        if isinstance(v, dict) and v.get("enable") is None:
+            return None
+        return v
+
 
 class _CustomProcessRule(BaseModel):
     model_config = ConfigDict(extra="allow")
@@ -178,6 +188,16 @@ class _CustomProcessRule(BaseModel):
     rules: _EstimateRules
     summary_index_setting: _SummaryIndexSetting | None = None
 
+    @field_validator("summary_index_setting", mode="before")
+    @classmethod
+    def _normalize_summary_index_setting(cls, v: Any) -> Any:
+        """Treat dicts with enable=None (or missing enable) as None (#36602)."""
+        if v is None:
+            return None
+        if isinstance(v, dict) and v.get("enable") is None:
+            return None
+        return v
+
 
 class _HierarchicalProcessRule(BaseModel):
     model_config = ConfigDict(extra="allow")
@@ -186,6 +206,16 @@ class _HierarchicalProcessRule(BaseModel):
     rules: _EstimateRules
     summary_index_setting: _SummaryIndexSetting | None = None
 
+    @field_validator("summary_index_setting", mode="before")
+    @classmethod
+    def _normalize_summary_index_setting(cls, v: Any) -> Any:
+        """Treat dicts with enable=None (or missing enable) as None (#36602)."""
+        if v is None:
+            return None
+        if isinstance(v, dict) and v.get("enable") is None:
+            return None
+        return v
+
 
 _EstimateProcessRule = Annotated[
     _AutomaticProcessRule | _CustomProcessRule | _HierarchicalProcessRule,

api/tests/unit_tests/controllers/console/datasets/test_datasets_segments.py

@@ -1036,6 +1036,48 @@ class TestSegmentListAdvancedCases:
         assert status == 200
         assert response["total"] == 1
 
+    def test_segment_list_postgres_keyword_filter_handles_scalar_keywords(self, app: Flask):
+        api = DatasetDocumentSegmentListApi()
+        method = unwrap(api.get)
+
+        dataset = MagicMock()
+        document = MagicMock()
+        pagination = MagicMock(items=[], total=0, pages=0)
+
+        with (
+            app.test_request_context("/?keyword=test"),
+            patch(
+                "controllers.console.datasets.datasets_segments.current_account_with_tenant",
+                return_value=(MagicMock(), "11111111-1111-1111-1111-111111111111"),
+            ),
+            patch(
+                "controllers.console.datasets.datasets_segments.DatasetService.get_dataset",
+                return_value=dataset,
+            ),
+            patch(
+                "controllers.console.datasets.datasets_segments.DatasetService.check_dataset_permission",
+                return_value=None,
+            ),
+            patch(
+                "controllers.console.datasets.datasets_segments.DocumentService.get_document",
+                return_value=document,
+            ),
+            patch(
+                "controllers.console.datasets.datasets_segments.dify_config",
+                SimpleNamespace(SQLALCHEMY_DATABASE_URI_SCHEME="postgresql"),
+            ),
+            patch(
+                "controllers.console.datasets.datasets_segments.db.paginate",
+                return_value=pagination,
+            ) as paginate_mock,
+        ):
+            method(api, "22222222-2222-2222-2222-222222222222", "33333333-3333-3333-3333-333333333333")
+
+        query = paginate_mock.call_args.kwargs["select"]
+        sql = str(query.compile(compile_kwargs={"literal_binds": True}))
+        assert "jsonb_array_elements_text(CASE" in sql
+        assert "ELSE CAST('[]' AS JSONB)" in sql
+
     def test_segment_list_permission_denied(self, app: Flask):
         """Test segment list with permission denied"""
         api = DatasetDocumentSegmentListApi()

api/tests/unit_tests/core/app/layers/test_trigger_post_layer.py

@@ -4,7 +4,11 @@ from unittest.mock import Mock, patch
 
 from core.app.layers.trigger_post_layer import TriggerPostLayer
 from core.workflow.system_variables import build_system_variables
-from graphon.graph_events import GraphRunFailedEvent, GraphRunSucceededEvent
+from graphon.graph_events import (
+    GraphRunAbortedEvent,
+    GraphRunFailedEvent,
+    GraphRunSucceededEvent,
+)
 from graphon.runtime import VariablePool
 from models.enums import WorkflowTriggerStatus
 
@@ -59,6 +63,57 @@ class TestTriggerPostLayer:
         repo.update.assert_called_once_with(trigger_log)
         session.commit.assert_called_once()
 
+    def test_on_event_updates_trigger_log_for_aborted_event(self):
+        trigger_log = SimpleNamespace(
+            status=None,
+            workflow_run_id=None,
+            outputs=None,
+            error=None,
+            elapsed_time=None,
+            total_tokens=None,
+            finished_at=None,
+        )
+        runtime_state = SimpleNamespace(
+            outputs={"partial": "ok"},
+            variable_pool=VariablePool.from_bootstrap(
+                system_variables=build_system_variables(workflow_execution_id="run-1")
+            ),
+            total_tokens=7,
+        )
+
+        with (
+            patch("core.app.layers.trigger_post_layer.session_factory") as mock_session_factory,
+            patch("core.app.layers.trigger_post_layer.SQLAlchemyWorkflowTriggerLogRepository") as mock_repo_cls,
+            patch("core.app.layers.trigger_post_layer.datetime") as mock_datetime,
+        ):
+            mock_datetime.now.return_value = datetime(2026, 2, 20, tzinfo=UTC)
+
+            session = Mock()
+            mock_session_factory.create_session.return_value.__enter__.return_value = session
+
+            repo = Mock()
+            repo.get_by_id.return_value = trigger_log
+            mock_repo_cls.return_value = repo
+
+            layer = TriggerPostLayer(
+                cfs_plan_scheduler_entity=Mock(),
+                start_time=datetime(2026, 2, 20, tzinfo=UTC) - timedelta(seconds=10),
+                trigger_log_id="log-1",
+            )
+            layer.initialize(runtime_state, Mock())
+
+            layer.on_event(GraphRunAbortedEvent(reason="timeout"))
+
+        assert trigger_log.status == WorkflowTriggerStatus.FAILED
+        assert trigger_log.workflow_run_id == "run-1"
+        assert trigger_log.outputs is not None
+        assert trigger_log.error == "timeout"
+        assert trigger_log.elapsed_time is not None
+        assert trigger_log.total_tokens == 7
+        assert trigger_log.finished_at is not None
+        repo.update.assert_called_once_with(trigger_log)
+        session.commit.assert_called_once()
+
     def test_on_event_handles_missing_trigger_log(self):
         runtime_state = SimpleNamespace(
             outputs={},

api/tests/unit_tests/core/plugin/impl/test_base_client_impl.py

@@ -1,11 +1,12 @@
 import json
+from urllib.parse import quote
 
 import pytest
 from pytest_mock import MockerFixture
 
 from core.plugin.endpoint.exc import EndpointSetupFailedError
 from core.plugin.entities.plugin_daemon import PluginDaemonInnerError
-from core.plugin.impl.base import BasePluginClient
+from core.plugin.impl.base import PLUGIN_DAEMON_MAX_PATH_LENGTH, BasePluginClient
 from core.trigger.errors import (
     EventIgnoreError,
     TriggerInvokeError,
@@ -67,6 +68,36 @@ class TestBasePluginClientImpl:
         assert result == ["hello", "world"]
         assert stream_mock.call_args.kwargs["data"] == {"k": "v"}
 
+    @pytest.mark.parametrize(
+        "path",
+        [
+            "plugin/tenant/%252e%252e%252ftarget",
+            "plugin/tenant/%2e%2e%252ftarget",
+        ],
+    )
+    def test_prepare_request_rejects_encoded_traversal_with_encoded_separator(self, path: str):
+        client = BasePluginClient()
+
+        with pytest.raises(ValueError, match="traversal sequence detected"):
+            client._prepare_request(path, None, None, None, None)
+
+    def test_prepare_request_rejects_path_exceeding_max_length(self):
+        client = BasePluginClient()
+        path = "a" * (PLUGIN_DAEMON_MAX_PATH_LENGTH + 1)
+
+        with pytest.raises(ValueError, match="path length exceeds"):
+            client._prepare_request(path, None, None, None, None)
+
+    def test_prepare_request_rejects_excessively_encoded_path(self):
+        client = BasePluginClient()
+        segment = "..%2Ftarget"
+        for _ in range(9):
+            segment = quote(segment, safe="")
+        path = f"plugin/tenant/{segment}"
+
+        with pytest.raises(ValueError, match="too deeply encoded"):
+            client._prepare_request(path, None, None, None, None)
+
     def test_request_with_plugin_daemon_response_handles_request_exception(self, mocker: MockerFixture):
         client = BasePluginClient()
         mocker.patch.object(client, "_request", side_effect=RuntimeError("boom"))

api/tests/unit_tests/services/test_account_service.py

@@ -436,7 +436,10 @@ class TestAccountService:
         mock_db_dependencies["db"].session.scalar.return_value = mock_tenant_join
 
         # Mock datetime
-        with patch("services.account_service.datetime") as mock_datetime:
+        with (
+            patch("services.account_service.datetime") as mock_datetime,
+            patch.object(AccountService, "_refresh_account_last_active") as mock_refresh_last_active,
+        ):
             mock_now = datetime.now()
             mock_datetime.now.return_value = mock_now
             mock_datetime.UTC = "UTC"
@@ -447,6 +450,7 @@ class TestAccountService:
             # Verify results
             assert result == mock_account
             assert mock_account.set_tenant_id.called
+            mock_refresh_last_active.assert_called_once_with(mock_account)
 
     def test_load_user_not_found(self, mock_db_dependencies):
         """Test user loading when user does not exist."""
@@ -483,7 +487,10 @@ class TestAccountService:
         mock_db_dependencies["db"].session.scalar.side_effect = [None, mock_available_tenant]
 
         # Mock datetime
-        with patch("services.account_service.datetime") as mock_datetime:
+        with (
+            patch("services.account_service.datetime") as mock_datetime,
+            patch.object(AccountService, "_refresh_account_last_active") as mock_refresh_last_active,
+        ):
             mock_now = datetime.now()
             mock_datetime.now.return_value = mock_now
             mock_datetime.UTC = "UTC"
@@ -495,6 +502,7 @@ class TestAccountService:
             assert result == mock_account
             assert mock_available_tenant.current is True
             self._assert_database_operations_called(mock_db_dependencies["db"])
+            mock_refresh_last_active.assert_called_once_with(mock_account)
 
     def test_load_user_no_tenants(self, mock_db_dependencies):
         """Test user loading when user has no tenants at all."""
@@ -517,6 +525,68 @@ class TestAccountService:
             # Verify results
             assert result is None
 
+    def test_refresh_account_last_active_uses_redis_gate_and_conditional_update(self, mock_db_dependencies):
+        """Test last-active refresh is gated in Redis and conditionally written to DB."""
+        mock_account = TestAccountAssociatedDataFactory.create_account_mock()
+        now = datetime(2026, 6, 2, 2, 45, 49)
+        mock_account.last_active_at = now - timedelta(minutes=15)
+
+        with (
+            patch("services.account_service.naive_utc_now", return_value=now),
+            patch("services.account_service.redis_client") as mock_redis_client,
+        ):
+            mock_redis_client.set.return_value = True
+
+            AccountService._refresh_account_last_active(mock_account)
+
+        mock_redis_client.set.assert_called_once_with(
+            "account_last_active_refresh:user-123",
+            1,
+            ex=600,
+            nx=True,
+        )
+        mock_db_dependencies["db"].session.execute.assert_called_once()
+        mock_db_dependencies["db"].session.commit.assert_called_once()
+
+    def test_refresh_account_last_active_skips_db_when_redis_gate_exists(self, mock_db_dependencies):
+        """Test concurrent refresh attempts do not enqueue duplicate DB updates."""
+        mock_account = TestAccountAssociatedDataFactory.create_account_mock()
+        now = datetime(2026, 6, 2, 2, 45, 49)
+        mock_account.last_active_at = now - timedelta(minutes=15)
+
+        with (
+            patch("services.account_service.naive_utc_now", return_value=now),
+            patch("services.account_service.redis_client") as mock_redis_client,
+        ):
+            mock_redis_client.set.return_value = None
+
+            AccountService._refresh_account_last_active(mock_account)
+
+        mock_redis_client.set.assert_called_once_with(
+            "account_last_active_refresh:user-123",
+            1,
+            ex=600,
+            nx=True,
+        )
+        mock_db_dependencies["db"].session.execute.assert_not_called()
+        mock_db_dependencies["db"].session.commit.assert_not_called()
+
+    def test_refresh_account_last_active_skips_recent_account(self, mock_db_dependencies):
+        """Test recent activity does not touch Redis or DB."""
+        mock_account = TestAccountAssociatedDataFactory.create_account_mock()
+        now = datetime(2026, 6, 2, 2, 45, 49)
+        mock_account.last_active_at = now - timedelta(minutes=5)
+
+        with (
+            patch("services.account_service.naive_utc_now", return_value=now),
+            patch("services.account_service.redis_client") as mock_redis_client,
+        ):
+            AccountService._refresh_account_last_active(mock_account)
+
+        mock_redis_client.set.assert_not_called()
+        mock_db_dependencies["db"].session.execute.assert_not_called()
+        mock_db_dependencies["db"].session.commit.assert_not_called()
+
 
 class TestTenantService:
     """

api/tests/unit_tests/services/test_billing_service.py

@@ -72,6 +72,23 @@ class TestBillingServiceSendRequest:
         assert call_args[1]["headers"]["Billing-Api-Secret-Key"] == "test-secret-key"
         assert call_args[1]["headers"]["Content-Type"] == "application/json"
 
+    def test_send_request_with_base_url_override(self, mock_httpx_request, mock_billing_config):
+        """Quota APIs can use the new billing service without changing legacy billing calls."""
+        # Arrange
+        expected_response = {"result": "success"}
+        mock_response = MagicMock()
+        mock_response.status_code = httpx.codes.OK
+        mock_response.json.return_value = expected_response
+        mock_httpx_request.return_value = mock_response
+
+        # Act
+        result = BillingService._send_request("GET", "/quota/balance", base_url="https://quota.example.com")
+
+        # Assert
+        assert result == expected_response
+        call_args = mock_httpx_request.call_args
+        assert call_args[0][1] == "https://quota.example.com/quota/balance"
+
     @pytest.mark.parametrize(
         "status_code", [httpx.codes.NOT_FOUND, httpx.codes.INTERNAL_SERVER_ERROR, httpx.codes.BAD_REQUEST]
     )
@@ -313,6 +330,99 @@ class TestBillingServiceSubscriptionInfo:
         assert result == expected_response
         mock_send_request.assert_called_once_with("GET", "/subscription/info", params={"tenant_id": tenant_id})
 
+    def test_get_info_exclude_vector_space(self, mock_send_request):
+        """When requested, get_info asks billing to skip vector_space."""
+        # Arrange
+        tenant_id = "tenant-123"
+        expected_response = {
+            "enabled": True,
+            "subscription": {"plan": "professional", "interval": "month", "education": False},
+            "members": {"size": 1, "limit": 50},
+            "apps": {"size": 1, "limit": 200},
+            "knowledge_rate_limit": {"limit": 1000},
+            "documents_upload_quota": {"size": 0, "limit": 1000},
+            "annotation_quota_limit": {"size": 0, "limit": 5000},
+            "docs_processing": "top-priority",
+            "can_replace_logo": True,
+            "model_load_balancing_enabled": True,
+            "knowledge_pipeline_publish_enabled": True,
+        }
+        mock_send_request.return_value = expected_response
+
+        # Act
+        result = BillingService.get_info(tenant_id, exclude_vector_space=True)
+
+        # Assert
+        assert "vector_space" not in result
+        mock_send_request.assert_called_once_with(
+            "GET",
+            "/subscription/info",
+            params={"tenant_id": tenant_id, "exclude_vector_space": "true"},
+        )
+
+    def test_get_info_exclude_vector_space_normalizes_null_field(self, mock_send_request):
+        """When billing serializes skipped vector_space as null, get_info treats it as absent."""
+        # Arrange
+        tenant_id = "tenant-123"
+        expected_response = {
+            "enabled": True,
+            "subscription": {"plan": "professional", "interval": "month", "education": False},
+            "members": {"size": 1, "limit": 50},
+            "apps": {"size": 1, "limit": 200},
+            "vector_space": None,
+            "knowledge_rate_limit": {"limit": 1000},
+            "documents_upload_quota": {"size": 0, "limit": 1000},
+            "annotation_quota_limit": {"size": 0, "limit": 5000},
+            "docs_processing": "top-priority",
+            "can_replace_logo": True,
+            "model_load_balancing_enabled": True,
+            "knowledge_pipeline_publish_enabled": True,
+        }
+        mock_send_request.return_value = expected_response
+
+        # Act
+        result = BillingService.get_info(tenant_id, exclude_vector_space=True)
+
+        # Assert
+        assert "vector_space" not in result
+        mock_send_request.assert_called_once_with(
+            "GET",
+            "/subscription/info",
+            params={"tenant_id": tenant_id, "exclude_vector_space": "true"},
+        )
+
+    def test_get_vector_space_success(self, mock_send_request):
+        """Test successful retrieval of vector-space usage and limit."""
+        # Arrange
+        tenant_id = "tenant-123"
+        expected_response = {"size": 5120.75, "limit": 20480}
+        mock_send_request.return_value = expected_response
+
+        # Act
+        result = BillingService.get_vector_space(tenant_id)
+
+        # Assert
+        assert result == expected_response
+        mock_send_request.assert_called_once_with(
+            "GET",
+            "/subscription/vector-space",
+            params={"tenant_id": tenant_id},
+        )
+
+    def test_quota_get_balance_uses_quota_request(self):
+        tenant_id = "tenant-123"
+        with patch.object(BillingService, "_send_quota_request") as mock_send_quota_request:
+            mock_send_quota_request.return_value = {"quota": "200", "usage": "6", "available": "194", "reserved": "0"}
+
+            result = BillingService.quota_get_balance(tenant_id, "credit_pool", bucket="trial")
+
+        assert result == {"quota": 200, "usage": 6, "available": 194, "reserved": 0}
+        mock_send_quota_request.assert_called_once_with(
+            "GET",
+            "/quota/balance",
+            params={"tenant_id": tenant_id, "feature_key": "credit_pool", "bucket": "trial"},
+        )
+
     def test_get_knowledge_rate_limit_with_defaults(self, mock_send_request):
         """Test knowledge rate limit retrieval with default values."""
         # Arrange
@@ -438,19 +548,20 @@ class TestBillingServiceUsageCalculation:
         assert result == expected_response
         mock_send_request.assert_called_once_with("GET", "/tenant-feature-usage/info", params={"tenant_id": tenant_id})
 
-    def test_get_quota_info(self, mock_send_request):
+    def test_get_quota_info(self):
         """Test retrieval of quota info from new endpoint."""
         # Arrange
         tenant_id = "tenant-123"
         expected_response = {"trigger_event": {"limit": 100, "usage": 30}, "api_rate_limit": {"limit": -1, "usage": 0}}
-        mock_send_request.return_value = expected_response
+        with patch.object(BillingService, "_send_quota_request") as mock_send_quota_request:
+            mock_send_quota_request.return_value = expected_response
 
-        # Act
-        result = BillingService.get_quota_info(tenant_id)
+            # Act
+            result = BillingService.get_quota_info(tenant_id)
 
         # Assert
         assert result == expected_response
-        mock_send_request.assert_called_once_with("GET", "/quota/info", params={"tenant_id": tenant_id})
+        mock_send_quota_request.assert_called_once_with("GET", "/quota/info", params={"tenant_id": tenant_id})
 
     def test_update_tenant_feature_plan_usage_positive_delta(self, mock_send_request):
         """Test updating tenant feature usage with positive delta (adding credits)."""
@@ -534,7 +645,7 @@ class TestBillingServiceQuotaOperations:
 
     @pytest.fixture
     def mock_send_request(self):
-        with patch.object(BillingService, "_send_request") as mock:
+        with patch.object(BillingService, "_send_quota_request") as mock:
             yield mock
 
     def test_quota_reserve_success(self, mock_send_request):
@@ -572,6 +683,16 @@ class TestBillingServiceQuotaOperations:
         call_json = mock_send_request.call_args[1]["json"]
         assert call_json["meta"] == {"source": "webhook"}
 
+    def test_quota_reserve_with_bucket(self, mock_send_request):
+        mock_send_request.return_value = {"reservation_id": "rid-2", "available": 98, "reserved": 1}
+
+        BillingService.quota_reserve(
+            tenant_id="t1", feature_key="credit_pool", request_id="req-2", amount=1, bucket="trial"
+        )
+
+        call_json = mock_send_request.call_args[1]["json"]
+        assert call_json["bucket"] == "trial"
+
     def test_quota_commit_success(self, mock_send_request):
         expected = {"available": 98, "reserved": 0, "refunded": 0}
         mock_send_request.return_value = expected
@@ -616,6 +737,20 @@ class TestBillingServiceQuotaOperations:
         call_json = mock_send_request.call_args[1]["json"]
         assert call_json["meta"] == {"reason": "partial"}
 
+    def test_quota_commit_with_bucket(self, mock_send_request):
+        mock_send_request.return_value = {"available": 97, "reserved": 0, "refunded": 0}
+
+        BillingService.quota_commit(
+            tenant_id="t1",
+            feature_key="credit_pool",
+            reservation_id="rid-1",
+            actual_amount=1,
+            bucket="paid",
+        )
+
+        call_json = mock_send_request.call_args[1]["json"]
+        assert call_json["bucket"] == "paid"
+
     def test_quota_release_success(self, mock_send_request):
         expected = {"available": 100, "reserved": 0, "released": 1}
         mock_send_request.return_value = expected
@@ -640,6 +775,64 @@ class TestBillingServiceQuotaOperations:
         assert result["released"] == 1
         assert isinstance(result["released"], int)
 
+    def test_quota_release_with_bucket(self, mock_send_request):
+        mock_send_request.return_value = {"available": 100, "reserved": 0, "released": 1}
+
+        BillingService.quota_release(tenant_id="t1", feature_key="credit_pool", reservation_id="rid-1", bucket="trial")
+
+        call_json = mock_send_request.call_args[1]["json"]
+        assert call_json["bucket"] == "trial"
+
+    def test_quota_consume_capped_success(self, mock_send_request):
+        mock_send_request.return_value = {
+            "deducted": "2",
+            "available": "8",
+            "reserved": "0",
+            "quota": "10",
+            "usage": "2",
+        }
+
+        result = BillingService.quota_consume_capped(
+            tenant_id="t1",
+            feature_key="credit_pool",
+            request_id="req-1",
+            amount=5,
+            bucket="paid",
+            meta={"source": "test"},
+        )
+
+        assert result == {"deducted": 2, "available": 8, "reserved": 0, "quota": 10, "usage": 2}
+        mock_send_request.assert_called_once_with(
+            "POST",
+            "/quota/consume-capped",
+            json={
+                "tenant_id": "t1",
+                "feature_key": "credit_pool",
+                "request_id": "req-1",
+                "amount": 5,
+                "bucket": "paid",
+                "meta": {"source": "test"},
+            },
+        )
+
+    def test_send_quota_request_uses_quota_base_url(self):
+        with (
+            patch.object(BillingService, "quota_base_url", "https://quota.example.com/v1"),
+            patch.object(BillingService, "_send_request") as mock_send_request,
+        ):
+            mock_send_request.return_value = {"ok": True}
+
+            result = BillingService._send_quota_request("GET", "/quota/info", params={"tenant_id": "t1"})
+
+        assert result == {"ok": True}
+        mock_send_request.assert_called_once_with(
+            "GET",
+            "/quota/info",
+            json=None,
+            params={"tenant_id": "t1"},
+            base_url="https://quota.example.com/v1",
+        )
+
     def test_get_quota_info_coerces_string_to_int(self, mock_send_request):
         """Test that TypeAdapter coerces string values to int for get_quota_info."""
         mock_send_request.return_value = {

api/tests/unit_tests/services/test_credit_pool_service.py

@@ -1,6 +1,7 @@
 from collections.abc import Generator
 from contextlib import contextmanager
-from unittest.mock import patch
+from types import SimpleNamespace
+from unittest.mock import ANY, MagicMock, patch
 from uuid import uuid4
 
 import pytest
@@ -11,7 +12,13 @@ from sqlalchemy.orm import sessionmaker
 from core.errors.error import QuotaExceededError
 from models import TenantCreditPool
 from models.enums import ProviderQuotaType
-from services.credit_pool_service import CreditPoolService
+from services.credit_pool_service import (
+    CREDIT_POOL_TENANT_LOCK_BLOCKING_TIMEOUT_SECONDS,
+    CREDIT_POOL_TENANT_LOCK_TIMEOUT_SECONDS,
+    FEATURE_KEY_CREDIT_POOL,
+    CreditPoolBalance,
+    CreditPoolService,
+)
 
 
 def _create_engine_with_pool(*, quota_limit: int, quota_used: int) -> tuple[Engine, str, str]:
@@ -45,6 +52,26 @@ def _get_quota_used(*, engine: Engine, pool_id: str) -> int | None:
         return connection.scalar(select(TenantCreditPool.quota_used).where(TenantCreditPool.id == pool_id))
 
 
+def _make_session_maker(session: MagicMock) -> MagicMock:
+    session_maker = MagicMock()
+    transaction = session_maker.begin.return_value
+    transaction.__enter__.return_value = session
+    transaction.__exit__.return_value = None
+    return session_maker
+
+
+def _make_redis_lock() -> MagicMock:
+    lock = MagicMock()
+    lock.acquire.return_value = True
+    return lock
+
+
+@pytest.fixture(autouse=True)
+def _disable_billing_quota_by_default() -> Generator[None, None, None]:
+    with patch("services.credit_pool_service.dify_config.BILLING_ENABLED", False):
+        yield
+
+
 def test_get_pool_uses_configured_session_factory_without_flask_app_context() -> None:
     engine, tenant_id, _ = _create_engine_with_pool(quota_limit=10, quota_used=2)
 
@@ -56,6 +83,13 @@ def test_get_pool_uses_configured_session_factory_without_flask_app_context() ->
     assert pool.quota_used == 2
 
 
+def test_credit_pool_balance_unlimited_remaining_and_sufficiency() -> None:
+    pool = CreditPoolBalance(tenant_id="tenant-1", pool_type="paid", quota_limit=-1, quota_used=999)
+
+    assert pool.remaining_credits == -1
+    assert pool.has_sufficient_credits(10_000)
+
+
 def test_check_and_deduct_credits_deducts_exact_amount_when_sufficient() -> None:
     engine, tenant_id, pool_id = _create_engine_with_pool(quota_limit=10, quota_used=2)
 
@@ -176,3 +210,255 @@ def test_deduct_credits_capped_reraises_quota_exceeded_errors() -> None:
         CreditPoolService.deduct_credits_capped(tenant_id=tenant_id, credits_required=1)
 
     assert _get_quota_used(engine=engine, pool_id=pool_id) == 2
+
+
+def test_check_and_deduct_credits_uses_tenant_redis_lock_before_db_deduction() -> None:
+    tenant_id = "tenant-1"
+    session = MagicMock()
+    session_maker = _make_session_maker(session)
+    pool = SimpleNamespace(remaining_credits=10, quota_used=2)
+    redis_lock = _make_redis_lock()
+
+    with (
+        patch("services.credit_pool_service.redis_client.lock", return_value=redis_lock) as lock,
+        patch("services.credit_pool_service.session_factory.get_session_maker", return_value=session_maker),
+        patch.object(CreditPoolService, "_get_locked_pool", return_value=pool) as get_locked_pool,
+    ):
+        result = CreditPoolService.check_and_deduct_credits(
+            tenant_id=tenant_id,
+            credits_required=3,
+            pool_type=ProviderQuotaType.TRIAL,
+        )
+
+    assert result == 3
+    assert pool.quota_used == 5
+    lock.assert_called_once_with(
+        "credit_pool:tenant:tenant-1:deduct_lock",
+        timeout=CREDIT_POOL_TENANT_LOCK_TIMEOUT_SECONDS,
+        blocking_timeout=CREDIT_POOL_TENANT_LOCK_BLOCKING_TIMEOUT_SECONDS,
+    )
+    redis_lock.acquire.assert_called_once_with(blocking=True)
+    redis_lock.release.assert_called_once_with()
+    get_locked_pool.assert_called_once_with(session=session, tenant_id=tenant_id, pool_type="trial")
+
+
+def test_deduct_credits_capped_uses_tenant_redis_lock_before_db_deduction() -> None:
+    tenant_id = "tenant-1"
+    session = MagicMock()
+    session_maker = _make_session_maker(session)
+    pool = SimpleNamespace(remaining_credits=2, quota_used=8)
+    redis_lock = _make_redis_lock()
+
+    with (
+        patch("services.credit_pool_service.redis_client.lock", return_value=redis_lock) as lock,
+        patch("services.credit_pool_service.session_factory.get_session_maker", return_value=session_maker),
+        patch.object(CreditPoolService, "_get_locked_pool", return_value=pool) as get_locked_pool,
+    ):
+        result = CreditPoolService.deduct_credits_capped(
+            tenant_id=tenant_id,
+            credits_required=5,
+            pool_type=ProviderQuotaType.PAID,
+        )
+
+    assert result == 2
+    assert pool.quota_used == 10
+    lock.assert_called_once_with(
+        "credit_pool:tenant:tenant-1:deduct_lock",
+        timeout=CREDIT_POOL_TENANT_LOCK_TIMEOUT_SECONDS,
+        blocking_timeout=CREDIT_POOL_TENANT_LOCK_BLOCKING_TIMEOUT_SECONDS,
+    )
+    redis_lock.acquire.assert_called_once_with(blocking=True)
+    redis_lock.release.assert_called_once_with()
+    get_locked_pool.assert_called_once_with(session=session, tenant_id=tenant_id, pool_type="paid")
+
+
+def test_get_pool_uses_billing_quota_balance_when_enabled() -> None:
+    tenant_id = "tenant-1"
+    with (
+        patch("services.credit_pool_service.dify_config.BILLING_ENABLED", True),
+        patch("services.billing_service.BillingService.quota_get_balance") as quota_get_balance,
+    ):
+        quota_get_balance.return_value = {"quota": 1000, "usage": 250, "available": 750, "reserved": 0}
+
+        pool = CreditPoolService.get_pool(tenant_id=tenant_id, pool_type=ProviderQuotaType.PAID)
+
+    assert isinstance(pool, CreditPoolBalance)
+    assert pool.quota_limit == 1000
+    assert pool.quota_used == 250
+    assert pool.remaining_credits == 750
+    quota_get_balance.assert_called_once_with(
+        tenant_id=tenant_id,
+        feature_key=FEATURE_KEY_CREDIT_POOL,
+        bucket="paid",
+    )
+
+
+def test_check_and_deduct_credits_uses_billing_reserve_and_commit_when_enabled() -> None:
+    tenant_id = "tenant-1"
+    with (
+        patch("services.credit_pool_service.dify_config.BILLING_ENABLED", True),
+        patch("services.billing_service.BillingService.quota_reserve") as quota_reserve,
+        patch("services.billing_service.BillingService.quota_commit") as quota_commit,
+        patch("services.billing_service.BillingService.quota_release") as quota_release,
+    ):
+        quota_reserve.return_value = {"reservation_id": "reservation-1", "available": 7, "reserved": 3}
+
+        result = CreditPoolService.check_and_deduct_credits(
+            tenant_id=tenant_id,
+            credits_required=3,
+            pool_type=ProviderQuotaType.TRIAL,
+        )
+
+    assert result == 3
+    quota_reserve.assert_called_once_with(
+        tenant_id=tenant_id,
+        feature_key=FEATURE_KEY_CREDIT_POOL,
+        bucket="trial",
+        request_id=ANY,
+        amount=3,
+        meta={"source": "credit_pool.check_and_deduct"},
+    )
+    quota_commit.assert_called_once_with(
+        tenant_id=tenant_id,
+        feature_key=FEATURE_KEY_CREDIT_POOL,
+        bucket="trial",
+        reservation_id="reservation-1",
+        actual_amount=3,
+        meta={"source": "credit_pool.check_and_deduct"},
+    )
+    quota_release.assert_not_called()
+
+
+def test_check_and_deduct_credits_raises_when_billing_reserve_is_insufficient() -> None:
+    with (
+        patch("services.credit_pool_service.dify_config.BILLING_ENABLED", True),
+        patch("services.billing_service.BillingService.quota_reserve") as quota_reserve,
+    ):
+        quota_reserve.return_value = {"reservation_id": "", "available": 1, "reserved": 0}
+
+        with pytest.raises(QuotaExceededError, match="Insufficient credits remaining"):
+            CreditPoolService.check_and_deduct_credits(tenant_id="tenant-1", credits_required=3)
+
+
+def test_check_and_deduct_credits_releases_billing_reservation_when_commit_fails() -> None:
+    with (
+        patch("services.credit_pool_service.dify_config.BILLING_ENABLED", True),
+        patch("services.billing_service.BillingService.quota_reserve") as quota_reserve,
+        patch("services.billing_service.BillingService.quota_commit", side_effect=RuntimeError("commit failed")),
+        patch("services.billing_service.BillingService.quota_release") as quota_release,
+    ):
+        quota_reserve.return_value = {"reservation_id": "reservation-1", "available": 7, "reserved": 3}
+
+        with pytest.raises(RuntimeError, match="commit failed"):
+            CreditPoolService.check_and_deduct_credits(tenant_id="tenant-1", credits_required=3)
+
+    quota_release.assert_called_once_with(
+        tenant_id="tenant-1",
+        feature_key=FEATURE_KEY_CREDIT_POOL,
+        bucket="trial",
+        reservation_id="reservation-1",
+    )
+
+
+def test_check_and_deduct_credits_logs_when_billing_release_fails() -> None:
+    with (
+        patch("services.credit_pool_service.dify_config.BILLING_ENABLED", True),
+        patch("services.billing_service.BillingService.quota_reserve") as quota_reserve,
+        patch("services.billing_service.BillingService.quota_commit", side_effect=RuntimeError("commit failed")),
+        patch(
+            "services.billing_service.BillingService.quota_release", side_effect=RuntimeError("release failed")
+        ) as quota_release,
+        patch("services.credit_pool_service.logger.warning") as logger_warning,
+    ):
+        quota_reserve.return_value = {"reservation_id": "reservation-1", "available": 7, "reserved": 3}
+
+        with pytest.raises(RuntimeError, match="commit failed"):
+            CreditPoolService.check_and_deduct_credits(tenant_id="tenant-1", credits_required=3)
+
+    quota_release.assert_called_once_with(
+        tenant_id="tenant-1",
+        feature_key=FEATURE_KEY_CREDIT_POOL,
+        bucket="trial",
+        reservation_id="reservation-1",
+    )
+    logger_warning.assert_called_once()
+    assert logger_warning.call_args.args[3] == "reservation-1"
+    assert logger_warning.call_args.kwargs["exc_info"] is True
+
+
+def test_deduct_credits_capped_uses_billing_consume_capped_when_enabled() -> None:
+    tenant_id = "tenant-1"
+    with (
+        patch("services.credit_pool_service.dify_config.BILLING_ENABLED", True),
+        patch("services.billing_service.BillingService.quota_consume_capped") as quota_consume_capped,
+    ):
+        quota_consume_capped.return_value = {
+            "deducted": 2,
+            "available": 0,
+            "reserved": 0,
+            "quota": 10,
+            "usage": 10,
+        }
+
+        result = CreditPoolService.deduct_credits_capped(
+            tenant_id=tenant_id,
+            credits_required=5,
+            pool_type=ProviderQuotaType.PAID,
+        )
+
+    assert result == 2
+    quota_consume_capped.assert_called_once_with(
+        tenant_id=tenant_id,
+        feature_key=FEATURE_KEY_CREDIT_POOL,
+        bucket="paid",
+        request_id=ANY,
+        amount=5,
+        meta={"source": "credit_pool.deduct_capped"},
+    )
+
+
+@pytest.mark.parametrize(
+    "deduct_method",
+    [
+        CreditPoolService.check_and_deduct_credits,
+        CreditPoolService.deduct_credits_capped,
+    ],
+)
+def test_non_positive_credit_request_skips_tenant_redis_lock(deduct_method) -> None:
+    with patch("services.credit_pool_service.redis_client.lock") as lock:
+        result = deduct_method(tenant_id="tenant-1", credits_required=0)
+
+    assert result == 0
+    lock.assert_not_called()
+
+
+def test_check_and_deduct_credits_wraps_redis_lock_errors_without_querying_db() -> None:
+    session_maker = MagicMock()
+
+    with (
+        patch("services.credit_pool_service.redis_client.lock", side_effect=RuntimeError("redis unavailable")),
+        patch("services.credit_pool_service.session_factory.get_session_maker", return_value=session_maker),
+        pytest.raises(QuotaExceededError, match="Failed to deduct credits"),
+    ):
+        CreditPoolService.check_and_deduct_credits(tenant_id="tenant-1", credits_required=1)
+
+    session_maker.begin.assert_not_called()
+
+
+def test_deduct_credits_capped_ignores_release_errors_after_successful_deduction() -> None:
+    session = MagicMock()
+    session_maker = _make_session_maker(session)
+    pool = SimpleNamespace(remaining_credits=3, quota_used=7)
+    redis_lock = _make_redis_lock()
+    redis_lock.release.side_effect = RuntimeError("release failed")
+
+    with (
+        patch("services.credit_pool_service.redis_client.lock", return_value=redis_lock),
+        patch("services.credit_pool_service.session_factory.get_session_maker", return_value=session_maker),
+        patch.object(CreditPoolService, "_get_locked_pool", return_value=pool),
+    ):
+        result = CreditPoolService.deduct_credits_capped(tenant_id="tenant-1", credits_required=2)
+
+    assert result == 2
+    assert pool.quota_used == 9
+    redis_lock.release.assert_called_once_with()

web/app/(humanInputLayout)/form/[token]/__tests__/form.spec.tsx

@@ -0,0 +1,402 @@
+import type { FormData } from '../form'
+import type { HumanInputFieldValue } from '@/app/components/base/chat/chat/answer/human-input-content/field-renderer'
+import { act, render, renderHook, screen } from '@testing-library/react'
+import userEvent from '@testing-library/user-event'
+import { UserActionButtonType } from '@/app/components/workflow/nodes/human-input/types'
+import { InputVarType, SupportUploadFileTypes } from '@/app/components/workflow/types'
+import { TransferMethod } from '@/types/app'
+import FormContent from '../form'
+import { useFormSubmit } from '../use-form-submit'
+
+const mockSubmitForm = vi.hoisted(() => vi.fn())
+const mockUseGetHumanInputForm = vi.hoisted(() => vi.fn())
+const mockContentItemState = vi.hoisted(() => ({
+  staleAttachmentInputChange: undefined as ((name: string, value: unknown) => void) | undefined,
+  uploadedFile: {
+    id: 'file-1',
+    name: 'review.pdf',
+    size: 128,
+    type: 'document',
+    progress: 100,
+    transferMethod: 'local_file',
+    supportFileType: 'document',
+    uploadedId: 'upload-file-1',
+  },
+  uploadingFile: {
+    id: 'file-1',
+    name: 'review.pdf',
+    size: 128,
+    type: 'document',
+    progress: 50,
+    transferMethod: 'local_file',
+    supportFileType: 'document',
+    uploadedId: undefined,
+  },
+}))
+
+vi.mock('@/next/navigation', () => ({
+  useParams: () => ({ token: 'token-123' }),
+}))
+
+vi.mock('@/service/use-share', () => ({
+  useGetHumanInputForm: (...args: unknown[]) => mockUseGetHumanInputForm(...args),
+  useSubmitHumanInputForm: () => ({
+    mutate: mockSubmitForm,
+    isPending: false,
+  }),
+}))
+
+vi.mock('@/hooks/use-document-title', () => ({
+  __esModule: true,
+  default: vi.fn(),
+}))
+
+vi.mock('@/app/components/base/chat/chat/answer/human-input-content/content-item', () => ({
+  __esModule: true,
+  default: ({ content, onInputChange }: { content: string, onInputChange: (name: string, value: unknown) => void }) => {
+    const isSummaryField = content.includes('summary')
+    const isAttachmentField = content.includes('attachments')
+
+    if (isAttachmentField && !mockContentItemState.staleAttachmentInputChange)
+      mockContentItemState.staleAttachmentInputChange = onInputChange
+
+    return (
+      <div data-testid="share-form-content-item">
+        {content}
+        {isSummaryField && (
+          <>
+            <button type="button" onClick={() => onInputChange('summary', '')}>
+              share-clear-summary
+            </button>
+            <button type="button" onClick={() => onInputChange('summary', 'updated summary')}>
+              share-update-summary
+            </button>
+          </>
+        )}
+        {isAttachmentField && (
+          <>
+            <button
+              type="button"
+              onClick={() => mockContentItemState.staleAttachmentInputChange?.('attachments', [mockContentItemState.uploadingFile])}
+            >
+              share-uploading-attachments
+            </button>
+            <button
+              type="button"
+              onClick={() => mockContentItemState.staleAttachmentInputChange?.('attachments', [mockContentItemState.uploadedFile])}
+            >
+              share-update-attachments
+            </button>
+          </>
+        )}
+      </div>
+    )
+  },
+}))
+
+vi.mock('@/app/components/base/chat/chat/answer/human-input-content/expiration-time', () => ({
+  __esModule: true,
+  default: () => <div>expiration-time</div>,
+}))
+
+vi.mock('@/app/components/base/loading', () => ({
+  __esModule: true,
+  default: () => <div>loading</div>,
+}))
+
+vi.mock('@/app/components/base/logo/dify-logo', () => ({
+  __esModule: true,
+  default: () => <div>dify-logo</div>,
+}))
+
+vi.mock('@/app/components/base/app-icon', () => ({
+  __esModule: true,
+  default: () => <div>app-icon</div>,
+}))
+
+describe('Human input share form', () => {
+  const formData: FormData = {
+    site: {
+      site: {
+        title: 'Review App',
+        icon_type: 'emoji',
+        icon: 'R',
+        icon_background: '#fff',
+        icon_url: '',
+        default_language: 'en-US',
+        description: '',
+        copyright: '',
+        privacy_policy: '',
+        custom_disclaimer: '',
+        prompt_public: false,
+        use_icon_as_answer_icon: false,
+      },
+    },
+    form_content: '{{#$output.summary#}} {{#$output.attachments#}}',
+    inputs: [
+      {
+        type: InputVarType.paragraph,
+        output_variable_name: 'summary',
+        default: {
+          type: 'constant',
+          value: 'initial summary',
+          selector: [],
+        },
+      },
+      {
+        type: InputVarType.multiFiles,
+        output_variable_name: 'attachments',
+        allowed_file_extensions: ['.pdf'],
+        allowed_file_types: [SupportUploadFileTypes.document],
+        allowed_file_upload_methods: [TransferMethod.local_file],
+        number_limits: 3,
+      },
+    ],
+    resolved_default_values: {},
+    user_actions: [
+      {
+        id: 'approve',
+        title: 'Approve',
+        button_style: UserActionButtonType.Primary,
+      },
+    ],
+    expiration_time: 60,
+  }
+
+  beforeEach(() => {
+    vi.clearAllMocks()
+    mockContentItemState.staleAttachmentInputChange = undefined
+    mockUseGetHumanInputForm.mockReturnValue({
+      data: formData,
+      isLoading: false,
+      error: null,
+    })
+  })
+
+  it('should render the loading state while the form is being fetched', () => {
+    mockUseGetHumanInputForm.mockReturnValue({
+      data: undefined,
+      isLoading: true,
+      error: null,
+    })
+
+    render(<FormContent />)
+
+    expect(screen.getByText('loading')).toBeInTheDocument()
+  })
+
+  it('should render status cards for terminal fetch states', () => {
+    const cases = [
+      {
+        error: { code: 'human_input_form_expired' },
+        title: 'share.humanInput.sorry',
+        subtitle: 'share.humanInput.expired',
+        submissionID: true,
+      },
+      {
+        error: { code: 'human_input_form_submitted' },
+        title: 'share.humanInput.sorry',
+        subtitle: 'share.humanInput.completed',
+        submissionID: true,
+      },
+      {
+        error: { code: 'web_form_rate_limit_exceeded' },
+        title: 'share.humanInput.rateLimitExceeded',
+        subtitle: undefined,
+        submissionID: false,
+      },
+      {
+        error: null,
+        title: 'share.humanInput.formNotFound',
+        subtitle: undefined,
+        submissionID: false,
+      },
+    ]
+
+    cases.forEach(({ error, title, subtitle, submissionID }) => {
+      mockUseGetHumanInputForm.mockReturnValue({
+        data: undefined,
+        isLoading: false,
+        error,
+      })
+      const { unmount } = render(<FormContent />)
+
+      expect(screen.getByText(title)).toBeInTheDocument()
+      if (subtitle)
+        expect(screen.getByText(subtitle)).toBeInTheDocument()
+      else
+        expect(screen.queryByText('share.humanInput.expired')).not.toBeInTheDocument()
+
+      if (submissionID)
+        expect(screen.getByText('share.humanInput.submissionID:{"id":"token-123"}')).toBeInTheDocument()
+      else
+        expect(screen.queryByText(/share\.humanInput\.submissionID/)).not.toBeInTheDocument()
+
+      expect(screen.getByText('share.chat.poweredBy')).toBeInTheDocument()
+      expect(screen.getByText('dify-logo')).toBeInTheDocument()
+      unmount()
+    })
+  })
+
+  it('submits typed human input values through the share form mutation', async () => {
+    const user = userEvent.setup()
+
+    render(<FormContent />)
+
+    await user.click(screen.getByRole('button', { name: 'share-update-summary' }))
+    await user.click(screen.getByRole('button', { name: 'share-update-attachments' }))
+    await user.click(screen.getByRole('button', { name: 'Approve' }))
+
+    expect(mockSubmitForm).toHaveBeenCalledWith({
+      token: 'token-123',
+      data: {
+        action: 'approve',
+        inputs: {
+          summary: 'updated summary',
+          attachments: [{
+            type: 'document',
+            transfer_method: TransferMethod.local_file,
+            url: '',
+            upload_file_id: 'upload-file-1',
+          }],
+        },
+      },
+    }, expect.objectContaining({
+      onSuccess: expect.any(Function),
+    }))
+  })
+
+  it('should show the success status after the submit mutation succeeds', async () => {
+    const user = userEvent.setup()
+
+    render(<FormContent />)
+
+    await user.click(screen.getByRole('button', { name: 'share-update-summary' }))
+    await user.click(screen.getByRole('button', { name: 'share-update-attachments' }))
+    await user.click(screen.getByRole('button', { name: 'Approve' }))
+
+    const options = mockSubmitForm.mock.calls[0]![1] as { onSuccess: () => void }
+    act(() => {
+      options.onSuccess()
+    })
+
+    expect(screen.getByText('share.humanInput.thanks')).toBeInTheDocument()
+    expect(screen.getByText('share.humanInput.recorded')).toBeInTheDocument()
+    expect(screen.getByText('share.humanInput.submissionID:{"id":"token-123"}')).toBeInTheDocument()
+  })
+
+  it('should submit empty inputs when there are no form values to process', () => {
+    const { result } = renderHook(() => useFormSubmit('token-empty'))
+
+    act(() => {
+      result.current.submit(
+        undefined as unknown as Record<string, HumanInputFieldValue>,
+        'reject',
+        [],
+      )
+    })
+
+    expect(mockSubmitForm).toHaveBeenCalledWith({
+      token: 'token-empty',
+      data: {
+        action: 'reject',
+        inputs: {},
+      },
+    }, expect.objectContaining({
+      onSuccess: expect.any(Function),
+    }))
+  })
+
+  it('should keep initialized defaults when file upload uses the initial change callback', async () => {
+    const user = userEvent.setup()
+
+    render(<FormContent />)
+
+    await user.click(screen.getByRole('button', { name: 'share-update-attachments' }))
+    await user.click(screen.getByRole('button', { name: 'Approve' }))
+
+    expect(mockSubmitForm).toHaveBeenCalledWith({
+      token: 'token-123',
+      data: {
+        action: 'approve',
+        inputs: {
+          summary: 'initial summary',
+          attachments: [{
+            type: 'document',
+            transfer_method: TransferMethod.local_file,
+            url: '',
+            upload_file_id: 'upload-file-1',
+          }],
+        },
+      },
+    }, expect.objectContaining({
+      onSuccess: expect.any(Function),
+    }))
+  })
+
+  it('should disable action buttons until every required field is filled and files are uploaded', async () => {
+    const user = userEvent.setup()
+
+    render(<FormContent />)
+
+    const approveButton = screen.getByRole('button', { name: 'Approve' })
+    expect(approveButton).toBeDisabled()
+
+    await user.click(screen.getByRole('button', { name: 'share-uploading-attachments' }))
+    expect(approveButton).toBeDisabled()
+
+    await user.click(screen.getByRole('button', { name: 'share-update-attachments' }))
+    expect(approveButton).toBeEnabled()
+
+    await user.click(screen.getByRole('button', { name: 'share-clear-summary' }))
+    expect(approveButton).toBeDisabled()
+
+    await user.click(screen.getByRole('button', { name: 'share-update-summary' }))
+    expect(approveButton).toBeEnabled()
+  })
+
+  it('should hide branding when remove_webapp_brand is enabled', () => {
+    mockUseGetHumanInputForm.mockReturnValue({
+      data: {
+        ...formData,
+        site: {
+          ...formData.site,
+          custom_config: {
+            remove_webapp_brand: true,
+            replace_webapp_logo: null,
+          },
+        },
+      },
+      isLoading: false,
+      error: null,
+    })
+
+    render(<FormContent />)
+
+    expect(screen.queryByText('share.chat.poweredBy')).not.toBeInTheDocument()
+    expect(screen.queryByText('dify-logo')).not.toBeInTheDocument()
+  })
+
+  it('should render the custom branding logo when replace_webapp_logo is provided', () => {
+    mockUseGetHumanInputForm.mockReturnValue({
+      data: {
+        ...formData,
+        site: {
+          ...formData.site,
+          custom_config: {
+            remove_webapp_brand: false,
+            replace_webapp_logo: 'https://example.com/custom-logo.png',
+          },
+        },
+      },
+      isLoading: false,
+      error: null,
+    })
+
+    render(<FormContent />)
+
+    expect(screen.getByText('share.chat.poweredBy')).toBeInTheDocument()
+    expect(screen.getByRole('img', { name: 'logo' })).toHaveAttribute('src', 'https://example.com/custom-logo.png')
+    expect(screen.queryByText('dify-logo')).not.toBeInTheDocument()
+  })
+})

web/app/(humanInputLayout)/form/[token]/branding-footer.tsx

@@ -0,0 +1,30 @@
+import { useTranslation } from 'react-i18next'
+import DifyLogo from '@/app/components/base/logo/dify-logo'
+
+type BrandingFooterProps = {
+  removeWebappBrand?: boolean
+  replaceWebappLogo?: string | null
+}
+
+const BrandingFooter = ({
+  removeWebappBrand,
+  replaceWebappLogo,
+}: BrandingFooterProps) => {
+  const { t } = useTranslation()
+
+  if (removeWebappBrand)
+    return null
+
+  return (
+    <div className="flex flex-row-reverse px-2 py-3">
+      <div className="flex shrink-0 items-center gap-1.5 px-1">
+        <div className="system-2xs-medium-uppercase text-text-tertiary">{t('chat.poweredBy', { ns: 'share' })}</div>
+        {replaceWebappLogo
+          ? <img src={replaceWebappLogo} alt="logo" className="block h-5 w-auto" />
+          : <DifyLogo size="small" />}
+      </div>
+    </div>
+  )
+}
+
+export default BrandingFooter

web/app/(humanInputLayout)/form/[token]/form-status-card.tsx

@@ -0,0 +1,47 @@
+import type { ReactNode } from 'react'
+import { cn } from '@langgenius/dify-ui/cn'
+import BrandingFooter from './branding-footer'
+
+type FormStatusCardProps = {
+  iconClassName: string
+  title: ReactNode
+  subtitle?: ReactNode
+  submissionID?: string
+  removeWebappBrand?: boolean
+  replaceWebappLogo?: string | null
+}
+
+const FormStatusCard = ({
+  iconClassName,
+  title,
+  subtitle,
+  submissionID,
+  removeWebappBrand,
+  replaceWebappLogo,
+}: FormStatusCardProps) => {
+  return (
+    <div className={cn('flex size-full flex-col items-center justify-center')}>
+      <div className="max-w-160 min-w-120">
+        <div className="flex h-80 flex-col gap-4 rounded-[20px] border border-divider-subtle bg-chat-bubble-bg p-10 pb-9 shadow-lg backdrop-blur-xs">
+          <div className="flex h-14 w-14 shrink-0 items-center justify-center rounded-2xl border border-components-panel-border-subtle bg-background-default-dodge p-3">
+            <span className={cn('size-8', iconClassName)} />
+          </div>
+          <div className="grow">
+            <div className="title-4xl-semi-bold text-text-primary">{title}</div>
+            {!!subtitle && (
+              <div className="title-4xl-semi-bold text-text-primary">{subtitle}</div>
+            )}
+          </div>
+          {submissionID && (
+            <div className="shrink-0 system-2xs-regular-uppercase text-text-tertiary">
+              {submissionID}
+            </div>
+          )}
+        </div>
+        <BrandingFooter removeWebappBrand={removeWebappBrand} replaceWebappLogo={replaceWebappLogo} />
+      </div>
+    </div>
+  )
+}
+
+export default FormStatusCard

web/app/(humanInputLayout)/form/[token]/form.tsx

@@ -1,34 +1,26 @@
 'use client'
-import type { ButtonProps } from '@langgenius/dify-ui/button'
 import type { FormInputItem, UserAction } from '@/app/components/workflow/nodes/human-input/types'
-import type { SiteInfo } from '@/models/share'
+import type { CustomConfigValueType, SiteInfo } from '@/models/share'
 import type { HumanInputFormError } from '@/service/use-share'
-import { Button } from '@langgenius/dify-ui/button'
-import { cn } from '@langgenius/dify-ui/cn'
-import {
-  RiCheckboxCircleFill,
-  RiErrorWarningFill,
-  RiInformation2Fill,
-} from '@remixicon/react'
-import { produce } from 'immer'
+import type { HumanInputResolvedValue } from '@/types/workflow'
 import * as React from 'react'
-import { useEffect, useMemo, useState } from 'react'
 import { useTranslation } from 'react-i18next'
-import AppIcon from '@/app/components/base/app-icon'
-import ContentItem from '@/app/components/base/chat/chat/answer/human-input-content/content-item'
-import ExpirationTime from '@/app/components/base/chat/chat/answer/human-input-content/expiration-time'
-import { getButtonStyle } from '@/app/components/base/chat/chat/answer/human-input-content/utils'
 import Loading from '@/app/components/base/loading'
-import DifyLogo from '@/app/components/base/logo/dify-logo'
 import useDocumentTitle from '@/hooks/use-document-title'
 import { useParams } from '@/next/navigation'
-import { useGetHumanInputForm, useSubmitHumanInputForm } from '@/service/use-share'
+import { useGetHumanInputForm } from '@/service/use-share'
+import FormStatusCard from './form-status-card'
+import LoadedFormContent from './loaded-form-content'
+import { useFormSubmit } from './use-form-submit'
 
 export type FormData = {
-  site: { site: SiteInfo }
+  site: {
+    site: SiteInfo
+    custom_config?: Record<string, CustomConfigValueType> | null
+  }
   form_content: string
   inputs: FormInputItem[]
-  resolved_default_values: Record<string, string>
+  resolved_default_values: Record<string, HumanInputResolvedValue>
   user_actions: UserAction[]
   expiration_time: number
 }
@@ -39,58 +31,18 @@ const FormContent = () => {
   const { token } = useParams<{ token: string }>()
   useDocumentTitle('')
 
-  const [inputs, setInputs] = useState<Record<string, string>>({})
-  const [success, setSuccess] = useState(false)
-
-  const { mutate: submitForm, isPending: isSubmitting } = useSubmitHumanInputForm()
-
   const { data: formData, isLoading, error } = useGetHumanInputForm(token)
+  const { isSubmitting, submit, success } = useFormSubmit(token)
+
+  const removeWebappBrand = formData?.site?.custom_config?.remove_webapp_brand === true
+  const replaceWebappLogo = typeof formData?.site?.custom_config?.replace_webapp_logo === 'string'
+    ? formData.site.custom_config.replace_webapp_logo
+    : null
 
   const expired = (error as HumanInputFormError | null)?.code === 'human_input_form_expired'
   const submitted = (error as HumanInputFormError | null)?.code === 'human_input_form_submitted'
   const rateLimitExceeded = (error as HumanInputFormError | null)?.code === 'web_form_rate_limit_exceeded'
 
-  const splitByOutputVar = (content: string): string[] => {
-    const outputVarRegex = /(\{\{#\$output\.[^#]+#\}\})/g
-    const parts = content.split(outputVarRegex)
-    return parts.filter(part => part.length > 0)
-  }
-
-  const contentList = useMemo(() => {
-    if (!formData?.form_content)
-      return []
-    return splitByOutputVar(formData.form_content)
-  }, [formData?.form_content])
-
-  useEffect(() => {
-    if (!formData?.inputs)
-      return
-    const initialInputs: Record<string, string> = {}
-    formData.inputs.forEach((item) => {
-      initialInputs[item.output_variable_name] = item.default.type === 'variable' ? formData.resolved_default_values[item.output_variable_name] || '' : item.default.value
-    })
-    setInputs(initialInputs)
-  }, [formData?.inputs, formData?.resolved_default_values])
-
-  // use immer
-  const handleInputsChange = (name: string, value: string) => {
-    const newInputs = produce(inputs, (draft) => {
-      draft[name] = value
-    })
-    setInputs(newInputs)
-  }
-
-  const submit = (actionID: string) => {
-    submitForm(
-      { token, data: { inputs, action: actionID } },
-      {
-        onSuccess: () => {
-          setSuccess(true)
-        },
-      },
-    )
-  }
-
   if (isLoading) {
     return (
       <Loading type="app" />
@@ -99,190 +51,66 @@ const FormContent = () => {
 
   if (success) {
     return (
-      <div className={cn('flex h-full w-full flex-col items-center justify-center')}>
-        <div className="max-w-[640px] min-w-[480px]">
-          <div className="border-components-divider-subtle flex h-[320px] flex-col gap-4 rounded-[20px] border bg-chat-bubble-bg p-10 pb-9 shadow-lg backdrop-blur-xs">
-            <div className="h-[56px] w-[56px] shrink-0 rounded-2xl border border-components-panel-border-subtle bg-background-default-dodge p-3">
-              <RiCheckboxCircleFill className="h-8 w-8 text-text-success" />
-            </div>
-            <div className="grow">
-              <div className="title-4xl-semi-bold text-text-primary">{t('humanInput.thanks', { ns: 'share' })}</div>
-              <div className="title-4xl-semi-bold text-text-primary">{t('humanInput.recorded', { ns: 'share' })}</div>
-            </div>
-            <div className="shrink-0 system-2xs-regular-uppercase text-text-tertiary">{t('humanInput.submissionID', { id: token, ns: 'share' })}</div>
-          </div>
-          <div className="flex flex-row-reverse px-2 py-3">
-            <div className={cn(
-              'flex shrink-0 items-center gap-1.5 px-1',
-            )}
-            >
-              <div className="system-2xs-medium-uppercase text-text-tertiary">{t('chat.poweredBy', { ns: 'share' })}</div>
-              <DifyLogo size="small" />
-            </div>
-          </div>
-        </div>
-      </div>
+      <FormStatusCard
+        iconClassName="i-ri-checkbox-circle-fill text-text-success"
+        title={t('humanInput.thanks', { ns: 'share' })}
+        subtitle={t('humanInput.recorded', { ns: 'share' })}
+        submissionID={token}
+        removeWebappBrand={removeWebappBrand}
+        replaceWebappLogo={replaceWebappLogo}
+      />
     )
   }
 
   if (expired) {
     return (
-      <div className={cn('flex h-full w-full flex-col items-center justify-center')}>
-        <div className="max-w-[640px] min-w-[480px]">
-          <div className="border-components-divider-subtle flex h-[320px] flex-col gap-4 rounded-[20px] border bg-chat-bubble-bg p-10 pb-9 shadow-lg backdrop-blur-xs">
-            <div className="flex h-14 w-14 shrink-0 items-center justify-center rounded-2xl border border-components-panel-border-subtle bg-background-default-dodge p-3">
-              <RiInformation2Fill className="h-8 w-8 text-text-accent" />
-            </div>
-            <div className="grow">
-              <div className="title-4xl-semi-bold text-text-primary">{t('humanInput.sorry', { ns: 'share' })}</div>
-              <div className="title-4xl-semi-bold text-text-primary">{t('humanInput.expired', { ns: 'share' })}</div>
-            </div>
-            <div className="shrink-0 system-2xs-regular-uppercase text-text-tertiary">{t('humanInput.submissionID', { id: token, ns: 'share' })}</div>
-          </div>
-          <div className="flex flex-row-reverse px-2 py-3">
-            <div className={cn(
-              'flex shrink-0 items-center gap-1.5 px-1',
-            )}
-            >
-              <div className="system-2xs-medium-uppercase text-text-tertiary">{t('chat.poweredBy', { ns: 'share' })}</div>
-              <DifyLogo size="small" />
-            </div>
-          </div>
-        </div>
-      </div>
+      <FormStatusCard
+        iconClassName="i-ri-information-2-fill text-text-accent"
+        title={t('humanInput.sorry', { ns: 'share' })}
+        subtitle={t('humanInput.expired', { ns: 'share' })}
+        submissionID={token}
+      />
     )
   }
 
   if (submitted) {
     return (
-      <div className={cn('flex h-full w-full flex-col items-center justify-center')}>
-        <div className="max-w-[640px] min-w-[480px]">
-          <div className="border-components-divider-subtle flex h-[320px] flex-col gap-4 rounded-[20px] border bg-chat-bubble-bg p-10 pb-9 shadow-lg backdrop-blur-xs">
-            <div className="flex h-14 w-14 shrink-0 items-center justify-center rounded-2xl border border-components-panel-border-subtle bg-background-default-dodge p-3">
-              <RiInformation2Fill className="h-8 w-8 text-text-accent" />
-            </div>
-            <div className="grow">
-              <div className="title-4xl-semi-bold text-text-primary">{t('humanInput.sorry', { ns: 'share' })}</div>
-              <div className="title-4xl-semi-bold text-text-primary">{t('humanInput.completed', { ns: 'share' })}</div>
-            </div>
-            <div className="shrink-0 system-2xs-regular-uppercase text-text-tertiary">{t('humanInput.submissionID', { id: token, ns: 'share' })}</div>
-          </div>
-          <div className="flex flex-row-reverse px-2 py-3">
-            <div className={cn(
-              'flex shrink-0 items-center gap-1.5 px-1',
-            )}
-            >
-              <div className="system-2xs-medium-uppercase text-text-tertiary">{t('chat.poweredBy', { ns: 'share' })}</div>
-              <DifyLogo size="small" />
-            </div>
-          </div>
-        </div>
-      </div>
+      <FormStatusCard
+        iconClassName="i-ri-information-2-fill text-text-accent"
+        title={t('humanInput.sorry', { ns: 'share' })}
+        subtitle={t('humanInput.completed', { ns: 'share' })}
+        submissionID={token}
+      />
     )
   }
 
   if (rateLimitExceeded) {
     return (
-      <div className={cn('flex h-full w-full flex-col items-center justify-center')}>
-        <div className="max-w-[640px] min-w-[480px]">
-          <div className="border-components-divider-subtle flex h-[320px] flex-col gap-4 rounded-[20px] border bg-chat-bubble-bg p-10 pb-9 shadow-lg backdrop-blur-xs">
-            <div className="flex h-14 w-14 shrink-0 items-center justify-center rounded-2xl border border-components-panel-border-subtle bg-background-default-dodge p-3">
-              <RiErrorWarningFill className="h-8 w-8 text-text-destructive" />
-            </div>
-            <div className="grow">
-              <div className="title-4xl-semi-bold text-text-primary">{t('humanInput.rateLimitExceeded', { ns: 'share' })}</div>
-            </div>
-          </div>
-          <div className="flex flex-row-reverse px-2 py-3">
-            <div className={cn(
-              'flex shrink-0 items-center gap-1.5 px-1',
-            )}
-            >
-              <div className="system-2xs-medium-uppercase text-text-tertiary">{t('chat.poweredBy', { ns: 'share' })}</div>
-              <DifyLogo size="small" />
-            </div>
-          </div>
-        </div>
-      </div>
+      <FormStatusCard
+        iconClassName="i-ri-error-warning-fill text-text-destructive"
+        title={t('humanInput.rateLimitExceeded', { ns: 'share' })}
+      />
     )
   }
 
   if (!formData) {
     return (
-      <div className={cn('flex h-full w-full flex-col items-center justify-center')}>
-        <div className="max-w-[640px] min-w-[480px]">
-          <div className="border-components-divider-subtle flex h-[320px] flex-col gap-4 rounded-[20px] border bg-chat-bubble-bg p-10 pb-9 shadow-lg backdrop-blur-xs">
-            <div className="flex h-14 w-14 shrink-0 items-center justify-center rounded-2xl border border-components-panel-border-subtle bg-background-default-dodge p-3">
-              <RiErrorWarningFill className="h-8 w-8 text-text-destructive" />
-            </div>
-            <div className="grow">
-              <div className="title-4xl-semi-bold text-text-primary">{t('humanInput.formNotFound', { ns: 'share' })}</div>
-            </div>
-          </div>
-          <div className="flex flex-row-reverse px-2 py-3">
-            <div className={cn(
-              'flex shrink-0 items-center gap-1.5 px-1',
-            )}
-            >
-              <div className="system-2xs-medium-uppercase text-text-tertiary">{t('chat.poweredBy', { ns: 'share' })}</div>
-              <DifyLogo size="small" />
-            </div>
-          </div>
-        </div>
-      </div>
+      <FormStatusCard
+        iconClassName="i-ri-error-warning-fill text-text-destructive"
+        title={t('humanInput.formNotFound', { ns: 'share' })}
+      />
     )
   }
 
-  const site = formData.site.site
-
   return (
-    <div className={cn('mx-auto flex h-full w-full max-w-[720px] flex-col items-center')}>
-      <div className="mt-4 flex w-full shrink-0 items-center gap-3 py-3">
-        <AppIcon
-          size="large"
-          iconType={site.icon_type}
-          icon={site.icon}
-          background={site.icon_background}
-          imageUrl={site.icon_url}
-        />
-        <div className="grow system-xl-semibold text-text-primary">{site.title}</div>
-      </div>
-      <div className="h-0 w-full grow overflow-y-auto">
-        <div className="border-components-divider-subtle rounded-[20px] border bg-chat-bubble-bg p-4 shadow-lg backdrop-blur-xs">
-          {contentList.map((content, index) => (
-            <ContentItem
-              key={index}
-              content={content}
-              formInputFields={formData.inputs}
-              inputs={inputs}
-              onInputChange={handleInputsChange}
-            />
-          ))}
-          <div className="flex flex-wrap gap-1 py-1">
-            {formData.user_actions.map((action: UserAction) => (
-              <Button
-                key={action.id}
-                disabled={isSubmitting}
-                variant={getButtonStyle(action.button_style) as ButtonProps['variant']}
-                onClick={() => submit(action.id)}
-              >
-                {action.title}
-              </Button>
-            ))}
-          </div>
-          <ExpirationTime expirationTime={formData.expiration_time * 1000} />
-        </div>
-        <div className="flex flex-row-reverse px-2 py-3">
-          <div className={cn(
-            'flex shrink-0 items-center gap-1.5 px-1',
-          )}
-          >
-            <div className="system-2xs-medium-uppercase text-text-tertiary">{t('chat.poweredBy', { ns: 'share' })}</div>
-            <DifyLogo size="small" />
-          </div>
-        </div>
-      </div>
-    </div>
+    <LoadedFormContent
+      key={token}
+      formData={formData}
+      isSubmitting={isSubmitting}
+      onSubmit={submit}
+      removeWebappBrand={removeWebappBrand}
+      replaceWebappLogo={replaceWebappLogo}
+    />
   )
 }
 

web/app/(humanInputLayout)/form/[token]/loaded-form-content.tsx

@@ -0,0 +1,95 @@
+import type { ButtonProps } from '@langgenius/dify-ui/button'
+import type { FormData } from './form'
+import type { HumanInputFieldValue } from '@/app/components/base/chat/chat/answer/human-input-content/field-renderer'
+import type { UserAction } from '@/app/components/workflow/nodes/human-input/types'
+import { Button } from '@langgenius/dify-ui/button'
+import { cn } from '@langgenius/dify-ui/cn'
+import { produce } from 'immer'
+import { useMemo, useState } from 'react'
+import AppIcon from '@/app/components/base/app-icon'
+import ContentItem from '@/app/components/base/chat/chat/answer/human-input-content/content-item'
+import ExpirationTime from '@/app/components/base/chat/chat/answer/human-input-content/expiration-time'
+import { getButtonStyle, getRenderedFormInputs, hasInvalidRequiredHumanInput, initializeInputs, splitByOutputVar } from '@/app/components/base/chat/chat/answer/human-input-content/utils'
+import BrandingFooter from './branding-footer'
+
+type LoadedFormContentProps = {
+  formData: FormData
+  isSubmitting: boolean
+  onSubmit: (inputs: Record<string, HumanInputFieldValue>, actionID: string, formInputs: FormData['inputs']) => void
+  removeWebappBrand?: boolean
+  replaceWebappLogo?: string | null
+}
+
+const LoadedFormContent = ({
+  formData,
+  isSubmitting,
+  onSubmit,
+  removeWebappBrand,
+  replaceWebappLogo,
+}: LoadedFormContentProps) => {
+  const renderedFormInputs = getRenderedFormInputs(formData.inputs, formData.form_content)
+  const [inputs, setInputs] = useState<Record<string, HumanInputFieldValue>>(() =>
+    initializeInputs(renderedFormInputs, formData.resolved_default_values),
+  )
+
+  const contentList = useMemo(() => {
+    return splitByOutputVar(formData.form_content)
+  }, [formData.form_content])
+
+  const handleInputsChange = (name: string, value: HumanInputFieldValue) => {
+    setInputs(prevInputs => produce(prevInputs, (draft) => {
+      draft[name] = value
+    }))
+  }
+
+  const submit = (actionID: string) => {
+    onSubmit(inputs, actionID, formData.inputs)
+  }
+
+  const isActionDisabled = isSubmitting || hasInvalidRequiredHumanInput(renderedFormInputs, inputs)
+  const site = formData.site.site
+
+  return (
+    <div className={cn('mx-auto flex size-full max-w-180 flex-col items-center')}>
+      <div className="mt-4 flex w-full shrink-0 items-center gap-3 py-3">
+        <AppIcon
+          size="large"
+          iconType={site.icon_type}
+          icon={site.icon}
+          background={site.icon_background}
+          imageUrl={site.icon_url}
+        />
+        <div className="grow system-xl-semibold text-text-primary">{site.title}</div>
+      </div>
+      <div className="h-0 w-full grow overflow-y-auto">
+        <div className="rounded-[20px] border border-divider-subtle bg-chat-bubble-bg p-4 shadow-lg backdrop-blur-xs">
+          {contentList.map((content, index) => (
+            <ContentItem
+              key={index}
+              content={content}
+              formInputFields={formData.inputs}
+              inputs={inputs}
+              onInputChange={handleInputsChange}
+            />
+          ))}
+          <div className="flex flex-wrap gap-1 py-1">
+            {formData.user_actions.map((action: UserAction) => (
+              <Button
+                key={action.id}
+                disabled={isActionDisabled}
+                variant={getButtonStyle(action.button_style) as ButtonProps['variant']}
+                onClick={() => submit(action.id)}
+              >
+                {action.title}
+              </Button>
+            ))}
+          </div>
+          <ExpirationTime expirationTime={formData.expiration_time * 1000} />
+        </div>
+        <BrandingFooter removeWebappBrand={removeWebappBrand} replaceWebappLogo={replaceWebappLogo} />
+      </div>
+    </div>
+  )
+}
+
+export default LoadedFormContent

web/app/(humanInputLayout)/form/[token]/use-form-submit.tsx

@@ -0,0 +1,33 @@
+import type { HumanInputFieldValue } from '@/app/components/base/chat/chat/answer/human-input-content/field-renderer'
+import type { FormInputItem } from '@/app/components/workflow/nodes/human-input/types'
+import { useCallback, useState } from 'react'
+import { getProcessedHumanInputFormInputs } from '@/app/components/base/chat/chat/answer/human-input-content/utils'
+import { useSubmitHumanInputForm } from '@/service/use-share'
+
+export const useFormSubmit = (token: string) => {
+  const [success, setSuccess] = useState(false)
+  const { mutate: submitForm, isPending: isSubmitting } = useSubmitHumanInputForm()
+
+  const submit = useCallback((inputs: Record<string, HumanInputFieldValue>, actionID: string, formInputs: FormInputItem[]) => {
+    submitForm(
+      {
+        token,
+        data: {
+          inputs: getProcessedHumanInputFormInputs(formInputs, inputs) || {},
+          action: actionID,
+        },
+      },
+      {
+        onSuccess: () => {
+          setSuccess(true)
+        },
+      },
+    )
+  }, [submitForm, token])
+
+  return {
+    isSubmitting,
+    submit,
+    success,
+  }
+}

web/app/components/base/chat/chat/answer/human-input-content/utils.ts

@@ -1,10 +1,20 @@
+import type { HumanInputFieldValue } from './field-renderer'
+import type { FileEntity } from '@/app/components/base/file-uploader/types'
 import type { FormInputItem } from '@/app/components/workflow/nodes/human-input/types'
 import type { Locale } from '@/i18n-config'
+import type { HumanInputResolvedValue } from '@/types/workflow'
 import dayjs from 'dayjs'
 import isSameOrAfter from 'dayjs/plugin/isSameOrAfter'
 import relativeTime from 'dayjs/plugin/relativeTime'
 import utc from 'dayjs/plugin/utc'
-import { UserActionButtonType } from '@/app/components/workflow/nodes/human-input/types'
+import { fileIsUploaded, getProcessedFiles } from '@/app/components/base/file-uploader/utils'
+import {
+  isFileFormInput,
+  isFileListFormInput,
+  isParagraphFormInput,
+  isSelectFormInput,
+  UserActionButtonType,
+} from '@/app/components/workflow/nodes/human-input/types'
 import 'dayjs/locale/en'
 import 'dayjs/locale/zh-cn'
 import 'dayjs/locale/ja'
@@ -31,17 +41,140 @@ export const splitByOutputVar = (content: string): string[] => {
   return parts.filter(part => part.length > 0)
 }
 
-export const initializeInputs = (formInputs: FormInputItem[], defaultValues: Record<string, string> = {}) => {
-  const initialInputs: Record<string, any> = {}
+export const getFormContentInputNames = (content: string) => {
+  const outputVarRegex = /\{\{#\$output\.([^#]+)#\}\}/g
+  return [...content.matchAll(outputVarRegex)].map(match => match[1]!)
+}
+
+export const getRenderedFormInputs = (formInputs: FormInputItem[], content: string) => {
+  const inputNames = new Set(getFormContentInputNames(content))
+  return formInputs.filter(input => inputNames.has(input.output_variable_name))
+}
+
+export const initializeInputs = (formInputs: FormInputItem[], defaultValues: Record<string, HumanInputResolvedValue> = {}) => {
+  const initialInputs: Record<string, HumanInputFieldValue> = {}
   formInputs.forEach((item) => {
-    if (item.type === 'text-input' || item.type === 'paragraph')
-      initialInputs[item.output_variable_name] = item.default.type === 'variable' ? defaultValues[item.output_variable_name] || '' : item.default.value
-    else
-      initialInputs[item.output_variable_name] = undefined
+    if (isParagraphFormInput(item)) {
+      const resolvedValue = defaultValues[item.output_variable_name]
+      initialInputs[item.output_variable_name] = item.default.type === 'variable' && typeof resolvedValue === 'string'
+        ? resolvedValue
+        : item.default.value
+      return
+    }
+
+    if (isSelectFormInput(item)) {
+      initialInputs[item.output_variable_name] = ''
+      return
+    }
+
+    if (isFileFormInput(item)) {
+      initialInputs[item.output_variable_name] = null
+      return
+    }
+
+    if (isFileListFormInput(item)) {
+      initialInputs[item.output_variable_name] = []
+    }
   })
   return initialInputs
 }
 
+const isHumanInputFileUploaded = (value: HumanInputFieldValue | undefined) => {
+  return !!value
+    && !Array.isArray(value)
+    && typeof value !== 'string'
+    && !!fileIsUploaded(value as FileEntity)
+}
+
+const hasUploadedHumanInputFiles = (value: HumanInputFieldValue | undefined) => {
+  return Array.isArray(value)
+    && value.length > 0
+    && value.every(file => !!fileIsUploaded(file))
+}
+
+export const hasInvalidSelectOrFileInput = (
+  formInputs: FormInputItem[],
+  values: Record<string, HumanInputFieldValue>,
+) => {
+  return formInputs.some((input) => {
+    if (!(input.output_variable_name in values))
+      return false
+
+    const value = values[input.output_variable_name]
+
+    if (isSelectFormInput(input))
+      return typeof value !== 'string' || value.length === 0
+
+    if (isFileFormInput(input))
+      return Array.isArray(value) ? !hasUploadedHumanInputFiles(value) : !isHumanInputFileUploaded(value)
+
+    if (isFileListFormInput(input))
+      return !hasUploadedHumanInputFiles(value)
+
+    return false
+  })
+}
+
+export const hasInvalidRequiredHumanInput = (
+  formInputs: FormInputItem[],
+  values: Record<string, HumanInputFieldValue>,
+) => {
+  return formInputs.some((input) => {
+    if (!(input.output_variable_name in values))
+      return false
+
+    const value = values[input.output_variable_name]
+
+    if (isParagraphFormInput(input))
+      return typeof value !== 'string' || value.trim().length === 0
+
+    if (isSelectFormInput(input))
+      return typeof value !== 'string' || value.length === 0
+
+    if (isFileFormInput(input))
+      return Array.isArray(value) ? !hasUploadedHumanInputFiles(value) : !isHumanInputFileUploaded(value)
+
+    if (isFileListFormInput(input))
+      return !hasUploadedHumanInputFiles(value)
+
+    return false
+  })
+}
+
+export const getProcessedHumanInputFormInputs = (
+  formInputs: FormInputItem[],
+  values: Record<string, HumanInputFieldValue> | undefined,
+) => {
+  if (!values)
+    return undefined
+
+  const processedInputs: Record<string, unknown> = { ...values }
+
+  formInputs.forEach((input) => {
+    const value = values[input.output_variable_name]
+
+    if (isFileListFormInput(input)) {
+      processedInputs[input.output_variable_name] = Array.isArray(value)
+        ? getProcessedFiles(value)
+        : []
+      return
+    }
+
+    if (isFileFormInput(input)) {
+      if (Array.isArray(value)) {
+        processedInputs[input.output_variable_name] = getProcessedFiles(value)[0]
+        return
+      }
+
+      processedInputs[input.output_variable_name] = value && typeof value !== 'string'
+        ? getProcessedFiles([value as FileEntity])[0]
+        : undefined
+    }
+  })
+
+  return processedInputs
+}
+
 const localeMap: Record<string, string> = {
   'en-US': 'en',
   'zh-Hans': 'zh-cn',

web/app/components/workflow/nodes/human-input/types.ts

@@ -1,8 +1,13 @@
 import type {
   CommonNodeType,
-  InputVarType,
+  UploadFileSetting,
   ValueSelector,
 } from '@/app/components/workflow/types'
+import {
+  InputVarType,
+  SupportUploadFileTypes,
+} from '@/app/components/workflow/types'
+import { TransferMethod } from '@/types/app'
 
 export type HumanInputNodeType = CommonNodeType & {
   delivery_methods: DeliveryMethod[]
@@ -59,14 +64,137 @@ export type UserAction = {
   button_style: UserActionButtonType
 }
 
-export type FormInputItemDefault = {
+type StringDefault = {
   selector: ValueSelector
   type: 'variable' | 'constant'
   value: string
 }
 
-export type FormInputItem = {
-  type: InputVarType
+type StringListSource = {
+  selector: ValueSelector
+  type: 'variable' | 'constant'
+  value: string[]
+}
+
+// Preserve the old export during the transition to the new schema names.
+export type FormInputItemDefault = StringDefault
+
+type BaseFormInputItem = {
   output_variable_name: string
-  default: FormInputItemDefault
+}
+
+export type ParagraphFormInput = BaseFormInputItem & {
+  type: InputVarType.paragraph
+  default: StringDefault
+}
+
+export type SelectFormInput = BaseFormInputItem & {
+  type: InputVarType.select
+  option_source: StringListSource
+}
+
+type SharedFileFormInput = Pick<
+  UploadFileSetting,
+  'allowed_file_extensions' | 'allowed_file_types' | 'allowed_file_upload_methods'
+>
+
+export type FileFormInput = BaseFormInputItem & SharedFileFormInput & {
+  type: InputVarType.singleFile
+}
+
+export type FileListFormInput = BaseFormInputItem & SharedFileFormInput & {
+  type: InputVarType.multiFiles
+  number_limits?: UploadFileSetting['number_limits']
+}
+
+export type FormInputItem
+  = | ParagraphFormInput
+    | SelectFormInput
+    | FileFormInput
+    | FileListFormInput
+
+export const isParagraphFormInput = (
+  input: FormInputItem,
+): input is ParagraphFormInput => {
+  return input.type === InputVarType.paragraph
+}
+
+export const isSelectFormInput = (
+  input: FormInputItem,
+): input is SelectFormInput => {
+  return input.type === InputVarType.select
+}
+
+export const isFileFormInput = (
+  input: FormInputItem,
+): input is FileFormInput => {
+  return input.type === InputVarType.singleFile
+}
+
+export const isFileListFormInput = (
+  input: FormInputItem,
+): input is FileListFormInput => {
+  return input.type === InputVarType.multiFiles
+}
+
+export const createDefaultParagraphFormInput = (
+  output_variable_name = '',
+): ParagraphFormInput => ({
+  type: InputVarType.paragraph,
+  output_variable_name,
+  default: {
+    type: 'constant',
+    selector: [],
+    value: '',
+  },
+})
+
+const createDefaultSelectFormInput = (
+  output_variable_name = '',
+): SelectFormInput => ({
+  type: InputVarType.select,
+  output_variable_name,
+  option_source: {
+    type: 'constant',
+    selector: [],
+    value: [],
+  },
+})
+
+const createDefaultFileFormInput = (
+  output_variable_name = '',
+): FileFormInput => ({
+  type: InputVarType.singleFile,
+  output_variable_name,
+  allowed_file_extensions: [],
+  allowed_file_types: [SupportUploadFileTypes.image],
+  allowed_file_upload_methods: [TransferMethod.local_file, TransferMethod.remote_url],
+})
+
+const createDefaultFileListFormInput = (
+  output_variable_name = '',
+): FileListFormInput => ({
+  type: InputVarType.multiFiles,
+  output_variable_name,
+  allowed_file_extensions: [],
+  allowed_file_types: [SupportUploadFileTypes.image],
+  allowed_file_upload_methods: [TransferMethod.local_file, TransferMethod.remote_url],
+  number_limits: 5,
+})
+
+export const createDefaultFormInputByType = (
+  type: FormInputItem['type'],
+  output_variable_name = '',
+): FormInputItem => {
+  switch (type) {
+    case InputVarType.select:
+      return createDefaultSelectFormInput(output_variable_name)
+    case InputVarType.singleFile:
+      return createDefaultFileFormInput(output_variable_name)
+    case InputVarType.multiFiles:
+      return createDefaultFileListFormInput(output_variable_name)
+    case InputVarType.paragraph:
+    default:
+      return createDefaultParagraphFormInput(output_variable_name)
+  }
 }

web/types/workflow.ts

@@ -39,6 +39,8 @@ export type AgentLogItemWithChildren = AgentLogItem & {
   children: AgentLogItemWithChildren[]
 }
 
+export type HumanInputResolvedValue = string | FileResponse | FileResponse[]
+
 export type NodeTracing = {
   id: string
   index: number