feat: summary index

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

.claude/settings.json

@@ -5,5 +5,18 @@
     "typescript-lsp@claude-plugins-official": true,
     "pyright-lsp@claude-plugins-official": true,
     "ralph-loop@claude-plugins-official": true
+  },
+  "hooks": {
+    "PreToolUse": [
+      {
+        "matcher": "Bash",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "npx -y block-no-verify@1.1.1"
+          }
+        ]
+      }
+    ]
   }
 }

.github/workflows/api-tests.yml

@@ -39,12 +39,6 @@ jobs:
       - name: Install dependencies
         run: uv sync --project api --dev
 
-      - name: Run pyrefly check
-        run: |
-          cd api
-          uv add --dev pyrefly
-          uv run pyrefly check || true
-
       - name: Run dify config tests
         run: uv run --project api dev/pytest/pytest_config_tests.py
 

.github/workflows/deploy-agent-dev.yml

@@ -1,4 +1,4 @@
-name: Deploy Trigger Dev
+name: Deploy Agent Dev
 
 permissions:
   contents: read
@@ -7,7 +7,7 @@ on:
   workflow_run:
     workflows: ["Build and Push API & Web"]
     branches:
-      - "deploy/trigger-dev"
+      - "deploy/agent-dev"
     types:
       - completed
 
@@ -16,12 +16,12 @@ jobs:
     runs-on: ubuntu-latest
     if: |
       github.event.workflow_run.conclusion == 'success' &&
-      github.event.workflow_run.head_branch == 'deploy/trigger-dev'
+      github.event.workflow_run.head_branch == 'deploy/agent-dev'
     steps:
       - name: Deploy to server
         uses: appleboy/ssh-action@v0.1.8
         with:
-          host: ${{ secrets.TRIGGER_SSH_HOST }}
+          host: ${{ secrets.AGENT_DEV_SSH_HOST }}
           username: ${{ secrets.SSH_USER }}
           key: ${{ secrets.SSH_PRIVATE_KEY }}
           script: |

.github/workflows/translate-i18n-base-on-english.yml

@@ -1,94 +0,0 @@
-name: Translate i18n Files Based on English
-
-on:
-  push:
-    branches: [main]
-    paths:
-      - 'web/i18n/en-US/*.json'
-  workflow_dispatch:
-
-permissions:
-  contents: write
-  pull-requests: write
-
-jobs:
-  check-and-update:
-    if: github.repository == 'langgenius/dify'
-    runs-on: ubuntu-latest
-    defaults:
-      run:
-        working-directory: web
-    steps:
-      # Keep use old checkout action version for https://github.com/peter-evans/create-pull-request/issues/4272
-      - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-          token: ${{ secrets.GITHUB_TOKEN }}
-
-      - name: Check for file changes in i18n/en-US
-        id: check_files
-        run: |
-          # Skip check for manual trigger, translate all files
-          if [ "${{ github.event_name }}" == "workflow_dispatch" ]; then
-            echo "FILES_CHANGED=true" >> $GITHUB_ENV
-            echo "FILE_ARGS=" >> $GITHUB_ENV
-            echo "Manual trigger: translating all files"
-          else
-            git fetch origin "${{ github.event.before }}" || true
-            git fetch origin "${{ github.sha }}" || true
-            changed_files=$(git diff --name-only "${{ github.event.before }}" "${{ github.sha }}" -- 'i18n/en-US/*.json')
-            echo "Changed files: $changed_files"
-            if [ -n "$changed_files" ]; then
-              echo "FILES_CHANGED=true" >> $GITHUB_ENV
-              file_args=""
-              for file in $changed_files; do
-                filename=$(basename "$file" .json)
-                file_args="$file_args --file $filename"
-              done
-              echo "FILE_ARGS=$file_args" >> $GITHUB_ENV
-              echo "File arguments: $file_args"
-            else
-              echo "FILES_CHANGED=false" >> $GITHUB_ENV
-            fi
-          fi
-
-      - name: Install pnpm
-        uses: pnpm/action-setup@v4
-        with:
-          package_json_file: web/package.json
-          run_install: false
-
-      - name: Set up Node.js
-        if: env.FILES_CHANGED == 'true'
-        uses: actions/setup-node@v6
-        with:
-          node-version: 'lts/*'
-          cache: pnpm
-          cache-dependency-path: ./web/pnpm-lock.yaml
-
-      - name: Install dependencies
-        if: env.FILES_CHANGED == 'true'
-        working-directory: ./web
-        run: pnpm install --frozen-lockfile
-
-      - name: Generate i18n translations
-        if: env.FILES_CHANGED == 'true'
-        working-directory: ./web
-        run: pnpm run i18n:gen ${{ env.FILE_ARGS }}
-
-      - name: Create Pull Request
-        if: env.FILES_CHANGED == 'true'
-        uses: peter-evans/create-pull-request@v6
-        with:
-          token: ${{ secrets.GITHUB_TOKEN }}
-          commit-message: 'chore(i18n): update translations based on en-US changes'
-          title: 'chore(i18n): translate i18n files based on en-US changes'
-          body: |
-            This PR was automatically created to update i18n translation files based on changes in en-US locale.
-
-            **Triggered by:** ${{ github.sha }}
-
-            **Changes included:**
-            - Updated translation files for all locales
-          branch: chore/automated-i18n-updates-${{ github.sha }}
-          delete-branch: true

.github/workflows/translate-i18n-claude.yml

@@ -0,0 +1,421 @@
+name: Translate i18n Files with Claude Code
+
+# Note: claude-code-action doesn't support push events directly.
+# Push events are handled by trigger-i18n-sync.yml which sends repository_dispatch.
+# See: https://github.com/langgenius/dify/issues/30743
+
+on:
+  repository_dispatch:
+    types: [i18n-sync]
+  workflow_dispatch:
+    inputs:
+      files:
+        description: 'Specific files to translate (space-separated, e.g., "app common"). Leave empty for all files.'
+        required: false
+        type: string
+      languages:
+        description: 'Specific languages to translate (space-separated, e.g., "zh-Hans ja-JP"). Leave empty for all supported languages.'
+        required: false
+        type: string
+      mode:
+        description: 'Sync mode: incremental (only changes) or full (re-check all keys)'
+        required: false
+        default: 'incremental'
+        type: choice
+        options:
+          - incremental
+          - full
+
+permissions:
+  contents: write
+  pull-requests: write
+
+jobs:
+  translate:
+    if: github.repository == 'langgenius/dify'
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v6
+        with:
+          fetch-depth: 0
+          token: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Configure Git
+        run: |
+          git config --global user.name "github-actions[bot]"
+          git config --global user.email "github-actions[bot]@users.noreply.github.com"
+
+      - name: Install pnpm
+        uses: pnpm/action-setup@v4
+        with:
+          package_json_file: web/package.json
+          run_install: false
+
+      - name: Set up Node.js
+        uses: actions/setup-node@v6
+        with:
+          node-version: 'lts/*'
+          cache: pnpm
+          cache-dependency-path: ./web/pnpm-lock.yaml
+
+      - name: Detect changed files and generate diff
+        id: detect_changes
+        run: |
+          if [ "${{ github.event_name }}" == "workflow_dispatch" ]; then
+            # Manual trigger
+            if [ -n "${{ github.event.inputs.files }}" ]; then
+              echo "CHANGED_FILES=${{ github.event.inputs.files }}" >> $GITHUB_OUTPUT
+            else
+              # Get all JSON files in en-US directory
+              files=$(ls web/i18n/en-US/*.json 2>/dev/null | xargs -n1 basename | sed 's/.json$//' | tr '\n' ' ')
+              echo "CHANGED_FILES=$files" >> $GITHUB_OUTPUT
+            fi
+            echo "TARGET_LANGS=${{ github.event.inputs.languages }}" >> $GITHUB_OUTPUT
+            echo "SYNC_MODE=${{ github.event.inputs.mode || 'incremental' }}" >> $GITHUB_OUTPUT
+
+            # For manual trigger with incremental mode, get diff from last commit
+            # For full mode, we'll do a complete check anyway
+            if [ "${{ github.event.inputs.mode }}" == "full" ]; then
+              echo "Full mode: will check all keys" > /tmp/i18n-diff.txt
+              echo "DIFF_AVAILABLE=false" >> $GITHUB_OUTPUT
+            else
+              git diff HEAD~1..HEAD -- 'web/i18n/en-US/*.json' > /tmp/i18n-diff.txt 2>/dev/null || echo "" > /tmp/i18n-diff.txt
+              if [ -s /tmp/i18n-diff.txt ]; then
+                echo "DIFF_AVAILABLE=true" >> $GITHUB_OUTPUT
+              else
+                echo "DIFF_AVAILABLE=false" >> $GITHUB_OUTPUT
+              fi
+            fi
+          elif [ "${{ github.event_name }}" == "repository_dispatch" ]; then
+            # Triggered by push via trigger-i18n-sync.yml workflow
+            # Validate required payload fields
+            if [ -z "${{ github.event.client_payload.changed_files }}" ]; then
+              echo "Error: repository_dispatch payload missing required 'changed_files' field" >&2
+              exit 1
+            fi
+            echo "CHANGED_FILES=${{ github.event.client_payload.changed_files }}" >> $GITHUB_OUTPUT
+            echo "TARGET_LANGS=" >> $GITHUB_OUTPUT
+            echo "SYNC_MODE=${{ github.event.client_payload.sync_mode || 'incremental' }}" >> $GITHUB_OUTPUT
+
+            # Decode the base64-encoded diff from the trigger workflow
+            if [ -n "${{ github.event.client_payload.diff_base64 }}" ]; then
+              if ! echo "${{ github.event.client_payload.diff_base64 }}" | base64 -d > /tmp/i18n-diff.txt 2>&1; then
+                echo "Warning: Failed to decode base64 diff payload" >&2
+                echo "" > /tmp/i18n-diff.txt
+                echo "DIFF_AVAILABLE=false" >> $GITHUB_OUTPUT
+              elif [ -s /tmp/i18n-diff.txt ]; then
+                echo "DIFF_AVAILABLE=true" >> $GITHUB_OUTPUT
+              else
+                echo "DIFF_AVAILABLE=false" >> $GITHUB_OUTPUT
+              fi
+            else
+              echo "" > /tmp/i18n-diff.txt
+              echo "DIFF_AVAILABLE=false" >> $GITHUB_OUTPUT
+            fi
+          else
+            echo "Unsupported event type: ${{ github.event_name }}"
+            exit 1
+          fi
+
+          # Truncate diff if too large (keep first 50KB)
+          if [ -f /tmp/i18n-diff.txt ]; then
+            head -c 50000 /tmp/i18n-diff.txt > /tmp/i18n-diff-truncated.txt
+            mv /tmp/i18n-diff-truncated.txt /tmp/i18n-diff.txt
+          fi
+
+          echo "Detected files: $(cat $GITHUB_OUTPUT | grep CHANGED_FILES || echo 'none')"
+
+      - name: Run Claude Code for Translation Sync
+        if: steps.detect_changes.outputs.CHANGED_FILES != ''
+        uses: anthropics/claude-code-action@v1
+        with:
+          anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+          prompt: |
+            You are a professional i18n synchronization engineer for the Dify project.
+            Your task is to keep all language translations in sync with the English source (en-US).
+
+            ## CRITICAL TOOL RESTRICTIONS
+            - Use **Read** tool to read files (NOT cat or bash)
+            - Use **Edit** tool to modify JSON files (NOT node, jq, or bash scripts)
+            - Use **Bash** ONLY for: git commands, gh commands, pnpm commands
+            - Run bash commands ONE BY ONE, never combine with && or ||
+            - NEVER use `$()` command substitution - it's not supported. Split into separate commands instead.
+
+            ## WORKING DIRECTORY & ABSOLUTE PATHS
+            Claude Code sandbox working directory may vary. Always use absolute paths:
+            - For pnpm: `pnpm --dir ${{ github.workspace }}/web <command>`
+            - For git: `git -C ${{ github.workspace }} <command>`
+            - For gh: `gh --repo ${{ github.repository }} <command>`
+            - For file paths: `${{ github.workspace }}/web/i18n/`
+
+            ## EFFICIENCY RULES
+            - **ONE Edit per language file** - batch all key additions into a single Edit
+            - Insert new keys at the beginning of JSON (after `{`), lint:fix will sort them
+            - Translate ALL keys for a language mentally first, then do ONE Edit
+
+            ## Context
+            - Changed/target files: ${{ steps.detect_changes.outputs.CHANGED_FILES }}
+            - Target languages (empty means all supported): ${{ steps.detect_changes.outputs.TARGET_LANGS }}
+            - Sync mode: ${{ steps.detect_changes.outputs.SYNC_MODE }}
+            - Translation files are located in: ${{ github.workspace }}/web/i18n/{locale}/{filename}.json
+            - Language configuration is in: ${{ github.workspace }}/web/i18n-config/languages.ts
+            - Git diff is available: ${{ steps.detect_changes.outputs.DIFF_AVAILABLE }}
+
+            ## CRITICAL DESIGN: Verify First, Then Sync
+
+            You MUST follow this three-phase approach:
+
+            ═══════════════════════════════════════════════════════════════
+            ║  PHASE 1: VERIFY - Analyze and Generate Change Report       ║
+            ═══════════════════════════════════════════════════════════════
+
+            ### Step 1.1: Analyze Git Diff (for incremental mode)
+            Use the Read tool to read `/tmp/i18n-diff.txt` to see the git diff.
+
+            Parse the diff to categorize changes:
+            - Lines with `+` (not `+++`): Added or modified values
+            - Lines with `-` (not `---`): Removed or old values
+            - Identify specific keys for each category:
+              * ADD: Keys that appear only in `+` lines (new keys)
+              * UPDATE: Keys that appear in both `-` and `+` lines (value changed)
+              * DELETE: Keys that appear only in `-` lines (removed keys)
+
+            ### Step 1.2: Read Language Configuration
+            Use the Read tool to read `${{ github.workspace }}/web/i18n-config/languages.ts`.
+            Extract all languages with `supported: true`.
+
+            ### Step 1.3: Run i18n:check for Each Language
+            ```bash
+            pnpm --dir ${{ github.workspace }}/web install --frozen-lockfile
+            ```
+            ```bash
+            pnpm --dir ${{ github.workspace }}/web run i18n:check
+            ```
+
+            This will report:
+            - Missing keys (need to ADD)
+            - Extra keys (need to DELETE)
+
+            ### Step 1.4: Generate Change Report
+
+            Create a structured report identifying:
+            ```
+            ╔══════════════════════════════════════════════════════════════╗
+            ║                    I18N SYNC CHANGE REPORT                   ║
+            ╠══════════════════════════════════════════════════════════════╣
+            ║ Files to process: [list]                                     ║
+            ║ Languages to sync: [list]                                    ║
+            ╠══════════════════════════════════════════════════════════════╣
+            ║ ADD (New Keys):                                              ║
+            ║   - [filename].[key]: "English value"                        ║
+            ║   ...                                                        ║
+            ╠══════════════════════════════════════════════════════════════╣
+            ║ UPDATE (Modified Keys - MUST re-translate):                  ║
+            ║   - [filename].[key]: "Old value" → "New value"              ║
+            ║   ...                                                        ║
+            ╠══════════════════════════════════════════════════════════════╣
+            ║ DELETE (Extra Keys):                                         ║
+            ║   - [language]/[filename].[key]                              ║
+            ║   ...                                                        ║
+            ╚══════════════════════════════════════════════════════════════╝
+            ```
+
+            **IMPORTANT**: For UPDATE detection, compare git diff to find keys where
+            the English value changed. These MUST be re-translated even if target
+            language already has a translation (it's now stale!).
+
+            ═══════════════════════════════════════════════════════════════
+            ║  PHASE 2: SYNC - Execute Changes Based on Report            ║
+            ═══════════════════════════════════════════════════════════════
+
+            ### Step 2.1: Process ADD Operations (BATCH per language file)
+
+            **CRITICAL WORKFLOW for efficiency:**
+            1. First, translate ALL new keys for ALL languages mentally
+            2. Then, for EACH language file, do ONE Edit operation:
+               - Read the file once
+               - Insert ALL new keys at the beginning (right after the opening `{`)
+               - Don't worry about alphabetical order - lint:fix will sort them later
+
+            Example Edit (adding 3 keys to zh-Hans/app.json):
+            ```
+            old_string: '{\n  "accessControl"'
+            new_string: '{\n  "newKey1": "translation1",\n  "newKey2": "translation2",\n  "newKey3": "translation3",\n  "accessControl"'
+            ```
+
+            **IMPORTANT**:
+            - ONE Edit per language file (not one Edit per key!)
+            - Always use the Edit tool. NEVER use bash scripts, node, or jq.
+
+            ### Step 2.2: Process UPDATE Operations
+
+            **IMPORTANT: Special handling for zh-Hans and ja-JP**
+            If zh-Hans or ja-JP files were ALSO modified in the same push:
+            - Run: `git -C ${{ github.workspace }} diff HEAD~1 --name-only` and check for zh-Hans or ja-JP files
+            - If found, it means someone manually translated them. Apply these rules:
+
+            1. **Missing keys**: Still ADD them (completeness required)
+            2. **Existing translations**: Compare with the NEW English value:
+               - If translation is **completely wrong** or **unrelated** → Update it
+               - If translation is **roughly correct** (captures the meaning) → Keep it, respect manual work
+               - When in doubt, **keep the manual translation**
+
+            Example:
+            - English changed: "Save" → "Save Changes"
+            - Manual translation: "保存更改" → Keep it (correct meaning)
+            - Manual translation: "删除" → Update it (completely wrong)
+
+            For other languages:
+            Use Edit tool to replace the old value with the new translation.
+            You can batch multiple updates in one Edit if they are adjacent.
+
+            ### Step 2.3: Process DELETE Operations
+            For extra keys reported by i18n:check:
+            - Run: `pnpm --dir ${{ github.workspace }}/web run i18n:check --auto-remove`
+            - Or manually remove from target language JSON files
+
+            ## Translation Guidelines
+
+            - PRESERVE all placeholders exactly as-is:
+              - `{{variable}}` - Mustache interpolation
+              - `${variable}` - Template literal
+              - `<tag>content</tag>` - HTML tags
+              - `_one`, `_other` - Pluralization suffixes (these are KEY suffixes, not values)
+            - Use appropriate language register (formal/informal) based on existing translations
+            - Match existing translation style in each language
+            - Technical terms: check existing conventions per language
+            - For CJK languages: no spaces between characters unless necessary
+            - For RTL languages (ar-TN, fa-IR): ensure proper text handling
+
+            ## Output Format Requirements
+            - Alphabetical key ordering (if original file uses it)
+            - 2-space indentation
+            - Trailing newline at end of file
+            - Valid JSON (use proper escaping for special characters)
+
+            ═══════════════════════════════════════════════════════════════
+            ║  PHASE 3: RE-VERIFY - Confirm All Issues Resolved           ║
+            ═══════════════════════════════════════════════════════════════
+
+            ### Step 3.1: Run Lint Fix (IMPORTANT!)
+            ```bash
+            pnpm --dir ${{ github.workspace }}/web lint:fix --quiet -- 'i18n/**/*.json'
+            ```
+            This ensures:
+            - JSON keys are sorted alphabetically (jsonc/sort-keys rule)
+            - Valid i18n keys (dify-i18n/valid-i18n-keys rule)
+            - No extra keys (dify-i18n/no-extra-keys rule)
+
+            ### Step 3.2: Run Final i18n Check
+            ```bash
+            pnpm --dir ${{ github.workspace }}/web run i18n:check
+            ```
+
+            ### Step 3.3: Fix Any Remaining Issues
+            If check reports issues:
+            - Go back to PHASE 2 for unresolved items
+            - Repeat until check passes
+
+            ### Step 3.4: Generate Final Summary
+            ```
+            ╔══════════════════════════════════════════════════════════════╗
+            ║                    SYNC COMPLETED SUMMARY                    ║
+            ╠══════════════════════════════════════════════════════════════╣
+            ║ Language      │ Added │ Updated │ Deleted │ Status          ║
+            ╠══════════════════════════════════════════════════════════════╣
+            ║ zh-Hans       │  5    │   2     │    1    │ ✓ Complete      ║
+            ║ ja-JP         │  5    │   2     │    1    │ ✓ Complete      ║
+            ║ ...           │ ...   │  ...    │   ...   │ ...             ║
+            ╠══════════════════════════════════════════════════════════════╣
+            ║ i18n:check    │ PASSED - All keys in sync                   ║
+            ╚══════════════════════════════════════════════════════════════╝
+            ```
+
+            ## Mode-Specific Behavior
+
+            **SYNC_MODE = "incremental"** (default):
+            - Focus on keys identified from git diff
+            - Also check i18n:check output for any missing/extra keys
+            - Efficient for small changes
+
+            **SYNC_MODE = "full"**:
+            - Compare ALL keys between en-US and each language
+            - Run i18n:check to identify all discrepancies
+            - Use for first-time sync or fixing historical issues
+
+            ## Important Notes
+
+            1. Always run i18n:check BEFORE and AFTER making changes
+            2. The check script is the source of truth for missing/extra keys
+            3. For UPDATE scenario: git diff is the source of truth for changed values
+            4. Create a single commit with all translation changes
+            5. If any translation fails, continue with others and report failures
+
+            ═══════════════════════════════════════════════════════════════
+            ║  PHASE 4: COMMIT AND CREATE PR                              ║
+            ═══════════════════════════════════════════════════════════════
+
+            After all translations are complete and verified:
+
+            ### Step 4.1: Check for changes
+            ```bash
+            git -C ${{ github.workspace }} status --porcelain
+            ```
+
+            If there are changes:
+
+            ### Step 4.2: Create a new branch and commit
+            Run these git commands ONE BY ONE (not combined with &&).
+            **IMPORTANT**: Do NOT use `$()` command substitution. Use two separate commands:
+
+            1. First, get the timestamp:
+            ```bash
+            date +%Y%m%d-%H%M%S
+            ```
+            (Note the output, e.g., "20260115-143052")
+
+            2. Then create branch using the timestamp value:
+            ```bash
+            git -C ${{ github.workspace }} checkout -b chore/i18n-sync-20260115-143052
+            ```
+            (Replace "20260115-143052" with the actual timestamp from step 1)
+
+            3. Stage changes:
+            ```bash
+            git -C ${{ github.workspace }} add web/i18n/
+            ```
+
+            4. Commit:
+            ```bash
+            git -C ${{ github.workspace }} commit -m "chore(i18n): sync translations with en-US - Mode: ${{ steps.detect_changes.outputs.SYNC_MODE }}"
+            ```
+
+            5. Push:
+            ```bash
+            git -C ${{ github.workspace }} push origin HEAD
+            ```
+
+            ### Step 4.3: Create Pull Request
+            ```bash
+            gh pr create --repo ${{ github.repository }} --title "chore(i18n): sync translations with en-US" --body "## Summary
+
+            This PR was automatically generated to sync i18n translation files.
+
+            ### Changes
+            - Mode: ${{ steps.detect_changes.outputs.SYNC_MODE }}
+            - Files processed: ${{ steps.detect_changes.outputs.CHANGED_FILES }}
+
+            ### Verification
+            - [x] \`i18n:check\` passed
+            - [x] \`lint:fix\` applied
+
+            🤖 Generated with Claude Code GitHub Action" --base main
+            ```
+
+          claude_args: |
+            --max-turns 150
+            --allowedTools "Read,Write,Edit,Bash(git *),Bash(git:*),Bash(gh *),Bash(gh:*),Bash(pnpm *),Bash(pnpm:*),Bash(date *),Bash(date:*),Glob,Grep"

.github/workflows/trigger-i18n-sync.yml

@@ -0,0 +1,66 @@
+name: Trigger i18n Sync on Push
+
+# This workflow bridges the push event to repository_dispatch
+# because claude-code-action doesn't support push events directly.
+# See: https://github.com/langgenius/dify/issues/30743
+
+on:
+  push:
+    branches: [main]
+    paths:
+      - 'web/i18n/en-US/*.json'
+
+permissions:
+  contents: write
+
+jobs:
+  trigger:
+    if: github.repository == 'langgenius/dify'
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Detect changed files and generate diff
+        id: detect
+        run: |
+          BEFORE_SHA="${{ github.event.before }}"
+          # Handle edge case: force push may have null/zero SHA
+          if [ -z "$BEFORE_SHA" ] || [ "$BEFORE_SHA" = "0000000000000000000000000000000000000000" ]; then
+            BEFORE_SHA="HEAD~1"
+          fi
+
+          # Detect changed i18n files
+          changed=$(git diff --name-only "$BEFORE_SHA" "${{ github.sha }}" -- 'web/i18n/en-US/*.json' 2>/dev/null | xargs -n1 basename 2>/dev/null | sed 's/.json$//' | tr '\n' ' ' || echo "")
+          echo "changed_files=$changed" >> $GITHUB_OUTPUT
+
+          # Generate diff for context
+          git diff "$BEFORE_SHA" "${{ github.sha }}" -- 'web/i18n/en-US/*.json' > /tmp/i18n-diff.txt 2>/dev/null || echo "" > /tmp/i18n-diff.txt
+
+          # Truncate if too large (keep first 50KB to match receiving workflow)
+          head -c 50000 /tmp/i18n-diff.txt > /tmp/i18n-diff-truncated.txt
+          mv /tmp/i18n-diff-truncated.txt /tmp/i18n-diff.txt
+
+          # Base64 encode the diff for safe JSON transport (portable, single-line)
+          diff_base64=$(base64 < /tmp/i18n-diff.txt | tr -d '\n')
+          echo "diff_base64=$diff_base64" >> $GITHUB_OUTPUT
+
+          if [ -n "$changed" ]; then
+            echo "has_changes=true" >> $GITHUB_OUTPUT
+            echo "Detected changed files: $changed"
+          else
+            echo "has_changes=false" >> $GITHUB_OUTPUT
+            echo "No i18n changes detected"
+          fi
+
+      - name: Trigger i18n sync workflow
+        if: steps.detect.outputs.has_changes == 'true'
+        uses: peter-evans/repository-dispatch@v3
+        with:
+          token: ${{ secrets.GITHUB_TOKEN }}
+          event-type: i18n-sync
+          client-payload: '{"changed_files": "${{ steps.detect.outputs.changed_files }}", "diff_base64": "${{ steps.detect.outputs.diff_base64 }}", "sync_mode": "incremental", "trigger_sha": "${{ github.sha }}"}'

.nvmrc

@@ -1 +0,0 @@
-22.11.0

api/.env.example

@@ -589,6 +589,7 @@ ENABLE_CLEAN_UNUSED_DATASETS_TASK=false
 ENABLE_CREATE_TIDB_SERVERLESS_TASK=false
 ENABLE_UPDATE_TIDB_SERVERLESS_STATUS_TASK=false
 ENABLE_CLEAN_MESSAGES=false
+ENABLE_WORKFLOW_RUN_CLEANUP_TASK=false
 ENABLE_MAIL_CLEAN_DOCUMENT_NOTIFY_TASK=false
 ENABLE_DATASETS_QUEUE_MONITOR=false
 ENABLE_CHECK_UPGRADABLE_PLUGIN_TASK=true

api/commands.py

@@ -1,8 +1,10 @@
 import base64
+import datetime
 import json
 import logging
 import secrets
 from typing import Any
+from uuid import uuid4
 
 import click
 import sqlalchemy as sa
@@ -22,6 +24,7 @@ from core.rag.datasource.vdb.vector_type import VectorType
 from core.rag.index_processor.constant.built_in_field import BuiltInField
 from core.rag.models.document import Document
 from core.tools.utils.system_oauth_encryption import encrypt_system_oauth_params
+from core.workflow.entities.pause_reason import PauseReasonType
 from events.app_event import app_was_created
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client
@@ -34,13 +37,25 @@ from libs.rsa import generate_key_pair
 from models import Tenant
 from models.dataset import Dataset, DatasetCollectionBinding, DatasetMetadata, DatasetMetadataBinding, DocumentSegment
 from models.dataset import Document as DatasetDocument
+from models.enums import CreatorUserRole, ExecutionOffLoadType
 from models.model import Account, App, AppAnnotationSetting, AppMode, Conversation, MessageAnnotation, UploadFile
 from models.oauth import DatasourceOauthParamConfig, DatasourceProvider
 from models.provider import Provider, ProviderModel
 from models.provider_ids import DatasourceProviderID, ToolProviderID
 from models.source import DataSourceApiKeyAuthBinding, DataSourceOauthBinding
 from models.tools import ToolOAuthSystemClient
+from models.trigger import AppTriggerType, WorkflowTriggerLog, WorkflowTriggerStatus
+from models.workflow import (
+    WorkflowAppLog,
+    WorkflowNodeExecutionModel,
+    WorkflowNodeExecutionOffload,
+    WorkflowNodeExecutionTriggeredFrom,
+    WorkflowPause,
+    WorkflowPauseReason,
+    WorkflowRun,
+)
 from services.account_service import AccountService, RegisterService, TenantService
+from services.clear_free_plan_expired_workflow_run_logs import WorkflowRunCleanup
 from services.clear_free_plan_tenant_expired_logs import ClearFreePlanTenantExpiredLogs
 from services.plugin.data_migration import PluginDataMigration
 from services.plugin.plugin_migration import PluginMigration
@@ -852,6 +867,218 @@ def clear_free_plan_tenant_expired_logs(days: int, batch: int, tenant_ids: list[
     click.echo(click.style("Clear free plan tenant expired logs completed.", fg="green"))
 
 
+@click.command("clean-workflow-runs", help="Clean expired workflow runs and related data for free tenants.")
+@click.option("--days", default=30, show_default=True, help="Delete workflow runs created before N days ago.")
+@click.option("--batch-size", default=200, show_default=True, help="Batch size for selecting workflow runs.")
+@click.option(
+    "--start-after",
+    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    default=None,
+    help="Optional lower bound (inclusive) for created_at; must be paired with --end-before.",
+)
+@click.option(
+    "--end-before",
+    type=click.DateTime(formats=["%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"]),
+    default=None,
+    help="Optional upper bound (exclusive) for created_at; must be paired with --start-after.",
+)
+def clean_workflow_runs(
+    days: int,
+    batch_size: int,
+    start_after: datetime.datetime | None,
+    end_before: datetime.datetime | None,
+):
+    """
+    Clean workflow runs and related workflow data for free tenants.
+    """
+    if (start_after is None) ^ (end_before is None):
+        raise click.UsageError("--start-after and --end-before must be provided together.")
+
+    click.echo(click.style("Starting workflow run cleanup.", fg="white"))
+
+    WorkflowRunCleanup(
+        days=days,
+        batch_size=batch_size,
+        start_after=start_after,
+        end_before=end_before,
+    ).run()
+
+    click.echo(click.style("Workflow run cleanup completed.", fg="green"))
+
+
+@click.command(
+    "seed-expired-workflow-runs",
+    help="Test-only: create synthetic expired workflow runs (with related records) for cleanup validation.",
+)
+@click.option("--days-ago", default=40, show_default=True, help="Backdate created_at to N days ago.")
+@click.option("--limit", default=5, show_default=True, help="Number of workflow runs to create.")
+@click.option(
+    "--tenant-id",
+    default=None,
+    help="Optional tenant id to use; if omitted, a random tenant id is generated.",
+)
+def seed_expired_workflow_runs(days_ago: int, limit: int, tenant_id: str | None) -> None:
+    base_time = datetime.datetime.now() - datetime.timedelta(days=days_ago)
+    tenant = tenant_id or str(uuid4())
+
+    with sessionmaker(db.engine, expire_on_commit=False).begin() as session:
+        for i in range(limit):
+            run_id = str(uuid4())
+            app_id = str(uuid4())
+            workflow_id = str(uuid4())
+            creator_id = str(uuid4())
+            created_at = base_time - datetime.timedelta(minutes=i)
+
+            run = WorkflowRun(
+                id=run_id,
+                tenant_id=tenant,
+                app_id=app_id,
+                workflow_id=workflow_id,
+                type="workflow",
+                triggered_from="workflow-run",
+                version="v1",
+                graph="{}",
+                inputs="{}",
+                status="succeeded",
+                outputs="{}",
+                error=None,
+                elapsed_time=0.1,
+                total_tokens=0,
+                total_steps=0,
+                created_by_role="account",
+                created_by=creator_id,
+                created_at=created_at,
+                finished_at=created_at + datetime.timedelta(seconds=1),
+                exceptions_count=0,
+            )
+
+            node_execution = WorkflowNodeExecutionModel(
+                id=str(uuid4()),
+                tenant_id=tenant,
+                app_id=app_id,
+                workflow_id=workflow_id,
+                triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN.value,
+                workflow_run_id=run_id,
+                index=0,
+                predecessor_node_id=None,
+                node_execution_id=None,
+                node_id="start",
+                node_type="start",
+                title="Seeded node",
+                inputs="{}",
+                process_data="{}",
+                outputs="{}",
+                status="succeeded",
+                error=None,
+                elapsed_time=0.1,
+                execution_metadata="{}",
+                created_by_role="account",
+                created_by=creator_id,
+                created_at=created_at,
+                finished_at=created_at + datetime.timedelta(seconds=1),
+            )
+
+            offload_files: list[UploadFile] = []
+            node_execution_offloads: list[WorkflowNodeExecutionOffload] = []
+            for offload_type in (
+                ExecutionOffLoadType.INPUTS,
+                ExecutionOffLoadType.OUTPUTS,
+                ExecutionOffLoadType.PROCESS_DATA,
+            ):
+                upload_file = UploadFile(
+                    tenant_id=tenant,
+                    storage_type=StorageType.LOCAL.value,
+                    key=f"seeded/workflow-node-offload/{run_id}-{offload_type.value}.json",
+                    name=f"{run_id}-{offload_type.value}.json",
+                    size=256,
+                    extension="json",
+                    mime_type="application/json",
+                    created_by_role=CreatorUserRole.ACCOUNT,
+                    created_by=creator_id,
+                    created_at=created_at,
+                    used=True,
+                    used_by=creator_id,
+                    used_at=created_at,
+                )
+                offload_files.append(upload_file)
+                node_execution_offloads.append(
+                    WorkflowNodeExecutionOffload(
+                        tenant_id=tenant,
+                        app_id=app_id,
+                        node_execution_id=node_execution.id,
+                        type_=offload_type,
+                        file_id=upload_file.id,
+                    )
+                )
+
+            app_log = WorkflowAppLog(
+                tenant_id=tenant,
+                app_id=app_id,
+                workflow_id=workflow_id,
+                workflow_run_id=run_id,
+                created_from="service-api",
+                created_by_role="account",
+                created_by=creator_id
+            )
+
+            trigger_log = WorkflowTriggerLog(
+                tenant_id=tenant,
+                app_id=app_id,
+                workflow_id=workflow_id,
+                workflow_run_id=run_id,
+                root_node_id=None,
+                trigger_metadata="{}",
+                trigger_type=AppTriggerType.TRIGGER_WEBHOOK,
+                trigger_data="{}",
+                inputs="{}",
+                outputs="{}",
+                status=WorkflowTriggerStatus.SUCCEEDED,
+                error=None,
+                queue_name="seed",
+                celery_task_id=None,
+                created_by_role="account",
+                created_by=creator_id,
+                retry_count=0,
+                elapsed_time=0.1,
+                total_tokens=0,
+                triggered_at=created_at,
+                finished_at=created_at + datetime.timedelta(seconds=1),
+            )
+
+            pause_id = str(uuid4())
+            pause = WorkflowPause(
+                id=pause_id,
+                workflow_id=workflow_id,
+                workflow_run_id=run_id,
+                resumed_at=None,
+                state_object_key="seeded-state",
+                created_at=created_at,
+                updated_at=created_at,
+            )
+
+            pause_reason = WorkflowPauseReason(
+                pause_id=pause_id,
+                type_=PauseReasonType.HUMAN_INPUT_REQUIRED,
+                form_id=str(uuid4()),
+                message="Seeded human input required.",
+                node_id="seeded-human-input",
+                created_at=created_at,
+                updated_at=created_at,
+            )
+
+            records = [run, node_execution, app_log, trigger_log, pause, pause_reason]
+            records.extend(offload_files)
+            records.extend(node_execution_offloads)
+            session.add_all(records)
+
+        click.echo(
+            click.style(
+                f"Created {limit} synthetic workflow runs (tenant={tenant}) backdated ~{days_ago} days.",
+                fg="green",
+            )
+        )
+
+
 @click.option("-f", "--force", is_flag=True, help="Skip user confirmation and force the command to execute.")
 @click.command("clear-orphaned-file-records", help="Clear orphaned file records.")
 def clear_orphaned_file_records(force: bool):

api/configs/feature/__init__.py

@@ -959,6 +959,16 @@ class MailConfig(BaseSettings):
         default=None,
     )
 
+    ENABLE_TRIAL_APP: bool = Field(
+        description="Enable trial app",
+        default=False,
+    )
+
+    ENABLE_EXPLORE_BANNER: bool = Field(
+        description="Enable explore banner",
+        default=False,
+    )
+
 
 class RagEtlConfig(BaseSettings):
     """
@@ -1101,6 +1111,10 @@ class CeleryScheduleTasksConfig(BaseSettings):
         description="Enable clean messages task",
         default=False,
     )
+    ENABLE_WORKFLOW_RUN_CLEANUP_TASK: bool = Field(
+        description="Enable scheduled workflow run cleanup task",
+        default=False,
+    )
     ENABLE_MAIL_CLEAN_DOCUMENT_NOTIFY_TASK: bool = Field(
         description="Enable mail clean document notify task",
         default=False,

api/configs/feature/hosted_service/__init__.py

@@ -8,6 +8,11 @@ class HostedCreditConfig(BaseSettings):
         default="",
     )
 
+    HOSTED_POOL_CREDITS: int = Field(
+        description="Pool credits for hosted service",
+        default=200,
+    )
+
     def get_model_credits(self, model_name: str) -> int:
         """
         Get credit value for a specific model name.
@@ -60,19 +65,46 @@ class HostedOpenAiConfig(BaseSettings):
 
     HOSTED_OPENAI_TRIAL_MODELS: str = Field(
         description="Comma-separated list of available models for trial access",
-        default="gpt-3.5-turbo,"
-        "gpt-3.5-turbo-1106,"
-        "gpt-3.5-turbo-instruct,"
+        default="gpt-4,"
+        "gpt-4-turbo-preview,"
+        "gpt-4-turbo-2024-04-09,"
+        "gpt-4-1106-preview,"
+        "gpt-4-0125-preview,"
+        "gpt-4-turbo,"
+        "gpt-4.1,"
+        "gpt-4.1-2025-04-14,"
+        "gpt-4.1-mini,"
+        "gpt-4.1-mini-2025-04-14,"
+        "gpt-4.1-nano,"
+        "gpt-4.1-nano-2025-04-14,"
+        "gpt-3.5-turbo,"
         "gpt-3.5-turbo-16k,"
         "gpt-3.5-turbo-16k-0613,"
+        "gpt-3.5-turbo-1106,"
         "gpt-3.5-turbo-0613,"
         "gpt-3.5-turbo-0125,"
-        "text-davinci-003",
-    )
-
-    HOSTED_OPENAI_QUOTA_LIMIT: NonNegativeInt = Field(
-        description="Quota limit for hosted OpenAI service usage",
-        default=200,
+        "gpt-3.5-turbo-instruct,"
+        "text-davinci-003,"
+        "chatgpt-4o-latest,"
+        "gpt-4o,"
+        "gpt-4o-2024-05-13,"
+        "gpt-4o-2024-08-06,"
+        "gpt-4o-2024-11-20,"
+        "gpt-4o-audio-preview,"
+        "gpt-4o-audio-preview-2025-06-03,"
+        "gpt-4o-mini,"
+        "gpt-4o-mini-2024-07-18,"
+        "o3-mini,"
+        "o3-mini-2025-01-31,"
+        "gpt-5-mini-2025-08-07,"
+        "gpt-5-mini,"
+        "o4-mini,"
+        "o4-mini-2025-04-16,"
+        "gpt-5-chat-latest,"
+        "gpt-5,"
+        "gpt-5-2025-08-07,"
+        "gpt-5-nano,"
+        "gpt-5-nano-2025-08-07",
     )
 
     HOSTED_OPENAI_PAID_ENABLED: bool = Field(
@@ -87,6 +119,13 @@ class HostedOpenAiConfig(BaseSettings):
         "gpt-4-turbo-2024-04-09,"
         "gpt-4-1106-preview,"
         "gpt-4-0125-preview,"
+        "gpt-4-turbo,"
+        "gpt-4.1,"
+        "gpt-4.1-2025-04-14,"
+        "gpt-4.1-mini,"
+        "gpt-4.1-mini-2025-04-14,"
+        "gpt-4.1-nano,"
+        "gpt-4.1-nano-2025-04-14,"
         "gpt-3.5-turbo,"
         "gpt-3.5-turbo-16k,"
         "gpt-3.5-turbo-16k-0613,"
@@ -94,7 +133,150 @@ class HostedOpenAiConfig(BaseSettings):
         "gpt-3.5-turbo-0613,"
         "gpt-3.5-turbo-0125,"
         "gpt-3.5-turbo-instruct,"
-        "text-davinci-003",
+        "text-davinci-003,"
+        "chatgpt-4o-latest,"
+        "gpt-4o,"
+        "gpt-4o-2024-05-13,"
+        "gpt-4o-2024-08-06,"
+        "gpt-4o-2024-11-20,"
+        "gpt-4o-audio-preview,"
+        "gpt-4o-audio-preview-2025-06-03,"
+        "gpt-4o-mini,"
+        "gpt-4o-mini-2024-07-18,"
+        "o3-mini,"
+        "o3-mini-2025-01-31,"
+        "gpt-5-mini-2025-08-07,"
+        "gpt-5-mini,"
+        "o4-mini,"
+        "o4-mini-2025-04-16,"
+        "gpt-5-chat-latest,"
+        "gpt-5,"
+        "gpt-5-2025-08-07,"
+        "gpt-5-nano,"
+        "gpt-5-nano-2025-08-07",
+    )
+
+
+class HostedGeminiConfig(BaseSettings):
+    """
+    Configuration for fetching Gemini service
+    """
+
+    HOSTED_GEMINI_API_KEY: str | None = Field(
+        description="API key for hosted Gemini service",
+        default=None,
+    )
+
+    HOSTED_GEMINI_API_BASE: str | None = Field(
+        description="Base URL for hosted Gemini API",
+        default=None,
+    )
+
+    HOSTED_GEMINI_API_ORGANIZATION: str | None = Field(
+        description="Organization ID for hosted Gemini service",
+        default=None,
+    )
+
+    HOSTED_GEMINI_TRIAL_ENABLED: bool = Field(
+        description="Enable trial access to hosted Gemini service",
+        default=False,
+    )
+
+    HOSTED_GEMINI_TRIAL_MODELS: str = Field(
+        description="Comma-separated list of available models for trial access",
+        default="gemini-2.5-flash,gemini-2.0-flash,gemini-2.0-flash-lite,",
+    )
+
+    HOSTED_GEMINI_PAID_ENABLED: bool = Field(
+        description="Enable paid access to hosted gemini service",
+        default=False,
+    )
+
+    HOSTED_GEMINI_PAID_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="gemini-2.5-flash,gemini-2.0-flash,gemini-2.0-flash-lite,",
+    )
+
+
+class HostedXAIConfig(BaseSettings):
+    """
+    Configuration for fetching XAI service
+    """
+
+    HOSTED_XAI_API_KEY: str | None = Field(
+        description="API key for hosted XAI service",
+        default=None,
+    )
+
+    HOSTED_XAI_API_BASE: str | None = Field(
+        description="Base URL for hosted XAI API",
+        default=None,
+    )
+
+    HOSTED_XAI_API_ORGANIZATION: str | None = Field(
+        description="Organization ID for hosted XAI service",
+        default=None,
+    )
+
+    HOSTED_XAI_TRIAL_ENABLED: bool = Field(
+        description="Enable trial access to hosted XAI service",
+        default=False,
+    )
+
+    HOSTED_XAI_TRIAL_MODELS: str = Field(
+        description="Comma-separated list of available models for trial access",
+        default="grok-3,grok-3-mini,grok-3-mini-fast",
+    )
+
+    HOSTED_XAI_PAID_ENABLED: bool = Field(
+        description="Enable paid access to hosted XAI service",
+        default=False,
+    )
+
+    HOSTED_XAI_PAID_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="grok-3,grok-3-mini,grok-3-mini-fast",
+    )
+
+
+class HostedDeepseekConfig(BaseSettings):
+    """
+    Configuration for fetching Deepseek service
+    """
+
+    HOSTED_DEEPSEEK_API_KEY: str | None = Field(
+        description="API key for hosted Deepseek service",
+        default=None,
+    )
+
+    HOSTED_DEEPSEEK_API_BASE: str | None = Field(
+        description="Base URL for hosted Deepseek API",
+        default=None,
+    )
+
+    HOSTED_DEEPSEEK_API_ORGANIZATION: str | None = Field(
+        description="Organization ID for hosted Deepseek service",
+        default=None,
+    )
+
+    HOSTED_DEEPSEEK_TRIAL_ENABLED: bool = Field(
+        description="Enable trial access to hosted Deepseek service",
+        default=False,
+    )
+
+    HOSTED_DEEPSEEK_TRIAL_MODELS: str = Field(
+        description="Comma-separated list of available models for trial access",
+        default="deepseek-chat,deepseek-reasoner",
+    )
+
+    HOSTED_DEEPSEEK_PAID_ENABLED: bool = Field(
+        description="Enable paid access to hosted Deepseek service",
+        default=False,
+    )
+
+    HOSTED_DEEPSEEK_PAID_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="deepseek-chat,deepseek-reasoner",
     )
 
 
@@ -144,16 +326,66 @@ class HostedAnthropicConfig(BaseSettings):
         default=False,
     )
 
-    HOSTED_ANTHROPIC_QUOTA_LIMIT: NonNegativeInt = Field(
-        description="Quota limit for hosted Anthropic service usage",
-        default=600000,
-    )
-
     HOSTED_ANTHROPIC_PAID_ENABLED: bool = Field(
         description="Enable paid access to hosted Anthropic service",
         default=False,
     )
 
+    HOSTED_ANTHROPIC_TRIAL_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="claude-opus-4-20250514,"
+        "claude-sonnet-4-20250514,"
+        "claude-3-5-haiku-20241022,"
+        "claude-3-opus-20240229,"
+        "claude-3-7-sonnet-20250219,"
+        "claude-3-haiku-20240307",
+    )
+    HOSTED_ANTHROPIC_PAID_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="claude-opus-4-20250514,"
+        "claude-sonnet-4-20250514,"
+        "claude-3-5-haiku-20241022,"
+        "claude-3-opus-20240229,"
+        "claude-3-7-sonnet-20250219,"
+        "claude-3-haiku-20240307",
+    )
+
+
+class HostedTongyiConfig(BaseSettings):
+    """
+    Configuration for hosted Tongyi service
+    """
+
+    HOSTED_TONGYI_API_KEY: str | None = Field(
+        description="API key for hosted Tongyi service",
+        default=None,
+    )
+
+    HOSTED_TONGYI_USE_INTERNATIONAL_ENDPOINT: bool = Field(
+        description="Use international endpoint for hosted Tongyi service",
+        default=False,
+    )
+
+    HOSTED_TONGYI_TRIAL_ENABLED: bool = Field(
+        description="Enable trial access to hosted Tongyi service",
+        default=False,
+    )
+
+    HOSTED_TONGYI_PAID_ENABLED: bool = Field(
+        description="Enable paid access to hosted Anthropic service",
+        default=False,
+    )
+
+    HOSTED_TONGYI_TRIAL_MODELS: str = Field(
+        description="Comma-separated list of available models for trial access",
+        default="",
+    )
+
+    HOSTED_TONGYI_PAID_MODELS: str = Field(
+        description="Comma-separated list of available models for paid access",
+        default="",
+    )
+
 
 class HostedMinmaxConfig(BaseSettings):
     """
@@ -246,9 +478,13 @@ class HostedServiceConfig(
     HostedOpenAiConfig,
     HostedSparkConfig,
     HostedZhipuAIConfig,
+    HostedTongyiConfig,
     # moderation
     HostedModerationConfig,
     # credit config
     HostedCreditConfig,
+    HostedGeminiConfig,
+    HostedXAIConfig,
+    HostedDeepseekConfig,
 ):
     pass

api/controllers/console/__init__.py

@@ -107,10 +107,12 @@ from .datasets.rag_pipeline import (
 
 # Import explore controllers
 from .explore import (
+    banner,
     installed_app,
     parameter,
     recommended_app,
     saved_message,
+    trial,
 )
 
 # Import tag controllers
@@ -145,6 +147,7 @@ __all__ = [
     "apikey",
     "app",
     "audio",
+    "banner",
     "billing",
     "bp",
     "completion",
@@ -198,6 +201,7 @@ __all__ = [
     "statistic",
     "tags",
     "tool_providers",
+    "trial",
     "trigger_providers",
     "version",
     "website",

api/controllers/console/admin.py

@@ -15,7 +15,7 @@ from controllers.console.wraps import only_edition_cloud
 from core.db.session_factory import session_factory
 from extensions.ext_database import db
 from libs.token import extract_access_token
-from models.model import App, InstalledApp, RecommendedApp
+from models.model import App, ExporleBanner, InstalledApp, RecommendedApp, TrialApp
 
 P = ParamSpec("P")
 R = TypeVar("R")
@@ -32,6 +32,8 @@ class InsertExploreAppPayload(BaseModel):
     language: str = Field(...)
     category: str = Field(...)
     position: int = Field(...)
+    can_trial: bool = Field(default=False)
+    trial_limit: int = Field(default=0)
 
     @field_validator("language")
     @classmethod
@@ -39,11 +41,33 @@ class InsertExploreAppPayload(BaseModel):
         return supported_language(value)
 
 
+class InsertExploreBannerPayload(BaseModel):
+    category: str = Field(...)
+    title: str = Field(...)
+    description: str = Field(...)
+    img_src: str = Field(..., alias="img-src")
+    language: str = Field(default="en-US")
+    link: str = Field(...)
+    sort: int = Field(...)
+
+    @field_validator("language")
+    @classmethod
+    def validate_language(cls, value: str) -> str:
+        return supported_language(value)
+
+    model_config = {"populate_by_name": True}
+
+
 console_ns.schema_model(
     InsertExploreAppPayload.__name__,
     InsertExploreAppPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
 )
 
+console_ns.schema_model(
+    InsertExploreBannerPayload.__name__,
+    InsertExploreBannerPayload.model_json_schema(ref_template=DEFAULT_REF_TEMPLATE_SWAGGER_2_0),
+)
+
 
 def admin_required(view: Callable[P, R]):
     @wraps(view)
@@ -109,6 +133,20 @@ class InsertExploreAppListApi(Resource):
                 )
 
                 db.session.add(recommended_app)
+                if payload.can_trial:
+                    trial_app = db.session.execute(
+                        select(TrialApp).where(TrialApp.app_id == payload.app_id)
+                    ).scalar_one_or_none()
+                    if not trial_app:
+                        db.session.add(
+                            TrialApp(
+                                app_id=payload.app_id,
+                                tenant_id=app.tenant_id,
+                                trial_limit=payload.trial_limit,
+                            )
+                        )
+                    else:
+                        trial_app.trial_limit = payload.trial_limit
 
                 app.is_public = True
                 db.session.commit()
@@ -123,6 +161,20 @@ class InsertExploreAppListApi(Resource):
                 recommended_app.category = payload.category
                 recommended_app.position = payload.position
 
+                if payload.can_trial:
+                    trial_app = db.session.execute(
+                        select(TrialApp).where(TrialApp.app_id == payload.app_id)
+                    ).scalar_one_or_none()
+                    if not trial_app:
+                        db.session.add(
+                            TrialApp(
+                                app_id=payload.app_id,
+                                tenant_id=app.tenant_id,
+                                trial_limit=payload.trial_limit,
+                            )
+                        )
+                    else:
+                        trial_app.trial_limit = payload.trial_limit
                 app.is_public = True
 
                 db.session.commit()
@@ -168,7 +220,62 @@ class InsertExploreAppApi(Resource):
             for installed_app in installed_apps:
                 session.delete(installed_app)
 
+            trial_app = session.execute(
+                select(TrialApp).where(TrialApp.app_id == recommended_app.app_id)
+            ).scalar_one_or_none()
+            if trial_app:
+                session.delete(trial_app)
+
         db.session.delete(recommended_app)
         db.session.commit()
 
         return {"result": "success"}, 204
+
+
+@console_ns.route("/admin/insert-explore-banner")
+class InsertExploreBannerApi(Resource):
+    @console_ns.doc("insert_explore_banner")
+    @console_ns.doc(description="Insert an explore banner")
+    @console_ns.expect(console_ns.models[InsertExploreBannerPayload.__name__])
+    @console_ns.response(201, "Banner inserted successfully")
+    @only_edition_cloud
+    @admin_required
+    def post(self):
+        payload = InsertExploreBannerPayload.model_validate(console_ns.payload)
+
+        content = {
+            "category": payload.category,
+            "title": payload.title,
+            "description": payload.description,
+            "img-src": payload.img_src,
+        }
+
+        banner = ExporleBanner(
+            content=content,
+            link=payload.link,
+            sort=payload.sort,
+            language=payload.language,
+        )
+        db.session.add(banner)
+        db.session.commit()
+
+        return {"result": "success"}, 201
+
+
+@console_ns.route("/admin/insert-explore-banner/<uuid:banner_id>")
+class DeleteExploreBannerApi(Resource):
+    @console_ns.doc("delete_explore_banner")
+    @console_ns.doc(description="Delete an explore banner")
+    @console_ns.doc(params={"banner_id": "Banner ID to delete"})
+    @console_ns.response(204, "Banner deleted successfully")
+    @only_edition_cloud
+    @admin_required
+    def delete(self, banner_id):
+        banner = db.session.execute(select(ExporleBanner).where(ExporleBanner.id == banner_id)).scalar_one_or_none()
+        if not banner:
+            raise NotFound(f"Banner '{banner_id}' is not found")
+
+        db.session.delete(banner)
+        db.session.commit()
+
+        return {"result": "success"}, 204

api/controllers/console/app/annotation.py

@@ -272,6 +272,7 @@ class AnnotationExportApi(Resource):
     @account_initialization_required
     @edit_permission_required
     def get(self, app_id):
+
         app_id = str(app_id)
         annotation_list = AppAnnotationService.export_annotation_list_by_app_id(app_id)
         response_data = {"data": marshal(annotation_list, annotation_fields)}
@@ -359,7 +360,6 @@ class AnnotationBatchImportApi(Resource):
         file.seek(0, 2)  # Seek to end of file
         file_size = file.tell()
         file.seek(0)  # Reset to beginning
-
         max_size_bytes = dify_config.ANNOTATION_IMPORT_FILE_SIZE_LIMIT * 1024 * 1024
         if file_size > max_size_bytes:
             abort(

api/controllers/console/app/error.py

@@ -115,3 +115,9 @@ class InvokeRateLimitError(BaseHTTPException):
     error_code = "rate_limit_error"
     description = "Rate Limit Error"
     code = 429
+
+
+class NeedAddIdsError(BaseHTTPException):
+    error_code = "need_add_ids"
+    description = "Need to add ids."
+    code = 400

api/controllers/console/app/wraps.py

@@ -23,6 +23,11 @@ def _load_app_model(app_id: str) -> App | None:
     return app_model
 
 
+def _load_app_model_with_trial(app_id: str) -> App | None:
+    app_model = db.session.query(App).where(App.id == app_id, App.status == "normal").first()
+    return app_model
+
+
 def get_app_model(view: Callable[P, R] | None = None, *, mode: Union[AppMode, list[AppMode], None] = None):
     def decorator(view_func: Callable[P1, R1]):
         @wraps(view_func)
@@ -62,3 +67,44 @@ def get_app_model(view: Callable[P, R] | None = None, *, mode: Union[AppMode, li
         return decorator
     else:
         return decorator(view)
+
+
+def get_app_model_with_trial(view: Callable[P, R] | None = None, *, mode: Union[AppMode, list[AppMode], None] = None):
+    def decorator(view_func: Callable[P, R]):
+        @wraps(view_func)
+        def decorated_view(*args: P.args, **kwargs: P.kwargs):
+            if not kwargs.get("app_id"):
+                raise ValueError("missing app_id in path parameters")
+
+            app_id = kwargs.get("app_id")
+            app_id = str(app_id)
+
+            del kwargs["app_id"]
+
+            app_model = _load_app_model_with_trial(app_id)
+
+            if not app_model:
+                raise AppNotFoundError()
+
+            app_mode = AppMode.value_of(app_model.mode)
+
+            if mode is not None:
+                if isinstance(mode, list):
+                    modes = mode
+                else:
+                    modes = [mode]
+
+                if app_mode not in modes:
+                    mode_values = {m.value for m in modes}
+                    raise AppNotFoundError(f"App mode is not in the supported list: {mode_values}")
+
+            kwargs["app_model"] = app_model
+
+            return view_func(*args, **kwargs)
+
+        return decorated_view
+
+    if view is None:
+        return decorator
+    else:
+        return decorator(view)

api/controllers/console/auth/oauth.py

@@ -159,10 +159,7 @@ class OAuthCallback(Resource):
             ip_address=extract_remote_ip(request),
         )
 
-        base_url = dify_config.CONSOLE_WEB_URL
-        query_char = "&" if "?" in base_url else "?"
-        target_url = f"{base_url}{query_char}oauth_new_user={str(oauth_new_user).lower()}"
-        response = redirect(target_url)
+        response = redirect(f"{dify_config.CONSOLE_WEB_URL}?oauth_new_user={str(oauth_new_user).lower()}")
 
         set_access_token_to_cookie(request, response, token_pair.access_token)
         set_refresh_token_to_cookie(request, response, token_pair.refresh_token)

api/controllers/console/datasets/datasets.py

@@ -146,6 +146,7 @@ class DatasetUpdatePayload(BaseModel):
     embedding_model: str | None = None
     embedding_model_provider: str | None = None
     retrieval_model: dict[str, Any] | None = None
+    summary_index_setting: dict[str, Any] | None = None
     partial_member_list: list[dict[str, str]] | None = None
     external_retrieval_model: dict[str, Any] | None = None
     external_knowledge_id: str | None = None

api/controllers/console/datasets/datasets_document.py

@@ -39,9 +39,10 @@ from fields.document_fields import (
 from libs.datetime_utils import naive_utc_now
 from libs.login import current_account_with_tenant, login_required
 from models import DatasetProcessRule, Document, DocumentSegment, UploadFile
-from models.dataset import DocumentPipelineExecutionLog
+from models.dataset import DocumentPipelineExecutionLog, DocumentSegmentSummary
 from services.dataset_service import DatasetService, DocumentService
 from services.entities.knowledge_entities.knowledge_entities import KnowledgeConfig, ProcessRule, RetrievalModel
+from tasks.generate_summary_index_task import generate_summary_index_task
 
 from ..app.error import (
     ProviderModelCurrentlyNotSupportError,
@@ -104,6 +105,10 @@ class DocumentRenamePayload(BaseModel):
     name: str
 
 
+class GenerateSummaryPayload(BaseModel):
+    document_list: list[str]
+
+
 register_schema_models(
     console_ns,
     KnowledgeConfig,
@@ -111,6 +116,7 @@ register_schema_models(
     RetrievalModel,
     DocumentRetryPayload,
     DocumentRenamePayload,
+    GenerateSummaryPayload,
 )
 
 
@@ -295,6 +301,97 @@ class DatasetDocumentListApi(Resource):
 
         paginated_documents = db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)
         documents = paginated_documents.items
+        
+        # Check if dataset has summary index enabled
+        has_summary_index = (
+            dataset.summary_index_setting
+            and dataset.summary_index_setting.get("enable") is True
+        )
+        
+        # Filter documents that need summary calculation
+        documents_need_summary = [doc for doc in documents if doc.need_summary is True]
+        document_ids_need_summary = [str(doc.id) for doc in documents_need_summary]
+        
+        # Calculate summary_index_status for documents that need summary (only if dataset summary index is enabled)
+        summary_status_map = {}
+        if has_summary_index and document_ids_need_summary:
+            # Get all segments for these documents (excluding qa_model and re_segment)
+            segments = (
+                db.session.query(DocumentSegment.id, DocumentSegment.document_id)
+                .where(
+                    DocumentSegment.document_id.in_(document_ids_need_summary),
+                    DocumentSegment.status != "re_segment",
+                    DocumentSegment.tenant_id == current_tenant_id,
+                )
+                .all()
+            )
+            
+            # Group segments by document_id
+            document_segments_map = {}
+            for segment in segments:
+                doc_id = str(segment.document_id)
+                if doc_id not in document_segments_map:
+                    document_segments_map[doc_id] = []
+                document_segments_map[doc_id].append(segment.id)
+            
+            # Get all summary records for these segments
+            all_segment_ids = [seg.id for seg in segments]
+            summaries = {}
+            if all_segment_ids:
+                summary_records = (
+                    db.session.query(DocumentSegmentSummary)
+                    .where(
+                        DocumentSegmentSummary.chunk_id.in_(all_segment_ids),
+                        DocumentSegmentSummary.dataset_id == dataset_id,
+                        DocumentSegmentSummary.enabled == True,  # Only count enabled summaries
+                    )
+                    .all()
+                )
+                summaries = {summary.chunk_id: summary.status for summary in summary_records}
+            
+            # Calculate summary_index_status for each document
+            for doc_id in document_ids_need_summary:
+                segment_ids = document_segments_map.get(doc_id, [])
+                if not segment_ids:
+                    # No segments, status is "GENERATING" (waiting to generate)
+                    summary_status_map[doc_id] = "GENERATING"
+                    continue
+                
+                # Count summary statuses for this document's segments
+                status_counts = {"completed": 0, "generating": 0, "error": 0, "not_started": 0}
+                for segment_id in segment_ids:
+                    status = summaries.get(segment_id, "not_started")
+                    if status in status_counts:
+                        status_counts[status] += 1
+                    else:
+                        status_counts["not_started"] += 1
+                
+                total_segments = len(segment_ids)
+                completed_count = status_counts["completed"]
+                generating_count = status_counts["generating"]
+                error_count = status_counts["error"]
+                
+                # Determine overall status (only three states: GENERATING, COMPLETED, ERROR)
+                if completed_count == total_segments:
+                    summary_status_map[doc_id] = "COMPLETED"
+                elif error_count > 0:
+                    # Has errors (even if some are completed or generating)
+                    summary_status_map[doc_id] = "ERROR"
+                elif generating_count > 0 or status_counts["not_started"] > 0:
+                    # Still generating or not started
+                    summary_status_map[doc_id] = "GENERATING"
+                else:
+                    # Default to generating
+                    summary_status_map[doc_id] = "GENERATING"
+        
+        # Add summary_index_status to each document
+        for document in documents:
+            if has_summary_index and document.need_summary is True:
+                document.summary_index_status = summary_status_map.get(str(document.id), "GENERATING")
+            else:
+                # Return null if summary index is not enabled or document doesn't need summary
+                document.summary_index_status = None
+        
         if fetch:
             for document in documents:
                 completed_segments = (
@@ -393,6 +490,7 @@ class DatasetDocumentListApi(Resource):
         return {"result": "success"}, 204
 
 
+
 @console_ns.route("/datasets/init")
 class DatasetInitApi(Resource):
     @console_ns.doc("init_dataset")
@@ -780,6 +878,7 @@ class DocumentApi(DocumentResource):
                 "display_status": document.display_status,
                 "doc_form": document.doc_form,
                 "doc_language": document.doc_language,
+                "need_summary": document.need_summary if document.need_summary is not None else False,
             }
         else:
             dataset_process_rules = DatasetService.get_process_rules(dataset_id)
@@ -815,6 +914,7 @@ class DocumentApi(DocumentResource):
                 "display_status": document.display_status,
                 "doc_form": document.doc_form,
                 "doc_language": document.doc_language,
+                "need_summary": document.need_summary if document.need_summary is not None else False,
             }
 
         return response, 200
@@ -1182,3 +1282,211 @@ class DocumentPipelineExecutionLogApi(DocumentResource):
             "input_data": log.input_data,
             "datasource_node_id": log.datasource_node_id,
         }, 200
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/generate-summary")
+class DocumentGenerateSummaryApi(Resource):
+    @console_ns.doc("generate_summary_for_documents")
+    @console_ns.doc(description="Generate summary index for documents")
+    @console_ns.doc(params={"dataset_id": "Dataset ID"})
+    @console_ns.expect(console_ns.models[GenerateSummaryPayload.__name__])
+    @console_ns.response(200, "Summary generation started successfully")
+    @console_ns.response(400, "Invalid request or dataset configuration")
+    @console_ns.response(403, "Permission denied")
+    @console_ns.response(404, "Dataset not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @cloud_edition_billing_rate_limit_check("knowledge")
+    def post(self, dataset_id):
+        """
+        Generate summary index for specified documents.
+        
+        This endpoint checks if the dataset configuration supports summary generation
+        (indexing_technique must be 'high_quality' and summary_index_setting.enable must be true),
+        then asynchronously generates summary indexes for the provided documents.
+        """
+        current_user, _ = current_account_with_tenant()
+        dataset_id = str(dataset_id)
+        
+        # Get dataset
+        dataset = DatasetService.get_dataset(dataset_id)
+        if not dataset:
+            raise NotFound("Dataset not found.")
+        
+        # Check permissions
+        if not current_user.is_dataset_editor:
+            raise Forbidden()
+        
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+        
+        # Validate request payload
+        payload = GenerateSummaryPayload.model_validate(console_ns.payload or {})
+        document_list = payload.document_list
+        
+        if not document_list:
+            raise ValueError("document_list cannot be empty.")
+        
+        # Check if dataset configuration supports summary generation
+        if dataset.indexing_technique != "high_quality":
+            raise ValueError(
+                f"Summary generation is only available for 'high_quality' indexing technique. "
+                f"Current indexing technique: {dataset.indexing_technique}"
+            )
+        
+        summary_index_setting = dataset.summary_index_setting
+        if not summary_index_setting or not summary_index_setting.get("enable"):
+            raise ValueError(
+                "Summary index is not enabled for this dataset. "
+                "Please enable it in the dataset settings."
+            )
+        
+        # Verify all documents exist and belong to the dataset
+        documents = (
+            db.session.query(Document)
+            .filter(
+                Document.id.in_(document_list),
+                Document.dataset_id == dataset_id,
+            )
+            .all()
+        )
+        
+        if len(documents) != len(document_list):
+            found_ids = {doc.id for doc in documents}
+            missing_ids = set(document_list) - found_ids
+            raise NotFound(f"Some documents not found: {list(missing_ids)}")
+        
+        # Dispatch async tasks for each document
+        for document in documents:
+            # Skip qa_model documents as they don't generate summaries
+            if document.doc_form == "qa_model":
+                logger.info(
+                    f"Skipping summary generation for qa_model document {document.id}"
+                )
+                continue
+            
+            # Dispatch async task
+            generate_summary_index_task(dataset_id, document.id)
+            logger.info(
+                f"Dispatched summary generation task for document {document.id} in dataset {dataset_id}"
+            )
+        
+        return {"result": "success"}, 200
+
+
+@console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/summary-status")
+class DocumentSummaryStatusApi(DocumentResource):
+    @console_ns.doc("get_document_summary_status")
+    @console_ns.doc(description="Get summary index generation status for a document")
+    @console_ns.doc(params={"dataset_id": "Dataset ID", "document_id": "Document ID"})
+    @console_ns.response(200, "Summary status retrieved successfully")
+    @console_ns.response(404, "Document not found")
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def get(self, dataset_id, document_id):
+        """
+        Get summary index generation status for a document.
+        
+        Returns:
+        - total_segments: Total number of segments in the document
+        - summary_status: Dictionary with status counts
+          - completed: Number of summaries completed
+          - generating: Number of summaries being generated
+          - error: Number of summaries with errors
+          - not_started: Number of segments without summary records
+        - summaries: List of summary records with status and content preview
+        """
+        current_user, _ = current_account_with_tenant()
+        dataset_id = str(dataset_id)
+        document_id = str(document_id)
+        
+        # Get document
+        document = self.get_document(dataset_id, document_id)
+        
+        # Get dataset
+        dataset = DatasetService.get_dataset(dataset_id)
+        if not dataset:
+            raise NotFound("Dataset not found.")
+        
+        # Check permissions
+        try:
+            DatasetService.check_dataset_permission(dataset, current_user)
+        except services.errors.account.NoPermissionError as e:
+            raise Forbidden(str(e))
+        
+        # Get all segments for this document
+        segments = (
+            db.session.query(DocumentSegment)
+            .filter(
+                DocumentSegment.document_id == document_id,
+                DocumentSegment.dataset_id == dataset_id,
+                DocumentSegment.status == "completed",
+                DocumentSegment.enabled == True,
+            )
+            .all()
+        )
+        
+        total_segments = len(segments)
+        
+        # Get all summary records for these segments
+        segment_ids = [segment.id for segment in segments]
+        summaries = []
+        if segment_ids:
+            summaries = (
+                db.session.query(DocumentSegmentSummary)
+                .filter(
+                    DocumentSegmentSummary.document_id == document_id,
+                    DocumentSegmentSummary.dataset_id == dataset_id,
+                    DocumentSegmentSummary.chunk_id.in_(segment_ids),
+                    DocumentSegmentSummary.enabled == True,  # Only return enabled summaries
+                )
+                .all()
+            )
+        
+        # Create a mapping of chunk_id to summary
+        summary_map = {summary.chunk_id: summary for summary in summaries}
+        
+        # Count statuses
+        status_counts = {
+            "completed": 0,
+            "generating": 0,
+            "error": 0,
+            "not_started": 0,
+        }
+        
+        summary_list = []
+        for segment in segments:
+            summary = summary_map.get(segment.id)
+            if summary:
+                status = summary.status
+                status_counts[status] = status_counts.get(status, 0) + 1
+                summary_list.append({
+                    "segment_id": segment.id,
+                    "segment_position": segment.position,
+                    "status": summary.status,
+                    "summary_preview": summary.summary_content[:100] + "..." if summary.summary_content and len(summary.summary_content) > 100 else summary.summary_content,
+                    "error": summary.error,
+                    "created_at": int(summary.created_at.timestamp()) if summary.created_at else None,
+                    "updated_at": int(summary.updated_at.timestamp()) if summary.updated_at else None,
+                })
+            else:
+                status_counts["not_started"] += 1
+                summary_list.append({
+                    "segment_id": segment.id,
+                    "segment_position": segment.position,
+                    "status": "not_started",
+                    "summary_preview": None,
+                    "error": None,
+                    "created_at": None,
+                    "updated_at": None,
+                })
+        
+        return {
+            "total_segments": total_segments,
+            "summary_status": status_counts,
+            "summaries": summary_list,
+        }, 200

api/controllers/console/datasets/datasets_segments.py

@@ -32,7 +32,7 @@ from extensions.ext_redis import redis_client
 from fields.segment_fields import child_chunk_fields, segment_fields
 from libs.helper import escape_like_pattern
 from libs.login import current_account_with_tenant, login_required
-from models.dataset import ChildChunk, DocumentSegment
+from models.dataset import ChildChunk, DocumentSegment, DocumentSegmentSummary
 from models.model import UploadFile
 from services.dataset_service import DatasetService, DocumentService, SegmentService
 from services.entities.knowledge_entities.knowledge_entities import ChildChunkUpdateArgs, SegmentUpdateArgs
@@ -41,6 +41,23 @@ from services.errors.chunk import ChildChunkIndexingError as ChildChunkIndexingS
 from tasks.batch_create_segment_to_index_task import batch_create_segment_to_index_task
 
 
+def _get_segment_with_summary(segment, dataset_id):
+    """Helper function to marshal segment and add summary information."""
+    segment_dict = marshal(segment, segment_fields)
+    # Query summary for this segment (only enabled summaries)
+    summary = (
+        db.session.query(DocumentSegmentSummary)
+        .where(
+            DocumentSegmentSummary.chunk_id == segment.id,
+            DocumentSegmentSummary.dataset_id == dataset_id,
+            DocumentSegmentSummary.enabled == True,  # Only return enabled summaries
+        )
+        .first()
+    )
+    segment_dict["summary"] = summary.summary_content if summary else None
+    return segment_dict
+
+
 class SegmentListQuery(BaseModel):
     limit: int = Field(default=20, ge=1, le=100)
     status: list[str] = Field(default_factory=list)
@@ -63,6 +80,7 @@ class SegmentUpdatePayload(BaseModel):
     keywords: list[str] | None = None
     regenerate_child_chunks: bool = False
     attachment_ids: list[str] | None = None
+    summary: str | None = None  # Summary content for summary index
 
 
 class BatchImportPayload(BaseModel):
@@ -180,8 +198,34 @@ class DatasetDocumentSegmentListApi(Resource):
 
         segments = db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)
 
+        # Query summaries for all segments in this page (batch query for efficiency)
+        segment_ids = [segment.id for segment in segments.items]
+        summaries = {}
+        if segment_ids:
+            summary_records = (
+                db.session.query(DocumentSegmentSummary)
+                .where(
+                    DocumentSegmentSummary.chunk_id.in_(segment_ids),
+                    DocumentSegmentSummary.dataset_id == dataset_id,
+                )
+                .all()
+            )
+            # Only include enabled summaries
+            summaries = {
+                summary.chunk_id: summary.summary_content 
+                for summary in summary_records 
+                if summary.enabled is True
+            }
+
+        # Add summary to each segment
+        segments_with_summary = []
+        for segment in segments.items:
+            segment_dict = marshal(segment, segment_fields)
+            segment_dict["summary"] = summaries.get(segment.id)
+            segments_with_summary.append(segment_dict)
+
         response = {
-            "data": marshal(segments.items, segment_fields),
+            "data": segments_with_summary,
             "limit": limit,
             "total": segments.total,
             "total_pages": segments.pages,
@@ -327,7 +371,7 @@ class DatasetDocumentSegmentAddApi(Resource):
         payload_dict = payload.model_dump(exclude_none=True)
         SegmentService.segment_create_args_validate(payload_dict, document)
         segment = SegmentService.create_segment(payload_dict, document, dataset)
-        return {"data": marshal(segment, segment_fields), "doc_form": document.doc_form}, 200
+        return {"data": _get_segment_with_summary(segment, dataset_id), "doc_form": document.doc_form}, 200
 
 
 @console_ns.route("/datasets/<uuid:dataset_id>/documents/<uuid:document_id>/segments/<uuid:segment_id>")
@@ -389,10 +433,12 @@ class DatasetDocumentSegmentUpdateApi(Resource):
         payload = SegmentUpdatePayload.model_validate(console_ns.payload or {})
         payload_dict = payload.model_dump(exclude_none=True)
         SegmentService.segment_create_args_validate(payload_dict, document)
+        
+        # Update segment (summary update with change detection is handled in SegmentService.update_segment)
         segment = SegmentService.update_segment(
             SegmentUpdateArgs.model_validate(payload.model_dump(exclude_none=True)), segment, document, dataset
         )
-        return {"data": marshal(segment, segment_fields), "doc_form": document.doc_form}, 200
+        return {"data": _get_segment_with_summary(segment, dataset_id), "doc_form": document.doc_form}, 200
 
     @setup_required
     @login_required

api/controllers/console/datasets/hit_testing.py

@@ -1,4 +1,4 @@
-from flask_restx import Resource
+from flask_restx import Resource, fields
 
 from controllers.common.schema import register_schema_model
 from libs.login import login_required
@@ -10,17 +10,56 @@ from ..wraps import (
     cloud_edition_billing_rate_limit_check,
     setup_required,
 )
+from fields.hit_testing_fields import (
+    child_chunk_fields,
+    document_fields,
+    files_fields,
+    hit_testing_record_fields,
+    segment_fields,
+)
 
 register_schema_model(console_ns, HitTestingPayload)
 
 
+def _get_or_create_model(model_name: str, field_def):
+    """Get or create a flask_restx model to avoid dict type issues in Swagger."""
+    existing = console_ns.models.get(model_name)
+    if existing is None:
+        existing = console_ns.model(model_name, field_def)
+    return existing
+
+
+# Register models for flask_restx to avoid dict type issues in Swagger
+document_model = _get_or_create_model("HitTestingDocument", document_fields)
+
+segment_fields_copy = segment_fields.copy()
+segment_fields_copy["document"] = fields.Nested(document_model)
+segment_model = _get_or_create_model("HitTestingSegment", segment_fields_copy)
+
+child_chunk_model = _get_or_create_model("HitTestingChildChunk", child_chunk_fields)
+files_model = _get_or_create_model("HitTestingFile", files_fields)
+
+hit_testing_record_fields_copy = hit_testing_record_fields.copy()
+hit_testing_record_fields_copy["segment"] = fields.Nested(segment_model)
+hit_testing_record_fields_copy["child_chunks"] = fields.List(fields.Nested(child_chunk_model))
+hit_testing_record_fields_copy["files"] = fields.List(fields.Nested(files_model))
+hit_testing_record_model = _get_or_create_model("HitTestingRecord", hit_testing_record_fields_copy)
+
+# Response model for hit testing API
+hit_testing_response_fields = {
+    "query": fields.String,
+    "records": fields.List(fields.Nested(hit_testing_record_model)),
+}
+hit_testing_response_model = _get_or_create_model("HitTestingResponse", hit_testing_response_fields)
+
+
 @console_ns.route("/datasets/<uuid:dataset_id>/hit-testing")
 class HitTestingApi(Resource, DatasetsHitTestingBase):
     @console_ns.doc("test_dataset_retrieval")
     @console_ns.doc(description="Test dataset knowledge retrieval")
     @console_ns.doc(params={"dataset_id": "Dataset ID"})
     @console_ns.expect(console_ns.models[HitTestingPayload.__name__])
-    @console_ns.response(200, "Hit testing completed successfully")
+    @console_ns.response(200, "Hit testing completed successfully", model=hit_testing_response_model)
     @console_ns.response(404, "Dataset not found")
     @console_ns.response(400, "Invalid parameters")
     @setup_required

api/controllers/console/explore/banner.py

@@ -0,0 +1,43 @@
+from flask import request
+from flask_restx import Resource
+
+from controllers.console import api
+from controllers.console.explore.wraps import explore_banner_enabled
+from extensions.ext_database import db
+from models.model import ExporleBanner
+
+
+class BannerApi(Resource):
+    """Resource for banner list."""
+
+    @explore_banner_enabled
+    def get(self):
+        """Get banner list."""
+        language = request.args.get("language", "en-US")
+
+        # Build base query for enabled banners
+        base_query = db.session.query(ExporleBanner).where(ExporleBanner.status == "enabled")
+
+        # Try to get banners in the requested language
+        banners = base_query.where(ExporleBanner.language == language).order_by(ExporleBanner.sort).all()
+
+        # Fallback to en-US if no banners found and language is not en-US
+        if not banners and language != "en-US":
+            banners = base_query.where(ExporleBanner.language == "en-US").order_by(ExporleBanner.sort).all()
+        # Convert banners to serializable format
+        result = []
+        for banner in banners:
+            banner_data = {
+                "id": banner.id,
+                "content": banner.content,  # Already parsed as JSON by SQLAlchemy
+                "link": banner.link,
+                "sort": banner.sort,
+                "status": banner.status,
+                "created_at": banner.created_at.isoformat() if banner.created_at else None,
+            }
+            result.append(banner_data)
+
+        return result
+
+
+api.add_resource(BannerApi, "/explore/banners")

api/controllers/console/explore/error.py

@@ -29,3 +29,25 @@ class AppAccessDeniedError(BaseHTTPException):
     error_code = "access_denied"
     description = "App access denied."
     code = 403
+
+
+class TrialAppNotAllowed(BaseHTTPException):
+    """*403* `Trial App Not Allowed`
+
+    Raise if the user has reached the trial app limit.
+    """
+
+    error_code = "trial_app_not_allowed"
+    code = 403
+    description = "the app is not allowed to be trial."
+
+
+class TrialAppLimitExceeded(BaseHTTPException):
+    """*403* `Trial App Limit Exceeded`
+
+    Raise if the user has exceeded the trial app limit.
+    """
+
+    error_code = "trial_app_limit_exceeded"
+    code = 403
+    description = "The user has exceeded the trial app limit."

api/controllers/console/explore/recommended_app.py

@@ -29,6 +29,7 @@ recommended_app_fields = {
     "category": fields.String,
     "position": fields.Integer,
     "is_listed": fields.Boolean,
+    "can_trial": fields.Boolean,
 }
 
 recommended_app_list_fields = {

api/controllers/console/explore/trial.py

@@ -0,0 +1,512 @@
+import logging
+from typing import Any, cast
+
+from flask import request
+from flask_restx import Resource, marshal, marshal_with, reqparse
+from werkzeug.exceptions import Forbidden, InternalServerError, NotFound
+
+import services
+from controllers.common.fields import Parameters as ParametersResponse
+from controllers.common.fields import Site as SiteResponse
+from controllers.console import api
+from controllers.console.app.error import (
+    AppUnavailableError,
+    AudioTooLargeError,
+    CompletionRequestError,
+    ConversationCompletedError,
+    NeedAddIdsError,
+    NoAudioUploadedError,
+    ProviderModelCurrentlyNotSupportError,
+    ProviderNotInitializeError,
+    ProviderNotSupportSpeechToTextError,
+    ProviderQuotaExceededError,
+    UnsupportedAudioTypeError,
+)
+from controllers.console.app.wraps import get_app_model_with_trial
+from controllers.console.explore.error import (
+    AppSuggestedQuestionsAfterAnswerDisabledError,
+    NotChatAppError,
+    NotCompletionAppError,
+    NotWorkflowAppError,
+)
+from controllers.console.explore.wraps import TrialAppResource, trial_feature_enable
+from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
+from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
+from core.app.apps.base_app_queue_manager import AppQueueManager
+from core.app.entities.app_invoke_entities import InvokeFrom
+from core.errors.error import (
+    ModelCurrentlyNotSupportError,
+    ProviderTokenNotInitError,
+    QuotaExceededError,
+)
+from core.model_runtime.errors.invoke import InvokeError
+from core.workflow.graph_engine.manager import GraphEngineManager
+from extensions.ext_database import db
+from fields.app_fields import app_detail_fields_with_site
+from fields.dataset_fields import dataset_fields
+from fields.workflow_fields import workflow_fields
+from libs import helper
+from libs.helper import uuid_value
+from libs.login import current_user
+from models import Account
+from models.account import TenantStatus
+from models.model import AppMode, Site
+from models.workflow import Workflow
+from services.app_generate_service import AppGenerateService
+from services.app_service import AppService
+from services.audio_service import AudioService
+from services.dataset_service import DatasetService
+from services.errors.audio import (
+    AudioTooLargeServiceError,
+    NoAudioUploadedServiceError,
+    ProviderNotSupportSpeechToTextServiceError,
+    UnsupportedAudioTypeServiceError,
+)
+from services.errors.conversation import ConversationNotExistsError
+from services.errors.llm import InvokeRateLimitError
+from services.errors.message import (
+    MessageNotExistsError,
+    SuggestedQuestionsAfterAnswerDisabledError,
+)
+from services.message_service import MessageService
+from services.recommended_app_service import RecommendedAppService
+
+logger = logging.getLogger(__name__)
+
+
+class TrialAppWorkflowRunApi(TrialAppResource):
+    def post(self, trial_app):
+        """
+        Run workflow
+        """
+        app_model = trial_app
+        if not app_model:
+            raise NotWorkflowAppError()
+        app_mode = AppMode.value_of(app_model.mode)
+        if app_mode != AppMode.WORKFLOW:
+            raise NotWorkflowAppError()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, nullable=False, location="json")
+        parser.add_argument("files", type=list, required=False, location="json")
+        args = parser.parse_args()
+        assert current_user is not None
+        try:
+            app_id = app_model.id
+            user_id = current_user.id
+            response = AppGenerateService.generate(
+                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=True
+            )
+            RecommendedAppService.add_trial_app_record(app_id, user_id)
+            return helper.compact_generate_response(response)
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+        except InvokeRateLimitError as ex:
+            raise InvokeRateLimitHttpError(ex.description)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+class TrialAppWorkflowTaskStopApi(TrialAppResource):
+    def post(self, trial_app, task_id: str):
+        """
+        Stop workflow task
+        """
+        app_model = trial_app
+        if not app_model:
+            raise NotWorkflowAppError()
+        app_mode = AppMode.value_of(app_model.mode)
+        if app_mode != AppMode.WORKFLOW:
+            raise NotWorkflowAppError()
+        assert current_user is not None
+
+        # Stop using both mechanisms for backward compatibility
+        # Legacy stop flag mechanism (without user check)
+        AppQueueManager.set_stop_flag_no_user_check(task_id)
+
+        # New graph engine command channel mechanism
+        GraphEngineManager.send_stop_command(task_id)
+
+        return {"result": "success"}
+
+
+class TrialChatApi(TrialAppResource):
+    @trial_feature_enable
+    def post(self, trial_app):
+        app_model = trial_app
+        app_mode = AppMode.value_of(app_model.mode)
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
+            raise NotChatAppError()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, location="json")
+        parser.add_argument("query", type=str, required=True, location="json")
+        parser.add_argument("files", type=list, required=False, location="json")
+        parser.add_argument("conversation_id", type=uuid_value, location="json")
+        parser.add_argument("parent_message_id", type=uuid_value, required=False, location="json")
+        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
+        args = parser.parse_args()
+
+        args["auto_generate_name"] = False
+
+        try:
+            if not isinstance(current_user, Account):
+                raise ValueError("current_user must be an Account instance")
+
+            # Get IDs before they might be detached from session
+            app_id = app_model.id
+            user_id = current_user.id
+
+            response = AppGenerateService.generate(
+                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=True
+            )
+            RecommendedAppService.add_trial_app_record(app_id, user_id)
+            return helper.compact_generate_response(response)
+        except services.errors.conversation.ConversationNotExistsError:
+            raise NotFound("Conversation Not Exists.")
+        except services.errors.conversation.ConversationCompletedError:
+            raise ConversationCompletedError()
+        except services.errors.app_model_config.AppModelConfigBrokenError:
+            logger.exception("App model config broken.")
+            raise AppUnavailableError()
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+        except InvokeRateLimitError as ex:
+            raise InvokeRateLimitHttpError(ex.description)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+class TrialMessageSuggestedQuestionApi(TrialAppResource):
+    @trial_feature_enable
+    def get(self, trial_app, message_id):
+        app_model = trial_app
+        app_mode = AppMode.value_of(app_model.mode)
+        if app_mode not in {AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT}:
+            raise NotChatAppError()
+
+        message_id = str(message_id)
+
+        try:
+            if not isinstance(current_user, Account):
+                raise ValueError("current_user must be an Account instance")
+            questions = MessageService.get_suggested_questions_after_answer(
+                app_model=app_model, user=current_user, message_id=message_id, invoke_from=InvokeFrom.EXPLORE
+            )
+        except MessageNotExistsError:
+            raise NotFound("Message not found")
+        except ConversationNotExistsError:
+            raise NotFound("Conversation not found")
+        except SuggestedQuestionsAfterAnswerDisabledError:
+            raise AppSuggestedQuestionsAfterAnswerDisabledError()
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+        return {"data": questions}
+
+
+class TrialChatAudioApi(TrialAppResource):
+    @trial_feature_enable
+    def post(self, trial_app):
+        app_model = trial_app
+
+        file = request.files["file"]
+
+        try:
+            if not isinstance(current_user, Account):
+                raise ValueError("current_user must be an Account instance")
+
+            # Get IDs before they might be detached from session
+            app_id = app_model.id
+            user_id = current_user.id
+
+            response = AudioService.transcript_asr(app_model=app_model, file=file, end_user=None)
+            RecommendedAppService.add_trial_app_record(app_id, user_id)
+            return response
+        except services.errors.app_model_config.AppModelConfigBrokenError:
+            logger.exception("App model config broken.")
+            raise AppUnavailableError()
+        except NoAudioUploadedServiceError:
+            raise NoAudioUploadedError()
+        except AudioTooLargeServiceError as e:
+            raise AudioTooLargeError(str(e))
+        except UnsupportedAudioTypeServiceError:
+            raise UnsupportedAudioTypeError()
+        except ProviderNotSupportSpeechToTextServiceError:
+            raise ProviderNotSupportSpeechToTextError()
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+        except ValueError as e:
+            raise e
+        except Exception as e:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+class TrialChatTextApi(TrialAppResource):
+    @trial_feature_enable
+    def post(self, trial_app):
+        app_model = trial_app
+        try:
+            parser = reqparse.RequestParser()
+            parser.add_argument("message_id", type=str, required=False, location="json")
+            parser.add_argument("voice", type=str, location="json")
+            parser.add_argument("text", type=str, location="json")
+            parser.add_argument("streaming", type=bool, location="json")
+            args = parser.parse_args()
+
+            message_id = args.get("message_id", None)
+            text = args.get("text", None)
+            voice = args.get("voice", None)
+            if not isinstance(current_user, Account):
+                raise ValueError("current_user must be an Account instance")
+
+            # Get IDs before they might be detached from session
+            app_id = app_model.id
+            user_id = current_user.id
+
+            response = AudioService.transcript_tts(app_model=app_model, text=text, voice=voice, message_id=message_id)
+            RecommendedAppService.add_trial_app_record(app_id, user_id)
+            return response
+        except services.errors.app_model_config.AppModelConfigBrokenError:
+            logger.exception("App model config broken.")
+            raise AppUnavailableError()
+        except NoAudioUploadedServiceError:
+            raise NoAudioUploadedError()
+        except AudioTooLargeServiceError as e:
+            raise AudioTooLargeError(str(e))
+        except UnsupportedAudioTypeServiceError:
+            raise UnsupportedAudioTypeError()
+        except ProviderNotSupportSpeechToTextServiceError:
+            raise ProviderNotSupportSpeechToTextError()
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+        except ValueError as e:
+            raise e
+        except Exception as e:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+class TrialCompletionApi(TrialAppResource):
+    @trial_feature_enable
+    def post(self, trial_app):
+        app_model = trial_app
+        if app_model.mode != "completion":
+            raise NotCompletionAppError()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("inputs", type=dict, required=True, location="json")
+        parser.add_argument("query", type=str, location="json", default="")
+        parser.add_argument("files", type=list, required=False, location="json")
+        parser.add_argument("response_mode", type=str, choices=["blocking", "streaming"], location="json")
+        parser.add_argument("retriever_from", type=str, required=False, default="explore_app", location="json")
+        args = parser.parse_args()
+
+        streaming = args["response_mode"] == "streaming"
+        args["auto_generate_name"] = False
+
+        try:
+            if not isinstance(current_user, Account):
+                raise ValueError("current_user must be an Account instance")
+
+            # Get IDs before they might be detached from session
+            app_id = app_model.id
+            user_id = current_user.id
+
+            response = AppGenerateService.generate(
+                app_model=app_model, user=current_user, args=args, invoke_from=InvokeFrom.EXPLORE, streaming=streaming
+            )
+
+            RecommendedAppService.add_trial_app_record(app_id, user_id)
+            return helper.compact_generate_response(response)
+        except services.errors.conversation.ConversationNotExistsError:
+            raise NotFound("Conversation Not Exists.")
+        except services.errors.conversation.ConversationCompletedError:
+            raise ConversationCompletedError()
+        except services.errors.app_model_config.AppModelConfigBrokenError:
+            logger.exception("App model config broken.")
+            raise AppUnavailableError()
+        except ProviderTokenNotInitError as ex:
+            raise ProviderNotInitializeError(ex.description)
+        except QuotaExceededError:
+            raise ProviderQuotaExceededError()
+        except ModelCurrentlyNotSupportError:
+            raise ProviderModelCurrentlyNotSupportError()
+        except InvokeError as e:
+            raise CompletionRequestError(e.description)
+        except ValueError as e:
+            raise e
+        except Exception:
+            logger.exception("internal server error.")
+            raise InternalServerError()
+
+
+class TrialSitApi(Resource):
+    """Resource for trial app sites."""
+
+    @trial_feature_enable
+    @get_app_model_with_trial
+    def get(self, app_model):
+        """Retrieve app site info.
+
+        Returns the site configuration for the application including theme, icons, and text.
+        """
+        site = db.session.query(Site).where(Site.app_id == app_model.id).first()
+
+        if not site:
+            raise Forbidden()
+
+        assert app_model.tenant
+        if app_model.tenant.status == TenantStatus.ARCHIVE:
+            raise Forbidden()
+
+        return SiteResponse.model_validate(site).model_dump(mode="json")
+
+
+class TrialAppParameterApi(Resource):
+    """Resource for app variables."""
+
+    @trial_feature_enable
+    @get_app_model_with_trial
+    def get(self, app_model):
+        """Retrieve app parameters."""
+
+        if app_model is None:
+            raise AppUnavailableError()
+
+        if app_model.mode in {AppMode.ADVANCED_CHAT, AppMode.WORKFLOW}:
+            workflow = app_model.workflow
+            if workflow is None:
+                raise AppUnavailableError()
+
+            features_dict = workflow.features_dict
+            user_input_form = workflow.user_input_form(to_old_structure=True)
+        else:
+            app_model_config = app_model.app_model_config
+            if app_model_config is None:
+                raise AppUnavailableError()
+
+            features_dict = app_model_config.to_dict()
+
+            user_input_form = features_dict.get("user_input_form", [])
+
+        parameters = get_parameters_from_feature_dict(features_dict=features_dict, user_input_form=user_input_form)
+        return ParametersResponse.model_validate(parameters).model_dump(mode="json")
+
+
+class AppApi(Resource):
+    @trial_feature_enable
+    @get_app_model_with_trial
+    @marshal_with(app_detail_fields_with_site)
+    def get(self, app_model):
+        """Get app detail"""
+
+        app_service = AppService()
+        app_model = app_service.get_app(app_model)
+
+        return app_model
+
+
+class AppWorkflowApi(Resource):
+    @trial_feature_enable
+    @get_app_model_with_trial
+    @marshal_with(workflow_fields)
+    def get(self, app_model):
+        """Get workflow detail"""
+        if not app_model.workflow_id:
+            raise AppUnavailableError()
+
+        workflow = (
+            db.session.query(Workflow)
+            .where(
+                Workflow.id == app_model.workflow_id,
+            )
+            .first()
+        )
+        return workflow
+
+
+class DatasetListApi(Resource):
+    @trial_feature_enable
+    @get_app_model_with_trial
+    def get(self, app_model):
+        page = request.args.get("page", default=1, type=int)
+        limit = request.args.get("limit", default=20, type=int)
+        ids = request.args.getlist("ids")
+
+        tenant_id = app_model.tenant_id
+        if ids:
+            datasets, total = DatasetService.get_datasets_by_ids(ids, tenant_id)
+        else:
+            raise NeedAddIdsError()
+
+        data = cast(list[dict[str, Any]], marshal(datasets, dataset_fields))
+
+        response = {"data": data, "has_more": len(datasets) == limit, "limit": limit, "total": total, "page": page}
+        return response
+
+
+api.add_resource(TrialChatApi, "/trial-apps/<uuid:app_id>/chat-messages", endpoint="trial_app_chat_completion")
+
+api.add_resource(
+    TrialMessageSuggestedQuestionApi,
+    "/trial-apps/<uuid:app_id>/messages/<uuid:message_id>/suggested-questions",
+    endpoint="trial_app_suggested_question",
+)
+
+api.add_resource(TrialChatAudioApi, "/trial-apps/<uuid:app_id>/audio-to-text", endpoint="trial_app_audio")
+api.add_resource(TrialChatTextApi, "/trial-apps/<uuid:app_id>/text-to-audio", endpoint="trial_app_text")
+
+api.add_resource(TrialCompletionApi, "/trial-apps/<uuid:app_id>/completion-messages", endpoint="trial_app_completion")
+
+api.add_resource(TrialSitApi, "/trial-apps/<uuid:app_id>/site")
+
+api.add_resource(TrialAppParameterApi, "/trial-apps/<uuid:app_id>/parameters", endpoint="trial_app_parameters")
+
+api.add_resource(AppApi, "/trial-apps/<uuid:app_id>", endpoint="trial_app")
+
+api.add_resource(TrialAppWorkflowRunApi, "/trial-apps/<uuid:app_id>/workflows/run", endpoint="trial_app_workflow_run")
+api.add_resource(TrialAppWorkflowTaskStopApi, "/trial-apps/<uuid:app_id>/workflows/tasks/<string:task_id>/stop")
+
+api.add_resource(AppWorkflowApi, "/trial-apps/<uuid:app_id>/workflows", endpoint="trial_app_workflow")
+api.add_resource(DatasetListApi, "/trial-apps/<uuid:app_id>/datasets", endpoint="trial_app_datasets")

api/controllers/console/explore/wraps.py

@@ -2,14 +2,15 @@ from collections.abc import Callable
 from functools import wraps
 from typing import Concatenate, ParamSpec, TypeVar
 
+from flask import abort
 from flask_restx import Resource
 from werkzeug.exceptions import NotFound
 
-from controllers.console.explore.error import AppAccessDeniedError
+from controllers.console.explore.error import AppAccessDeniedError, TrialAppLimitExceeded, TrialAppNotAllowed
 from controllers.console.wraps import account_initialization_required
 from extensions.ext_database import db
 from libs.login import current_account_with_tenant, login_required
-from models import InstalledApp
+from models import AccountTrialAppRecord, App, InstalledApp, TrialApp
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
 
@@ -71,6 +72,61 @@ def user_allowed_to_access_app(view: Callable[Concatenate[InstalledApp, P], R] |
     return decorator
 
 
+def trial_app_required(view: Callable[Concatenate[App, P], R] | None = None):
+    def decorator(view: Callable[Concatenate[App, P], R]):
+        @wraps(view)
+        def decorated(app_id: str, *args: P.args, **kwargs: P.kwargs):
+            current_user, _ = current_account_with_tenant()
+
+            trial_app = db.session.query(TrialApp).where(TrialApp.app_id == str(app_id)).first()
+
+            if trial_app is None:
+                raise TrialAppNotAllowed()
+            app = trial_app.app
+
+            if app is None:
+                raise TrialAppNotAllowed()
+
+            account_trial_app_record = (
+                db.session.query(AccountTrialAppRecord)
+                .where(AccountTrialAppRecord.account_id == current_user.id, AccountTrialAppRecord.app_id == app_id)
+                .first()
+            )
+            if account_trial_app_record:
+                if account_trial_app_record.count >= trial_app.trial_limit:
+                    raise TrialAppLimitExceeded()
+
+            return view(app, *args, **kwargs)
+
+        return decorated
+
+    if view:
+        return decorator(view)
+    return decorator
+
+
+def trial_feature_enable(view: Callable[..., R]) -> Callable[..., R]:
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        features = FeatureService.get_system_features()
+        if not features.enable_trial_app:
+            abort(403, "Trial app feature is not enabled.")
+        return view(*args, **kwargs)
+
+    return decorated
+
+
+def explore_banner_enabled(view: Callable[..., R]) -> Callable[..., R]:
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        features = FeatureService.get_system_features()
+        if not features.enable_explore_banner:
+            abort(403, "Explore banner feature is not enabled.")
+        return view(*args, **kwargs)
+
+    return decorated
+
+
 class InstalledAppResource(Resource):
     # must be reversed if there are multiple decorators
 
@@ -80,3 +136,13 @@ class InstalledAppResource(Resource):
         account_initialization_required,
         login_required,
     ]
+
+
+class TrialAppResource(Resource):
+    # must be reversed if there are multiple decorators
+
+    method_decorators = [
+        trial_app_required,
+        account_initialization_required,
+        login_required,
+    ]

api/controllers/console/workspace/trigger_providers.py

@@ -1,14 +1,14 @@
 import logging
-from collections.abc import Mapping
 from typing import Any
 
 from flask import make_response, redirect, request
-from flask_restx import Resource, reqparse
-from pydantic import BaseModel, Field, model_validator
+from flask_restx import Resource
+from pydantic import BaseModel, model_validator
 from sqlalchemy.orm import Session
 from werkzeug.exceptions import BadRequest, Forbidden
 
 from configs import dify_config
+from controllers.common.schema import register_schema_models
 from controllers.web.error import NotFoundError
 from core.model_runtime.utils.encoders import jsonable_encoder
 from core.plugin.entities.plugin_daemon import CredentialType
@@ -35,35 +35,38 @@ from ..wraps import (
 logger = logging.getLogger(__name__)
 
 
-class TriggerSubscriptionUpdateRequest(BaseModel):
-    """Request payload for updating a trigger subscription"""
+class TriggerSubscriptionBuilderCreatePayload(BaseModel):
+    credential_type: str = CredentialType.UNAUTHORIZED
 
-    name: str | None = Field(default=None, description="The name for the subscription")
-    credentials: Mapping[str, Any] | None = Field(default=None, description="The credentials for the subscription")
-    parameters: Mapping[str, Any] | None = Field(default=None, description="The parameters for the subscription")
-    properties: Mapping[str, Any] | None = Field(default=None, description="The properties for the subscription")
+
+class TriggerSubscriptionBuilderVerifyPayload(BaseModel):
+    credentials: dict[str, Any]
+
+
+class TriggerSubscriptionBuilderUpdatePayload(BaseModel):
+    name: str | None = None
+    parameters: dict[str, Any] | None = None
+    properties: dict[str, Any] | None = None
+    credentials: dict[str, Any] | None = None
 
     @model_validator(mode="after")
     def check_at_least_one_field(self):
-        if all(v is None for v in (self.name, self.credentials, self.parameters, self.properties)):
+        if all(v is None for v in self.model_dump().values()):
             raise ValueError("At least one of name, credentials, parameters, or properties must be provided")
         return self
 
 
-class TriggerSubscriptionVerifyRequest(BaseModel):
-    """Request payload for verifying subscription credentials."""
-
-    credentials: Mapping[str, Any] = Field(description="The credentials to verify")
+class TriggerOAuthClientPayload(BaseModel):
+    client_params: dict[str, Any] | None = None
+    enabled: bool | None = None
 
 
-console_ns.schema_model(
-    TriggerSubscriptionUpdateRequest.__name__,
-    TriggerSubscriptionUpdateRequest.model_json_schema(ref_template="#/definitions/{model}"),
-)
-
-console_ns.schema_model(
-    TriggerSubscriptionVerifyRequest.__name__,
-    TriggerSubscriptionVerifyRequest.model_json_schema(ref_template="#/definitions/{model}"),
+register_schema_models(
+    console_ns,
+    TriggerSubscriptionBuilderCreatePayload,
+    TriggerSubscriptionBuilderVerifyPayload,
+    TriggerSubscriptionBuilderUpdatePayload,
+    TriggerOAuthClientPayload,
 )
 
 
@@ -132,16 +135,11 @@ class TriggerSubscriptionListApi(Resource):
             raise
 
 
-parser = reqparse.RequestParser().add_argument(
-    "credential_type", type=str, required=False, nullable=True, location="json"
-)
-
-
 @console_ns.route(
     "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/create",
 )
 class TriggerSubscriptionBuilderCreateApi(Resource):
-    @console_ns.expect(parser)
+    @console_ns.expect(console_ns.models[TriggerSubscriptionBuilderCreatePayload.__name__])
     @setup_required
     @login_required
     @edit_permission_required
@@ -151,10 +149,10 @@ class TriggerSubscriptionBuilderCreateApi(Resource):
         user = current_user
         assert user.current_tenant_id is not None
 
-        args = parser.parse_args()
+        payload = TriggerSubscriptionBuilderCreatePayload.model_validate(console_ns.payload or {})
 
         try:
-            credential_type = CredentialType.of(args.get("credential_type") or CredentialType.UNAUTHORIZED.value)
+            credential_type = CredentialType.of(payload.credential_type)
             subscription_builder = TriggerSubscriptionBuilderService.create_trigger_subscription_builder(
                 tenant_id=user.current_tenant_id,
                 user_id=user.id,
@@ -182,18 +180,11 @@ class TriggerSubscriptionBuilderGetApi(Resource):
         )
 
 
-parser_api = (
-    reqparse.RequestParser()
-    # The credentials of the subscription builder
-    .add_argument("credentials", type=dict, required=False, nullable=True, location="json")
-)
-
-
 @console_ns.route(
     "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/verify-and-update/<path:subscription_builder_id>",
 )
-class TriggerSubscriptionBuilderVerifyAndUpdateApi(Resource):
-    @console_ns.expect(parser_api)
+class TriggerSubscriptionBuilderVerifyApi(Resource):
+    @console_ns.expect(console_ns.models[TriggerSubscriptionBuilderVerifyPayload.__name__])
     @setup_required
     @login_required
     @edit_permission_required
@@ -203,7 +194,7 @@ class TriggerSubscriptionBuilderVerifyAndUpdateApi(Resource):
         user = current_user
         assert user.current_tenant_id is not None
 
-        args = parser_api.parse_args()
+        payload = TriggerSubscriptionBuilderVerifyPayload.model_validate(console_ns.payload or {})
 
         try:
             # Use atomic update_and_verify to prevent race conditions
@@ -213,7 +204,7 @@ class TriggerSubscriptionBuilderVerifyAndUpdateApi(Resource):
                 provider_id=TriggerProviderID(provider),
                 subscription_builder_id=subscription_builder_id,
                 subscription_builder_updater=SubscriptionBuilderUpdater(
-                    credentials=args.get("credentials", None),
+                    credentials=payload.credentials,
                 ),
             )
         except Exception as e:
@@ -221,24 +212,11 @@ class TriggerSubscriptionBuilderVerifyAndUpdateApi(Resource):
             raise ValueError(str(e)) from e
 
 
-parser_update_api = (
-    reqparse.RequestParser()
-    # The name of the subscription builder
-    .add_argument("name", type=str, required=False, nullable=True, location="json")
-    # The parameters of the subscription builder
-    .add_argument("parameters", type=dict, required=False, nullable=True, location="json")
-    # The properties of the subscription builder
-    .add_argument("properties", type=dict, required=False, nullable=True, location="json")
-    # The credentials of the subscription builder
-    .add_argument("credentials", type=dict, required=False, nullable=True, location="json")
-)
-
-
 @console_ns.route(
     "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/update/<path:subscription_builder_id>",
 )
 class TriggerSubscriptionBuilderUpdateApi(Resource):
-    @console_ns.expect(parser_update_api)
+    @console_ns.expect(console_ns.models[TriggerSubscriptionBuilderUpdatePayload.__name__])
     @setup_required
     @login_required
     @edit_permission_required
@@ -249,7 +227,7 @@ class TriggerSubscriptionBuilderUpdateApi(Resource):
         assert isinstance(user, Account)
         assert user.current_tenant_id is not None
 
-        args = parser_update_api.parse_args()
+        payload = TriggerSubscriptionBuilderUpdatePayload.model_validate(console_ns.payload or {})
         try:
             return jsonable_encoder(
                 TriggerSubscriptionBuilderService.update_trigger_subscription_builder(
@@ -257,10 +235,10 @@ class TriggerSubscriptionBuilderUpdateApi(Resource):
                     provider_id=TriggerProviderID(provider),
                     subscription_builder_id=subscription_builder_id,
                     subscription_builder_updater=SubscriptionBuilderUpdater(
-                        name=args.get("name", None),
-                        parameters=args.get("parameters", None),
-                        properties=args.get("properties", None),
-                        credentials=args.get("credentials", None),
+                        name=payload.name,
+                        parameters=payload.parameters,
+                        properties=payload.properties,
+                        credentials=payload.credentials,
                     ),
                 )
             )
@@ -295,7 +273,7 @@ class TriggerSubscriptionBuilderLogsApi(Resource):
     "/workspaces/current/trigger-provider/<path:provider>/subscriptions/builder/build/<path:subscription_builder_id>",
 )
 class TriggerSubscriptionBuilderBuildApi(Resource):
-    @console_ns.expect(parser_update_api)
+    @console_ns.expect(console_ns.models[TriggerSubscriptionBuilderUpdatePayload.__name__])
     @setup_required
     @login_required
     @edit_permission_required
@@ -304,7 +282,7 @@ class TriggerSubscriptionBuilderBuildApi(Resource):
         """Build a subscription instance for a trigger provider"""
         user = current_user
         assert user.current_tenant_id is not None
-        args = parser_update_api.parse_args()
+        payload = TriggerSubscriptionBuilderUpdatePayload.model_validate(console_ns.payload or {})
         try:
             # Use atomic update_and_build to prevent race conditions
             TriggerSubscriptionBuilderService.update_and_build_builder(
@@ -313,9 +291,9 @@ class TriggerSubscriptionBuilderBuildApi(Resource):
                 provider_id=TriggerProviderID(provider),
                 subscription_builder_id=subscription_builder_id,
                 subscription_builder_updater=SubscriptionBuilderUpdater(
-                    name=args.get("name", None),
-                    parameters=args.get("parameters", None),
-                    properties=args.get("properties", None),
+                    name=payload.name,
+                    parameters=payload.parameters,
+                    properties=payload.properties,
                 ),
             )
             return 200
@@ -328,7 +306,7 @@ class TriggerSubscriptionBuilderBuildApi(Resource):
     "/workspaces/current/trigger-provider/<path:subscription_id>/subscriptions/update",
 )
 class TriggerSubscriptionUpdateApi(Resource):
-    @console_ns.expect(console_ns.models[TriggerSubscriptionUpdateRequest.__name__])
+    @console_ns.expect(console_ns.models[TriggerSubscriptionBuilderUpdatePayload.__name__])
     @setup_required
     @login_required
     @edit_permission_required
@@ -338,7 +316,7 @@ class TriggerSubscriptionUpdateApi(Resource):
         user = current_user
         assert user.current_tenant_id is not None
 
-        request = TriggerSubscriptionUpdateRequest.model_validate(console_ns.payload)
+        request = TriggerSubscriptionBuilderUpdatePayload.model_validate(console_ns.payload or {})
 
         subscription = TriggerProviderService.get_subscription_by_id(
             tenant_id=user.current_tenant_id,
@@ -568,13 +546,6 @@ class TriggerOAuthCallbackApi(Resource):
         return redirect(f"{dify_config.CONSOLE_WEB_URL}/oauth-callback")
 
 
-parser_oauth_client = (
-    reqparse.RequestParser()
-    .add_argument("client_params", type=dict, required=False, nullable=True, location="json")
-    .add_argument("enabled", type=bool, required=False, nullable=True, location="json")
-)
-
-
 @console_ns.route("/workspaces/current/trigger-provider/<path:provider>/oauth/client")
 class TriggerOAuthClientManageApi(Resource):
     @setup_required
@@ -622,7 +593,7 @@ class TriggerOAuthClientManageApi(Resource):
             logger.exception("Error getting OAuth client", exc_info=e)
             raise
 
-    @console_ns.expect(parser_oauth_client)
+    @console_ns.expect(console_ns.models[TriggerOAuthClientPayload.__name__])
     @setup_required
     @login_required
     @is_admin_or_owner_required
@@ -632,15 +603,15 @@ class TriggerOAuthClientManageApi(Resource):
         user = current_user
         assert user.current_tenant_id is not None
 
-        args = parser_oauth_client.parse_args()
+        payload = TriggerOAuthClientPayload.model_validate(console_ns.payload or {})
 
         try:
             provider_id = TriggerProviderID(provider)
             return TriggerProviderService.save_custom_oauth_client_params(
                 tenant_id=user.current_tenant_id,
                 provider_id=provider_id,
-                client_params=args.get("client_params"),
-                enabled=args.get("enabled"),
+                client_params=payload.client_params,
+                enabled=payload.enabled,
             )
 
         except ValueError as e:
@@ -676,7 +647,7 @@ class TriggerOAuthClientManageApi(Resource):
     "/workspaces/current/trigger-provider/<path:provider>/subscriptions/verify/<path:subscription_id>",
 )
 class TriggerSubscriptionVerifyApi(Resource):
-    @console_ns.expect(console_ns.models[TriggerSubscriptionVerifyRequest.__name__])
+    @console_ns.expect(console_ns.models[TriggerSubscriptionBuilderVerifyPayload.__name__])
     @setup_required
     @login_required
     @edit_permission_required
@@ -686,9 +657,7 @@ class TriggerSubscriptionVerifyApi(Resource):
         user = current_user
         assert user.current_tenant_id is not None
 
-        verify_request: TriggerSubscriptionVerifyRequest = TriggerSubscriptionVerifyRequest.model_validate(
-            console_ns.payload
-        )
+        verify_request = TriggerSubscriptionBuilderVerifyPayload.model_validate(console_ns.payload or {})
 
         try:
             result = TriggerProviderService.verify_subscription_credentials(

api/controllers/console/workspace/workspace.py

@@ -80,6 +80,9 @@ tenant_fields = {
     "in_trial": fields.Boolean,
     "trial_end_reason": fields.String,
     "custom_config": fields.Raw(attribute="custom_config"),
+    "trial_credits": fields.Integer,
+    "trial_credits_used": fields.Integer,
+    "next_credit_reset_date": fields.Integer,
 }
 
 tenants_fields = {

api/controllers/console/wraps.py

@@ -358,14 +358,12 @@ def annotation_import_rate_limit(view: Callable[P, R]):
     def decorated(*args: P.args, **kwargs: P.kwargs):
         _, current_tenant_id = current_account_with_tenant()
         current_time = int(time.time() * 1000)
-
         # Check per-minute rate limit
         minute_key = f"annotation_import_rate_limit:{current_tenant_id}:1min"
         redis_client.zadd(minute_key, {current_time: current_time})
         redis_client.zremrangebyscore(minute_key, 0, current_time - 60000)
         minute_count = redis_client.zcard(minute_key)
         redis_client.expire(minute_key, 120)  # 2 minutes TTL
-
         if minute_count > dify_config.ANNOTATION_IMPORT_RATE_LIMIT_PER_MINUTE:
             abort(
                 429,
@@ -379,7 +377,6 @@ def annotation_import_rate_limit(view: Callable[P, R]):
         redis_client.zremrangebyscore(hour_key, 0, current_time - 3600000)
         hour_count = redis_client.zcard(hour_key)
         redis_client.expire(hour_key, 7200)  # 2 hours TTL
-
         if hour_count > dify_config.ANNOTATION_IMPORT_RATE_LIMIT_PER_HOUR:
             abort(
                 429,

api/controllers/web/login.py

@@ -10,7 +10,12 @@ from controllers.console.auth.error import (
     InvalidEmailError,
 )
 from controllers.console.error import AccountBannedError
-from controllers.console.wraps import only_edition_enterprise, setup_required
+from controllers.console.wraps import (
+    decrypt_code_field,
+    decrypt_password_field,
+    only_edition_enterprise,
+    setup_required,
+)
 from controllers.web import web_ns
 from controllers.web.wraps import decode_jwt_token
 from libs.helper import email
@@ -42,6 +47,7 @@ class LoginApi(Resource):
             404: "Account not found",
         }
     )
+    @decrypt_password_field
     def post(self):
         """Authenticate user and login."""
         parser = (
@@ -181,6 +187,7 @@ class EmailCodeLoginApi(Resource):
             404: "Account not found",
         }
     )
+    @decrypt_code_field
     def post(self):
         parser = (
             reqparse.RequestParser()

api/core/agent/base_agent_runner.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import uuid
+from decimal import Decimal
 from typing import Union, cast
 
 from sqlalchemy import select
@@ -41,6 +42,7 @@ from core.tools.tool_manager import ToolManager
 from core.tools.utils.dataset_retriever_tool import DatasetRetrieverTool
 from extensions.ext_database import db
 from factories import file_factory
+from models.enums import CreatorUserRole
 from models.model import Conversation, Message, MessageAgentThought, MessageFile
 
 logger = logging.getLogger(__name__)
@@ -300,6 +302,7 @@ class BaseAgentRunner(AppRunner):
         thought = MessageAgentThought(
             message_id=message_id,
             message_chain_id=None,
+            tool_process_data=None,
             thought="",
             tool=tool_name,
             tool_labels_str="{}",
@@ -307,20 +310,20 @@ class BaseAgentRunner(AppRunner):
             tool_input=tool_input,
             message=message,
             message_token=0,
-            message_unit_price=0,
-            message_price_unit=0,
+            message_unit_price=Decimal(0),
+            message_price_unit=Decimal("0.001"),
             message_files=json.dumps(messages_ids) if messages_ids else "",
             answer="",
             observation="",
             answer_token=0,
-            answer_unit_price=0,
-            answer_price_unit=0,
+            answer_unit_price=Decimal(0),
+            answer_price_unit=Decimal("0.001"),
             tokens=0,
-            total_price=0,
+            total_price=Decimal(0),
             position=self.agent_thought_count + 1,
             currency="USD",
             latency=0,
-            created_by_role="account",
+            created_by_role=CreatorUserRole.ACCOUNT,
             created_by=self.user_id,
         )
 
@@ -353,7 +356,8 @@ class BaseAgentRunner(AppRunner):
             raise ValueError("agent thought not found")
 
         if thought:
-            agent_thought.thought += thought
+            existing_thought = agent_thought.thought or ""
+            agent_thought.thought = f"{existing_thought}{thought}"
 
         if tool_name:
             agent_thought.tool = tool_name
@@ -451,21 +455,30 @@ class BaseAgentRunner(AppRunner):
             agent_thoughts: list[MessageAgentThought] = message.agent_thoughts
             if agent_thoughts:
                 for agent_thought in agent_thoughts:
-                    tools = agent_thought.tool
-                    if tools:
-                        tools = tools.split(";")
+                    tool_names_raw = agent_thought.tool
+                    if tool_names_raw:
+                        tool_names = tool_names_raw.split(";")
                         tool_calls: list[AssistantPromptMessage.ToolCall] = []
                         tool_call_response: list[ToolPromptMessage] = []
-                        try:
-                            tool_inputs = json.loads(agent_thought.tool_input)
-                        except Exception:
-                            tool_inputs = {tool: {} for tool in tools}
-                        try:
-                            tool_responses = json.loads(agent_thought.observation)
-                        except Exception:
-                            tool_responses = dict.fromkeys(tools, agent_thought.observation)
+                        tool_input_payload = agent_thought.tool_input
+                        if tool_input_payload:
+                            try:
+                                tool_inputs = json.loads(tool_input_payload)
+                            except Exception:
+                                tool_inputs = {tool: {} for tool in tool_names}
+                        else:
+                            tool_inputs = {tool: {} for tool in tool_names}
 
-                        for tool in tools:
+                        observation_payload = agent_thought.observation
+                        if observation_payload:
+                            try:
+                                tool_responses = json.loads(observation_payload)
+                            except Exception:
+                                tool_responses = dict.fromkeys(tool_names, observation_payload)
+                        else:
+                            tool_responses = dict.fromkeys(tool_names, observation_payload)
+
+                        for tool in tool_names:
                             # generate a uuid for tool call
                             tool_call_id = str(uuid.uuid4())
                             tool_calls.append(
@@ -495,7 +508,7 @@ class BaseAgentRunner(AppRunner):
                                 *tool_call_response,
                             ]
                         )
-                    if not tools:
+                    if not tool_names_raw:
                         result.append(AssistantPromptMessage(content=agent_thought.thought))
             else:
                 if message.answer:

api/core/app/app_config/entities.py

@@ -1,4 +1,3 @@
-import json
 from collections.abc import Sequence
 from enum import StrEnum, auto
 from typing import Any, Literal
@@ -121,7 +120,7 @@ class VariableEntity(BaseModel):
     allowed_file_types: Sequence[FileType] | None = Field(default_factory=list)
     allowed_file_extensions: Sequence[str] | None = Field(default_factory=list)
     allowed_file_upload_methods: Sequence[FileTransferMethod] | None = Field(default_factory=list)
-    json_schema: str | None = Field(default=None)
+    json_schema: dict | None = Field(default=None)
 
     @field_validator("description", mode="before")
     @classmethod
@@ -135,17 +134,11 @@ class VariableEntity(BaseModel):
 
     @field_validator("json_schema")
     @classmethod
-    def validate_json_schema(cls, schema: str | None) -> str | None:
+    def validate_json_schema(cls, schema: dict | None) -> dict | None:
         if schema is None:
             return None
-
         try:
-            json_schema = json.loads(schema)
-        except json.JSONDecodeError:
-            raise ValueError(f"invalid json_schema value {schema}")
-
-        try:
-            Draft7Validator.check_schema(json_schema)
+            Draft7Validator.check_schema(schema)
         except SchemaError as e:
             raise ValueError(f"Invalid JSON schema: {e.message}")
         return schema

api/core/app/apps/advanced_chat/app_config_manager.py

@@ -26,7 +26,6 @@ class AdvancedChatAppConfigManager(BaseAppConfigManager):
     @classmethod
     def get_app_config(cls, app_model: App, workflow: Workflow) -> AdvancedChatAppConfig:
         features_dict = workflow.features_dict
-
         app_mode = AppMode.value_of(app_model.mode)
         app_config = AdvancedChatAppConfig(
             tenant_id=app_model.tenant_id,

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

@@ -39,7 +39,6 @@ from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from extensions.otel import WorkflowAppRunnerHandler, trace_span
 from models import Workflow
-from models.enums import UserFrom
 from models.model import App, Conversation, Message, MessageAnnotation
 from models.workflow import ConversationVariable
 from services.conversation_variable_updater import ConversationVariableUpdater
@@ -106,6 +105,11 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
         if not app_record:
             raise ValueError("App not found")
 
+        invoke_from = self.application_generate_entity.invoke_from
+        if self.application_generate_entity.single_iteration_run or self.application_generate_entity.single_loop_run:
+            invoke_from = InvokeFrom.DEBUGGER
+        user_from = self._resolve_user_from(invoke_from)
+
         if self.application_generate_entity.single_iteration_run or self.application_generate_entity.single_loop_run:
             # Handle single iteration or single loop run
             graph, variable_pool, graph_runtime_state = self._prepare_single_node_execution(
@@ -158,6 +162,8 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                 workflow_id=self._workflow.id,
                 tenant_id=self._workflow.tenant_id,
                 user_id=self.application_generate_entity.user_id,
+                user_from=user_from,
+                invoke_from=invoke_from,
             )
 
         db.session.close()
@@ -175,12 +181,8 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
             graph=graph,
             graph_config=self._workflow.graph_dict,
             user_id=self.application_generate_entity.user_id,
-            user_from=(
-                UserFrom.ACCOUNT
-                if self.application_generate_entity.invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}
-                else UserFrom.END_USER
-            ),
-            invoke_from=self.application_generate_entity.invoke_from,
+            user_from=user_from,
+            invoke_from=invoke_from,
             call_depth=self.application_generate_entity.call_depth,
             variable_pool=variable_pool,
             graph_runtime_state=graph_runtime_state,

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

@@ -471,25 +471,6 @@ class AdvancedChatAppGenerateTaskPipeline(GraphRuntimeStateSupport):
         if node_finish_resp:
             yield node_finish_resp
 
-        # For ANSWER nodes, check if we need to send a message_replace event
-        # Only send if the final output differs from the accumulated task_state.answer
-        # This happens when variables were updated by variable_assigner during workflow execution
-        if event.node_type == NodeType.ANSWER and event.outputs:
-            final_answer = event.outputs.get("answer")
-            if final_answer is not None and final_answer != self._task_state.answer:
-                logger.info(
-                    "ANSWER node final output '%s' differs from accumulated answer '%s', sending message_replace event",
-                    final_answer,
-                    self._task_state.answer,
-                )
-                # Update the task state answer
-                self._task_state.answer = str(final_answer)
-                # Send message_replace event to update the UI
-                yield self._message_cycle_manager.message_replace_to_stream_response(
-                    answer=str(final_answer),
-                    reason="variable_update",
-                )
-
     def _handle_node_failed_events(
         self,
         event: Union[QueueNodeFailedEvent, QueueNodeExceptionEvent],

api/core/app/apps/base_app_generator.py

@@ -1,4 +1,3 @@
-import json
 from collections.abc import Generator, Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Union, final
 
@@ -76,12 +75,24 @@ class BaseAppGenerator:
         user_inputs = {**user_inputs, **files_inputs, **file_list_inputs}
 
         # Check if all files are converted to File
-        if any(filter(lambda v: isinstance(v, dict), user_inputs.values())):
-            raise ValueError("Invalid input type")
-        if any(
-            filter(lambda v: isinstance(v, dict), filter(lambda item: isinstance(item, list), user_inputs.values()))
-        ):
-            raise ValueError("Invalid input type")
+        invalid_dict_keys = [
+            k
+            for k, v in user_inputs.items()
+            if isinstance(v, dict)
+            and entity_dictionary[k].type not in {VariableEntityType.FILE, VariableEntityType.JSON_OBJECT}
+        ]
+        if invalid_dict_keys:
+            raise ValueError(f"Invalid input type for {invalid_dict_keys}")
+
+        invalid_list_dict_keys = [
+            k
+            for k, v in user_inputs.items()
+            if isinstance(v, list)
+            and any(isinstance(item, dict) for item in v)
+            and entity_dictionary[k].type != VariableEntityType.FILE_LIST
+        ]
+        if invalid_list_dict_keys:
+            raise ValueError(f"Invalid input type for {invalid_list_dict_keys}")
 
         return user_inputs
 
@@ -178,12 +189,8 @@ class BaseAppGenerator:
                     elif value == 0:
                         value = False
             case VariableEntityType.JSON_OBJECT:
-                if not isinstance(value, str):
-                    raise ValueError(f"{variable_entity.variable} in input form must be a string")
-                try:
-                    json.loads(value)
-                except json.JSONDecodeError:
-                    raise ValueError(f"{variable_entity.variable} in input form must be a valid JSON object")
+                if not isinstance(value, dict):
+                    raise ValueError(f"{variable_entity.variable} in input form must be a dict")
             case _:
                 raise AssertionError("this statement should be unreachable.")
 

api/core/app/apps/pipeline/pipeline_runner.py

@@ -73,9 +73,15 @@ class PipelineRunner(WorkflowBasedAppRunner):
         """
         app_config = self.application_generate_entity.app_config
         app_config = cast(PipelineConfig, app_config)
+        invoke_from = self.application_generate_entity.invoke_from
+
+        if self.application_generate_entity.single_iteration_run or self.application_generate_entity.single_loop_run:
+            invoke_from = InvokeFrom.DEBUGGER
+
+        user_from = self._resolve_user_from(invoke_from)
 
         user_id = None
-        if self.application_generate_entity.invoke_from in {InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API}:
+        if invoke_from in {InvokeFrom.WEB_APP, InvokeFrom.SERVICE_API}:
             end_user = db.session.query(EndUser).where(EndUser.id == self.application_generate_entity.user_id).first()
             if end_user:
                 user_id = end_user.session_id
@@ -117,7 +123,7 @@ class PipelineRunner(WorkflowBasedAppRunner):
                 dataset_id=self.application_generate_entity.dataset_id,
                 datasource_type=self.application_generate_entity.datasource_type,
                 datasource_info=self.application_generate_entity.datasource_info,
-                invoke_from=self.application_generate_entity.invoke_from.value,
+                invoke_from=invoke_from.value,
             )
 
             rag_pipeline_variables = []
@@ -149,6 +155,8 @@ class PipelineRunner(WorkflowBasedAppRunner):
                 graph_runtime_state=graph_runtime_state,
                 start_node_id=self.application_generate_entity.start_node_id,
                 workflow=workflow,
+                user_from=user_from,
+                invoke_from=invoke_from,
             )
 
         # RUN WORKFLOW
@@ -159,12 +167,8 @@ class PipelineRunner(WorkflowBasedAppRunner):
             graph=graph,
             graph_config=workflow.graph_dict,
             user_id=self.application_generate_entity.user_id,
-            user_from=(
-                UserFrom.ACCOUNT
-                if self.application_generate_entity.invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}
-                else UserFrom.END_USER
-            ),
-            invoke_from=self.application_generate_entity.invoke_from,
+            user_from=user_from,
+            invoke_from=invoke_from,
             call_depth=self.application_generate_entity.call_depth,
             graph_runtime_state=graph_runtime_state,
             variable_pool=variable_pool,
@@ -210,7 +214,12 @@ class PipelineRunner(WorkflowBasedAppRunner):
         return workflow
 
     def _init_rag_pipeline_graph(
-        self, workflow: Workflow, graph_runtime_state: GraphRuntimeState, start_node_id: str | None = None
+        self,
+        workflow: Workflow,
+        graph_runtime_state: GraphRuntimeState,
+        start_node_id: str | None = None,
+        user_from: UserFrom = UserFrom.ACCOUNT,
+        invoke_from: InvokeFrom = InvokeFrom.SERVICE_API,
     ) -> Graph:
         """
         Init pipeline graph
@@ -253,8 +262,8 @@ class PipelineRunner(WorkflowBasedAppRunner):
             workflow_id=workflow.id,
             graph_config=graph_config,
             user_id=self.application_generate_entity.user_id,
-            user_from=UserFrom.ACCOUNT,
-            invoke_from=InvokeFrom.SERVICE_API,
+            user_from=user_from,
+            invoke_from=invoke_from,
             call_depth=0,
         )
 

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

@@ -20,7 +20,6 @@ from core.workflow.workflow_entry import WorkflowEntry
 from extensions.ext_redis import redis_client
 from extensions.otel import WorkflowAppRunnerHandler, trace_span
 from libs.datetime_utils import naive_utc_now
-from models.enums import UserFrom
 from models.workflow import Workflow
 
 logger = logging.getLogger(__name__)
@@ -74,7 +73,12 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
             workflow_execution_id=self.application_generate_entity.workflow_execution_id,
         )
 
+        invoke_from = self.application_generate_entity.invoke_from
         # if only single iteration or single loop run is requested
+        if self.application_generate_entity.single_iteration_run or self.application_generate_entity.single_loop_run:
+            invoke_from = InvokeFrom.DEBUGGER
+        user_from = self._resolve_user_from(invoke_from)
+
         if self.application_generate_entity.single_iteration_run or self.application_generate_entity.single_loop_run:
             graph, variable_pool, graph_runtime_state = self._prepare_single_node_execution(
                 workflow=self._workflow,
@@ -102,6 +106,8 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
                 workflow_id=self._workflow.id,
                 tenant_id=self._workflow.tenant_id,
                 user_id=self.application_generate_entity.user_id,
+                user_from=user_from,
+                invoke_from=invoke_from,
                 root_node_id=self._root_node_id,
             )
 
@@ -120,12 +126,8 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
             graph=graph,
             graph_config=self._workflow.graph_dict,
             user_id=self.application_generate_entity.user_id,
-            user_from=(
-                UserFrom.ACCOUNT
-                if self.application_generate_entity.invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}
-                else UserFrom.END_USER
-            ),
-            invoke_from=self.application_generate_entity.invoke_from,
+            user_from=user_from,
+            invoke_from=invoke_from,
             call_depth=self.application_generate_entity.call_depth,
             variable_pool=variable_pool,
             graph_runtime_state=graph_runtime_state,

api/core/app/apps/workflow_app_runner.py

@@ -77,10 +77,18 @@ class WorkflowBasedAppRunner:
         self._app_id = app_id
         self._graph_engine_layers = graph_engine_layers
 
+    @staticmethod
+    def _resolve_user_from(invoke_from: InvokeFrom) -> UserFrom:
+        if invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}:
+            return UserFrom.ACCOUNT
+        return UserFrom.END_USER
+
     def _init_graph(
         self,
         graph_config: Mapping[str, Any],
         graph_runtime_state: GraphRuntimeState,
+        user_from: UserFrom,
+        invoke_from: InvokeFrom,
         workflow_id: str = "",
         tenant_id: str = "",
         user_id: str = "",
@@ -105,8 +113,8 @@ class WorkflowBasedAppRunner:
             workflow_id=workflow_id,
             graph_config=graph_config,
             user_id=user_id,
-            user_from=UserFrom.ACCOUNT,
-            invoke_from=InvokeFrom.SERVICE_API,
+            user_from=user_from,
+            invoke_from=invoke_from,
             call_depth=0,
         )
 
@@ -250,7 +258,7 @@ class WorkflowBasedAppRunner:
             graph_config=graph_config,
             user_id="",
             user_from=UserFrom.ACCOUNT,
-            invoke_from=InvokeFrom.SERVICE_API,
+            invoke_from=InvokeFrom.DEBUGGER,
             call_depth=0,
         )
 

api/core/entities/knowledge_entities.py

@@ -3,6 +3,7 @@ from pydantic import BaseModel, Field, field_validator
 
 class PreviewDetail(BaseModel):
     content: str
+    summary: str | None = None
     child_chunks: list[str] | None = None
 
 

api/core/hosting_configuration.py

@@ -56,6 +56,10 @@ class HostingConfiguration:
         self.provider_map[f"{DEFAULT_PLUGIN_ID}/minimax/minimax"] = self.init_minimax()
         self.provider_map[f"{DEFAULT_PLUGIN_ID}/spark/spark"] = self.init_spark()
         self.provider_map[f"{DEFAULT_PLUGIN_ID}/zhipuai/zhipuai"] = self.init_zhipuai()
+        self.provider_map[f"{DEFAULT_PLUGIN_ID}/gemini/google"] = self.init_gemini()
+        self.provider_map[f"{DEFAULT_PLUGIN_ID}/x/x"] = self.init_xai()
+        self.provider_map[f"{DEFAULT_PLUGIN_ID}/deepseek/deepseek"] = self.init_deepseek()
+        self.provider_map[f"{DEFAULT_PLUGIN_ID}/tongyi/tongyi"] = self.init_tongyi()
 
         self.moderation_config = self.init_moderation_config()
 
@@ -128,7 +132,7 @@ class HostingConfiguration:
         quotas: list[HostingQuota] = []
 
         if dify_config.HOSTED_OPENAI_TRIAL_ENABLED:
-            hosted_quota_limit = dify_config.HOSTED_OPENAI_QUOTA_LIMIT
+            hosted_quota_limit = 0
             trial_models = self.parse_restrict_models_from_env("HOSTED_OPENAI_TRIAL_MODELS")
             trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trial_models)
             quotas.append(trial_quota)
@@ -156,18 +160,49 @@ class HostingConfiguration:
             quota_unit=quota_unit,
         )
 
-    @staticmethod
-    def init_anthropic() -> HostingProvider:
-        quota_unit = QuotaUnit.TOKENS
+    def init_gemini(self) -> HostingProvider:
+        quota_unit = QuotaUnit.CREDITS
+        quotas: list[HostingQuota] = []
+
+        if dify_config.HOSTED_GEMINI_TRIAL_ENABLED:
+            hosted_quota_limit = 0
+            trial_models = self.parse_restrict_models_from_env("HOSTED_GEMINI_TRIAL_MODELS")
+            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trial_models)
+            quotas.append(trial_quota)
+
+        if dify_config.HOSTED_GEMINI_PAID_ENABLED:
+            paid_models = self.parse_restrict_models_from_env("HOSTED_GEMINI_PAID_MODELS")
+            paid_quota = PaidHostingQuota(restrict_models=paid_models)
+            quotas.append(paid_quota)
+
+        if len(quotas) > 0:
+            credentials = {
+                "google_api_key": dify_config.HOSTED_GEMINI_API_KEY,
+            }
+
+            if dify_config.HOSTED_GEMINI_API_BASE:
+                credentials["google_base_url"] = dify_config.HOSTED_GEMINI_API_BASE
+
+            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
+
+        return HostingProvider(
+            enabled=False,
+            quota_unit=quota_unit,
+        )
+
+    def init_anthropic(self) -> HostingProvider:
+        quota_unit = QuotaUnit.CREDITS
         quotas: list[HostingQuota] = []
 
         if dify_config.HOSTED_ANTHROPIC_TRIAL_ENABLED:
-            hosted_quota_limit = dify_config.HOSTED_ANTHROPIC_QUOTA_LIMIT
-            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit)
+            hosted_quota_limit = 0
+            trail_models = self.parse_restrict_models_from_env("HOSTED_ANTHROPIC_TRIAL_MODELS")
+            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
             quotas.append(trial_quota)
 
         if dify_config.HOSTED_ANTHROPIC_PAID_ENABLED:
-            paid_quota = PaidHostingQuota()
+            paid_models = self.parse_restrict_models_from_env("HOSTED_ANTHROPIC_PAID_MODELS")
+            paid_quota = PaidHostingQuota(restrict_models=paid_models)
             quotas.append(paid_quota)
 
         if len(quotas) > 0:
@@ -185,6 +220,94 @@ class HostingConfiguration:
             quota_unit=quota_unit,
         )
 
+    def init_tongyi(self) -> HostingProvider:
+        quota_unit = QuotaUnit.CREDITS
+        quotas: list[HostingQuota] = []
+
+        if dify_config.HOSTED_TONGYI_TRIAL_ENABLED:
+            hosted_quota_limit = 0
+            trail_models = self.parse_restrict_models_from_env("HOSTED_TONGYI_TRIAL_MODELS")
+            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
+            quotas.append(trial_quota)
+
+        if dify_config.HOSTED_TONGYI_PAID_ENABLED:
+            paid_models = self.parse_restrict_models_from_env("HOSTED_TONGYI_PAID_MODELS")
+            paid_quota = PaidHostingQuota(restrict_models=paid_models)
+            quotas.append(paid_quota)
+
+        if len(quotas) > 0:
+            credentials = {
+                "dashscope_api_key": dify_config.HOSTED_TONGYI_API_KEY,
+                "use_international_endpoint": dify_config.HOSTED_TONGYI_USE_INTERNATIONAL_ENDPOINT,
+            }
+
+            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
+
+        return HostingProvider(
+            enabled=False,
+            quota_unit=quota_unit,
+        )
+
+    def init_xai(self) -> HostingProvider:
+        quota_unit = QuotaUnit.CREDITS
+        quotas: list[HostingQuota] = []
+
+        if dify_config.HOSTED_XAI_TRIAL_ENABLED:
+            hosted_quota_limit = 0
+            trail_models = self.parse_restrict_models_from_env("HOSTED_XAI_TRIAL_MODELS")
+            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
+            quotas.append(trial_quota)
+
+        if dify_config.HOSTED_XAI_PAID_ENABLED:
+            paid_models = self.parse_restrict_models_from_env("HOSTED_XAI_PAID_MODELS")
+            paid_quota = PaidHostingQuota(restrict_models=paid_models)
+            quotas.append(paid_quota)
+
+        if len(quotas) > 0:
+            credentials = {
+                "api_key": dify_config.HOSTED_XAI_API_KEY,
+            }
+
+            if dify_config.HOSTED_XAI_API_BASE:
+                credentials["endpoint_url"] = dify_config.HOSTED_XAI_API_BASE
+
+            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
+
+        return HostingProvider(
+            enabled=False,
+            quota_unit=quota_unit,
+        )
+
+    def init_deepseek(self) -> HostingProvider:
+        quota_unit = QuotaUnit.CREDITS
+        quotas: list[HostingQuota] = []
+
+        if dify_config.HOSTED_DEEPSEEK_TRIAL_ENABLED:
+            hosted_quota_limit = 0
+            trail_models = self.parse_restrict_models_from_env("HOSTED_DEEPSEEK_TRIAL_MODELS")
+            trial_quota = TrialHostingQuota(quota_limit=hosted_quota_limit, restrict_models=trail_models)
+            quotas.append(trial_quota)
+
+        if dify_config.HOSTED_DEEPSEEK_PAID_ENABLED:
+            paid_models = self.parse_restrict_models_from_env("HOSTED_DEEPSEEK_PAID_MODELS")
+            paid_quota = PaidHostingQuota(restrict_models=paid_models)
+            quotas.append(paid_quota)
+
+        if len(quotas) > 0:
+            credentials = {
+                "api_key": dify_config.HOSTED_DEEPSEEK_API_KEY,
+            }
+
+            if dify_config.HOSTED_DEEPSEEK_API_BASE:
+                credentials["endpoint_url"] = dify_config.HOSTED_DEEPSEEK_API_BASE
+
+            return HostingProvider(enabled=True, credentials=credentials, quota_unit=quota_unit, quotas=quotas)
+
+        return HostingProvider(
+            enabled=False,
+            quota_unit=quota_unit,
+        )
+
     @staticmethod
     def init_minimax() -> HostingProvider:
         quota_unit = QuotaUnit.TOKENS

api/core/indexing_runner.py

@@ -311,14 +311,18 @@ class IndexingRunner:
         qa_preview_texts: list[QAPreviewDetail] = []
 
         total_segments = 0
+        # doc_form represents the segmentation method (general, parent-child, QA)
         index_type = doc_form
         index_processor = IndexProcessorFactory(index_type).init_index_processor()
+        # one extract_setting is one source document
         for extract_setting in extract_settings:
             # extract
             processing_rule = DatasetProcessRule(
                 mode=tmp_processing_rule["mode"], rules=json.dumps(tmp_processing_rule["rules"])
             )
+            # Extract document content
             text_docs = index_processor.extract(extract_setting, process_rule_mode=tmp_processing_rule["mode"])
+            # Cleaning and segmentation
             documents = index_processor.transform(
                 text_docs,
                 current_user=None,
@@ -361,6 +365,12 @@ class IndexingRunner:
 
         if doc_form and doc_form == "qa_model":
             return IndexingEstimate(total_segments=total_segments * 20, qa_preview=qa_preview_texts, preview=[])
+
+        # Generate summary preview
+        summary_index_setting = tmp_processing_rule["summary_index_setting"] if "summary_index_setting" in tmp_processing_rule else None
+        if summary_index_setting and summary_index_setting.get('enable') and preview_texts:
+            preview_texts = index_processor.generate_summary_preview(tenant_id, preview_texts, summary_index_setting)
+
         return IndexingEstimate(total_segments=total_segments, preview=preview_texts)
 
     def _extract(

api/core/llm_generator/prompts.py

@@ -434,3 +434,6 @@ INSTRUCTION_GENERATE_TEMPLATE_PROMPT = """The output of this prompt is not as ex
 You should edit the prompt according to the IDEAL OUTPUT."""
 
 INSTRUCTION_GENERATE_TEMPLATE_CODE = """Please fix the errors in the {{#error_message#}}."""
+
+DEFAULT_GENERATOR_SUMMARY_PROMPT = """
+You are a helpful assistant that summarizes long pieces of text into concise summaries. Given the following text, generate a brief summary that captures the main points and key information. The summary should be clear, concise, and written in complete sentences. """

api/core/provider_manager.py

@@ -618,18 +618,18 @@ class ProviderManager:
                 )
 
             for quota in configuration.quotas:
-                if quota.quota_type == ProviderQuotaType.TRIAL:
+                if quota.quota_type in (ProviderQuotaType.TRIAL, ProviderQuotaType.PAID):
                     # Init trial provider records if not exists
-                    if ProviderQuotaType.TRIAL not in provider_quota_to_provider_record_dict:
+                    if quota.quota_type not in provider_quota_to_provider_record_dict:
                         try:
                             # FIXME ignore the type error, only TrialHostingQuota has limit need to change the logic
                             new_provider_record = Provider(
                                 tenant_id=tenant_id,
                                 # TODO: Use provider name with prefix after the data migration.
                                 provider_name=ModelProviderID(provider_name).provider_name,
-                                provider_type=ProviderType.SYSTEM,
-                                quota_type=ProviderQuotaType.TRIAL,
-                                quota_limit=quota.quota_limit,  # type: ignore
+                                provider_type=ProviderType.SYSTEM.value,
+                                quota_type=quota.quota_type,
+                                quota_limit=0,  # type: ignore
                                 quota_used=0,
                                 is_valid=True,
                             )
@@ -641,8 +641,8 @@ class ProviderManager:
                             stmt = select(Provider).where(
                                 Provider.tenant_id == tenant_id,
                                 Provider.provider_name == ModelProviderID(provider_name).provider_name,
-                                Provider.provider_type == ProviderType.SYSTEM,
-                                Provider.quota_type == ProviderQuotaType.TRIAL,
+                                Provider.provider_type == ProviderType.SYSTEM.value,
+                                Provider.quota_type == quota.quota_type,
                             )
                             existed_provider_record = db.session.scalar(stmt)
                             if not existed_provider_record:
@@ -912,6 +912,22 @@ class ProviderManager:
                 provider_record
             )
         quota_configurations = []
+
+        if dify_config.EDITION == "CLOUD":
+            from services.credit_pool_service import CreditPoolService
+
+            trail_pool = CreditPoolService.get_pool(
+                tenant_id=tenant_id,
+                pool_type=ProviderQuotaType.TRIAL.value,
+            )
+            paid_pool = CreditPoolService.get_pool(
+                tenant_id=tenant_id,
+                pool_type=ProviderQuotaType.PAID.value,
+            )
+        else:
+            trail_pool = None
+            paid_pool = None
+
         for provider_quota in provider_hosting_configuration.quotas:
             if provider_quota.quota_type not in quota_type_to_provider_records_dict:
                 if provider_quota.quota_type == ProviderQuotaType.FREE:
@@ -932,16 +948,36 @@ class ProviderManager:
                     raise ValueError("quota_used is None")
                 if provider_record.quota_limit is None:
                     raise ValueError("quota_limit is None")
+                if provider_quota.quota_type == ProviderQuotaType.TRIAL and trail_pool is not None:
+                    quota_configuration = QuotaConfiguration(
+                        quota_type=provider_quota.quota_type,
+                        quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
+                        quota_used=trail_pool.quota_used,
+                        quota_limit=trail_pool.quota_limit,
+                        is_valid=trail_pool.quota_limit > trail_pool.quota_used or trail_pool.quota_limit == -1,
+                        restrict_models=provider_quota.restrict_models,
+                    )
 
-                quota_configuration = QuotaConfiguration(
-                    quota_type=provider_quota.quota_type,
-                    quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
-                    quota_used=provider_record.quota_used,
-                    quota_limit=provider_record.quota_limit,
-                    is_valid=provider_record.quota_limit > provider_record.quota_used
-                    or provider_record.quota_limit == -1,
-                    restrict_models=provider_quota.restrict_models,
-                )
+                elif provider_quota.quota_type == ProviderQuotaType.PAID and paid_pool is not None:
+                    quota_configuration = QuotaConfiguration(
+                        quota_type=provider_quota.quota_type,
+                        quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
+                        quota_used=paid_pool.quota_used,
+                        quota_limit=paid_pool.quota_limit,
+                        is_valid=paid_pool.quota_limit > paid_pool.quota_used or paid_pool.quota_limit == -1,
+                        restrict_models=provider_quota.restrict_models,
+                    )
+
+                else:
+                    quota_configuration = QuotaConfiguration(
+                        quota_type=provider_quota.quota_type,
+                        quota_unit=provider_hosting_configuration.quota_unit or QuotaUnit.TOKENS,
+                        quota_used=provider_record.quota_used,
+                        quota_limit=provider_record.quota_limit,
+                        is_valid=provider_record.quota_limit > provider_record.quota_used
+                        or provider_record.quota_limit == -1,
+                        restrict_models=provider_quota.restrict_models,
+                    )
 
             quota_configurations.append(quota_configuration)
 

api/core/rag/datasource/retrieval_service.py

@@ -392,6 +392,69 @@ class RetrievalService:
             records = []
             include_segment_ids = set()
             segment_child_map = {}
+            segment_file_map = {}
+            segment_summary_map = {}  # Map segment_id to summary content
+            summary_segment_ids = set()  # Track segments retrieved via summary
+            with Session(bind=db.engine, expire_on_commit=False) as session:
+                # Process documents
+                for document in documents:
+                    segment_id = None
+                    attachment_info = None
+                    child_chunk = None
+                    document_id = document.metadata.get("document_id")
+                    if document_id not in dataset_documents:
+                        continue
+
+                    dataset_document = dataset_documents[document_id]
+                    if not dataset_document:
+                        continue
+
+                    if dataset_document.doc_form == IndexStructureType.PARENT_CHILD_INDEX:
+                        # Handle parent-child documents
+                        if document.metadata.get("doc_type") == DocType.IMAGE:
+                            attachment_info_dict = cls.get_segment_attachment_info(
+                                dataset_document.dataset_id,
+                                dataset_document.tenant_id,
+                                document.metadata.get("doc_id") or "",
+                                session,
+                            )
+                            if attachment_info_dict:
+                                attachment_info = attachment_info_dict["attachment_info"]
+                                segment_id = attachment_info_dict["segment_id"]
+                        else:
+                            # Check if this is a summary document
+                            is_summary = document.metadata.get("is_summary", False)
+                            if is_summary:
+                                # For summary documents, find the original chunk via original_chunk_id
+                                original_chunk_id = document.metadata.get("original_chunk_id")
+                                if not original_chunk_id:
+                                    continue
+                                segment_id = original_chunk_id
+                                # Track that this segment was retrieved via summary
+                                summary_segment_ids.add(segment_id)
+                            else:
+                                # For normal documents, find by child chunk index_node_id
+                                child_index_node_id = document.metadata.get("doc_id")
+                                child_chunk_stmt = select(ChildChunk).where(ChildChunk.index_node_id == child_index_node_id)
+                                child_chunk = session.scalar(child_chunk_stmt)
+
+                                if not child_chunk:
+                                    continue
+                                segment_id = child_chunk.segment_id
+
+                        if not segment_id:
+                            continue
+
+                        segment = (
+                            session.query(DocumentSegment)
+                            .where(
+                                DocumentSegment.dataset_id == dataset_document.dataset_id,
+                                DocumentSegment.enabled == True,
+                                DocumentSegment.status == "completed",
+                                DocumentSegment.id == segment_id,
+                            )
+                            .first()
+                        )
 
             valid_dataset_documents = {}
             image_doc_ids: list[Any] = []
@@ -507,7 +570,47 @@ class RetrievalService:
                                 max_score = max(
                                     max_score, file_document.metadata.get("score", 0.0) if file_document else 0.0
                                 )
+                                segment = session.scalar(document_segment_stmt)
+                                if segment:
+                                    segment_file_map[segment.id] = [attachment_info]
+                        else:
+                            # Check if this is a summary document
+                            is_summary = document.metadata.get("is_summary", False)
+                            if is_summary:
+                                # For summary documents, find the original chunk via original_chunk_id
+                                original_chunk_id = document.metadata.get("original_chunk_id")
+                                if not original_chunk_id:
+                                    continue
+                                # Track that this segment was retrieved via summary
+                                summary_segment_ids.add(original_chunk_id)
+                                document_segment_stmt = select(DocumentSegment).where(
+                                    DocumentSegment.dataset_id == dataset_document.dataset_id,
+                                    DocumentSegment.enabled == True,
+                                    DocumentSegment.status == "completed",
+                                    DocumentSegment.id == original_chunk_id,
+                                )
+                                segment = session.scalar(document_segment_stmt)
+                            else:
+                                # For normal documents, find by index_node_id
+                                index_node_id = document.metadata.get("doc_id")
+                                if not index_node_id:
+                                    continue
+                                document_segment_stmt = select(DocumentSegment).where(
+                                    DocumentSegment.dataset_id == dataset_document.dataset_id,
+                                    DocumentSegment.enabled == True,
+                                    DocumentSegment.status == "completed",
+                                    DocumentSegment.index_node_id == index_node_id,
+                                )
+                                segment = session.scalar(document_segment_stmt)
 
+                        if not segment:
+                            continue
+                        if segment.id not in include_segment_ids:
+                            include_segment_ids.add(segment.id)
+                            record = {
+                                "segment": segment,
+                                "score": document.metadata.get("score"),  # type: ignore
+                            }
                             map_detail = {
                                 "max_score": max_score,
                                 "child_chunks": child_chunk_details,
@@ -542,6 +645,23 @@ class RetrievalService:
                 if record["segment"].id in attachment_map:
                     record["files"] = attachment_map[record["segment"].id]  # type: ignore[assignment]
 
+            # Batch query summaries for segments retrieved via summary (only enabled summaries)
+            if summary_segment_ids:
+                from models.dataset import DocumentSegmentSummary
+
+                summaries = (
+                    session.query(DocumentSegmentSummary)
+                    .filter(
+                        DocumentSegmentSummary.chunk_id.in_(summary_segment_ids),
+                        DocumentSegmentSummary.status == "completed",
+                        DocumentSegmentSummary.enabled == True,  # Only retrieve enabled summaries
+                    )
+                    .all()
+                )
+                for summary in summaries:
+                    if summary.summary_content:
+                        segment_summary_map[summary.chunk_id] = summary.summary_content
+
             result: list[RetrievalSegments] = []
             for record in records:
                 # Extract segment
@@ -576,9 +696,16 @@ class RetrievalService:
                     else None
                 )
 
+                # Extract summary if this segment was retrieved via summary
+                summary_content = segment_summary_map.get(segment.id)
+
                 # Create RetrievalSegments object
                 retrieval_segment = RetrievalSegments(
-                    segment=segment, child_chunks=child_chunks_list, score=score, files=files
+                    segment=segment,
+                    child_chunks=child_chunks_list,
+                    score=score,
+                    files=files,
+                    summary=summary_content
                 )
                 result.append(retrieval_segment)
 

api/core/rag/embedding/retrieval.py

@@ -20,3 +20,4 @@ class RetrievalSegments(BaseModel):
     child_chunks: list[RetrievalChildChunk] | None = None
     score: float | None = None
     files: list[dict[str, str | int]] | None = None
+    summary: str | None = None  # Summary content if retrieved via summary index

api/core/rag/index_processor/index_processor_base.py

@@ -13,6 +13,7 @@ from urllib.parse import unquote, urlparse
 import httpx
 
 from configs import dify_config
+from core.entities.knowledge_entities import PreviewDetail
 from core.helper import ssrf_proxy
 from core.rag.extractor.entity.extract_setting import ExtractSetting
 from core.rag.index_processor.constant.doc_type import DocType
@@ -45,6 +46,15 @@ class BaseIndexProcessor(ABC):
     def transform(self, documents: list[Document], current_user: Account | None = None, **kwargs) -> list[Document]:
         raise NotImplementedError
 
+    @abstractmethod
+    def generate_summary_preview(self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict) -> list[PreviewDetail]:
+        """
+        For each segment in preview_texts, generate a summary using LLM and attach it to the segment.
+        The summary can be stored in a new attribute, e.g., summary.
+        This method should be implemented by subclasses.
+        """
+        raise NotImplementedError
+
     @abstractmethod
     def load(
         self,

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

@@ -1,9 +1,13 @@
 """Paragraph index processor."""
 
+import logging
 import uuid
 from collections.abc import Mapping
 from typing import Any
 
+logger = logging.getLogger(__name__)
+
+from core.entities.knowledge_entities import PreviewDetail
 from core.rag.cleaner.clean_processor import CleanProcessor
 from core.rag.datasource.keyword.keyword_factory import Keyword
 from core.rag.datasource.retrieval_service import RetrievalService
@@ -17,12 +21,19 @@ from core.rag.index_processor.index_processor_base import BaseIndexProcessor
 from core.rag.models.document import AttachmentDocument, Document, MultimodalGeneralStructureChunk
 from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.tools.utils.text_processing_utils import remove_leading_symbols
+from extensions.ext_database import db
 from libs import helper
 from models.account import Account
-from models.dataset import Dataset, DatasetProcessRule
+from models.dataset import Dataset, DatasetProcessRule, DocumentSegment
 from models.dataset import Document as DatasetDocument
 from services.account_service import AccountService
 from services.entities.knowledge_entities.knowledge_entities import Rule
+from services.summary_index_service import SummaryIndexService
+from core.llm_generator.prompts import DEFAULT_GENERATOR_SUMMARY_PROMPT
+from core.model_runtime.entities.message_entities import UserPromptMessage
+from core.model_runtime.entities.model_entities import ModelType
+from core.provider_manager import ProviderManager
+from core.model_manager import ModelInstance
 
 
 class ParagraphIndexProcessor(BaseIndexProcessor):
@@ -108,6 +119,29 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
                 keyword.add_texts(documents)
 
     def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
+        # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
+        # This method is called for actual deletion scenarios (e.g., when segment is deleted).
+        # For disable operations, disable_summaries_for_segments is called directly in the task.
+        # Only delete summaries if explicitly requested (e.g., when segment is actually deleted)
+        delete_summaries = kwargs.get("delete_summaries", False)
+        if delete_summaries:
+            if node_ids:
+                # Find segments by index_node_id
+                segments = (
+                    db.session.query(DocumentSegment)
+                    .filter(
+                        DocumentSegment.dataset_id == dataset.id,
+                        DocumentSegment.index_node_id.in_(node_ids),
+                    )
+                    .all()
+                )
+                segment_ids = [segment.id for segment in segments]
+                if segment_ids:
+                    SummaryIndexService.delete_summaries_for_segments(dataset, segment_ids)
+            else:
+                # Delete all summaries for the dataset
+                SummaryIndexService.delete_summaries_for_segments(dataset, None)
+
         if dataset.indexing_technique == "high_quality":
             vector = Vector(dataset)
             if node_ids:
@@ -227,3 +261,70 @@ class ParagraphIndexProcessor(BaseIndexProcessor):
             }
         else:
             raise ValueError("Chunks is not a list")
+
+    def generate_summary_preview(self, tenant_id: str, preview_texts: list[PreviewDetail], summary_index_setting: dict) -> list[PreviewDetail]:
+        """
+        For each segment, concurrently call generate_summary to generate a summary
+        and write it to the summary attribute of PreviewDetail.
+        """
+        import concurrent.futures
+        from flask import current_app
+
+        # Capture Flask app context for worker threads
+        flask_app = None
+        try:
+            flask_app = current_app._get_current_object()  # type: ignore
+        except RuntimeError:
+            logger.warning("No Flask application context available, summary generation may fail")
+
+        def process(preview: PreviewDetail) -> None:
+            """Generate summary for a single preview item."""
+            try:
+                if flask_app:
+                    # Ensure Flask app context in worker thread
+                    with flask_app.app_context():
+                        summary = self.generate_summary(tenant_id, preview.content, summary_index_setting)
+                        preview.summary = summary
+                else:
+                    # Fallback: try without app context (may fail)
+                    summary = self.generate_summary(tenant_id, preview.content, summary_index_setting)
+                    preview.summary = summary
+            except Exception as e:
+                logger.error(f"Failed to generate summary for preview: {str(e)}")
+                # Don't fail the entire preview if summary generation fails
+                preview.summary = None
+
+        with concurrent.futures.ThreadPoolExecutor() as executor:
+            list(executor.map(process, preview_texts))
+        return preview_texts
+
+    @staticmethod
+    def generate_summary(tenant_id: str, text: str, summary_index_setting: dict = None) -> str:
+        """
+        Generate summary for the given text using ModelInstance.invoke_llm and the default or custom summary prompt.
+        """
+        if not summary_index_setting or not summary_index_setting.get("enable"):
+            raise ValueError("summary_index_setting is required and must be enabled to generate summary.")
+
+        model_name = summary_index_setting.get("model_name")
+        model_provider_name = summary_index_setting.get("model_provider_name")
+        summary_prompt = summary_index_setting.get("summary_prompt")
+
+        # Import default summary prompt
+        if not summary_prompt:
+            summary_prompt = DEFAULT_GENERATOR_SUMMARY_PROMPT
+
+        prompt = f"{summary_prompt}\n{text}"
+
+        provider_manager = ProviderManager()
+        provider_model_bundle = provider_manager.get_provider_model_bundle(tenant_id, model_provider_name, ModelType.LLM)
+        model_instance = ModelInstance(provider_model_bundle, model_name)
+        prompt_messages = [UserPromptMessage(content=prompt)]
+
+        result = model_instance.invoke_llm(
+            prompt_messages=prompt_messages,
+            model_parameters={},
+            stream=False
+        )
+
+        return getattr(result.message, "content", "")

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

@@ -25,6 +25,7 @@ from models.dataset import ChildChunk, Dataset, DatasetProcessRule, DocumentSegm
 from models.dataset import Document as DatasetDocument
 from services.account_service import AccountService
 from services.entities.knowledge_entities.knowledge_entities import ParentMode, Rule
+from services.summary_index_service import SummaryIndexService
 
 
 class ParentChildIndexProcessor(BaseIndexProcessor):
@@ -135,6 +136,29 @@ class ParentChildIndexProcessor(BaseIndexProcessor):
 
     def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
         # node_ids is segment's node_ids
+        # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
+        # This method is called for actual deletion scenarios (e.g., when segment is deleted).
+        # For disable operations, disable_summaries_for_segments is called directly in the task.
+        # Only delete summaries if explicitly requested (e.g., when segment is actually deleted)
+        delete_summaries = kwargs.get("delete_summaries", False)
+        if delete_summaries:
+            if node_ids:
+                # Find segments by index_node_id
+                segments = (
+                    db.session.query(DocumentSegment)
+                    .filter(
+                        DocumentSegment.dataset_id == dataset.id,
+                        DocumentSegment.index_node_id.in_(node_ids),
+                    )
+                    .all()
+                )
+                segment_ids = [segment.id for segment in segments]
+                if segment_ids:
+                    SummaryIndexService.delete_summaries_for_segments(dataset, segment_ids)
+            else:
+                # Delete all summaries for the dataset
+                SummaryIndexService.delete_summaries_for_segments(dataset, None)
+
         if dataset.indexing_technique == "high_quality":
             delete_child_chunks = kwargs.get("delete_child_chunks") or False
             precomputed_child_node_ids = kwargs.get("precomputed_child_node_ids")

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

@@ -25,9 +25,10 @@ from core.rag.retrieval.retrieval_methods import RetrievalMethod
 from core.tools.utils.text_processing_utils import remove_leading_symbols
 from libs import helper
 from models.account import Account
-from models.dataset import Dataset
+from models.dataset import Dataset, DocumentSegment
 from models.dataset import Document as DatasetDocument
 from services.entities.knowledge_entities.knowledge_entities import Rule
+from services.summary_index_service import SummaryIndexService
 
 logger = logging.getLogger(__name__)
 
@@ -144,6 +145,30 @@ class QAIndexProcessor(BaseIndexProcessor):
                 vector.create_multimodal(multimodal_documents)
 
     def clean(self, dataset: Dataset, node_ids: list[str] | None, with_keywords: bool = True, **kwargs):
+        # Note: Summary indexes are now disabled (not deleted) when segments are disabled.
+        # This method is called for actual deletion scenarios (e.g., when segment is deleted).
+        # For disable operations, disable_summaries_for_segments is called directly in the task.
+        # Note: qa_model doesn't generate summaries, but we clean them for completeness
+        # Only delete summaries if explicitly requested (e.g., when segment is actually deleted)
+        delete_summaries = kwargs.get("delete_summaries", False)
+        if delete_summaries:
+            if node_ids:
+                # Find segments by index_node_id
+                segments = (
+                    db.session.query(DocumentSegment)
+                    .filter(
+                        DocumentSegment.dataset_id == dataset.id,
+                        DocumentSegment.index_node_id.in_(node_ids),
+                    )
+                    .all()
+                )
+                segment_ids = [segment.id for segment in segments]
+                if segment_ids:
+                    SummaryIndexService.delete_summaries_for_segments(dataset, segment_ids)
+            else:
+                # Delete all summaries for the dataset
+                SummaryIndexService.delete_summaries_for_segments(dataset, None)
+
         vector = Vector(dataset)
         if node_ids:
             vector.delete_by_ids(node_ids)

api/core/workflow/nodes/document_extractor/node.py

@@ -62,6 +62,21 @@ class DocumentExtractorNode(Node[DocumentExtractorNodeData]):
         inputs = {"variable_selector": variable_selector}
         process_data = {"documents": value if isinstance(value, list) else [value]}
 
+        # Ensure storage_key is loaded for File objects
+        files_to_check = value if isinstance(value, list) else [value]
+        files_needing_storage_key = [
+            f for f in files_to_check
+            if isinstance(f, File) and not f.storage_key and f.related_id
+        ]
+        if files_needing_storage_key:
+            from factories.file_factory import StorageKeyLoader
+            from extensions.ext_database import db
+            from sqlalchemy.orm import Session
+            
+            with Session(bind=db.engine) as session:
+                storage_key_loader = StorageKeyLoader(session, tenant_id=self.tenant_id)
+                storage_key_loader.load_storage_keys(files_needing_storage_key)
+
         try:
             if isinstance(value, list):
                 extracted_text_list = list(map(_extract_text_from_file, value))
@@ -415,6 +430,15 @@ def _download_file_content(file: File) -> bytes:
             response.raise_for_status()
             return response.content
         else:
+            # Check if storage_key is set
+            if not file.storage_key:
+                raise FileDownloadError(f"File storage_key is missing for file: {file.filename}")
+            
+            # Check if file exists before downloading
+            from extensions.ext_storage import storage
+            if not storage.exists(file.storage_key):
+                raise FileDownloadError(f"File not found in storage: {file.storage_key}")
+            
             return file_manager.download(file)
     except Exception as e:
         raise FileDownloadError(f"Error downloading file: {str(e)}") from e

api/core/workflow/nodes/knowledge_index/entities.py

@@ -158,3 +158,5 @@ class KnowledgeIndexNodeData(BaseNodeData):
     type: str = "knowledge-index"
     chunk_structure: str
     index_chunk_variable_selector: list[str]
+    indexing_technique: str | None = None
+    summary_index_setting: dict | None = None

api/core/workflow/nodes/knowledge_index/knowledge_index_node.py

@@ -1,9 +1,11 @@
+import concurrent.futures
 import datetime
 import logging
 import time
 from collections.abc import Mapping
 from typing import Any
 
+from flask import current_app
 from sqlalchemy import func, select
 
 from core.app.entities.app_invoke_entities import InvokeFrom
@@ -16,7 +18,9 @@ from core.workflow.nodes.base.node import Node
 from core.workflow.nodes.base.template import Template
 from core.workflow.runtime import VariablePool
 from extensions.ext_database import db
-from models.dataset import Dataset, Document, DocumentSegment
+from models.dataset import Dataset, Document, DocumentSegment, DocumentSegmentSummary
+from services.summary_index_service import SummaryIndexService
+from tasks.generate_summary_index_task import generate_summary_index_task
 
 from .entities import KnowledgeIndexNodeData
 from .exc import (
@@ -67,7 +71,18 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
         # index knowledge
         try:
             if is_preview:
-                outputs = self._get_preview_output(node_data.chunk_structure, chunks)
+                # Preview mode: generate summaries for chunks directly without saving to database
+                # Format preview and generate summaries on-the-fly
+                # Get indexing_technique and summary_index_setting from node_data (workflow graph config)
+                # or fallback to dataset if not available in node_data
+                indexing_technique = node_data.indexing_technique or dataset.indexing_technique
+                summary_index_setting = node_data.summary_index_setting or dataset.summary_index_setting
+                
+                outputs = self._get_preview_output_with_summaries(
+                    node_data.chunk_structure, chunks, dataset=dataset,
+                    indexing_technique=indexing_technique,
+                    summary_index_setting=summary_index_setting
+                )
                 return NodeRunResult(
                     status=WorkflowNodeExecutionStatus.SUCCEEDED,
                     inputs=variables,
@@ -163,6 +178,9 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
 
         db.session.commit()
 
+        # Generate summary index if enabled
+        self._handle_summary_index_generation(dataset, document, variable_pool)
+
         return {
             "dataset_id": ds_id_value,
             "dataset_name": dataset_name_value,
@@ -173,9 +191,269 @@ class KnowledgeIndexNode(Node[KnowledgeIndexNodeData]):
             "display_status": "completed",
         }
 
-    def _get_preview_output(self, chunk_structure: str, chunks: Any) -> Mapping[str, Any]:
+    def _handle_summary_index_generation(
+        self,
+        dataset: Dataset,
+        document: Document,
+        variable_pool: VariablePool,
+    ) -> None:
+        """
+        Handle summary index generation based on mode (debug/preview or production).
+
+        Args:
+            dataset: Dataset containing the document
+            document: Document to generate summaries for
+            variable_pool: Variable pool to check invoke_from
+        """
+        # Only generate summary index for high_quality indexing technique
+        if dataset.indexing_technique != "high_quality":
+            return
+
+        # Check if summary index is enabled
+        summary_index_setting = dataset.summary_index_setting
+        if not summary_index_setting or not summary_index_setting.get("enable"):
+            return
+
+        # Skip qa_model documents
+        if document.doc_form == "qa_model":
+            return
+
+        # Determine if in preview/debug mode
+        invoke_from = variable_pool.get(["sys", SystemVariableKey.INVOKE_FROM])
+        is_preview = invoke_from and invoke_from.value == InvokeFrom.DEBUGGER
+
+        # Determine if only parent chunks should be processed
+        only_parent_chunks = dataset.chunk_structure == "parent_child_index"
+
+        if is_preview:
+            try:
+                # Query segments that need summary generation
+                query = db.session.query(DocumentSegment).filter_by(
+                    dataset_id=dataset.id,
+                    document_id=document.id,
+                    status="completed",
+                    enabled=True,
+                )
+                segments = query.all()
+
+                if not segments:
+                    logger.info(f"No segments found for document {document.id}")
+                    return
+
+                # Filter segments based on mode
+                segments_to_process = []
+                for segment in segments:
+                    # Skip if summary already exists
+                    existing_summary = (
+                        db.session.query(DocumentSegmentSummary)
+                        .filter_by(chunk_id=segment.id, dataset_id=dataset.id, status="completed")
+                        .first()
+                    )
+                    if existing_summary:
+                        continue
+
+                    # For parent-child mode, all segments are parent chunks, so process all
+                    segments_to_process.append(segment)
+
+                if not segments_to_process:
+                    logger.info(f"No segments need summary generation for document {document.id}")
+                    return
+
+                # Use ThreadPoolExecutor for concurrent generation
+                flask_app = current_app._get_current_object()  # type: ignore
+                max_workers = min(10, len(segments_to_process))  # Limit to 10 workers
+
+                def process_segment(segment: DocumentSegment) -> None:
+                    """Process a single segment in a thread with Flask app context."""
+                    with flask_app.app_context():
+                        try:
+                            SummaryIndexService.generate_and_vectorize_summary(
+                                segment, dataset, summary_index_setting
+                            )
+                        except Exception as e:
+                            logger.error(f"Failed to generate summary for segment {segment.id}: {str(e)}")
+                            # Continue processing other segments
+
+                with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+                    futures = [
+                        executor.submit(process_segment, segment) for segment in segments_to_process
+                    ]
+                    # Wait for all tasks to complete
+                    concurrent.futures.wait(futures, timeout=300)
+
+                logger.info(
+                    f"Successfully generated summary index for {len(segments_to_process)} segments "
+                    f"in document {document.id}"
+                )
+            except Exception as e:
+                logger.exception(f"Failed to generate summary index for document {document.id}: {str(e)}")
+                # Don't fail the entire indexing process if summary generation fails
+        else:
+            # Production mode: asynchronous generation
+            logger.info(f"Queuing summary index generation task for document {document.id} (production mode)")
+            try:
+                generate_summary_index_task.delay(dataset.id, document.id, None)
+                logger.info(f"Summary index generation task queued for document {document.id}")
+            except Exception as e:
+                logger.exception(f"Failed to queue summary index generation task for document {document.id}: {str(e)}")
+                # Don't fail the entire indexing process if task queuing fails
+
+    def _get_preview_output_with_summaries(
+        self, chunk_structure: str, chunks: Any, dataset: Dataset,
+        indexing_technique: str | None = None,
+        summary_index_setting: dict | None = None
+    ) -> Mapping[str, Any]:
+        """
+        Generate preview output with summaries for chunks in preview mode.
+        This method generates summaries on-the-fly without saving to database.
+        
+        Args:
+            chunk_structure: Chunk structure type
+            chunks: Chunks to generate preview for
+            dataset: Dataset object (for tenant_id)
+            indexing_technique: Indexing technique from node config or dataset
+            summary_index_setting: Summary index setting from node config or dataset
+        """
         index_processor = IndexProcessorFactory(chunk_structure).init_index_processor()
-        return index_processor.format_preview(chunks)
+        preview_output = index_processor.format_preview(chunks)
+        
+        # Check if summary index is enabled
+        if indexing_technique != "high_quality":
+            return preview_output
+        
+        if not summary_index_setting or not summary_index_setting.get("enable"):
+            return preview_output
+        
+        # Generate summaries for chunks
+        if "preview" in preview_output and isinstance(preview_output["preview"], list):
+            chunk_count = len(preview_output["preview"])
+            logger.info(
+                f"Generating summaries for {chunk_count} chunks in preview mode "
+                f"(dataset: {dataset.id})"
+            )
+            # Use ParagraphIndexProcessor's generate_summary method
+            from core.rag.index_processor.processor.paragraph_index_processor import ParagraphIndexProcessor
+            
+            # Get Flask app for application context in worker threads
+            flask_app = None
+            try:
+                flask_app = current_app._get_current_object()  # type: ignore
+            except RuntimeError:
+                logger.warning("No Flask application context available, summary generation may fail")
+            
+            def generate_summary_for_chunk(preview_item: dict) -> None:
+                """Generate summary for a single chunk."""
+                if "content" in preview_item:
+                    try:
+                        # Set Flask application context in worker thread
+                        if flask_app:
+                            with flask_app.app_context():
+                                summary = ParagraphIndexProcessor.generate_summary(
+                                    tenant_id=dataset.tenant_id,
+                                    text=preview_item["content"],
+                                    summary_index_setting=summary_index_setting,
+                                )
+                                if summary:
+                                    preview_item["summary"] = summary
+                        else:
+                            # Fallback: try without app context (may fail)
+                            summary = ParagraphIndexProcessor.generate_summary(
+                                tenant_id=dataset.tenant_id,
+                                text=preview_item["content"],
+                                summary_index_setting=summary_index_setting,
+                            )
+                            if summary:
+                                preview_item["summary"] = summary
+                    except Exception as e:
+                        logger.error(f"Failed to generate summary for chunk: {str(e)}")
+                        # Don't fail the entire preview if summary generation fails
+            
+            # Generate summaries concurrently using ThreadPoolExecutor
+            # Set a reasonable timeout to prevent hanging (60 seconds per chunk, max 5 minutes total)
+            timeout_seconds = min(300, 60 * len(preview_output["preview"]))
+            with concurrent.futures.ThreadPoolExecutor(max_workers=min(10, len(preview_output["preview"]))) as executor:
+                futures = [
+                    executor.submit(generate_summary_for_chunk, preview_item)
+                    for preview_item in preview_output["preview"]
+                ]
+                # Wait for all tasks to complete with timeout
+                done, not_done = concurrent.futures.wait(futures, timeout=timeout_seconds)
+                
+                # Cancel tasks that didn't complete in time
+                if not_done:
+                    logger.warning(
+                        f"Summary generation timeout: {len(not_done)} chunks did not complete within {timeout_seconds}s. "
+                        "Cancelling remaining tasks..."
+                    )
+                    for future in not_done:
+                        future.cancel()
+                    # Wait a bit for cancellation to take effect
+                    concurrent.futures.wait(not_done, timeout=5)
+            
+            completed_count = sum(1 for item in preview_output["preview"] if item.get("summary") is not None)
+            logger.info(
+                f"Completed summary generation for preview chunks: {completed_count}/{len(preview_output['preview'])} succeeded"
+            )
+        
+        return preview_output
+
+    def _get_preview_output(
+        self, chunk_structure: str, chunks: Any, dataset: Dataset | None = None, variable_pool: VariablePool | None = None
+    ) -> Mapping[str, Any]:
+        index_processor = IndexProcessorFactory(chunk_structure).init_index_processor()
+        preview_output = index_processor.format_preview(chunks)
+        
+        # If dataset is provided, try to enrich preview with summaries
+        if dataset and variable_pool:
+            document_id = variable_pool.get(["sys", SystemVariableKey.DOCUMENT_ID])
+            if document_id:
+                document = db.session.query(Document).filter_by(id=document_id.value).first()
+                if document:
+                    # Query summaries for this document
+                    summaries = (
+                        db.session.query(DocumentSegmentSummary)
+                        .filter_by(
+                            dataset_id=dataset.id,
+                            document_id=document.id,
+                            status="completed",
+                            enabled=True,
+                        )
+                        .all()
+                    )
+                    
+                    if summaries:
+                        # Create a map of segment content to summary for matching
+                        # Use content matching as chunks in preview might not be indexed yet
+                        summary_by_content = {}
+                        for summary in summaries:
+                            segment = (
+                                db.session.query(DocumentSegment)
+                                .filter_by(id=summary.chunk_id, dataset_id=dataset.id)
+                                .first()
+                            )
+                            if segment:
+                                # Normalize content for matching (strip whitespace)
+                                normalized_content = segment.content.strip()
+                                summary_by_content[normalized_content] = summary.summary_content
+                        
+                        # Enrich preview with summaries by content matching
+                        if "preview" in preview_output and isinstance(preview_output["preview"], list):
+                            matched_count = 0
+                            for preview_item in preview_output["preview"]:
+                                if "content" in preview_item:
+                                    # Normalize content for matching
+                                    normalized_chunk_content = preview_item["content"].strip()
+                                    if normalized_chunk_content in summary_by_content:
+                                        preview_item["summary"] = summary_by_content[normalized_chunk_content]
+                                        matched_count += 1
+                            
+                            if matched_count > 0:
+                                logger.info(
+                                    f"Enriched preview with {matched_count} existing summaries "
+                                    f"(dataset: {dataset.id}, document: {document.id})"
+                                )
+        
+        return preview_output
 
     @classmethod
     def version(cls) -> str:

api/core/workflow/nodes/llm/llm_utils.py

@@ -6,7 +6,7 @@ from sqlalchemy.orm import Session
 
 from configs import dify_config
 from core.app.entities.app_invoke_entities import ModelConfigWithCredentialsEntity
-from core.entities.provider_entities import QuotaUnit
+from core.entities.provider_entities import ProviderQuotaType, QuotaUnit
 from core.file.models import File
 from core.memory.token_buffer_memory import TokenBufferMemory
 from core.model_manager import ModelInstance, ModelManager
@@ -136,21 +136,37 @@ def deduct_llm_quota(tenant_id: str, model_instance: ModelInstance, usage: LLMUs
             used_quota = 1
 
     if used_quota is not None and system_configuration.current_quota_type is not None:
-        with Session(db.engine) as session:
-            stmt = (
-                update(Provider)
-                .where(
-                    Provider.tenant_id == tenant_id,
-                    # TODO: Use provider name with prefix after the data migration.
-                    Provider.provider_name == ModelProviderID(model_instance.provider).provider_name,
-                    Provider.provider_type == ProviderType.SYSTEM,
-                    Provider.quota_type == system_configuration.current_quota_type.value,
-                    Provider.quota_limit > Provider.quota_used,
-                )
-                .values(
-                    quota_used=Provider.quota_used + used_quota,
-                    last_used=naive_utc_now(),
-                )
+        if system_configuration.current_quota_type == ProviderQuotaType.TRIAL:
+            from services.credit_pool_service import CreditPoolService
+
+            CreditPoolService.check_and_deduct_credits(
+                tenant_id=tenant_id,
+                credits_required=used_quota,
             )
-            session.execute(stmt)
-            session.commit()
+        elif system_configuration.current_quota_type == ProviderQuotaType.PAID:
+            from services.credit_pool_service import CreditPoolService
+
+            CreditPoolService.check_and_deduct_credits(
+                tenant_id=tenant_id,
+                credits_required=used_quota,
+                pool_type="paid",
+            )
+        else:
+            with Session(db.engine) as session:
+                stmt = (
+                    update(Provider)
+                    .where(
+                        Provider.tenant_id == tenant_id,
+                        # TODO: Use provider name with prefix after the data migration.
+                        Provider.provider_name == ModelProviderID(model_instance.provider).provider_name,
+                        Provider.provider_type == ProviderType.SYSTEM.value,
+                        Provider.quota_type == system_configuration.current_quota_type.value,
+                        Provider.quota_limit > Provider.quota_used,
+                    )
+                    .values(
+                        quota_used=Provider.quota_used + used_quota,
+                        last_used=naive_utc_now(),
+                    )
+                )
+                session.execute(stmt)
+                session.commit()

api/core/workflow/nodes/llm/node.py

@@ -1529,7 +1529,7 @@ class LLMNode(Node[LLMNodeData]):
         model_instance: ModelInstance,
         prompt_messages: Sequence[PromptMessage],
         stop: Sequence[str] | None,
-        files: Sequence["File"],
+        files: Sequence[File],
         variable_pool: VariablePool,
         node_inputs: dict[str, Any],
         process_data: dict[str, Any],
@@ -1636,7 +1636,7 @@ class LLMNode(Node[LLMNodeData]):
 
         return tool_instances
 
-    def _extract_prompt_files(self, variable_pool: VariablePool) -> list["File"]:
+    def _extract_prompt_files(self, variable_pool: VariablePool) -> list[File]:
         """Extract files from prompt template variables."""
         from core.variables import ArrayFileVariable, FileVariable
 

api/core/workflow/nodes/start/start_node.py

@@ -1,4 +1,3 @@
-import json
 from typing import Any
 
 from jsonschema import Draft7Validator, ValidationError
@@ -43,25 +42,22 @@ class StartNode(Node[StartNodeData]):
             if value is None and variable.required:
                 raise ValueError(f"{key} is required in input form")
 
+            # If no value provided, skip further processing for this key
+            if not value:
+                continue
+
+            if not isinstance(value, dict):
+                raise ValueError(f"JSON object for '{key}' must be an object")
+
+            # Overwrite with normalized dict to ensure downstream consistency
+            node_inputs[key] = value
+
+            # If schema exists, then validate against it
             schema = variable.json_schema
             if not schema:
                 continue
 
-            if not value:
-                continue
-
             try:
-                json_schema = json.loads(schema)
-            except json.JSONDecodeError as e:
-                raise ValueError(f"{schema} must be a valid JSON object")
-
-            try:
-                json_value = json.loads(value)
-            except json.JSONDecodeError as e:
-                raise ValueError(f"{value} must be a valid JSON object")
-
-            try:
-                Draft7Validator(json_schema).validate(json_value)
+                Draft7Validator(schema).validate(value)
             except ValidationError as e:
                 raise ValueError(f"JSON object for '{key}' does not match schema: {e.message}")
-            node_inputs[key] = json_value

api/events/event_handlers/update_provider_when_message_created.py

@@ -10,7 +10,7 @@ from sqlalchemy.orm import Session
 
 from configs import dify_config
 from core.app.entities.app_invoke_entities import AgentChatAppGenerateEntity, ChatAppGenerateEntity
-from core.entities.provider_entities import QuotaUnit, SystemConfiguration
+from core.entities.provider_entities import ProviderQuotaType, QuotaUnit, SystemConfiguration
 from events.message_event import message_was_created
 from extensions.ext_database import db
 from extensions.ext_redis import redis_client, redis_fallback
@@ -134,22 +134,38 @@ def handle(sender: Message, **kwargs):
             system_configuration=system_configuration,
             model_name=model_config.model,
         )
-
         if used_quota is not None:
-            quota_update = _ProviderUpdateOperation(
-                filters=_ProviderUpdateFilters(
+            if provider_configuration.system_configuration.current_quota_type == ProviderQuotaType.TRIAL:
+                from services.credit_pool_service import CreditPoolService
+
+                CreditPoolService.check_and_deduct_credits(
                     tenant_id=tenant_id,
-                    provider_name=ModelProviderID(model_config.provider).provider_name,
-                    provider_type=ProviderType.SYSTEM,
-                    quota_type=provider_configuration.system_configuration.current_quota_type.value,
-                ),
-                values=_ProviderUpdateValues(quota_used=Provider.quota_used + used_quota, last_used=current_time),
-                additional_filters=_ProviderUpdateAdditionalFilters(
-                    quota_limit_check=True  # Provider.quota_limit > Provider.quota_used
-                ),
-                description="quota_deduction_update",
-            )
-            updates_to_perform.append(quota_update)
+                    credits_required=used_quota,
+                    pool_type="trial",
+                )
+            elif provider_configuration.system_configuration.current_quota_type == ProviderQuotaType.PAID:
+                from services.credit_pool_service import CreditPoolService
+
+                CreditPoolService.check_and_deduct_credits(
+                    tenant_id=tenant_id,
+                    credits_required=used_quota,
+                    pool_type="paid",
+                )
+            else:
+                quota_update = _ProviderUpdateOperation(
+                    filters=_ProviderUpdateFilters(
+                        tenant_id=tenant_id,
+                        provider_name=ModelProviderID(model_config.provider).provider_name,
+                        provider_type=ProviderType.SYSTEM.value,
+                        quota_type=provider_configuration.system_configuration.current_quota_type.value,
+                    ),
+                    values=_ProviderUpdateValues(quota_used=Provider.quota_used + used_quota, last_used=current_time),
+                    additional_filters=_ProviderUpdateAdditionalFilters(
+                        quota_limit_check=True  # Provider.quota_limit > Provider.quota_used
+                    ),
+                    description="quota_deduction_update",
+                )
+                updates_to_perform.append(quota_update)
 
     # Execute all updates
     start_time = time_module.perf_counter()

api/extensions/ext_celery.py

@@ -102,6 +102,8 @@ def init_app(app: DifyApp) -> Celery:
     imports = [
         "tasks.async_workflow_tasks",  # trigger workers
         "tasks.trigger_processing_tasks",  # async trigger processing
+        "tasks.generate_summary_index_task",  # summary index generation
+        "tasks.regenerate_summary_index_task",  # summary index regeneration
     ]
     day = dify_config.CELERY_BEAT_SCHEDULER_TIME
 
@@ -163,6 +165,13 @@ def init_app(app: DifyApp) -> Celery:
             "task": "schedule.clean_workflow_runlogs_precise.clean_workflow_runlogs_precise",
             "schedule": crontab(minute="0", hour="2"),
         }
+    if dify_config.ENABLE_WORKFLOW_RUN_CLEANUP_TASK:
+        # for saas only
+        imports.append("schedule.clean_workflow_runs_task")
+        beat_schedule["clean_workflow_runs_task"] = {
+            "task": "schedule.clean_workflow_runs_task.clean_workflow_runs_task",
+            "schedule": crontab(minute="0", hour="0"),
+        }
     if dify_config.ENABLE_WORKFLOW_SCHEDULE_POLLER_TASK:
         imports.append("schedule.workflow_schedule_task")
         beat_schedule["workflow_schedule_task"] = {

api/extensions/ext_commands.py

@@ -4,6 +4,7 @@ from dify_app import DifyApp
 def init_app(app: DifyApp):
     from commands import (
         add_qdrant_index,
+        clean_workflow_runs,
         cleanup_orphaned_draft_variables,
         clear_free_plan_tenant_expired_logs,
         clear_orphaned_file_records,
@@ -22,6 +23,7 @@ def init_app(app: DifyApp):
         reset_email,
         reset_encrypt_key_pair,
         reset_password,
+        seed_expired_workflow_runs,
         setup_datasource_oauth_client,
         setup_system_tool_oauth_client,
         setup_system_trigger_oauth_client,
@@ -56,6 +58,8 @@ def init_app(app: DifyApp):
         setup_datasource_oauth_client,
         transform_datasource_credentials,
         install_rag_pipeline_plugins,
+        clean_workflow_runs,
+        seed_expired_workflow_runs,
     ]
     for cmd in cmds_to_register:
         app.cli.add_command(cmd)

api/fields/dataset_fields.py

@@ -39,6 +39,14 @@ dataset_retrieval_model_fields = {
     "score_threshold_enabled": fields.Boolean,
     "score_threshold": fields.Float,
 }
+
+dataset_summary_index_fields = {
+    "enable": fields.Boolean,
+    "model_name": fields.String,
+    "model_provider_name": fields.String,
+    "summary_prompt": fields.String,
+}
+
 external_retrieval_model_fields = {
     "top_k": fields.Integer,
     "score_threshold": fields.Float,
@@ -83,6 +91,7 @@ dataset_detail_fields = {
     "embedding_model_provider": fields.String,
     "embedding_available": fields.Boolean,
     "retrieval_model_dict": fields.Nested(dataset_retrieval_model_fields),
+    "summary_index_setting": fields.Nested(dataset_summary_index_fields),
     "tags": fields.List(fields.Nested(tag_fields)),
     "doc_form": fields.String,
     "external_knowledge_info": fields.Nested(external_knowledge_info_fields),

api/fields/document_fields.py

@@ -33,6 +33,8 @@ document_fields = {
     "hit_count": fields.Integer,
     "doc_form": fields.String,
     "doc_metadata": fields.List(fields.Nested(document_metadata_fields), attribute="doc_metadata_details"),
+    "summary_index_status": fields.String,  # Summary index generation status: "waiting", "generating", "completed", "partial_error", or null if not enabled
+    "need_summary": fields.Boolean,  # Whether this document needs summary index generation
 }
 
 document_with_segments_fields = {
@@ -60,6 +62,8 @@ document_with_segments_fields = {
     "completed_segments": fields.Integer,
     "total_segments": fields.Integer,
     "doc_metadata": fields.List(fields.Nested(document_metadata_fields), attribute="doc_metadata_details"),
+    "summary_index_status": fields.String,  # Summary index generation status: "waiting", "generating", "completed", "partial_error", or null if not enabled
+    "need_summary": fields.Boolean,  # Whether this document needs summary index generation
 }
 
 dataset_and_document_fields = {

api/fields/hit_testing_fields.py

@@ -58,4 +58,5 @@ hit_testing_record_fields = {
     "score": fields.Float,
     "tsne_position": fields.Raw,
     "files": fields.List(fields.Nested(files_fields)),
+    "summary": fields.String,  # Summary content if retrieved via summary index
 }

api/fields/segment_fields.py

@@ -49,4 +49,5 @@ segment_fields = {
     "stopped_at": TimestampField,
     "child_chunks": fields.List(fields.Nested(child_chunk_fields)),
     "attachments": fields.List(fields.Nested(attachment_fields)),
+    "summary": fields.String,  # Summary content for the segment
 }

api/migrations/versions/2025_12_10_1504-8a7f2ad7c23e_add_workflow_runs_created_at_idx.py

@@ -0,0 +1,29 @@
+"""Add index on workflow_runs.created_at
+
+Revision ID: 8a7f2ad7c23e
+Revises: d57accd375ae
+Create Date: 2025-12-10 15:04:00.000000
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = "8a7f2ad7c23e"
+down_revision = "d57accd375ae"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    with op.batch_alter_table("workflow_runs", schema=None) as batch_op:
+        batch_op.create_index(
+            batch_op.f("workflow_runs_created_at_idx"),
+            ["created_at"],
+            unique=False,
+        )
+
+
+def downgrade():
+    with op.batch_alter_table("workflow_runs", schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f("workflow_runs_created_at_idx"))

api/migrations/versions/2025_12_15_1614-6bb0832495f0_alter_table_pipeline_recommended_.py

@@ -0,0 +1,64 @@
+"""Alter table pipeline_recommended_plugins add column type
+
+Revision ID: 6bb0832495f0
+Revises: 7bb281b7a422
+Create Date: 2025-12-15 16:14:38.482072
+
+"""
+from alembic import op
+import models as models
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = '6bb0832495f0'
+down_revision = '7bb281b7a422'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('app_triggers', schema=None) as batch_op:
+        batch_op.alter_column('provider_name',
+               existing_type=sa.VARCHAR(length=255),
+               nullable=False,
+               existing_server_default=sa.text("''::character varying"))
+
+    with op.batch_alter_table('operation_logs', schema=None) as batch_op:
+        batch_op.alter_column('content',
+               existing_type=postgresql.JSON(astext_type=sa.Text()),
+               nullable=False)
+
+    with op.batch_alter_table('pipeline_recommended_plugins', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('type', sa.String(length=50), nullable=True))
+
+    with op.batch_alter_table('providers', schema=None) as batch_op:
+        batch_op.alter_column('quota_used',
+               existing_type=sa.BIGINT(),
+               nullable=False)
+
+    # ### end Alembic commands ###
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('providers', schema=None) as batch_op:
+        batch_op.alter_column('quota_used',
+               existing_type=sa.BIGINT(),
+               nullable=True)
+
+    with op.batch_alter_table('pipeline_recommended_plugins', schema=None) as batch_op:
+        batch_op.drop_column('type')
+
+    with op.batch_alter_table('operation_logs', schema=None) as batch_op:
+        batch_op.alter_column('content',
+               existing_type=postgresql.JSON(astext_type=sa.Text()),
+               nullable=True)
+
+    with op.batch_alter_table('app_triggers', schema=None) as batch_op:
+        batch_op.alter_column('provider_name',
+               existing_type=sa.VARCHAR(length=255),
+               nullable=True,
+               existing_server_default=sa.text("''::character varying"))
+
+    # ### end Alembic commands ###

api/migrations/versions/2025_12_16_1424-2536f83803a8_add_type_column_not_null_default_tool.py

@@ -0,0 +1,33 @@
+"""add type column not null default tool
+
+Revision ID: 2536f83803a8
+Revises: 6bb0832495f0
+Create Date: 2025-12-16 14:24:40.740253
+
+"""
+from alembic import op
+import models as models
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '2536f83803a8'
+down_revision = '6bb0832495f0'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('pipeline_recommended_plugins', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('type', sa.String(length=50), nullable=False, server_default='tool'))
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('pipeline_recommended_plugins', schema=None) as batch_op:
+        batch_op.drop_column('type')
+
+    # ### end Alembic commands ###

api/migrations/versions/2025_12_25_1039-7df29de0f6be_add_credit_pool.py

@@ -0,0 +1,46 @@
+"""add credit pool
+
+Revision ID: 7df29de0f6be
+Revises: 03ea244985ce
+Create Date: 2025-12-25 10:39:15.139304
+
+"""
+from alembic import op
+import models as models
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = '7df29de0f6be'
+down_revision = '03ea244985ce'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('tenant_credit_pools',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+    sa.Column('pool_type', sa.String(length=40), server_default='trial', nullable=False),
+    sa.Column('quota_limit', sa.BigInteger(), nullable=False),
+    sa.Column('quota_used', sa.BigInteger(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='tenant_credit_pool_pkey')
+    )
+    with op.batch_alter_table('tenant_credit_pools', schema=None) as batch_op:
+        batch_op.create_index('tenant_credit_pool_pool_type_idx', ['pool_type'], unique=False)
+        batch_op.create_index('tenant_credit_pool_tenant_id_idx', ['tenant_id'], unique=False)
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+   
+    with op.batch_alter_table('tenant_credit_pools', schema=None) as batch_op:
+        batch_op.drop_index('tenant_credit_pool_tenant_id_idx')
+        batch_op.drop_index('tenant_credit_pool_pool_type_idx')
+
+    op.drop_table('tenant_credit_pools')
+    # ### end Alembic commands ###

api/migrations/versions/2025_12_30_1617-85c8b4a64f53_add_llm_generation_detail_table.py

@@ -12,7 +12,7 @@ from sqlalchemy.dialects import postgresql
 
 # revision identifiers, used by Alembic.
 revision = '85c8b4a64f53'
-down_revision = '03ea244985ce'
+down_revision = '7df29de0f6be'
 branch_labels = None
 depends_on = None
 

api/migrations/versions/2026_01_09_1110-f9f6d18a37f9_add_table_explore_banner_and_trial.py

@@ -0,0 +1,73 @@
+"""add table explore banner and trial
+
+Revision ID: f9f6d18a37f9
+Revises: 7df29de0f6be
+Create Date: 2026-01-09 11:10:18.079355
+
+"""
+from alembic import op
+import models as models
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = 'f9f6d18a37f9'
+down_revision = '7df29de0f6be'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('account_trial_app_records',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('account_id', models.types.StringUUID(), nullable=False),
+    sa.Column('app_id', models.types.StringUUID(), nullable=False),
+    sa.Column('count', sa.Integer(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='user_trial_app_pkey'),
+    sa.UniqueConstraint('account_id', 'app_id', name='unique_account_trial_app_record')
+    )
+    with op.batch_alter_table('account_trial_app_records', schema=None) as batch_op:
+        batch_op.create_index('account_trial_app_record_account_id_idx', ['account_id'], unique=False)
+        batch_op.create_index('account_trial_app_record_app_id_idx', ['app_id'], unique=False)
+
+    op.create_table('exporle_banners',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('content', sa.JSON(), nullable=False),
+    sa.Column('link', sa.String(length=255), nullable=False),
+    sa.Column('sort', sa.Integer(), nullable=False),
+    sa.Column('status', sa.String(length=255), server_default=sa.text("'enabled'::character varying"), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('language', sa.String(length=255), server_default=sa.text("'en-US'::character varying"), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='exporler_banner_pkey')
+    )
+    op.create_table('trial_apps',
+    sa.Column('id', models.types.StringUUID(), server_default=sa.text('uuid_generate_v4()'), nullable=False),
+    sa.Column('app_id', models.types.StringUUID(), nullable=False),
+    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('trial_limit', sa.Integer(), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='trial_app_pkey'),
+    sa.UniqueConstraint('app_id', name='unique_trail_app_id')
+    )
+    with op.batch_alter_table('trial_apps', schema=None) as batch_op:
+        batch_op.create_index('trial_app_app_id_idx', ['app_id'], unique=False)
+        batch_op.create_index('trial_app_tenant_id_idx', ['tenant_id'], unique=False)
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('trial_apps', schema=None) as batch_op:
+        batch_op.drop_index('trial_app_tenant_id_idx')
+        batch_op.drop_index('trial_app_app_id_idx')
+
+    op.drop_table('trial_apps')
+    op.drop_table('exporle_banners')
+    with op.batch_alter_table('account_trial_app_records', schema=None) as batch_op:
+        batch_op.drop_index('account_trial_app_record_app_id_idx')
+        batch_op.drop_index('account_trial_app_record_account_id_idx')
+
+    op.drop_table('account_trial_app_records')
+    # ### end Alembic commands ###

api/migrations/versions/2026_01_12_1358-562dcce7d77c_add_summaryindex_feature.py

@@ -0,0 +1,69 @@
+"""add SummaryIndex feature
+
+Revision ID: 562dcce7d77c
+Revises: 03ea244985ce
+Create Date: 2026-01-12 13:58:40.584802
+
+"""
+from alembic import op
+import models as models
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '562dcce7d77c'
+down_revision = '03ea244985ce'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('document_segment_summary',
+    sa.Column('id', models.types.StringUUID(), nullable=False),
+    sa.Column('dataset_id', models.types.StringUUID(), nullable=False),
+    sa.Column('document_id', models.types.StringUUID(), nullable=False),
+    sa.Column('chunk_id', models.types.StringUUID(), nullable=False),
+    sa.Column('summary_content', models.types.LongText(), nullable=True),
+    sa.Column('summary_index_node_id', sa.String(length=255), nullable=True),
+    sa.Column('summary_index_node_hash', sa.String(length=255), nullable=True),
+    sa.Column('status', sa.String(length=32), server_default=sa.text("'generating'"), nullable=False),
+    sa.Column('error', models.types.LongText(), nullable=True),
+    sa.Column('enabled', sa.Boolean(), server_default=sa.text('true'), nullable=False),
+    sa.Column('disabled_at', sa.DateTime(), nullable=True),
+    sa.Column('disabled_by', models.types.StringUUID(), nullable=True),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='document_segment_summary_pkey')
+    )
+    with op.batch_alter_table('document_segment_summary', schema=None) as batch_op:
+        batch_op.create_index('document_segment_summary_chunk_id_idx', ['chunk_id'], unique=False)
+        batch_op.create_index('document_segment_summary_dataset_id_idx', ['dataset_id'], unique=False)
+        batch_op.create_index('document_segment_summary_document_id_idx', ['document_id'], unique=False)
+        batch_op.create_index('document_segment_summary_status_idx', ['status'], unique=False)
+
+    with op.batch_alter_table('datasets', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('summary_index_setting', models.types.AdjustedJSON(), nullable=True))
+
+    with op.batch_alter_table('documents', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('need_summary', sa.Boolean(), server_default=sa.text('false'), nullable=True))
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('documents', schema=None) as batch_op:
+        batch_op.drop_column('need_summary')
+
+    with op.batch_alter_table('datasets', schema=None) as batch_op:
+        batch_op.drop_column('summary_index_setting')
+
+    with op.batch_alter_table('document_segment_summary', schema=None) as batch_op:
+        batch_op.drop_index('document_segment_summary_status_idx')
+        batch_op.drop_index('document_segment_summary_document_id_idx')
+        batch_op.drop_index('document_segment_summary_dataset_id_idx')
+        batch_op.drop_index('document_segment_summary_chunk_id_idx')
+
+    op.drop_table('document_segment_summary')
+    # ### end Alembic commands ###

api/models/__init__.py

@@ -35,6 +35,7 @@ from .enums import (
     WorkflowTriggerStatus,
 )
 from .model import (
+    AccountTrialAppRecord,
     ApiRequest,
     ApiToken,
     App,
@@ -47,6 +48,7 @@ from .model import (
     DatasetRetrieverResource,
     DifySetup,
     EndUser,
+    ExporleBanner,
     IconType,
     InstalledApp,
     LLMGenerationDetail,
@@ -61,7 +63,9 @@ from .model import (
     Site,
     Tag,
     TagBinding,
+    TenantCreditPool,
     TraceAppConfig,
+    TrialApp,
     UploadFile,
 )
 from .oauth import DatasourceOauthParamConfig, DatasourceProvider
@@ -114,6 +118,7 @@ __all__ = [
     "Account",
     "AccountIntegrate",
     "AccountStatus",
+    "AccountTrialAppRecord",
     "ApiRequest",
     "ApiToken",
     "ApiToolProvider",
@@ -150,6 +155,7 @@ __all__ = [
     "DocumentSegment",
     "Embedding",
     "EndUser",
+    "ExporleBanner",
     "ExternalKnowledgeApis",
     "ExternalKnowledgeBindings",
     "IconType",
@@ -179,6 +185,7 @@ __all__ = [
     "Tenant",
     "TenantAccountJoin",
     "TenantAccountRole",
+    "TenantCreditPool",
     "TenantDefaultModel",
     "TenantPreferredModelProvider",
     "TenantStatus",
@@ -188,6 +195,7 @@ __all__ = [
     "ToolLabelBinding",
     "ToolModelInvoke",
     "TraceAppConfig",
+    "TrialApp",
     "TriggerOAuthSystemClient",
     "TriggerOAuthTenantClient",
     "TriggerSubscription",

api/models/dataset.py

@@ -72,6 +72,7 @@ class Dataset(Base):
     keyword_number = mapped_column(sa.Integer, nullable=True, server_default=sa.text("10"))
     collection_binding_id = mapped_column(StringUUID, nullable=True)
     retrieval_model = mapped_column(AdjustedJSON, nullable=True)
+    summary_index_setting = mapped_column(AdjustedJSON, nullable=True)
     built_in_field_enabled = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("false"))
     icon_info = mapped_column(AdjustedJSON, nullable=True)
     runtime_mode = mapped_column(sa.String(255), nullable=True, server_default=sa.text("'general'"))
@@ -419,6 +420,7 @@ class Document(Base):
     doc_metadata = mapped_column(AdjustedJSON, nullable=True)
     doc_form = mapped_column(String(255), nullable=False, server_default=sa.text("'text_model'"))
     doc_language = mapped_column(String(255), nullable=True)
+    need_summary: Mapped[bool | None] = mapped_column(sa.Boolean, nullable=True, server_default=sa.text("false"))
 
     DATA_SOURCES = ["upload_file", "notion_import", "website_crawl"]
 
@@ -1567,3 +1569,34 @@ class SegmentAttachmentBinding(Base):
     segment_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     attachment_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     created_at: Mapped[datetime] = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
+
+
+class DocumentSegmentSummary(Base):
+    __tablename__ = "document_segment_summary"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="document_segment_summary_pkey"),
+        sa.Index("document_segment_summary_dataset_id_idx", "dataset_id"),
+        sa.Index("document_segment_summary_document_id_idx", "document_id"),
+        sa.Index("document_segment_summary_chunk_id_idx", "chunk_id"),
+        sa.Index("document_segment_summary_status_idx", "status"),
+    )
+
+    id: Mapped[str] = mapped_column(StringUUID, nullable=False, default=lambda: str(uuid4()))
+    dataset_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    document_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    # corresponds to DocumentSegment.id or parent chunk id
+    chunk_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    summary_content: Mapped[str] = mapped_column(LongText, nullable=True)
+    summary_index_node_id: Mapped[str] = mapped_column(String(255), nullable=True)
+    summary_index_node_hash: Mapped[str] = mapped_column(String(255), nullable=True)
+    status: Mapped[str] = mapped_column(String(32), nullable=False, server_default=sa.text("'generating'"))
+    error: Mapped[str] = mapped_column(LongText, nullable=True)
+    enabled: Mapped[bool] = mapped_column(sa.Boolean, nullable=False, server_default=sa.text("true"))
+    disabled_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    disabled_by = mapped_column(StringUUID, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp())
+    updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp())
+
+    def __repr__(self):
+        return f"<DocumentSegmentSummary id={self.id} chunk_id={self.chunk_id} status={self.status}>"
+

api/models/model.py

@@ -12,8 +12,8 @@ from uuid import uuid4
 
 import sqlalchemy as sa
 from flask import request
-from flask_login import UserMixin
-from sqlalchemy import Float, Index, PrimaryKeyConstraint, String, exists, func, select, text
+from flask_login import UserMixin  # type: ignore[import-untyped]
+from sqlalchemy import BigInteger, Float, Index, PrimaryKeyConstraint, String, exists, func, select, text
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from configs import dify_config
@@ -605,6 +605,64 @@ class InstalledApp(TypeBase):
         return tenant
 
 
+class TrialApp(Base):
+    __tablename__ = "trial_apps"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="trial_app_pkey"),
+        sa.Index("trial_app_app_id_idx", "app_id"),
+        sa.Index("trial_app_tenant_id_idx", "tenant_id"),
+        sa.UniqueConstraint("app_id", name="unique_trail_app_id"),
+    )
+
+    id = mapped_column(StringUUID, server_default=sa.text("uuid_generate_v4()"))
+    app_id = mapped_column(StringUUID, nullable=False)
+    tenant_id = mapped_column(StringUUID, nullable=False)
+    created_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
+    trial_limit = mapped_column(sa.Integer, nullable=False, default=3)
+
+    @property
+    def app(self) -> App | None:
+        app = db.session.query(App).where(App.id == self.app_id).first()
+        return app
+
+
+class AccountTrialAppRecord(Base):
+    __tablename__ = "account_trial_app_records"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="user_trial_app_pkey"),
+        sa.Index("account_trial_app_record_account_id_idx", "account_id"),
+        sa.Index("account_trial_app_record_app_id_idx", "app_id"),
+        sa.UniqueConstraint("account_id", "app_id", name="unique_account_trial_app_record"),
+    )
+    id = mapped_column(StringUUID, server_default=sa.text("uuid_generate_v4()"))
+    account_id = mapped_column(StringUUID, nullable=False)
+    app_id = mapped_column(StringUUID, nullable=False)
+    count = mapped_column(sa.Integer, nullable=False, default=0)
+    created_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
+
+    @property
+    def app(self) -> App | None:
+        app = db.session.query(App).where(App.id == self.app_id).first()
+        return app
+
+    @property
+    def user(self) -> Account | None:
+        user = db.session.query(Account).where(Account.id == self.account_id).first()
+        return user
+
+
+class ExporleBanner(Base):
+    __tablename__ = "exporle_banners"
+    __table_args__ = (sa.PrimaryKeyConstraint("id", name="exporler_banner_pkey"),)
+    id = mapped_column(StringUUID, server_default=sa.text("uuid_generate_v4()"))
+    content = mapped_column(sa.JSON, nullable=False)
+    link = mapped_column(String(255), nullable=False)
+    sort = mapped_column(sa.Integer, nullable=False)
+    status = mapped_column(sa.String(255), nullable=False, server_default=sa.text("'enabled'::character varying"))
+    created_at = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
+    language = mapped_column(String(255), nullable=False, server_default=sa.text("'en-US'::character varying"))
+
+
 class OAuthProviderApp(TypeBase):
     """
     Globally shared OAuth provider app information.
@@ -1857,7 +1915,7 @@ class MessageChain(TypeBase):
     )
 
 
-class MessageAgentThought(Base):
+class MessageAgentThought(TypeBase):
     __tablename__ = "message_agent_thoughts"
     __table_args__ = (
         sa.PrimaryKeyConstraint("id", name="message_agent_thought_pkey"),
@@ -1865,34 +1923,42 @@ class MessageAgentThought(Base):
         sa.Index("message_agent_thought_message_chain_id_idx", "message_chain_id"),
     )
 
-    id = mapped_column(StringUUID, default=lambda: str(uuid4()))
-    message_id = mapped_column(StringUUID, nullable=False)
-    message_chain_id = mapped_column(StringUUID, nullable=True)
+    id: Mapped[str] = mapped_column(
+        StringUUID, insert_default=lambda: str(uuid4()), default_factory=lambda: str(uuid4()), init=False
+    )
+    message_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
     position: Mapped[int] = mapped_column(sa.Integer, nullable=False)
-    thought = mapped_column(LongText, nullable=True)
-    tool = mapped_column(LongText, nullable=True)
-    tool_labels_str = mapped_column(LongText, nullable=False, default=sa.text("'{}'"))
-    tool_meta_str = mapped_column(LongText, nullable=False, default=sa.text("'{}'"))
-    tool_input = mapped_column(LongText, nullable=True)
-    observation = mapped_column(LongText, nullable=True)
+    created_by_role: Mapped[str] = mapped_column(String(255), nullable=False)
+    created_by: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    message_chain_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True, default=None)
+    thought: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
+    tool: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
+    tool_labels_str: Mapped[str] = mapped_column(LongText, nullable=False, default=sa.text("'{}'"))
+    tool_meta_str: Mapped[str] = mapped_column(LongText, nullable=False, default=sa.text("'{}'"))
+    tool_input: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
+    observation: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
     # plugin_id = mapped_column(StringUUID, nullable=True)  ## for future design
-    tool_process_data = mapped_column(LongText, nullable=True)
-    message = mapped_column(LongText, nullable=True)
-    message_token: Mapped[int | None] = mapped_column(sa.Integer, nullable=True)
-    message_unit_price = mapped_column(sa.Numeric, nullable=True)
-    message_price_unit = mapped_column(sa.Numeric(10, 7), nullable=False, server_default=sa.text("0.001"))
-    message_files = mapped_column(LongText, nullable=True)
-    answer = mapped_column(LongText, nullable=True)
-    answer_token: Mapped[int | None] = mapped_column(sa.Integer, nullable=True)
-    answer_unit_price = mapped_column(sa.Numeric, nullable=True)
-    answer_price_unit = mapped_column(sa.Numeric(10, 7), nullable=False, server_default=sa.text("0.001"))
-    tokens: Mapped[int | None] = mapped_column(sa.Integer, nullable=True)
-    total_price = mapped_column(sa.Numeric, nullable=True)
-    currency = mapped_column(String(255), nullable=True)
-    latency: Mapped[float | None] = mapped_column(sa.Float, nullable=True)
-    created_by_role = mapped_column(String(255), nullable=False)
-    created_by = mapped_column(StringUUID, nullable=False)
-    created_at = mapped_column(sa.DateTime, nullable=False, server_default=sa.func.current_timestamp())
+    tool_process_data: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
+    message: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
+    message_token: Mapped[int | None] = mapped_column(sa.Integer, nullable=True, default=None)
+    message_unit_price: Mapped[Decimal | None] = mapped_column(sa.Numeric, nullable=True, default=None)
+    message_price_unit: Mapped[Decimal] = mapped_column(
+        sa.Numeric(10, 7), nullable=False, default=Decimal("0.001"), server_default=sa.text("0.001")
+    )
+    message_files: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
+    answer: Mapped[str | None] = mapped_column(LongText, nullable=True, default=None)
+    answer_token: Mapped[int | None] = mapped_column(sa.Integer, nullable=True, default=None)
+    answer_unit_price: Mapped[Decimal | None] = mapped_column(sa.Numeric, nullable=True, default=None)
+    answer_price_unit: Mapped[Decimal] = mapped_column(
+        sa.Numeric(10, 7), nullable=False, default=Decimal("0.001"), server_default=sa.text("0.001")
+    )
+    tokens: Mapped[int | None] = mapped_column(sa.Integer, nullable=True, default=None)
+    total_price: Mapped[Decimal | None] = mapped_column(sa.Numeric, nullable=True, default=None)
+    currency: Mapped[str | None] = mapped_column(String(255), nullable=True, default=None)
+    latency: Mapped[float | None] = mapped_column(sa.Float, nullable=True, default=None)
+    created_at: Mapped[datetime] = mapped_column(
+        sa.DateTime, nullable=False, init=False, server_default=sa.func.current_timestamp()
+    )
 
     @property
     def files(self) -> list[Any]:
@@ -2089,6 +2155,32 @@ class TraceAppConfig(TypeBase):
         }
 
 
+class TenantCreditPool(Base):
+    __tablename__ = "tenant_credit_pools"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="tenant_credit_pool_pkey"),
+        sa.Index("tenant_credit_pool_tenant_id_idx", "tenant_id"),
+        sa.Index("tenant_credit_pool_pool_type_idx", "pool_type"),
+    )
+
+    id = mapped_column(StringUUID, primary_key=True, server_default=text("uuid_generate_v4()"))
+    tenant_id = mapped_column(StringUUID, nullable=False)
+    pool_type = mapped_column(String(40), nullable=False, default="trial", server_default="trial")
+    quota_limit = mapped_column(BigInteger, nullable=False, default=0)
+    quota_used = mapped_column(BigInteger, nullable=False, default=0)
+    created_at = mapped_column(sa.DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"))
+    updated_at = mapped_column(
+        sa.DateTime, nullable=False, server_default=func.current_timestamp(), onupdate=func.current_timestamp()
+    )
+
+    @property
+    def remaining_credits(self) -> int:
+        return max(0, self.quota_limit - self.quota_used)
+
+    def has_sufficient_credits(self, required_credits: int) -> bool:
+        return self.remaining_credits >= required_credits
+
+
 class LLMGenerationDetail(Base):
     """
     Store LLM generation details including reasoning process and tool calls.
@@ -2127,7 +2219,7 @@ class LLMGenerationDetail(Base):
 
     created_at: Mapped[datetime] = mapped_column(sa.DateTime, nullable=False, server_default=func.current_timestamp())
 
-    def to_domain_model(self) -> "LLMGenerationDetailData":
+    def to_domain_model(self) -> LLMGenerationDetailData:
         """Convert to Pydantic domain model with proper validation."""
         from core.app.entities.llm_generation_entities import LLMGenerationDetailData
 
@@ -2144,14 +2236,14 @@ class LLMGenerationDetail(Base):
     @classmethod
     def from_domain_model(
         cls,
-        data: "LLMGenerationDetailData",
+        data: LLMGenerationDetailData,
         *,
         tenant_id: str,
         app_id: str,
         message_id: str | None = None,
         workflow_run_id: str | None = None,
         node_id: str | None = None,
-    ) -> "LLMGenerationDetail":
+    ) -> LLMGenerationDetail:
         """Create from Pydantic domain model."""
         # Enforce association mode at object creation time as well.
         message_mode = message_id is not None

api/pyproject.toml

@@ -189,7 +189,7 @@ storage = [
     "opendal~=0.46.0",
     "oss2==2.18.5",
     "supabase~=2.18.1",
-    "tos~=2.7.1",
+    "tos~=2.9.0",
 ]
 
 ############################################################

api/repositories/api_workflow_run_repository.py

@@ -34,10 +34,12 @@ Example:
     ```
 """
 
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from datetime import datetime
 from typing import Protocol
 
+from sqlalchemy.orm import Session
+
 from core.workflow.entities.pause_reason import PauseReason
 from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
@@ -253,6 +255,30 @@ class APIWorkflowRunRepository(WorkflowExecutionRepository, Protocol):
         """
         ...
 
+    def get_runs_batch_by_time_range(
+        self,
+        start_after: datetime | None,
+        end_before: datetime,
+        last_seen: tuple[datetime, str] | None,
+        batch_size: int,
+    ) -> Sequence[WorkflowRun]:
+        """
+        Fetch a batch of workflow runs within a time window using keyset pagination.
+        """
+        ...
+
+    def delete_runs_with_related(
+        self,
+        runs: Sequence[WorkflowRun],
+        delete_node_executions: Callable[[Session, Sequence[WorkflowRun]], tuple[int, int]] | None = None,
+        delete_trigger_logs: Callable[[Session, Sequence[str]], int] | None = None,
+    ) -> dict[str, int]:
+        """
+        Delete workflow runs and their related records (node executions, offloads, app logs,
+        trigger logs, pauses, pause reasons).
+        """
+        ...
+
     def create_workflow_pause(
         self,
         workflow_run_id: str,

api/repositories/sqlalchemy_api_workflow_node_execution_repository.py

@@ -7,13 +7,13 @@ using SQLAlchemy 2.0 style queries for WorkflowNodeExecutionModel operations.
 
 from collections.abc import Sequence
 from datetime import datetime
-from typing import cast
+from typing import TypedDict, cast
 
-from sqlalchemy import asc, delete, desc, select
+from sqlalchemy import asc, delete, desc, select, tuple_
 from sqlalchemy.engine import CursorResult
 from sqlalchemy.orm import Session, sessionmaker
 
-from models.workflow import WorkflowNodeExecutionModel
+from models.workflow import WorkflowNodeExecutionModel, WorkflowNodeExecutionOffload
 from repositories.api_workflow_node_execution_repository import DifyAPIWorkflowNodeExecutionRepository
 
 
@@ -290,3 +290,64 @@ class DifyAPISQLAlchemyWorkflowNodeExecutionRepository(DifyAPIWorkflowNodeExecut
             result = cast(CursorResult, session.execute(stmt))
             session.commit()
             return result.rowcount
+
+    class RunContext(TypedDict):
+        run_id: str
+        tenant_id: str
+        app_id: str
+        workflow_id: str
+        triggered_from: str
+
+    @staticmethod
+    def delete_by_runs(session: Session, runs: Sequence[RunContext]) -> tuple[int, int]:
+        """
+        Delete node executions (and offloads) for the given workflow runs using indexed columns.
+
+        Uses the composite index on (tenant_id, app_id, workflow_id, triggered_from, workflow_run_id)
+        by filtering on those columns with tuple IN.
+        """
+        if not runs:
+            return 0, 0
+
+        tuple_values = [
+            (run["tenant_id"], run["app_id"], run["workflow_id"], run["triggered_from"], run["run_id"]) for run in runs
+        ]
+
+        node_execution_ids = session.scalars(
+            select(WorkflowNodeExecutionModel.id).where(
+                tuple_(
+                    WorkflowNodeExecutionModel.tenant_id,
+                    WorkflowNodeExecutionModel.app_id,
+                    WorkflowNodeExecutionModel.workflow_id,
+                    WorkflowNodeExecutionModel.triggered_from,
+                    WorkflowNodeExecutionModel.workflow_run_id,
+                ).in_(tuple_values)
+            )
+        ).all()
+
+        if not node_execution_ids:
+            return 0, 0
+
+        offloads_deleted = (
+            cast(
+                CursorResult,
+                session.execute(
+                    delete(WorkflowNodeExecutionOffload).where(
+                        WorkflowNodeExecutionOffload.node_execution_id.in_(node_execution_ids)
+                    )
+                ),
+            ).rowcount
+            or 0
+        )
+
+        node_executions_deleted = (
+            cast(
+                CursorResult,
+                session.execute(
+                    delete(WorkflowNodeExecutionModel).where(WorkflowNodeExecutionModel.id.in_(node_execution_ids))
+                ),
+            ).rowcount
+            or 0
+        )
+
+        return node_executions_deleted, offloads_deleted

api/repositories/sqlalchemy_api_workflow_run_repository.py

@@ -21,7 +21,7 @@ Implementation Notes:
 
 import logging
 import uuid
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from datetime import datetime
 from decimal import Decimal
 from typing import Any, cast
@@ -40,8 +40,14 @@ from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from libs.time_parser import get_time_threshold
 from libs.uuid_utils import uuidv7
 from models.enums import WorkflowRunTriggeredFrom
-from models.workflow import WorkflowPause as WorkflowPauseModel
-from models.workflow import WorkflowPauseReason, WorkflowRun
+from models.workflow import (
+    WorkflowAppLog,
+    WorkflowPauseReason,
+    WorkflowRun,
+)
+from models.workflow import (
+    WorkflowPause as WorkflowPauseModel,
+)
 from repositories.api_workflow_run_repository import APIWorkflowRunRepository
 from repositories.entities.workflow_pause import WorkflowPauseEntity
 from repositories.types import (
@@ -314,6 +320,102 @@ class DifyAPISQLAlchemyWorkflowRunRepository(APIWorkflowRunRepository):
         logger.info("Total deleted %s workflow runs for app %s", total_deleted, app_id)
         return total_deleted
 
+    def get_runs_batch_by_time_range(
+        self,
+        start_after: datetime | None,
+        end_before: datetime,
+        last_seen: tuple[datetime, str] | None,
+        batch_size: int,
+    ) -> Sequence[WorkflowRun]:
+        with self._session_maker() as session:
+            stmt = (
+                select(WorkflowRun)
+                .where(
+                    WorkflowRun.created_at < end_before,
+                    WorkflowRun.status.in_(
+                        [
+                            WorkflowExecutionStatus.SUCCEEDED.value,
+                            WorkflowExecutionStatus.FAILED.value,
+                            WorkflowExecutionStatus.STOPPED.value,
+                            WorkflowExecutionStatus.PARTIAL_SUCCEEDED.value,
+                        ]
+                    ),
+                )
+                .order_by(WorkflowRun.created_at.asc(), WorkflowRun.id.asc())
+                .limit(batch_size)
+            )
+
+            if start_after:
+                stmt = stmt.where(WorkflowRun.created_at >= start_after)
+
+            if last_seen:
+                stmt = stmt.where(
+                    or_(
+                        WorkflowRun.created_at > last_seen[0],
+                        and_(WorkflowRun.created_at == last_seen[0], WorkflowRun.id > last_seen[1]),
+                    )
+                )
+
+            return session.scalars(stmt).all()
+
+    def delete_runs_with_related(
+        self,
+        runs: Sequence[WorkflowRun],
+        delete_node_executions: Callable[[Session, Sequence[WorkflowRun]], tuple[int, int]] | None = None,
+        delete_trigger_logs: Callable[[Session, Sequence[str]], int] | None = None,
+    ) -> dict[str, int]:
+        if not runs:
+            return {
+                "runs": 0,
+                "node_executions": 0,
+                "offloads": 0,
+                "app_logs": 0,
+                "trigger_logs": 0,
+                "pauses": 0,
+                "pause_reasons": 0,
+            }
+
+        with self._session_maker() as session:
+            run_ids = [run.id for run in runs]
+            if delete_node_executions:
+                node_executions_deleted, offloads_deleted = delete_node_executions(session, runs)
+            else:
+                node_executions_deleted, offloads_deleted = 0, 0
+
+            app_logs_result = session.execute(delete(WorkflowAppLog).where(WorkflowAppLog.workflow_run_id.in_(run_ids)))
+            app_logs_deleted = cast(CursorResult, app_logs_result).rowcount or 0
+
+            pause_ids = session.scalars(
+                select(WorkflowPauseModel.id).where(WorkflowPauseModel.workflow_run_id.in_(run_ids))
+            ).all()
+            pause_reasons_deleted = 0
+            pauses_deleted = 0
+
+            if pause_ids:
+                pause_reasons_result = session.execute(
+                    delete(WorkflowPauseReason).where(WorkflowPauseReason.pause_id.in_(pause_ids))
+                )
+                pause_reasons_deleted = cast(CursorResult, pause_reasons_result).rowcount or 0
+                pauses_result = session.execute(delete(WorkflowPauseModel).where(WorkflowPauseModel.id.in_(pause_ids)))
+                pauses_deleted = cast(CursorResult, pauses_result).rowcount or 0
+
+            trigger_logs_deleted = delete_trigger_logs(session, run_ids) if delete_trigger_logs else 0
+
+            runs_result = session.execute(delete(WorkflowRun).where(WorkflowRun.id.in_(run_ids)))
+            runs_deleted = cast(CursorResult, runs_result).rowcount or 0
+
+            session.commit()
+
+            return {
+                "runs": runs_deleted,
+                "node_executions": node_executions_deleted,
+                "offloads": offloads_deleted,
+                "app_logs": app_logs_deleted,
+                "trigger_logs": trigger_logs_deleted,
+                "pauses": pauses_deleted,
+                "pause_reasons": pause_reasons_deleted,
+            }
+
     def create_workflow_pause(
         self,
         workflow_run_id: str,

api/repositories/sqlalchemy_workflow_trigger_log_repository.py

@@ -4,8 +4,10 @@ SQLAlchemy implementation of WorkflowTriggerLogRepository.
 
 from collections.abc import Sequence
 from datetime import UTC, datetime, timedelta
+from typing import cast
 
-from sqlalchemy import and_, select
+from sqlalchemy import and_, delete, select
+from sqlalchemy.engine import CursorResult
 from sqlalchemy.orm import Session
 
 from models.enums import WorkflowTriggerStatus
@@ -84,3 +86,19 @@ class SQLAlchemyWorkflowTriggerLogRepository(WorkflowTriggerLogRepository):
         )
 
         return list(self.session.scalars(query).all())
+
+    def delete_by_run_ids(self, run_ids: Sequence[str]) -> int:
+        """
+        Delete trigger logs associated with the given workflow run ids.
+
+        Args:
+            run_ids: Collection of workflow run identifiers.
+
+        Returns:
+            Number of rows deleted.
+        """
+        if not run_ids:
+            return 0
+
+        result = self.session.execute(delete(WorkflowTriggerLog).where(WorkflowTriggerLog.workflow_run_id.in_(run_ids)))
+        return cast(CursorResult, result).rowcount or 0

api/schedule/clean_workflow_runs_task.py

@@ -0,0 +1,30 @@
+import click
+
+import app
+from configs import dify_config
+from services.clear_free_plan_expired_workflow_run_logs import WorkflowRunCleanup
+
+CLEANUP_QUEUE = "retention"
+
+
+@app.celery.task(queue=CLEANUP_QUEUE)
+def clean_workflow_runs_task() -> None:
+    """
+    Scheduled cleanup for workflow runs and related records (sandbox tenants only).
+    """
+    click.echo(
+        click.style(
+            f"Scheduled workflow run cleanup starting: cutoff={dify_config.WORKFLOW_LOG_RETENTION_DAYS} days, "
+            f"batch={dify_config.WORKFLOW_LOG_CLEANUP_BATCH_SIZE}",
+            fg="green",
+        )
+    )
+
+    WorkflowRunCleanup(
+        days=dify_config.WORKFLOW_LOG_RETENTION_DAYS,
+        batch_size=dify_config.WORKFLOW_LOG_CLEANUP_BATCH_SIZE,
+        start_after=None,
+        end_before=None,
+    ).run()
+
+    click.echo(click.style("Scheduled workflow run cleanup finished.", fg="green"))

api/services/account_service.py

@@ -999,6 +999,11 @@ class TenantService:
 
         tenant.encrypt_public_key = generate_key_pair(tenant.id)
         db.session.commit()
+
+        from services.credit_pool_service import CreditPoolService
+
+        CreditPoolService.create_default_pool(tenant.id)
+
         return tenant
 
     @staticmethod

api/services/annotation_service.py

@@ -355,7 +355,6 @@ class AppAnnotationService:
     def batch_import_app_annotations(cls, app_id, file: FileStorage):
         """
         Batch import annotations from CSV file with enhanced security checks.
-
         Security features:
         - File size validation
         - Row count limits (min/max)
@@ -364,7 +363,6 @@ class AppAnnotationService:
         - Concurrency tracking
         """
         from configs import dify_config
-
         # get app info
         current_user, current_tenant_id = current_account_with_tenant()
         app = (
@@ -448,27 +446,31 @@ class AppAnnotationService:
                     f"The CSV file must contain at least {min_records} valid annotation record(s). "
                     f"Found {len(result)} valid record(s)."
                 )
-
             # Check annotation quota limit
             features = FeatureService.get_features(current_tenant_id)
             if features.billing.enabled:
                 annotation_quota_limit = features.annotation_quota_limit
                 if annotation_quota_limit.limit < len(result) + annotation_quota_limit.size:
-                    raise ValueError("The number of annotations exceeds the limit of your subscription.")
-            # async job
+                    raise ValueError(
+                        f"The number of annotations ({len(result)}) would exceed your subscription limit. "
+                        f"Current usage: {annotation_quota_limit.size}/{annotation_quota_limit.limit}. "
+                        f"Available: {annotation_quota_limit.limit - annotation_quota_limit.size}."
+                    )
+
+            # Create async job
             job_id = str(uuid.uuid4())
             indexing_cache_key = f"app_annotation_batch_import_{str(job_id)}"
-
             # Register job in active tasks list for concurrency tracking
             current_time = int(naive_utc_now().timestamp() * 1000)
             active_jobs_key = f"annotation_import_active:{current_tenant_id}"
             redis_client.zadd(active_jobs_key, {job_id: current_time})
             redis_client.expire(active_jobs_key, 7200)  # 2 hours TTL
-
             # Set job status
             redis_client.setnx(indexing_cache_key, "waiting")
-            batch_import_annotations_task.delay(str(job_id), result, app_id, current_tenant_id, current_user.id)
+            redis_client.expire(indexing_cache_key, 3600)  # 1 hour TTL
 
+            # Send batch import task
+            batch_import_annotations_task.delay(str(job_id), result, app_id, current_tenant_id, current_user.id)
         except ValueError as e:
             return {"error_msg": str(e)}
         except Exception as e:

api/services/billing_service.py

@@ -44,6 +44,33 @@ class BillingService:
         billing_info = cls._send_request("GET", "/subscription/info", params=params)
         return billing_info
 
+    @classmethod
+    def get_info_bulk(cls, tenant_ids: Sequence[str]) -> dict[str, str]:
+        """
+        Bulk billing info fetch via billing API.
+
+        Payload: {"tenant_ids": ["t1", "t2", ...]} (max 200 per request)
+
+        Returns:
+            Mapping of tenant_id -> plan
+        """
+        results: dict[str, str] = {}
+
+        chunk_size = 200
+        for i in range(0, len(tenant_ids), chunk_size):
+            chunk = tenant_ids[i : i + chunk_size]
+            try:
+                resp = cls._send_request("POST", "/subscription/plan/batch", json={"tenant_ids": chunk})
+                data = resp.get("data", {})
+                for tenant_id, plan in data.items():
+                    if isinstance(plan, str):
+                        results[tenant_id] = plan
+            except Exception:
+                logger.exception("Failed to fetch billing info batch for tenants: %s", chunk)
+                continue
+
+        return results
+
     @classmethod
     def get_tenant_feature_plan_usage_info(cls, tenant_id: str):
         params = {"tenant_id": tenant_id}

api/services/clear_free_plan_expired_workflow_run_logs.py

@@ -0,0 +1,171 @@
+import datetime
+import logging
+from collections.abc import Iterable, Sequence
+
+import click
+from sqlalchemy.orm import Session, sessionmaker
+
+from configs import dify_config
+from enums.cloud_plan import CloudPlan
+from extensions.ext_database import db
+from models.workflow import WorkflowRun
+from repositories.api_workflow_run_repository import APIWorkflowRunRepository
+from repositories.sqlalchemy_api_workflow_node_execution_repository import (
+    DifyAPISQLAlchemyWorkflowNodeExecutionRepository,
+)
+from repositories.sqlalchemy_workflow_trigger_log_repository import SQLAlchemyWorkflowTriggerLogRepository
+from services.billing_service import BillingService
+
+logger = logging.getLogger(__name__)
+
+
+class WorkflowRunCleanup:
+    def __init__(
+        self,
+        days: int,
+        batch_size: int,
+        start_after: datetime.datetime | None = None,
+        end_before: datetime.datetime | None = None,
+        workflow_run_repo: APIWorkflowRunRepository | None = None,
+    ):
+        if (start_after is None) ^ (end_before is None):
+            raise ValueError("start_after and end_before must be both set or both omitted.")
+
+        computed_cutoff = datetime.datetime.now() - datetime.timedelta(days=days)
+        self.window_start = start_after
+        self.window_end = end_before or computed_cutoff
+
+        if self.window_start and self.window_end <= self.window_start:
+            raise ValueError("end_before must be greater than start_after.")
+
+        self.batch_size = batch_size
+        self.billing_cache: dict[str, CloudPlan | None] = {}
+        self.workflow_run_repo: APIWorkflowRunRepository
+        if workflow_run_repo:
+            self.workflow_run_repo = workflow_run_repo
+        else:
+            # Lazy import to avoid circular dependencies during module import
+            from repositories.factory import DifyAPIRepositoryFactory
+
+            session_maker = sessionmaker(bind=db.engine, expire_on_commit=False)
+            self.workflow_run_repo = DifyAPIRepositoryFactory.create_api_workflow_run_repository(session_maker)
+
+    def run(self) -> None:
+        click.echo(
+            click.style(
+                f"Cleaning workflow runs "
+                f"{'between ' + self.window_start.isoformat() + ' and ' if self.window_start else 'before '}"
+                f"{self.window_end.isoformat()} (batch={self.batch_size})",
+                fg="white",
+            )
+        )
+
+        total_runs_deleted = 0
+        batch_index = 0
+        last_seen: tuple[datetime.datetime, str] | None = None
+
+        while True:
+            run_rows = self.workflow_run_repo.get_runs_batch_by_time_range(
+                start_after=self.window_start,
+                end_before=self.window_end,
+                last_seen=last_seen,
+                batch_size=self.batch_size,
+            )
+            if not run_rows:
+                break
+
+            batch_index += 1
+            last_seen = (run_rows[-1].created_at, run_rows[-1].id)
+            tenant_ids = {row.tenant_id for row in run_rows}
+            free_tenants = self._filter_free_tenants(tenant_ids)
+            free_runs = [row for row in run_rows if row.tenant_id in free_tenants]
+            paid_or_skipped = len(run_rows) - len(free_runs)
+
+            if not free_runs:
+                click.echo(
+                    click.style(
+                        f"[batch #{batch_index}] skipped (no sandbox runs in batch, {paid_or_skipped} paid/unknown)",
+                        fg="yellow",
+                    )
+                )
+                continue
+
+            try:
+                counts = self.workflow_run_repo.delete_runs_with_related(
+                    free_runs,
+                    delete_node_executions=self._delete_node_executions,
+                    delete_trigger_logs=self._delete_trigger_logs,
+                )
+            except Exception:
+                logger.exception("Failed to delete workflow runs batch ending at %s", last_seen[0])
+                raise
+
+            total_runs_deleted += counts["runs"]
+            click.echo(
+                click.style(
+                    f"[batch #{batch_index}] deleted runs: {counts['runs']} "
+                    f"(nodes {counts['node_executions']}, offloads {counts['offloads']}, "
+                    f"app_logs {counts['app_logs']}, trigger_logs {counts['trigger_logs']}, "
+                    f"pauses {counts['pauses']}, pause_reasons {counts['pause_reasons']}); "
+                    f"skipped {paid_or_skipped} paid/unknown",
+                    fg="green",
+                )
+            )
+
+        if self.window_start:
+            summary_message = (
+                f"Cleanup complete. Deleted {total_runs_deleted} workflow runs "
+                f"between {self.window_start.isoformat()} and {self.window_end.isoformat()}"
+            )
+        else:
+            summary_message = (
+                f"Cleanup complete. Deleted {total_runs_deleted} workflow runs before {self.window_end.isoformat()}"
+            )
+
+        click.echo(click.style(summary_message, fg="white"))
+
+    def _filter_free_tenants(self, tenant_ids: Iterable[str]) -> set[str]:
+        if not dify_config.BILLING_ENABLED:
+            return set(tenant_ids)
+
+        tenant_id_list = list(tenant_ids)
+        uncached_tenants = [tenant_id for tenant_id in tenant_id_list if tenant_id not in self.billing_cache]
+
+        if uncached_tenants:
+            try:
+                bulk_info = BillingService.get_info_bulk(uncached_tenants)
+            except Exception:
+                bulk_info = {}
+                logger.exception("Failed to fetch billing plans in bulk for tenants: %s", uncached_tenants)
+
+            for tenant_id in uncached_tenants:
+                plan: CloudPlan | None = None
+                info = bulk_info.get(tenant_id)
+                if info:
+                    try:
+                        plan = CloudPlan(info)
+                    except Exception:
+                        logger.exception("Failed to parse billing plan for tenant %s", tenant_id)
+                else:
+                    logger.warning("Missing billing info for tenant %s in bulk resp; treating as non-free", tenant_id)
+
+                self.billing_cache[tenant_id] = plan
+
+        return {tenant_id for tenant_id in tenant_id_list if self.billing_cache.get(tenant_id) == CloudPlan.SANDBOX}
+
+    def _delete_trigger_logs(self, session: Session, run_ids: Sequence[str]) -> int:
+        trigger_repo = SQLAlchemyWorkflowTriggerLogRepository(session)
+        return trigger_repo.delete_by_run_ids(run_ids)
+
+    def _delete_node_executions(self, session: Session, runs: Sequence[WorkflowRun]) -> tuple[int, int]:
+        run_contexts: list[DifyAPISQLAlchemyWorkflowNodeExecutionRepository.RunContext] = [
+            {
+                "run_id": run.id,
+                "tenant_id": run.tenant_id,
+                "app_id": run.app_id,
+                "workflow_id": run.workflow_id,
+                "triggered_from": run.triggered_from,
+            }
+            for run in runs
+        ]
+        return DifyAPISQLAlchemyWorkflowNodeExecutionRepository.delete_by_runs(session, run_contexts)

api/services/credit_pool_service.py

@@ -0,0 +1,85 @@
+import logging
+
+from sqlalchemy import update
+from sqlalchemy.orm import Session
+
+from configs import dify_config
+from core.errors.error import QuotaExceededError
+from extensions.ext_database import db
+from models import TenantCreditPool
+
+logger = logging.getLogger(__name__)
+
+
+class CreditPoolService:
+    @classmethod
+    def create_default_pool(cls, tenant_id: str) -> TenantCreditPool:
+        """create default credit pool for new tenant"""
+        credit_pool = TenantCreditPool(
+            tenant_id=tenant_id, quota_limit=dify_config.HOSTED_POOL_CREDITS, quota_used=0, pool_type="trial"
+        )
+        db.session.add(credit_pool)
+        db.session.commit()
+        return credit_pool
+
+    @classmethod
+    def get_pool(cls, tenant_id: str, pool_type: str = "trial") -> TenantCreditPool | None:
+        """get tenant credit pool"""
+        return (
+            db.session.query(TenantCreditPool)
+            .filter_by(
+                tenant_id=tenant_id,
+                pool_type=pool_type,
+            )
+            .first()
+        )
+
+    @classmethod
+    def check_credits_available(
+        cls,
+        tenant_id: str,
+        credits_required: int,
+        pool_type: str = "trial",
+    ) -> bool:
+        """check if credits are available without deducting"""
+        pool = cls.get_pool(tenant_id, pool_type)
+        if not pool:
+            return False
+        return pool.remaining_credits >= credits_required
+
+    @classmethod
+    def check_and_deduct_credits(
+        cls,
+        tenant_id: str,
+        credits_required: int,
+        pool_type: str = "trial",
+    ) -> int:
+        """check and deduct credits, returns actual credits deducted"""
+
+        pool = cls.get_pool(tenant_id, pool_type)
+        if not pool:
+            raise QuotaExceededError("Credit pool not found")
+
+        if pool.remaining_credits <= 0:
+            raise QuotaExceededError("No credits remaining")
+
+        # deduct all remaining credits if less than required
+        actual_credits = min(credits_required, pool.remaining_credits)
+
+        try:
+            with Session(db.engine) as session:
+                stmt = (
+                    update(TenantCreditPool)
+                    .where(
+                        TenantCreditPool.tenant_id == tenant_id,
+                        TenantCreditPool.pool_type == pool_type,
+                    )
+                    .values(quota_used=TenantCreditPool.quota_used + actual_credits)
+                )
+                session.execute(stmt)
+                session.commit()
+        except Exception:
+            logger.exception("Failed to deduct credits for tenant %s", tenant_id)
+            raise QuotaExceededError("Failed to deduct credits")
+
+        return actual_credits

api/services/dataset_service.py

@@ -89,6 +89,7 @@ from tasks.enable_segments_to_index_task import enable_segments_to_index_task
 from tasks.recover_document_indexing_task import recover_document_indexing_task
 from tasks.remove_document_from_index_task import remove_document_from_index_task
 from tasks.retry_document_indexing_task import retry_document_indexing_task
+from tasks.regenerate_summary_index_task import regenerate_summary_index_task
 from tasks.sync_website_document_indexing_task import sync_website_document_indexing_task
 
 logger = logging.getLogger(__name__)
@@ -474,6 +475,11 @@ class DatasetService:
         if external_retrieval_model:
             dataset.retrieval_model = external_retrieval_model
 
+        # Update summary index setting if provided
+        summary_index_setting = data.get("summary_index_setting", None)
+        if summary_index_setting is not None:
+            dataset.summary_index_setting = summary_index_setting
+
         # Update basic dataset properties
         dataset.name = data.get("name", dataset.name)
         dataset.description = data.get("description", dataset.description)
@@ -556,12 +562,20 @@ class DatasetService:
         # Handle indexing technique changes and embedding model updates
         action = DatasetService._handle_indexing_technique_change(dataset, data, filtered_data)
 
+        # Check if summary_index_setting model changed (before updating database)
+        summary_model_changed = DatasetService._check_summary_index_setting_model_changed(
+            dataset, data
+        )
+
         # Add metadata fields
         filtered_data["updated_by"] = user.id
         filtered_data["updated_at"] = naive_utc_now()
         # update Retrieval model
         if data.get("retrieval_model"):
             filtered_data["retrieval_model"] = data["retrieval_model"]
+        # update summary index setting
+        if data.get("summary_index_setting"):
+            filtered_data["summary_index_setting"] = data.get("summary_index_setting")
         # update icon info
         if data.get("icon_info"):
             filtered_data["icon_info"] = data.get("icon_info")
@@ -570,12 +584,30 @@ class DatasetService:
         db.session.query(Dataset).filter_by(id=dataset.id).update(filtered_data)
         db.session.commit()
 
+        # Reload dataset to get updated values
+        db.session.refresh(dataset)
+
         # update pipeline knowledge base node data
         DatasetService._update_pipeline_knowledge_base_node_data(dataset, user.id)
 
         # Trigger vector index task if indexing technique changed
         if action:
             deal_dataset_vector_index_task.delay(dataset.id, action)
+            # If embedding_model changed, also regenerate summary vectors
+            if action == "update":
+                regenerate_summary_index_task.delay(
+                    dataset.id,
+                    regenerate_reason="embedding_model_changed",
+                    regenerate_vectors_only=True,
+                )
+
+        # Trigger summary index regeneration if summary model changed
+        if summary_model_changed:
+            regenerate_summary_index_task.delay(
+                dataset.id,
+                regenerate_reason="summary_model_changed",
+                regenerate_vectors_only=False,
+            )
 
         return dataset
 
@@ -614,6 +646,7 @@ class DatasetService:
                             knowledge_index_node_data["chunk_structure"] = dataset.chunk_structure
                             knowledge_index_node_data["indexing_technique"] = dataset.indexing_technique  # pyright: ignore[reportAttributeAccessIssue]
                             knowledge_index_node_data["keyword_number"] = dataset.keyword_number
+                            knowledge_index_node_data["summary_index_setting"] = dataset.summary_index_setting
                             node["data"] = knowledge_index_node_data
                             updated = True
                         except Exception:
@@ -852,6 +885,49 @@ class DatasetService:
         )
         filtered_data["collection_binding_id"] = dataset_collection_binding.id
 
+    @staticmethod
+    def _check_summary_index_setting_model_changed(dataset: Dataset, data: dict[str, Any]) -> bool:
+        """
+        Check if summary_index_setting model (model_name or model_provider_name) has changed.
+
+        Args:
+            dataset: Current dataset object
+            data: Update data dictionary
+
+        Returns:
+            bool: True if summary model changed, False otherwise
+        """
+        # Check if summary_index_setting is being updated
+        if "summary_index_setting" not in data or data.get("summary_index_setting") is None:
+            return False
+
+        new_summary_setting = data.get("summary_index_setting")
+        old_summary_setting = dataset.summary_index_setting
+
+        # If old setting doesn't exist or is disabled, no need to regenerate
+        if not old_summary_setting or not old_summary_setting.get("enable"):
+            return False
+
+        # If new setting is disabled, no need to regenerate
+        if not new_summary_setting or not new_summary_setting.get("enable"):
+            return False
+
+        # Compare model_name and model_provider_name
+        old_model_name = old_summary_setting.get("model_name")
+        old_model_provider = old_summary_setting.get("model_provider_name")
+        new_model_name = new_summary_setting.get("model_name")
+        new_model_provider = new_summary_setting.get("model_provider_name")
+
+        # Check if model changed
+        if old_model_name != new_model_name or old_model_provider != new_model_provider:
+            logger.info(
+                f"Summary index setting model changed for dataset {dataset.id}: "
+                f"old={old_model_provider}/{old_model_name}, new={new_model_provider}/{new_model_name}"
+            )
+            return True
+
+        return False
+
     @staticmethod
     def update_rag_pipeline_dataset_settings(
         session: Session, dataset: Dataset, knowledge_configuration: KnowledgeConfiguration, has_published: bool = False
@@ -1824,6 +1900,8 @@ class DocumentService:
                         DuplicateDocumentIndexingTaskProxy(
                             dataset.tenant_id, dataset.id, duplicate_document_ids
                         ).delay()
+                    # Note: Summary index generation is triggered in document_indexing_task after indexing completes
+                    # to ensure segments are available. See tasks/document_indexing_task.py
             except LockNotOwnedError:
                 pass
 
@@ -2128,6 +2206,14 @@ class DocumentService:
         name: str,
         batch: str,
     ):
+        # Set need_summary based on dataset's summary_index_setting
+        need_summary = False
+        if (
+            dataset.summary_index_setting
+            and dataset.summary_index_setting.get("enable") is True
+        ):
+            need_summary = True
+        
         document = Document(
             tenant_id=dataset.tenant_id,
             dataset_id=dataset.id,
@@ -2141,6 +2227,7 @@ class DocumentService:
             created_by=account.id,
             doc_form=document_form,
             doc_language=document_language,
+            need_summary=need_summary,
         )
         doc_metadata = {}
         if dataset.built_in_field_enabled:
@@ -2365,6 +2452,7 @@ class DocumentService:
             embedding_model_provider=knowledge_config.embedding_model_provider,
             collection_binding_id=dataset_collection_binding_id,
             retrieval_model=retrieval_model.model_dump() if retrieval_model else None,
+            summary_index_setting=knowledge_config.summary_index_setting,
             is_multimodal=knowledge_config.is_multimodal,
         )
 
@@ -2546,6 +2634,14 @@ class DocumentService:
             if not isinstance(args["process_rule"]["rules"]["segmentation"]["max_tokens"], int):
                 raise ValueError("Process rule segmentation max_tokens is invalid")
 
+        # valid summary index setting
+        if args["process_rule"]["summary_index_setting"] and args["process_rule"]["summary_index_setting"]["enable"]:
+            summary_index_setting = args["process_rule"]["summary_index_setting"]
+            if "model_name" not in summary_index_setting or not summary_index_setting["model_name"]:
+                raise ValueError("Summary index model name is required")
+            if "model_provider_name" not in summary_index_setting or not summary_index_setting["model_provider_name"]:
+                raise ValueError("Summary index model provider name is required")
+
     @staticmethod
     def batch_update_document_status(
         dataset: Dataset, document_ids: list[str], action: Literal["enable", "disable", "archive", "un_archive"], user
@@ -3014,6 +3110,37 @@ class SegmentService:
                     if args.enabled or keyword_changed:
                         # update segment vector index
                         VectorService.update_segment_vector(args.keywords, segment, dataset)
+                # update summary index if summary is provided and has changed
+                if args.summary is not None:
+                    # Check if summary index is enabled
+                    has_summary_index = (
+                        dataset.indexing_technique == "high_quality"
+                        and dataset.summary_index_setting
+                        and dataset.summary_index_setting.get("enable") is True
+                    )
+                    
+                    if has_summary_index:
+                        # Query existing summary from database
+                        from models.dataset import DocumentSegmentSummary
+                        existing_summary = (
+                            db.session.query(DocumentSegmentSummary)
+                            .where(
+                                DocumentSegmentSummary.chunk_id == segment.id,
+                                DocumentSegmentSummary.dataset_id == dataset.id,
+                            )
+                            .first()
+                        )
+                        
+                        # Check if summary has changed
+                        existing_summary_content = existing_summary.summary_content if existing_summary else None
+                        if existing_summary_content != args.summary:
+                            # Summary has changed, update it
+                            from services.summary_index_service import SummaryIndexService
+                            try:
+                                SummaryIndexService.update_summary_for_segment(segment, dataset, args.summary)
+                            except Exception as e:
+                                logger.exception(f"Failed to update summary for segment {segment.id}: {str(e)}")
+                                # Don't fail the entire update if summary update fails
             else:
                 segment_hash = helper.generate_text_hash(content)
                 tokens = 0
@@ -3088,6 +3215,15 @@ class SegmentService:
                 elif document.doc_form in (IndexStructureType.PARAGRAPH_INDEX, IndexStructureType.QA_INDEX):
                     # update segment vector index
                     VectorService.update_segment_vector(args.keywords, segment, dataset)
+            # update summary index if summary is provided
+            if args.summary is not None:
+                from services.summary_index_service import SummaryIndexService
+
+                try:
+                    SummaryIndexService.update_summary_for_segment(segment, dataset, args.summary)
+                except Exception as e:
+                    logger.exception(f"Failed to update summary for segment {segment.id}: {str(e)}")
+                    # Don't fail the entire update if summary update fails
             # update multimodel vector index
             VectorService.update_multimodel_vector(segment, args.attachment_ids or [], dataset)
         except Exception as e:

api/services/entities/knowledge_entities/knowledge_entities.py

@@ -119,6 +119,7 @@ class KnowledgeConfig(BaseModel):
     data_source: DataSource | None = None
     process_rule: ProcessRule | None = None
     retrieval_model: RetrievalModel | None = None
+    summary_index_setting: dict | None = None
     doc_form: str = "text_model"
     doc_language: str = "English"
     embedding_model: str | None = None
@@ -141,6 +142,7 @@ class SegmentUpdateArgs(BaseModel):
     regenerate_child_chunks: bool = False
     enabled: bool | None = None
     attachment_ids: list[str] | None = None
+    summary: str | None = None  # Summary content for summary index
 
 
 class ChildChunkUpdateArgs(BaseModel):

api/services/feature_service.py

@@ -140,6 +140,7 @@ class FeatureModel(BaseModel):
     # pydantic configs
     model_config = ConfigDict(protected_namespaces=())
     knowledge_pipeline: KnowledgePipeline = KnowledgePipeline()
+    next_credit_reset_date: int = 0
 
 
 class KnowledgeRateLimitModel(BaseModel):
@@ -169,6 +170,8 @@ class SystemFeatureModel(BaseModel):
     plugin_installation_permission: PluginInstallationPermissionModel = PluginInstallationPermissionModel()
     enable_change_email: bool = True
     plugin_manager: PluginManagerModel = PluginManagerModel()
+    enable_trial_app: bool = False
+    enable_explore_banner: bool = False
 
 
 class FeatureService:
@@ -224,6 +227,8 @@ class FeatureService:
         system_features.is_allow_register = dify_config.ALLOW_REGISTER
         system_features.is_allow_create_workspace = dify_config.ALLOW_CREATE_WORKSPACE
         system_features.is_email_setup = dify_config.MAIL_TYPE is not None and dify_config.MAIL_TYPE != ""
+        system_features.enable_trial_app = dify_config.ENABLE_TRIAL_APP
+        system_features.enable_explore_banner = dify_config.ENABLE_EXPLORE_BANNER
 
     @classmethod
     def _fulfill_params_from_env(cls, features: FeatureModel):
@@ -301,6 +306,9 @@ class FeatureService:
         if "knowledge_pipeline_publish_enabled" in billing_info:
             features.knowledge_pipeline.publish_enabled = billing_info["knowledge_pipeline_publish_enabled"]
 
+        if "next_credit_reset_date" in billing_info:
+            features.next_credit_reset_date = billing_info["next_credit_reset_date"]
+
     @classmethod
     def _fulfill_params_from_enterprise(cls, features: SystemFeatureModel):
         enterprise_info = EnterpriseService.get_info()

api/services/recommended_app_service.py

@@ -1,4 +1,7 @@
 from configs import dify_config
+from extensions.ext_database import db
+from models.model import AccountTrialAppRecord, TrialApp
+from services.feature_service import FeatureService
 from services.recommend_app.recommend_app_factory import RecommendAppRetrievalFactory
 
 
@@ -20,6 +23,15 @@ class RecommendedAppService:
                 )
             )
 
+        if FeatureService.get_system_features().enable_trial_app:
+            apps = result["recommended_apps"]
+            for app in apps:
+                app_id = app["app_id"]
+                trial_app_model = db.session.query(TrialApp).where(TrialApp.app_id == app_id).first()
+                if trial_app_model:
+                    app["can_trial"] = True
+                else:
+                    app["can_trial"] = False
         return result
 
     @classmethod
@@ -32,4 +44,30 @@ class RecommendedAppService:
         mode = dify_config.HOSTED_FETCH_APP_TEMPLATES_MODE
         retrieval_instance = RecommendAppRetrievalFactory.get_recommend_app_factory(mode)()
         result: dict = retrieval_instance.get_recommend_app_detail(app_id)
+        if FeatureService.get_system_features().enable_trial_app:
+            app_id = result["id"]
+            trial_app_model = db.session.query(TrialApp).where(TrialApp.app_id == app_id).first()
+            if trial_app_model:
+                result["can_trial"] = True
+            else:
+                result["can_trial"] = False
         return result
+
+    @classmethod
+    def add_trial_app_record(cls, app_id: str, account_id: str):
+        """
+        Add trial app record.
+        :param app_id: app id
+        :return:
+        """
+        account_trial_app_record = (
+            db.session.query(AccountTrialAppRecord)
+            .where(AccountTrialAppRecord.app_id == app_id, AccountTrialAppRecord.account_id == account_id)
+            .first()
+        )
+        if account_trial_app_record:
+            account_trial_app_record.count += 1
+            db.session.commit()
+        else:
+            db.session.add(AccountTrialAppRecord(app_id=app_id, count=1, account_id=account_id))
+            db.session.commit()

api/services/summary_index_service.py

@@ -0,0 +1,612 @@
+"""Summary index service for generating and managing document segment summaries."""
+
+import logging
+import time
+import uuid
+from typing import Any
+
+from core.rag.datasource.vdb.vector_factory import Vector
+from core.rag.index_processor.constant.doc_type import DocType
+from core.rag.models.document import Document
+from extensions.ext_database import db
+from libs import helper
+from models.dataset import Dataset, DocumentSegment, DocumentSegmentSummary
+from models.dataset import Document as DatasetDocument
+
+logger = logging.getLogger(__name__)
+
+
+class SummaryIndexService:
+    """Service for generating and managing summary indexes."""
+
+    @staticmethod
+    def generate_summary_for_segment(
+        segment: DocumentSegment,
+        dataset: Dataset,
+        summary_index_setting: dict,
+    ) -> str:
+        """
+        Generate summary for a single segment.
+
+        Args:
+            segment: DocumentSegment to generate summary for
+            dataset: Dataset containing the segment
+            summary_index_setting: Summary index configuration
+
+        Returns:
+            Generated summary text
+
+        Raises:
+            ValueError: If summary_index_setting is invalid or generation fails
+        """
+        # Reuse the existing generate_summary method from ParagraphIndexProcessor
+        # Use lazy import to avoid circular import
+        from core.rag.index_processor.processor.paragraph_index_processor import ParagraphIndexProcessor
+
+        summary_content = ParagraphIndexProcessor.generate_summary(
+            tenant_id=dataset.tenant_id,
+            text=segment.content,
+            summary_index_setting=summary_index_setting,
+        )
+
+        if not summary_content:
+            raise ValueError("Generated summary is empty")
+
+        return summary_content
+
+    @staticmethod
+    def create_summary_record(
+        segment: DocumentSegment,
+        dataset: Dataset,
+        summary_content: str,
+        status: str = "generating",
+    ) -> DocumentSegmentSummary:
+        """
+        Create or update a DocumentSegmentSummary record.
+        If a summary record already exists for this segment, it will be updated instead of creating a new one.
+
+        Args:
+            segment: DocumentSegment to create summary for
+            dataset: Dataset containing the segment
+            summary_content: Generated summary content
+            status: Summary status (default: "generating")
+
+        Returns:
+            Created or updated DocumentSegmentSummary instance
+        """
+        # Check if summary record already exists
+        existing_summary = (
+            db.session.query(DocumentSegmentSummary)
+            .filter_by(chunk_id=segment.id, dataset_id=dataset.id)
+            .first()
+        )
+        
+        if existing_summary:
+            # Update existing record
+            existing_summary.summary_content = summary_content
+            existing_summary.status = status
+            existing_summary.error = None  # Clear any previous errors
+            # Re-enable if it was disabled
+            if not existing_summary.enabled:
+                existing_summary.enabled = True
+                existing_summary.disabled_at = None
+                existing_summary.disabled_by = None
+            db.session.add(existing_summary)
+            db.session.flush()
+            return existing_summary
+        else:
+            # Create new record (enabled by default)
+            summary_record = DocumentSegmentSummary(
+                dataset_id=dataset.id,
+                document_id=segment.document_id,
+                chunk_id=segment.id,
+                summary_content=summary_content,
+                status=status,
+                enabled=True,  # Explicitly set enabled to True
+            )
+            db.session.add(summary_record)
+            db.session.flush()
+            return summary_record
+
+    @staticmethod
+    def vectorize_summary(
+        summary_record: DocumentSegmentSummary,
+        segment: DocumentSegment,
+        dataset: Dataset,
+    ) -> None:
+        """
+        Vectorize summary and store in vector database.
+
+        Args:
+            summary_record: DocumentSegmentSummary record
+            segment: Original DocumentSegment
+            dataset: Dataset containing the segment
+        """
+        if dataset.indexing_technique != "high_quality":
+            logger.warning(
+                f"Summary vectorization skipped for dataset {dataset.id}: "
+                "indexing_technique is not high_quality"
+            )
+            return
+
+        # Reuse existing index_node_id if available (like segment does), otherwise generate new one
+        old_summary_node_id = summary_record.summary_index_node_id
+        if old_summary_node_id:
+            # Reuse existing index_node_id (like segment behavior)
+            summary_index_node_id = old_summary_node_id
+        else:
+            # Generate new index node ID only for new summaries
+            summary_index_node_id = str(uuid.uuid4())
+        
+        # Always regenerate hash (in case summary content changed)
+        summary_hash = helper.generate_text_hash(summary_record.summary_content)
+        
+        # Delete old vector only if we're reusing the same index_node_id (to overwrite)
+        # If index_node_id changed, the old vector should have been deleted elsewhere
+        if old_summary_node_id and old_summary_node_id == summary_index_node_id:
+            try:
+                vector = Vector(dataset)
+                vector.delete_by_ids([old_summary_node_id])
+            except Exception as e:
+                logger.warning(
+                    f"Failed to delete old summary vector for segment {segment.id}: {str(e)}. "
+                    "Continuing with new vectorization."
+                )
+
+        # Create document with summary content and metadata
+        summary_document = Document(
+            page_content=summary_record.summary_content,
+            metadata={
+                "doc_id": summary_index_node_id,
+                "doc_hash": summary_hash,
+                "dataset_id": dataset.id,
+                "document_id": segment.document_id,
+                "original_chunk_id": segment.id,  # Key: link to original chunk
+                "doc_type": DocType.TEXT,
+                "is_summary": True,  # Identifier for summary documents
+            },
+        )
+
+        # Vectorize and store with retry mechanism for connection errors
+        max_retries = 3
+        retry_delay = 2.0
+        
+        for attempt in range(max_retries):
+            try:
+                vector = Vector(dataset)
+                vector.add_texts([summary_document], duplicate_check=True)
+                
+                # Success - update summary record with index node info
+                summary_record.summary_index_node_id = summary_index_node_id
+                summary_record.summary_index_node_hash = summary_hash
+                summary_record.status = "completed"
+                db.session.add(summary_record)
+                db.session.flush()
+                return  # Success, exit function
+                
+            except (ConnectionError, Exception) as e:
+                error_str = str(e).lower()
+                # Check if it's a connection-related error that might be transient
+                is_connection_error = any(keyword in error_str for keyword in [
+                    "connection", "disconnected", "timeout", "network", 
+                    "could not connect", "server disconnected", "weaviate"
+                ])
+                
+                if is_connection_error and attempt < max_retries - 1:
+                    # Retry for connection errors
+                    wait_time = retry_delay * (2 ** attempt)  # Exponential backoff
+                    logger.warning(
+                        f"Vectorization attempt {attempt + 1}/{max_retries} failed for segment {segment.id}: {str(e)}. "
+                        f"Retrying in {wait_time:.1f} seconds..."
+                    )
+                    time.sleep(wait_time)
+                    continue
+                else:
+                    # Final attempt failed or non-connection error - log and update status
+                    logger.error(
+                        f"Failed to vectorize summary for segment {segment.id} after {attempt + 1} attempts: {str(e)}",
+                        exc_info=True
+                    )
+                    summary_record.status = "error"
+                    summary_record.error = f"Vectorization failed: {str(e)}"
+                    db.session.add(summary_record)
+                    db.session.flush()
+                    raise
+
+    @staticmethod
+    def generate_and_vectorize_summary(
+        segment: DocumentSegment,
+        dataset: Dataset,
+        summary_index_setting: dict,
+    ) -> DocumentSegmentSummary:
+        """
+        Generate summary for a segment and vectorize it.
+
+        Args:
+            segment: DocumentSegment to generate summary for
+            dataset: Dataset containing the segment
+            summary_index_setting: Summary index configuration
+
+        Returns:
+            Created DocumentSegmentSummary instance
+
+        Raises:
+            ValueError: If summary generation fails
+        """
+        try:
+            # Generate summary
+            summary_content = SummaryIndexService.generate_summary_for_segment(
+                segment, dataset, summary_index_setting
+            )
+
+            # Create or update summary record (will handle overwrite internally)
+            summary_record = SummaryIndexService.create_summary_record(
+                segment, dataset, summary_content, status="generating"
+            )
+
+            # Vectorize summary (will delete old vector if exists before creating new one)
+            SummaryIndexService.vectorize_summary(summary_record, segment, dataset)
+
+            db.session.commit()
+            logger.info(f"Successfully generated and vectorized summary for segment {segment.id}")
+            return summary_record
+
+        except Exception as e:
+            logger.exception(f"Failed to generate summary for segment {segment.id}: {str(e)}")
+            # Update summary record with error status if it exists
+            summary_record = (
+                db.session.query(DocumentSegmentSummary)
+                .filter_by(chunk_id=segment.id, dataset_id=dataset.id)
+                .first()
+            )
+            if summary_record:
+                summary_record.status = "error"
+                summary_record.error = str(e)
+                db.session.add(summary_record)
+                db.session.commit()
+            raise
+
+    @staticmethod
+    def generate_summaries_for_document(
+        dataset: Dataset,
+        document: DatasetDocument,
+        summary_index_setting: dict,
+        segment_ids: list[str] | None = None,
+        only_parent_chunks: bool = False,
+    ) -> list[DocumentSegmentSummary]:
+        """
+        Generate summaries for all segments in a document including vectorization.
+
+        Args:
+            dataset: Dataset containing the document
+            document: DatasetDocument to generate summaries for
+            summary_index_setting: Summary index configuration
+            segment_ids: Optional list of specific segment IDs to process
+            only_parent_chunks: If True, only process parent chunks (for parent-child mode)
+
+        Returns:
+            List of created DocumentSegmentSummary instances
+        """
+        # Only generate summary index for high_quality indexing technique
+        if dataset.indexing_technique != "high_quality":
+            logger.info(
+                f"Skipping summary generation for dataset {dataset.id}: "
+                f"indexing_technique is {dataset.indexing_technique}, not 'high_quality'"
+            )
+            return []
+
+        if not summary_index_setting or not summary_index_setting.get("enable"):
+            logger.info(f"Summary index is disabled for dataset {dataset.id}")
+            return []
+
+        # Skip qa_model documents
+        if document.doc_form == "qa_model":
+            logger.info(f"Skipping summary generation for qa_model document {document.id}")
+            return []
+
+        logger.info(
+            f"Starting summary generation for document {document.id} in dataset {dataset.id}, "
+            f"segment_ids: {len(segment_ids) if segment_ids else 'all'}, "
+            f"only_parent_chunks: {only_parent_chunks}"
+        )
+
+        # Query segments (only enabled segments)
+        query = db.session.query(DocumentSegment).filter_by(
+            dataset_id=dataset.id,
+            document_id=document.id,
+            status="completed",
+            enabled=True,  # Only generate summaries for enabled segments
+        )
+
+        if segment_ids:
+            query = query.filter(DocumentSegment.id.in_(segment_ids))
+
+        segments = query.all()
+
+        if not segments:
+            logger.info(f"No segments found for document {document.id}")
+            return []
+
+        summary_records = []
+
+        for segment in segments:
+            # For parent-child mode, only process parent chunks
+            # In parent-child mode, all DocumentSegments are parent chunks,
+            # so we process all of them. Child chunks are stored in ChildChunk table
+            # and are not DocumentSegments, so they won't be in the segments list.
+            # This check is mainly for clarity and future-proofing.
+            if only_parent_chunks:
+                # In parent-child mode, all segments in the query are parent chunks
+                # Child chunks are not DocumentSegments, so they won't appear here
+                # We can process all segments
+                pass
+
+            try:
+                summary_record = SummaryIndexService.generate_and_vectorize_summary(
+                    segment, dataset, summary_index_setting
+                )
+                summary_records.append(summary_record)
+            except Exception as e:
+                logger.error(f"Failed to generate summary for segment {segment.id}: {str(e)}")
+                # Continue with other segments
+                continue
+
+        logger.info(
+            f"Completed summary generation for document {document.id}: "
+            f"{len(summary_records)} summaries generated and vectorized"
+        )
+        return summary_records
+
+    @staticmethod
+    def disable_summaries_for_segments(
+        dataset: Dataset,
+        segment_ids: list[str] | None = None,
+        disabled_by: str | None = None,
+    ) -> None:
+        """
+        Disable summary records and remove vectors from vector database for segments.
+        Unlike delete, this preserves the summary records but marks them as disabled.
+
+        Args:
+            dataset: Dataset containing the segments
+            segment_ids: List of segment IDs to disable summaries for. If None, disable all.
+            disabled_by: User ID who disabled the summaries
+        """
+        from libs.datetime_utils import naive_utc_now
+        
+        query = db.session.query(DocumentSegmentSummary).filter_by(
+            dataset_id=dataset.id,
+            enabled=True,  # Only disable enabled summaries
+        )
+
+        if segment_ids:
+            query = query.filter(DocumentSegmentSummary.chunk_id.in_(segment_ids))
+
+        summaries = query.all()
+
+        if not summaries:
+            return
+
+        logger.info(
+            f"Disabling {len(summaries)} summary records for dataset {dataset.id}, "
+            f"segment_ids: {len(segment_ids) if segment_ids else 'all'}"
+        )
+
+        # Remove from vector database (but keep records)
+        if dataset.indexing_technique == "high_quality":
+            summary_node_ids = [
+                s.summary_index_node_id for s in summaries if s.summary_index_node_id
+            ]
+            if summary_node_ids:
+                try:
+                    vector = Vector(dataset)
+                    vector.delete_by_ids(summary_node_ids)
+                except Exception as e:
+                    logger.warning(f"Failed to remove summary vectors: {str(e)}")
+
+        # Disable summary records (don't delete)
+        now = naive_utc_now()
+        for summary in summaries:
+            summary.enabled = False
+            summary.disabled_at = now
+            summary.disabled_by = disabled_by
+            db.session.add(summary)
+
+        db.session.commit()
+        logger.info(f"Disabled {len(summaries)} summary records for dataset {dataset.id}")
+
+    @staticmethod
+    def enable_summaries_for_segments(
+        dataset: Dataset,
+        segment_ids: list[str] | None = None,
+    ) -> None:
+        """
+        Enable summary records and re-add vectors to vector database for segments.
+
+        Args:
+            dataset: Dataset containing the segments
+            segment_ids: List of segment IDs to enable summaries for. If None, enable all.
+        """
+        # Only enable summary index for high_quality indexing technique
+        if dataset.indexing_technique != "high_quality":
+            return
+
+        # Check if summary index is enabled
+        summary_index_setting = dataset.summary_index_setting
+        if not summary_index_setting or not summary_index_setting.get("enable"):
+            return
+
+        query = db.session.query(DocumentSegmentSummary).filter_by(
+            dataset_id=dataset.id,
+            enabled=False,  # Only enable disabled summaries
+        )
+
+        if segment_ids:
+            query = query.filter(DocumentSegmentSummary.chunk_id.in_(segment_ids))
+
+        summaries = query.all()
+
+        if not summaries:
+            return
+
+        logger.info(
+            f"Enabling {len(summaries)} summary records for dataset {dataset.id}, "
+            f"segment_ids: {len(segment_ids) if segment_ids else 'all'}"
+        )
+
+        # Re-vectorize and re-add to vector database
+        enabled_count = 0
+        for summary in summaries:
+            # Get the original segment
+            segment = db.session.query(DocumentSegment).filter_by(
+                id=summary.chunk_id,
+                dataset_id=dataset.id,
+            ).first()
+            
+            if not segment or not segment.enabled or segment.status != "completed":
+                continue
+
+            if not summary.summary_content:
+                continue
+
+            try:
+                # Re-vectorize summary
+                SummaryIndexService.vectorize_summary(summary, segment, dataset)
+                
+                # Enable summary record
+                summary.enabled = True
+                summary.disabled_at = None
+                summary.disabled_by = None
+                db.session.add(summary)
+                enabled_count += 1
+            except Exception as e:
+                logger.error(f"Failed to re-vectorize summary {summary.id}: {str(e)}")
+                # Keep it disabled if vectorization fails
+                continue
+
+        db.session.commit()
+        logger.info(f"Enabled {enabled_count} summary records for dataset {dataset.id}")
+
+    @staticmethod
+    def delete_summaries_for_segments(
+        dataset: Dataset,
+        segment_ids: list[str] | None = None,
+    ) -> None:
+        """
+        Delete summary records and vectors for segments (used only for actual deletion scenarios).
+        For disable/enable operations, use disable_summaries_for_segments/enable_summaries_for_segments.
+
+        Args:
+            dataset: Dataset containing the segments
+            segment_ids: List of segment IDs to delete summaries for. If None, delete all.
+        """
+        query = db.session.query(DocumentSegmentSummary).filter_by(dataset_id=dataset.id)
+
+        if segment_ids:
+            query = query.filter(DocumentSegmentSummary.chunk_id.in_(segment_ids))
+
+        summaries = query.all()
+
+        if not summaries:
+            return
+
+        # Delete from vector database
+        if dataset.indexing_technique == "high_quality":
+            summary_node_ids = [
+                s.summary_index_node_id for s in summaries if s.summary_index_node_id
+            ]
+            if summary_node_ids:
+                vector = Vector(dataset)
+                vector.delete_by_ids(summary_node_ids)
+
+        # Delete summary records
+        for summary in summaries:
+            db.session.delete(summary)
+
+        db.session.commit()
+        logger.info(f"Deleted {len(summaries)} summary records for dataset {dataset.id}")
+
+    @staticmethod
+    def update_summary_for_segment(
+        segment: DocumentSegment,
+        dataset: Dataset,
+        summary_content: str,
+    ) -> DocumentSegmentSummary | None:
+        """
+        Update summary for a segment and re-vectorize it.
+
+        Args:
+            segment: DocumentSegment to update summary for
+            dataset: Dataset containing the segment
+            summary_content: New summary content
+
+        Returns:
+            Updated DocumentSegmentSummary instance, or None if summary index is not enabled
+        """
+        # Only update summary index for high_quality indexing technique
+        if dataset.indexing_technique != "high_quality":
+            return None
+
+        # Check if summary index is enabled
+        summary_index_setting = dataset.summary_index_setting
+        if not summary_index_setting or not summary_index_setting.get("enable"):
+            return None
+
+        # Skip qa_model documents
+        if segment.document and segment.document.doc_form == "qa_model":
+            return None
+
+        try:
+            # Find existing summary record
+            summary_record = (
+                db.session.query(DocumentSegmentSummary)
+                .filter_by(chunk_id=segment.id, dataset_id=dataset.id)
+                .first()
+            )
+
+            if summary_record:
+                # Update existing summary
+                old_summary_node_id = summary_record.summary_index_node_id
+
+                # Update summary content
+                summary_record.summary_content = summary_content
+                summary_record.status = "generating"
+                db.session.add(summary_record)
+                db.session.flush()
+
+                # Delete old vector if exists
+                if old_summary_node_id:
+                    vector = Vector(dataset)
+                    vector.delete_by_ids([old_summary_node_id])
+
+                # Re-vectorize summary
+                SummaryIndexService.vectorize_summary(summary_record, segment, dataset)
+
+                db.session.commit()
+                logger.info(f"Successfully updated and re-vectorized summary for segment {segment.id}")
+                return summary_record
+            else:
+                # Create new summary record if doesn't exist
+                summary_record = SummaryIndexService.create_summary_record(
+                    segment, dataset, summary_content, status="generating"
+                )
+                SummaryIndexService.vectorize_summary(summary_record, segment, dataset)
+                db.session.commit()
+                logger.info(f"Successfully created and vectorized summary for segment {segment.id}")
+                return summary_record
+
+        except Exception as e:
+            logger.exception(f"Failed to update summary for segment {segment.id}: {str(e)}")
+            # Update summary record with error status if it exists
+            summary_record = (
+                db.session.query(DocumentSegmentSummary)
+                .filter_by(chunk_id=segment.id, dataset_id=dataset.id)
+                .first()
+            )
+            if summary_record:
+                summary_record.status = "error"
+                summary_record.error = str(e)
+                db.session.add(summary_record)
+                db.session.commit()
+            raise
+

api/services/trigger/trigger_provider_service.py

@@ -799,7 +799,7 @@ class TriggerProviderService:
         user_id: str,
         provider_id: TriggerProviderID,
         subscription_id: str,
-        credentials: Mapping[str, Any],
+        credentials: dict[str, Any],
     ) -> dict[str, Any]:
         """
         Verify credentials for an existing subscription without updating it.

api/services/workspace_service.py

@@ -31,7 +31,8 @@ class WorkspaceService:
         assert tenant_account_join is not None, "TenantAccountJoin not found"
         tenant_info["role"] = tenant_account_join.role
 
-        can_replace_logo = FeatureService.get_features(tenant.id).can_replace_logo
+        feature = FeatureService.get_features(tenant.id)
+        can_replace_logo = feature.can_replace_logo
 
         if can_replace_logo and TenantService.has_roles(tenant, [TenantAccountRole.OWNER, TenantAccountRole.ADMIN]):
             base_url = dify_config.FILES_URL
@@ -46,5 +47,19 @@ class WorkspaceService:
                 "remove_webapp_brand": remove_webapp_brand,
                 "replace_webapp_logo": replace_webapp_logo,
             }
+        if dify_config.EDITION == "CLOUD":
+            tenant_info["next_credit_reset_date"] = feature.next_credit_reset_date
+
+            from services.credit_pool_service import CreditPoolService
+
+            paid_pool = CreditPoolService.get_pool(tenant_id=tenant.id, pool_type="paid")
+            if paid_pool:
+                tenant_info["trial_credits"] = paid_pool.quota_limit
+                tenant_info["trial_credits_used"] = paid_pool.quota_used
+            else:
+                trial_pool = CreditPoolService.get_pool(tenant_id=tenant.id, pool_type="trial")
+                if trial_pool:
+                    tenant_info["trial_credits"] = trial_pool.quota_limit
+                    tenant_info["trial_credits_used"] = trial_pool.quota_used
 
         return tenant_info

api/tasks/add_document_to_index_task.py

@@ -117,6 +117,18 @@ def add_document_to_index_task(dataset_document_id: str):
         )
         db.session.commit()
 
+        # Enable summary indexes for all segments in this document
+        from services.summary_index_service import SummaryIndexService
+        segment_ids_list = [segment.id for segment in segments]
+        if segment_ids_list:
+            try:
+                SummaryIndexService.enable_summaries_for_segments(
+                    dataset=dataset,
+                    segment_ids=segment_ids_list,
+                )
+            except Exception as e:
+                logger.warning(f"Failed to enable summaries for document {dataset_document.id}: {str(e)}")
+
         end_at = time.perf_counter()
         logger.info(
             click.style(f"Document added to index: {dataset_document.id} latency: {end_at - start_at}", fg="green")

api/tasks/annotation/batch_import_annotations_task.py

@@ -31,7 +31,6 @@ def batch_import_annotations_task(job_id: str, content_list: list[dict], app_id:
     start_at = time.perf_counter()
     indexing_cache_key = f"app_annotation_batch_import_{str(job_id)}"
     active_jobs_key = f"annotation_import_active:{tenant_id}"
-
     # get app info
     app = db.session.query(App).where(App.id == app_id, App.tenant_id == tenant_id, App.status == "normal").first()
 
@@ -100,6 +99,5 @@ def batch_import_annotations_task(job_id: str, content_list: list[dict], app_id:
             except Exception as cleanup_error:
                 # Log but don't fail if cleanup fails - the job will be auto-expired
                 logger.warning("Failed to clean up active job tracking for %s: %s", job_id, cleanup_error)
-
             # Close database session
             db.session.close()

api/tasks/delete_segment_from_index_task.py

@@ -42,6 +42,7 @@ def delete_segment_from_index_task(
         doc_form = dataset_document.doc_form
 
         # Proceed with index cleanup using the index_node_ids directly
+        # For actual deletion, we should delete summaries (not just disable them)
         index_processor = IndexProcessorFactory(doc_form).init_index_processor()
         index_processor.clean(
             dataset,
@@ -49,6 +50,7 @@ def delete_segment_from_index_task(
             with_keywords=True,
             delete_child_chunks=True,
             precomputed_child_node_ids=child_node_ids,
+            delete_summaries=True,  # Actually delete summaries when segment is deleted
         )
         if dataset.is_multimodal:
             # delete segment attachment binding

api/tasks/disable_segment_from_index_task.py

@@ -53,6 +53,17 @@ def disable_segment_from_index_task(segment_id: str):
             logger.info(click.style(f"Segment {segment.id} document status is invalid, pass.", fg="cyan"))
             return
 
+        # Disable summary index for this segment
+        from services.summary_index_service import SummaryIndexService
+        try:
+            SummaryIndexService.disable_summaries_for_segments(
+                dataset=dataset,
+                segment_ids=[segment.id],
+                disabled_by=segment.disabled_by,
+            )
+        except Exception as e:
+            logger.warning(f"Failed to disable summary for segment {segment.id}: {str(e)}")
+
         index_type = dataset_document.doc_form
         index_processor = IndexProcessorFactory(index_type).init_index_processor()
         index_processor.clean(dataset, [segment.index_node_id])

api/tasks/disable_segments_from_index_task.py

@@ -58,12 +58,25 @@ def disable_segments_from_index_task(segment_ids: list, dataset_id: str, documen
         return
 
     try:
+        # Disable summary indexes for these segments
+        from services.summary_index_service import SummaryIndexService
+        segment_ids_list = [segment.id for segment in segments]
+        try:
+            # Get disabled_by from first segment (they should all have the same disabled_by)
+            disabled_by = segments[0].disabled_by if segments else None
+            SummaryIndexService.disable_summaries_for_segments(
+                dataset=dataset,
+                segment_ids=segment_ids_list,
+                disabled_by=disabled_by,
+            )
+        except Exception as e:
+            logger.warning(f"Failed to disable summaries for segments: {str(e)}")
+
         index_node_ids = [segment.index_node_id for segment in segments]
         if dataset.is_multimodal:
-            segment_ids = [segment.id for segment in segments]
             segment_attachment_bindings = (
                 db.session.query(SegmentAttachmentBinding)
-                .where(SegmentAttachmentBinding.segment_id.in_(segment_ids))
+                .where(SegmentAttachmentBinding.segment_id.in_(segment_ids_list))
                 .all()
             )
             if segment_attachment_bindings:

api/tasks/document_indexing_task.py

@@ -8,6 +8,7 @@ from celery import shared_task
 from configs import dify_config
 from core.entities.document_task import DocumentTask
 from core.indexing_runner import DocumentIsPausedError, IndexingRunner
+from tasks.generate_summary_index_task import generate_summary_index_task
 from core.rag.pipeline.queue import TenantIsolatedTaskQueue
 from enums.cloud_plan import CloudPlan
 from extensions.ext_database import db
@@ -100,6 +101,60 @@ def _document_indexing(dataset_id: str, document_ids: Sequence[str]):
         indexing_runner.run(documents)
         end_at = time.perf_counter()
         logger.info(click.style(f"Processed dataset: {dataset_id} latency: {end_at - start_at}", fg="green"))
+        
+        # Trigger summary index generation for completed documents if enabled
+        # Only generate for high_quality indexing technique and when summary_index_setting is enabled
+        # Re-query dataset to get latest summary_index_setting (in case it was updated)
+        dataset = db.session.query(Dataset).where(Dataset.id == dataset_id).first()
+        if not dataset:
+            logger.warning(f"Dataset {dataset_id} not found after indexing")
+            return
+        
+        if dataset.indexing_technique == "high_quality":
+            summary_index_setting = dataset.summary_index_setting
+            if summary_index_setting and summary_index_setting.get("enable"):
+                # Check each document's indexing status and trigger summary generation if completed
+                for document_id in document_ids:
+                    # Re-query document to get latest status (IndexingRunner may have updated it)
+                    document = (
+                        db.session.query(Document)
+                        .where(Document.id == document_id, Document.dataset_id == dataset_id)
+                        .first()
+                    )
+                    if document:
+                        logger.info(
+                            f"Checking document {document_id} for summary generation: "
+                            f"status={document.indexing_status}, doc_form={document.doc_form}"
+                        )
+                        if document.indexing_status == "completed" and document.doc_form != "qa_model":
+                            try:
+                                generate_summary_index_task.delay(dataset.id, document_id, None)
+                                logger.info(
+                                    f"Queued summary index generation task for document {document_id} "
+                                    f"in dataset {dataset.id} after indexing completed"
+                                )
+                            except Exception as e:
+                                logger.exception(
+                                    f"Failed to queue summary index generation task for document {document_id}: {str(e)}"
+                                )
+                                # Don't fail the entire indexing process if summary task queuing fails
+                        else:
+                            logger.info(
+                                f"Skipping summary generation for document {document_id}: "
+                                f"status={document.indexing_status}, doc_form={document.doc_form}"
+                            )
+                    else:
+                        logger.warning(f"Document {document_id} not found after indexing")
+            else:
+                logger.info(
+                    f"Summary index generation skipped for dataset {dataset.id}: "
+                    f"summary_index_setting.enable={summary_index_setting.get('enable') if summary_index_setting else None}"
+                )
+        else:
+            logger.info(
+                f"Summary index generation skipped for dataset {dataset.id}: "
+                f"indexing_technique={dataset.indexing_technique} (not 'high_quality')"
+            )
     except DocumentIsPausedError as ex:
         logger.info(click.style(str(ex), fg="yellow"))
     except Exception:

api/tasks/enable_segment_to_index_task.py

@@ -103,6 +103,16 @@ def enable_segment_to_index_task(segment_id: str):
         # save vector index
         index_processor.load(dataset, [document], multimodal_documents=multimodel_documents)
 
+        # Enable summary index for this segment
+        from services.summary_index_service import SummaryIndexService
+        try:
+            SummaryIndexService.enable_summaries_for_segments(
+                dataset=dataset,
+                segment_ids=[segment.id],
+            )
+        except Exception as e:
+            logger.warning(f"Failed to enable summary for segment {segment.id}: {str(e)}")
+
         end_at = time.perf_counter()
         logger.info(click.style(f"Segment enabled to index: {segment.id} latency: {end_at - start_at}", fg="green"))
     except Exception as e:

api/tasks/enable_segments_to_index_task.py

@@ -108,6 +108,17 @@ def enable_segments_to_index_task(segment_ids: list, dataset_id: str, document_i
         # save vector index
         index_processor.load(dataset, documents, multimodal_documents=multimodal_documents)
 
+        # Enable summary indexes for these segments
+        from services.summary_index_service import SummaryIndexService
+        segment_ids_list = [segment.id for segment in segments]
+        try:
+            SummaryIndexService.enable_summaries_for_segments(
+                dataset=dataset,
+                segment_ids=segment_ids_list,
+            )
+        except Exception as e:
+            logger.warning(f"Failed to enable summaries for segments: {str(e)}")
+
         end_at = time.perf_counter()
         logger.info(click.style(f"Segments enabled to index latency: {end_at - start_at}", fg="green"))
     except Exception as e:

api/tasks/generate_summary_index_task.py

@@ -0,0 +1,113 @@
+"""Async task for generating summary indexes."""
+
+import logging
+import time
+
+import click
+from celery import shared_task
+
+from extensions.ext_database import db
+from models.dataset import Dataset, DocumentSegment
+from models.dataset import Document as DatasetDocument
+from services.summary_index_service import SummaryIndexService
+
+logger = logging.getLogger(__name__)
+
+
+@shared_task(queue="dataset")
+def generate_summary_index_task(dataset_id: str, document_id: str, segment_ids: list[str] | None = None):
+    """
+    Async generate summary index for document segments.
+
+    Args:
+        dataset_id: Dataset ID
+        document_id: Document ID
+        segment_ids: Optional list of specific segment IDs to process. If None, process all segments.
+
+    Usage:
+        generate_summary_index_task.delay(dataset_id, document_id)
+        generate_summary_index_task.delay(dataset_id, document_id, segment_ids)
+    """
+    logger.info(
+        click.style(
+            f"Start generating summary index for document {document_id} in dataset {dataset_id}",
+            fg="green",
+        )
+    )
+    start_at = time.perf_counter()
+
+    try:
+        dataset = db.session.query(Dataset).where(Dataset.id == dataset_id).first()
+        if not dataset:
+            logger.error(click.style(f"Dataset not found: {dataset_id}", fg="red"))
+            db.session.close()
+            return
+
+        document = db.session.query(DatasetDocument).where(DatasetDocument.id == document_id).first()
+        if not document:
+            logger.error(click.style(f"Document not found: {document_id}", fg="red"))
+            db.session.close()
+            return
+
+        # Only generate summary index for high_quality indexing technique
+        if dataset.indexing_technique != "high_quality":
+            logger.info(
+                click.style(
+                    f"Skipping summary generation for dataset {dataset_id}: "
+                    f"indexing_technique is {dataset.indexing_technique}, not 'high_quality'",
+                    fg="cyan",
+                )
+            )
+            db.session.close()
+            return
+
+        # Check if summary index is enabled
+        summary_index_setting = dataset.summary_index_setting
+        if not summary_index_setting or not summary_index_setting.get("enable"):
+            logger.info(
+                click.style(
+                    f"Summary index is disabled for dataset {dataset_id}",
+                    fg="cyan",
+                )
+            )
+            db.session.close()
+            return
+
+        # Determine if only parent chunks should be processed
+        only_parent_chunks = dataset.chunk_structure == "parent_child_index"
+
+        # Generate summaries
+        summary_records = SummaryIndexService.generate_summaries_for_document(
+            dataset=dataset,
+            document=document,
+            summary_index_setting=summary_index_setting,
+            segment_ids=segment_ids,
+            only_parent_chunks=only_parent_chunks,
+        )
+
+        end_at = time.perf_counter()
+        logger.info(
+            click.style(
+                f"Summary index generation completed for document {document_id}: "
+                f"{len(summary_records)} summaries generated, latency: {end_at - start_at}",
+                fg="green",
+            )
+        )
+
+    except Exception as e:
+        logger.exception(f"Failed to generate summary index for document {document_id}: {str(e)}")
+        # Update document segments with error status if needed
+        if segment_ids:
+            db.session.query(DocumentSegment).filter(
+                DocumentSegment.id.in_(segment_ids),
+                DocumentSegment.dataset_id == dataset_id,
+            ).update(
+                {
+                    DocumentSegment.error: f"Summary generation failed: {str(e)}",
+                },
+                synchronize_session=False,
+            )
+            db.session.commit()
+    finally:
+        db.session.close()
+