chore(package): Bump version to 1.4.2 (#20897 )

Signed-off-by: -LAN- <laipz8200@outlook.com>
refactor: replace compact response generation with length-prefixed response for backwards invocation api (#20903 )
2025-06-11 16:28:36 +08:00 · 2025-06-11 16:01:50 +08:00 · 2025-06-11 15:38:51 +08:00 · 2025-06-11 01:01:33 +08:00 · 2025-06-10 17:34:26 +08:00 · 2025-06-10 10:46:17 +02:00
1040 changed files with 30399 additions and 9236 deletions
@@ -1,5 +1,4 @@
 FROM mcr.microsoft.com/devcontainers/python:3.12

-# [Optional] Uncomment this section to install additional OS packages.
-# RUN apt-get update && export DEBIAN_FRONTEND=noninteractive \
-#     && apt-get -y install --no-install-recommends <your-package-list-here>
+RUN apt-get update && export DEBIAN_FRONTEND=noninteractive \
+     && apt-get -y install libgmp-dev libmpfr-dev libmpc-dev
@@ -1,12 +1,13 @@
 #!/bin/bash

-npm add -g pnpm@10.8.0
+npm add -g pnpm@10.11.1
 cd web && pnpm install
 pipx install uv

 echo 'alias start-api="cd /workspaces/dify/api && uv run python -m flask run --host 0.0.0.0 --port=5001 --debug"' >> ~/.bashrc
 echo 'alias start-worker="cd /workspaces/dify/api && uv run python -m celery -A app.celery worker -P gevent -c 1 --loglevel INFO -Q dataset,generation,mail,ops_trace,app_deletion"' >> ~/.bashrc
 echo 'alias start-web="cd /workspaces/dify/web && pnpm dev"' >> ~/.bashrc
+echo 'alias start-web-prod="cd /workspaces/dify/web && pnpm build && pnpm start"' >> ~/.bashrc
 echo 'alias start-containers="cd /workspaces/dify/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env up -d"' >> ~/.bashrc
 echo 'alias stop-containers="cd /workspaces/dify/docker && docker-compose -f docker-compose.middleware.yaml -p dify --env-file middleware.env down"' >> ~/.bashrc

@@ -8,7 +8,7 @@ inputs:
  uv-version:
    description: UV version to set up
    required: true
-    default: '0.6.14'
+    default: '~=0.7.11'
  uv-lockfile:
    description: Path to the UV lockfile to restore cache from
    required: true
@@ -1,25 +1,23 @@
-# Summary
+> [!IMPORTANT]
+>
+> 1. Make sure you have read our [contribution guidelines](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md)
+> 2. Ensure there is an associated issue and you have been assigned to it
+> 3. Use the correct syntax to link this PR: `Fixes #<issue number>`.

-Please include a summary of the change and which issue is fixed. Please also include relevant motivation and context. List any dependencies that are required for this change.
+## Summary

-> [!Tip]
-> Close issue syntax: `Fixes #<issue number>` or `Resolves #<issue number>`, see [documentation](https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword) for more details.
+<!-- Please include a summary of the change and which issue is fixed. Please also include relevant motivation and context. List any dependencies that are required for this change. -->

-
-# Screenshots
+## Screenshots

 | Before | After |
 |--------|-------|
 | ...    | ...   |

-# Checklist
-
-> [!IMPORTANT]  
-> Please review the checklist below before submitting your pull request.
+## Checklist

 - [ ] This change requires a documentation update, included: [Dify Document](https://github.com/langgenius/dify-docs)
 - [x] I understand that this PR may be closed in case there was no previous discussion or issues. (This doesn't apply to typos!)
 - [x] I've added a test for each change that was introduced, and I tried as much as possible to make a single atomic change.
 - [x] I've updated the documentation accordingly.
 - [x] I ran `dev/reformat`(backend) and `cd web && npx lint-staged`(frontend) to appease the lint gods
-
@@ -139,6 +139,7 @@ jobs:
      - name: Checkout code
        uses: actions/checkout@v4
        with:
+          fetch-depth: 0
          persist-credentials: false

      - name: Check changed files
@@ -31,11 +31,19 @@ jobs:
            echo "FILES_CHANGED=false" >> $GITHUB_ENV
          fi

+      - name: Install pnpm
+        uses: pnpm/action-setup@v4
+        with:
+          version: 10
+          run_install: false
+
      - name: Set up Node.js
        if: env.FILES_CHANGED == 'true'
        uses: actions/setup-node@v4
        with:
          node-version: 'lts/*'
+          cache: pnpm
+          cache-dependency-path: ./web/package.json

      - name: Install dependencies
        if: env.FILES_CHANGED == 'true'
@@ -192,12 +192,12 @@ sdks/python-client/dist
 sdks/python-client/dify_client.egg-info

 .vscode/*
-!.vscode/launch.json
+!.vscode/launch.json.template
+!.vscode/README.md
 pyrightconfig.json
 api/.vscode

 .idea/
-.vscode

 # pnpm
 /.pnpm-store
@@ -207,3 +207,6 @@ plugins.jsonl

 # mise
 mise.toml
+
+# Next.js build output
+.next/
@@ -0,0 +1,14 @@
+# Debugging with VS Code
+
+This `launch.json.template` file provides various debug configurations for the Dify project within VS Code / Cursor. To use these configurations, you should copy the contents of this file into a new file named `launch.json` in the same `.vscode` directory.
+
+## How to Use
+
+1.  **Create `launch.json`**: If you don't have one, create a file named `launch.json` inside the `.vscode` directory.
+2.  **Copy Content**: Copy the entire content from `launch.json.template` into your newly created `launch.json` file.
+3.  **Select Debug Configuration**: Go to the Run and Debug view in VS Code / Cursor (Ctrl+Shift+D or Cmd+Shift+D).
+4.  **Start Debugging**: Select the desired configuration from the dropdown menu and click the green play button.
+
+## Tips
+
+- If you need to debug with Edge browser instead of Chrome, modify the `serverReadyAction` configuration in the "Next.js: debug full stack" section, change `"debugWithChrome"` to `"debugWithEdge"` to use Microsoft Edge for debugging.
@@ -0,0 +1,68 @@
+{
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Python: Flask API",
+            "type": "debugpy",
+            "request": "launch",
+            "module": "flask",
+            "env": {
+                "FLASK_APP": "app.py",
+                "FLASK_ENV": "development",
+                "GEVENT_SUPPORT": "True"
+            },
+            "args": [
+                "run",
+                "--host=0.0.0.0",
+                "--port=5001",
+                "--no-debugger",
+                "--no-reload"
+            ],
+            "jinja": true,
+            "justMyCode": true,
+            "cwd": "${workspaceFolder}/api",
+            "python": "${workspaceFolder}/api/.venv/bin/python"
+        },
+        {
+            "name": "Python: Celery Worker (Solo)",
+            "type": "debugpy",
+            "request": "launch",
+            "module": "celery",
+            "env": {
+                "GEVENT_SUPPORT": "True"
+            },
+            "args": [
+                "-A",
+                "app.celery",
+                "worker",
+                "-P",
+                "solo",
+                "-c",
+                "1",
+                "-Q",
+                "dataset,generation,mail,ops_trace",
+                "--loglevel",
+                "INFO"
+            ],
+            "justMyCode": false,
+            "cwd": "${workspaceFolder}/api",
+            "python": "${workspaceFolder}/api/.venv/bin/python"
+        },
+        {
+            "name": "Next.js: debug full stack",
+            "type": "node",
+            "request": "launch",
+            "program": "${workspaceFolder}/web/node_modules/next/dist/bin/next",
+            "runtimeArgs": ["--inspect"],
+            "skipFiles": ["<node_internals>/**"],
+            "serverReadyAction": {
+              "action": "debugWithChrome",
+              "killOnServerStop": true,
+              "pattern": "- Local:.+(https?://.+)",
+              "uriFormat": "%s",
+              "webRoot": "${workspaceFolder}/web"
+            },
+            "cwd": "${workspaceFolder}/web"
+          }
+    ]
+}
@@ -1,4 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
+![cover-v5-optimized](./images/GitHub_README_if.png)

 <p align="center">
  📌 <a href="https://dify.ai/blog/introducing-dify-workflow-file-upload-a-demo-on-ai-podcast">Introducing Dify Workflow File Upload: Recreate Google NotebookLM Podcast</a>
@@ -87,8 +87,6 @@ Please refer to our [FAQ](https://docs.dify.ai/getting-started/install-self-host
 **1. Workflow**:
 Build and test powerful AI workflows on a visual canvas, leveraging all the following features and beyond.

-https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa
-
 **2. Comprehensive model support**:
 Seamless integration with hundreds of proprietary / open-source LLMs from dozens of inference providers and self-hosted solutions, covering GPT, Mistral, Llama3, and any OpenAI API-compatible models. A full list of supported model providers can be found [here](https://docs.dify.ai/getting-started/readme/model-providers).

@@ -237,7 +235,7 @@ At the same time, please consider supporting Dify by sharing it on social media

 ## Community & contact

- [Github Discussion](https://github.com/langgenius/dify/discussions). Best for: sharing feedback and asking questions.
+- [GitHub Discussion](https://github.com/langgenius/dify/discussions). Best for: sharing feedback and asking questions.
 - [GitHub Issues](https://github.com/langgenius/dify/issues). Best for: bugs you encounter using Dify.AI, and feature proposals. See our [Contribution Guide](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md).
 - [Discord](https://discord.gg/FngNHpbcY7). Best for: sharing your applications and hanging out with the community.
 - [X(Twitter)](https://twitter.com/dify_ai). Best for: sharing your applications and hanging out with the community.
@@ -1,4 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
+![cover-v5-optimized](./images/GitHub_README_if.png)

 <p align="center">
  <a href="https://cloud.dify.ai">Dify Cloud</a> ·
@@ -54,8 +54,6 @@

 **1. سير العمل**:  قم ببناء واختبار سير عمل الذكاء الاصطناعي القوي على قماش بصري، مستفيدًا من جميع الميزات التالية وأكثر.

-  <https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa>
-
 **2. الدعم الشامل للنماذج**: تكامل سلس مع مئات من LLMs الخاصة / مفتوحة المصدر من عشرات من موفري التحليل والحلول المستضافة ذاتيًا، مما يغطي GPT و Mistral و Llama3 وأي نماذج متوافقة مع واجهة OpenAI API. يمكن العثور على قائمة كاملة بمزودي النموذج المدعومين [هنا](https://docs.dify.ai/getting-started/readme/model-providers).

 ![providers-v5](https://github.com/langgenius/dify/assets/13230914/5a17bdbe-097a-4100-8363-40255b70f6e3)
@@ -225,7 +223,7 @@ docker compose up -d
 </a>

 ## المجتمع والاتصال
- [مناقشة Github](https://github.com/langgenius/dify/discussions). الأفضل لـ: مشاركة التعليقات وطرح الأسئلة.
+- [مناقشة GitHub](https://github.com/langgenius/dify/discussions). الأفضل لـ: مشاركة التعليقات وطرح الأسئلة.
 - [المشكلات على GitHub](https://github.com/langgenius/dify/issues). الأفضل لـ: الأخطاء التي تواجهها في استخدام Dify.AI، واقتراحات الميزات. انظر [دليل المساهمة](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md).
 - [Discord](https://discord.gg/FngNHpbcY7). الأفضل لـ: مشاركة تطبيقاتك والترفيه مع المجتمع.
 - [تويتر](https://twitter.com/dify_ai). الأفضل لـ: مشاركة تطبيقاتك والترفيه مع المجتمع.
@@ -1,4 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
+![cover-v5-optimized](./images/GitHub_README_if.png)

 <p align="center">
  📌 <a href="https://dify.ai/blog/introducing-dify-workflow-file-upload-a-demo-on-ai-podcast">ডিফাই ওয়ার্কফ্লো ফাইল আপলোড পরিচিতি: গুগল নোটবুক-এলএম পডকাস্ট পুনর্নির্মাণ</a>
@@ -84,8 +84,6 @@ docker compose up -d
 **১. ওয়ার্কফ্লো**:
    ভিজ্যুয়াল ক্যানভাসে  AI ওয়ার্কফ্লো তৈরি এবং পরীক্ষা করুন, নিম্নলিখিত সব ফিচার এবং  তার বাইরেও আরও অনেক কিছু ব্যবহার করে।

-  <https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa>
-
 **২.  মডেল সাপোর্ট**:
  GPT, Mistral, Llama3, এবং যেকোনো OpenAI API-সামঞ্জস্যপূর্ণ মডেলসহ, কয়েক ডজন ইনফারেন্স প্রদানকারী এবং সেল্ফ-হোস্টেড সমাধান থেকে শুরু করে প্রোপ্রাইটরি/ওপেন-সোর্স LLM-এর সাথে সহজে ইন্টিগ্রেশন। সমর্থিত মডেল প্রদানকারীদের একটি সম্পূর্ণ তালিকা পাওয়া যাবে [এখানে](https://docs.dify.ai/getting-started/readme/model-providers)।

@@ -236,7 +234,7 @@ GitHub-এ ডিফাইকে স্টার দিয়ে রাখুন

 ## কমিউনিটি এবং যোগাযোগ

- [Github Discussion](https://github.com/langgenius/dify/discussions) ফিডব্যাক এবং প্রতিক্রিয়া জানানোর মাধ্যম।
+- [GitHub Discussion](https://github.com/langgenius/dify/discussions) ফিডব্যাক এবং প্রতিক্রিয়া জানানোর মাধ্যম।
 - [GitHub Issues](https://github.com/langgenius/dify/issues). Dify.AI ব্যবহার করে আপনি যেসব বাগের সম্মুখীন হন এবং ফিচার প্রস্তাবনা।  আমাদের [অবদান নির্দেশিকা](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md) দেখুন।
 - [Discord](https://discord.gg/FngNHpbcY7) আপনার এপ্লিকেশন শেয়ার এবং কমিউনিটি আড্ডার মাধ্যম। 
 - [X(Twitter)](https://twitter.com/dify_ai) আপনার এপ্লিকেশন শেয়ার এবং কমিউনিটি আড্ডার মাধ্যম। 
@@ -1,4 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
+![cover-v5-optimized](./images/GitHub_README_if.png)

 <div align="center">
  <a href="https://cloud.dify.ai">Dify 云服务</a> ·
@@ -61,11 +61,6 @@ Dify 是一个开源的 LLM 应用开发平台。其直观的界面结合了 AI
 **1. 工作流**: 
  在画布上构建和测试功能强大的 AI 工作流程，利用以下所有功能以及更多功能。

-
-  https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa
-
-
-
 **2. 全面的模型支持**: 
  与数百种专有/开源 LLMs 以及数十种推理提供商和自托管解决方案无缝集成，涵盖 GPT、Mistral、Llama3 以及任何与 OpenAI API 兼容的模型。完整的支持模型提供商列表可在[此处](https://docs.dify.ai/getting-started/readme/model-providers)找到。

@@ -248,7 +243,7 @@ docker compose up -d

 我们欢迎您为 Dify 做出贡献，以帮助改善 Dify。包括：提交代码、问题、新想法，或分享您基于 Dify 创建的有趣且有用的 AI 应用程序。同时，我们也欢迎您在不同的活动、会议和社交媒体上分享 Dify。

- [Github Discussion](https://github.com/langgenius/dify/discussions). 👉：分享您的应用程序并与社区交流。
+- [GitHub Discussion](https://github.com/langgenius/dify/discussions). 👉：分享您的应用程序并与社区交流。
 - [GitHub Issues](https://github.com/langgenius/dify/issues)。👉：使用 Dify.AI 时遇到的错误和问题，请参阅[贡献指南](CONTRIBUTING.md)。
 - [电子邮件支持](mailto:hello@dify.ai?subject=[GitHub]Questions%20About%20Dify)。👉：关于使用 Dify.AI 的问题。
 - [Discord](https://discord.gg/FngNHpbcY7)。👉：分享您的应用程序并与社区交流。
@@ -1,4 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
+![cover-v5-optimized](./images/GitHub_README_if.png)

 <p align="center">
  📌 <a href="https://dify.ai/blog/introducing-dify-workflow-file-upload-a-demo-on-ai-podcast">Einführung in Dify Workflow File Upload: Google NotebookLM Podcast nachbilden</a>
@@ -83,11 +83,6 @@ Bitte beachten Sie unsere [FAQ](https://docs.dify.ai/getting-started/install-sel
 **1. Workflow**: 
  Erstellen und testen Sie leistungsstarke KI-Workflows auf einer visuellen Oberfläche, wobei Sie alle der folgenden Funktionen und darüber hinaus nutzen können.

-
-  https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa
-
-
-
 **2. Umfassende Modellunterstützung**: 
  Nahtlose Integration mit Hunderten von proprietären und Open-Source-LLMs von Dutzenden Inferenzanbietern und selbstgehosteten Lösungen, die GPT, Mistral, Llama3 und alle mit der OpenAI API kompatiblen Modelle abdecken. Eine vollständige Liste der unterstützten Modellanbieter finden Sie [hier](https://docs.dify.ai/getting-started/readme/model-providers).

@@ -235,7 +230,7 @@ Falls Sie Code beitragen möchten, lesen Sie bitte unseren [Contribution Guide](

 ## Gemeinschaft & Kontakt

-* [Github Discussion](https://github.com/langgenius/dify/discussions). Am besten geeignet für: den Austausch von Feedback und das Stellen von Fragen.
+* [GitHub Discussion](https://github.com/langgenius/dify/discussions). Am besten geeignet für: den Austausch von Feedback und das Stellen von Fragen.
 * [GitHub Issues](https://github.com/langgenius/dify/issues). Am besten für: Fehler, auf die Sie bei der Verwendung von Dify.AI stoßen, und Funktionsvorschläge. Siehe unseren [Contribution Guide](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md).
 * [Discord](https://discord.gg/FngNHpbcY7).  Am besten geeignet für: den Austausch von Bewerbungen und den Austausch mit der Community.
 * [X(Twitter)](https://twitter.com/dify_ai). Am besten geeignet für: den Austausch von Bewerbungen und den Austausch mit der Community.
@@ -1,4 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
+![cover-v5-optimized](./images/GitHub_README_if.png)

 <p align="center">
  <a href="https://cloud.dify.ai">Dify Cloud</a> ·
@@ -59,11 +59,6 @@ Dify es una plataforma de desarrollo de aplicaciones de LLM de código abierto.
 **1. Flujo de trabajo**: 
  Construye y prueba potentes flujos de trabajo de IA en un lienzo visual, aprovechando todas las siguientes características y más.

-
-  https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa
-
-
-
 **2. Soporte de modelos completo**: 
  Integración perfecta con cientos de LLMs propietarios / de código abierto de docenas de proveedores de inferencia y soluciones auto-alojadas, que cubren GPT, Mistral, Llama3 y cualquier modelo compatible con la API de OpenAI. Se puede encontrar una lista completa de proveedores de modelos admitidos [aquí](https://docs.dify.ai/getting-started/readme/model-providers).

@@ -1,4 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
+![cover-v5-optimized](./images/GitHub_README_if.png)

 <p align="center">
  <a href="https://cloud.dify.ai">Dify Cloud</a> ·
@@ -59,11 +59,6 @@ Dify est une plateforme de développement d'applications LLM open source. Son in
 **1. Flux de travail** : 
  Construisez et testez des flux de travail d'IA puissants sur un canevas visuel, en utilisant toutes les fonctionnalités suivantes et plus encore.

-
-  https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa
-
-
-
 **2. Prise en charge complète des modèles** : 
  Intégration transparente avec des centaines de LLM propriétaires / open source provenant de dizaines de fournisseurs d'inférence et de solutions auto-hébergées, couvrant GPT, Mistral, Llama3, et tous les modèles compatibles avec l'API OpenAI. Une liste complète des fournisseurs de modèles pris en charge se trouve [ici](https://docs.dify.ai/getting-started/readme/model-providers).

@@ -1,4 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
+![cover-v5-optimized](./images/GitHub_README_if.png)

 <p align="center">
  <a href="https://cloud.dify.ai">Dify Cloud</a> ·
@@ -60,11 +60,6 @@ DifyはオープンソースのLLMアプリケーション開発プラットフ
 **1. ワークフロー**:
  強力なAIワークフローをビジュアルキャンバス上で構築し、テストできます。すべての機能、および以下の機能を使用できます。

-
-  https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa
-
-
-
 **2. 総合的なモデルサポート**:
  数百ものプロプライエタリ/オープンソースのLLMと、数十もの推論プロバイダーおよびセルフホスティングソリューションとのシームレスな統合を提供します。GPT、Mistral、Llama3、OpenAI APIと互換性のあるすべてのモデルを統合されています。サポートされているモデルプロバイダーの完全なリストは[こちら](https://docs.dify.ai/getting-started/readme/model-providers)をご覧ください。

@@ -241,7 +236,7 @@ docker compose up -d

 ## コミュニティ & お問い合わせ

-* [Github Discussion](https://github.com/langgenius/dify/discussions). 主に: フィードバックの共有や質問。
+* [GitHub Discussion](https://github.com/langgenius/dify/discussions). 主に: フィードバックの共有や質問。
 * [GitHub Issues](https://github.com/langgenius/dify/issues). 主に: Dify.AIを使用する際に発生するエラーや問題については、[貢献ガイド](CONTRIBUTING_JA.md)を参照してください
 * [Discord](https://discord.gg/FngNHpbcY7). 主に: アプリケーションの共有やコミュニティとの交流。
 * [X(Twitter)](https://twitter.com/dify_ai). 主に: アプリケーションの共有やコミュニティとの交流。
@@ -1,4 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
+![cover-v5-optimized](./images/GitHub_README_if.png)

 <p align="center">
  <a href="https://cloud.dify.ai">Dify Cloud</a> ·
@@ -59,11 +59,6 @@ Dify is an open-source LLM app development platform. Its intuitive interface com
 **1. Workflow**: 
  Build and test powerful AI workflows on a visual canvas, leveraging all the following features and beyond.

-
-  https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa
-
-
-
 **2. Comprehensive model support**: 
  Seamless integration with hundreds of proprietary / open-source LLMs from dozens of inference providers and self-hosted solutions, covering GPT, Mistral, Llama3, and any OpenAI API-compatible models. A full list of supported model providers can be found [here](https://docs.dify.ai/getting-started/readme/model-providers).

@@ -240,7 +235,7 @@ At the same time, please consider supporting Dify by sharing it on social media

 ## Community & Contact

-* [Github Discussion](https://github.com/langgenius/dify/discussions
+* [GitHub Discussion](https://github.com/langgenius/dify/discussions

 ). Best for: sharing feedback and asking questions.
 * [GitHub Issues](https://github.com/langgenius/dify/issues). Best for: bugs you encounter using Dify.AI, and feature proposals. See our [Contribution Guide](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md).
@@ -1,4 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
+![cover-v5-optimized](./images/GitHub_README_if.png)

 <p align="center">
  <a href="https://cloud.dify.ai">Dify 클라우드</a> ·
@@ -54,11 +54,6 @@
 **1. 워크플로우**:
    다음 기능들을 비롯한 다양한 기능을 활용하여 시각적 캔버스에서 강력한 AI 워크플로우를 구축하고 테스트하세요.

-
-  https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa
-
-
-
 **2. 포괄적인 모델 지원:**: 

 수십 개의 추론 제공업체와 자체 호스팅 솔루션에서 제공하는 수백 개의 독점 및 오픈 소스 LLM과 원활하게 통합되며, GPT, Mistral, Llama3 및 모든 OpenAI API 호환 모델을 포함합니다. 지원되는 모델 제공업체의 전체 목록은 [여기](https://docs.dify.ai/getting-started/readme/model-providers)에서 확인할 수 있습니다.
@@ -234,7 +229,7 @@ Dify를 Kubernetes에 배포하고 프리미엄 스케일링 설정을 구성했

 ## 커뮤니티 & 연락처

-* [Github 토론](https://github.com/langgenius/dify/discussions). 피드백 공유 및 질문하기에 적합합니다.
+* [GitHub 토론](https://github.com/langgenius/dify/discussions). 피드백 공유 및 질문하기에 적합합니다.
 * [GitHub 이슈](https://github.com/langgenius/dify/issues). Dify.AI 사용 중 발견한 버그와 기능 제안에 적합합니다. [기여 가이드](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md)를 참조하세요. 
 * [디스코드](https://discord.gg/FngNHpbcY7). 애플리케이션 공유 및 커뮤니티와 소통하기에 적합합니다.
 * [트위터](https://twitter.com/dify_ai). 애플리케이션 공유 및 커뮤니티와 소통하기에 적합합니다.
@@ -1,5 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
-
+![cover-v5-optimized](./images/GitHub_README_if.png)
 <p align="center">
  📌 <a href="https://dify.ai/blog/introducing-dify-workflow-file-upload-a-demo-on-ai-podcast">Introduzindo o Dify Workflow com Upload de Arquivo: Recrie o Podcast Google NotebookLM</a>
 </p>
@@ -59,11 +58,6 @@ Dify é uma plataforma de desenvolvimento de aplicativos LLM de código aberto.
 **1. Workflow**: 
  Construa e teste workflows poderosos de IA em uma interface visual, aproveitando todos os recursos a seguir e muito mais.

-
-  https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa
-
-
-
 **2. Suporte abrangente a modelos**: 
  Integração perfeita com centenas de LLMs proprietários e de código aberto de diversas provedoras e soluções auto-hospedadas, abrangendo GPT, Mistral, Llama3 e qualquer modelo compatível com a API da OpenAI. A lista completa de provedores suportados pode ser encontrada [aqui](https://docs.dify.ai/getting-started/readme/model-providers).

@@ -1,4 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
+![cover-v5-optimized](./images/GitHub_README_if.png)

 <p align="center">
  📌 <a href="https://dify.ai/blog/introducing-dify-workflow-file-upload-a-demo-on-ai-podcast">Predstavljamo nalaganje datotek Dify Workflow: znova ustvarite Google NotebookLM Podcast</a>
@@ -81,11 +81,6 @@ Prosimo, glejte naša pogosta vprašanja [FAQ](https://docs.dify.ai/getting-star
 **1. Potek dela**: 
  Zgradite in preizkusite zmogljive poteke dela AI na vizualnem platnu, pri čemer izkoristite vse naslednje funkcije in več.

-
-  https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa
-
-
-
 **2. Celovita podpora za modele**: 
  Brezhibna integracija s stotinami lastniških/odprtokodnih LLM-jev ducatov ponudnikov sklepanja in samostojnih rešitev, ki pokrivajo GPT, Mistral, Llama3 in vse modele, združljive z API-jem OpenAI. Celoten seznam podprtih ponudnikov modelov najdete [tukaj](https://docs.dify.ai/getting-started/readme/model-providers).

@@ -234,7 +229,7 @@ Za tiste, ki bi radi prispevali kodo, si oglejte naš vodnik za prispevke . Hkra

 ## Skupnost in stik

-* [Github Discussion](https://github.com/langgenius/dify/discussions). Najboljše za: izmenjavo povratnih informacij in postavljanje vprašanj.
+* [GitHub Discussion](https://github.com/langgenius/dify/discussions). Najboljše za: izmenjavo povratnih informacij in postavljanje vprašanj.
 * [GitHub Issues](https://github.com/langgenius/dify/issues). Najboljše za: hrošče, na katere naletite pri uporabi Dify.AI, in predloge funkcij. Oglejte si naš [vodnik za prispevke](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md).
 * [Discord](https://discord.gg/FngNHpbcY7). Najboljše za: deljenje vaših aplikacij in druženje s skupnostjo.
 * [X(Twitter)](https://twitter.com/dify_ai). Najboljše za: deljenje vaših aplikacij in druženje s skupnostjo.
@@ -1,4 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
+![cover-v5-optimized](./images/GitHub_README_if.png)

 <p align="center">
  <a href="https://cloud.dify.ai">Dify Bulut</a> ·
@@ -55,11 +55,6 @@ Dify, açık kaynaklı bir LLM uygulama geliştirme platformudur. Sezgisel aray
 **1. Workflow**: 
 Görsel bir arayüz üzerinde güçlü AI iş akışları oluşturun ve test edin, aşağıdaki tüm özellikleri ve daha fazlasını kullanarak.

-
-  https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa
-
-
-
 **2. Kapsamlı model desteği**:
 Çok sayıda çıkarım sağlayıcısı ve kendi kendine barındırılan çözümlerden yüzlerce özel / açık kaynaklı LLM ile sorunsuz entegrasyon sağlar. GPT, Mistral, Llama3 ve OpenAI API uyumlu tüm modelleri kapsar. Desteklenen model sağlayıcılarının tam listesine [buradan](https://docs.dify.ai/getting-started/readme/model-providers) ulaşabilirsiniz.

@@ -232,7 +227,7 @@ Aynı zamanda, lütfen Dify'ı sosyal medyada, etkinliklerde ve konferanslarda p

 ## Topluluk & iletişim

-* [Github Tartışmaları](https://github.com/langgenius/dify/discussions). En uygun: geri bildirim paylaşmak ve soru sormak için.
+* [GitHub Tartışmaları](https://github.com/langgenius/dify/discussions). En uygun: geri bildirim paylaşmak ve soru sormak için.
 * [GitHub Sorunları](https://github.com/langgenius/dify/issues). En uygun: Dify.AI kullanırken karşılaştığınız hatalar ve özellik önerileri için. [Katkı Kılavuzumuza](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md) bakın.
 * [Discord](https://discord.gg/FngNHpbcY7). En uygun: uygulamalarınızı paylaşmak ve toplulukla vakit geçirmek için.
 * [X(Twitter)](https://twitter.com/dify_ai). En uygun: uygulamalarınızı paylaşmak ve toplulukla vakit geçirmek için.
@@ -1,4 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
+![cover-v5-optimized](./images/GitHub_README_if.png)

 <p align="center">
  📌 <a href="https://dify.ai/blog/introducing-dify-workflow-file-upload-a-demo-on-ai-podcast">介紹 Dify 工作流程檔案上傳功能：重現 Google NotebookLM Podcast</a>
@@ -86,8 +86,6 @@ docker compose up -d
 **1. 工作流程**：
 在視覺化畫布上建立和測試強大的 AI 工作流程，利用以下所有功能及更多。

-https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa
-
 **2. 全面的模型支援**：
 無縫整合來自數十個推理提供商和自託管解決方案的數百個專有/開源 LLM，涵蓋 GPT、Mistral、Llama3 和任何與 OpenAI API 兼容的模型。您可以在[此處](https://docs.dify.ai/getting-started/readme/model-providers)找到支援的模型提供商完整列表。

@@ -235,7 +233,7 @@ Dify 的所有功能都提供相應的 API，因此您可以輕鬆地將 Dify

 ## 社群與聯絡方式

- [Github Discussion](https://github.com/langgenius/dify/discussions)：最適合分享反饋和提問。
+- [GitHub Discussion](https://github.com/langgenius/dify/discussions)：最適合分享反饋和提問。
 - [GitHub Issues](https://github.com/langgenius/dify/issues)：最適合報告使用 Dify.AI 時遇到的問題和提出功能建議。請參閱我們的[貢獻指南](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md)。
 - [Discord](https://discord.gg/FngNHpbcY7)：最適合分享您的應用程式並與社群互動。
 - [X(Twitter)](https://twitter.com/dify_ai)：最適合分享您的應用程式並與社群互動。
@@ -1,4 +1,4 @@
-![cover-v5-optimized](https://github.com/langgenius/dify/assets/13230914/f9e19af5-61ba-4119-b926-d10c4c06ebab)
+![cover-v5-optimized](./images/GitHub_README_if.png)

 <p align="center">
  <a href="https://cloud.dify.ai">Dify Cloud</a> ·
@@ -55,11 +55,6 @@ Dify là một nền tảng phát triển ứng dụng LLM mã nguồn mở. Gia
 **1. Quy trình làm việc**: 
  Xây dựng và kiểm tra các quy trình làm việc AI mạnh mẽ trên một canvas trực quan, tận dụng tất cả các tính năng sau đây và hơn thế nữa.

-
-  https://github.com/langgenius/dify/assets/13230914/356df23e-1604-483d-80a6-9517ece318aa
-
-
-
 **2. Hỗ trợ mô hình toàn diện**: 
  Tích hợp liền mạch với hàng trăm mô hình LLM độc quyền / mã nguồn mở từ hàng chục nhà cung cấp suy luận và giải pháp tự lưu trữ, bao gồm GPT, Mistral, Llama3, và bất kỳ mô hình tương thích API OpenAI nào. Danh sách đầy đủ các nhà cung cấp mô hình được hỗ trợ có thể được tìm thấy [tại đây](https://docs.dify.ai/getting-started/readme/model-providers).

@@ -152,6 +152,7 @@ QDRANT_API_KEY=difyai123456
 QDRANT_CLIENT_TIMEOUT=20
 QDRANT_GRPC_ENABLED=false
 QDRANT_GRPC_PORT=6334
+QDRANT_REPLICATION_FACTOR=1

 #Couchbase configuration
 COUCHBASE_CONNECTION_STRING=127.0.0.1
@@ -269,6 +270,7 @@ OPENSEARCH_PORT=9200
 OPENSEARCH_USER=admin
 OPENSEARCH_PASSWORD=admin
 OPENSEARCH_SECURE=true
+OPENSEARCH_VERIFY_CERTS=true

 # Baidu configuration
 BAIDU_VECTOR_DB_ENDPOINT=http://127.0.0.1:5287
@@ -348,6 +350,7 @@ SENTRY_DSN=

 # DEBUG
 DEBUG=false
+ENABLE_REQUEST_LOGGING=False
 SQLALCHEMY_ECHO=false

 # Notion import configuration, support public and internal
@@ -476,6 +479,7 @@ LOGIN_LOCKOUT_DURATION=86400
 ENABLE_OTEL=false
 OTLP_BASE_ENDPOINT=http://localhost:4318
 OTLP_API_KEY=
+OTEL_EXPORTER_OTLP_PROTOCOL=
 OTEL_EXPORTER_TYPE=otlp
 OTEL_SAMPLING_RATE=0.1
 OTEL_BATCH_EXPORT_SCHEDULE_DELAY=5000
@@ -487,3 +491,10 @@ OTEL_METRIC_EXPORT_TIMEOUT=30000

 # Prevent Clickjacking
 ALLOW_EMBED=false
+
+# Dataset queue monitor configuration
+QUEUE_MONITOR_THRESHOLD=200
+# You can configure multiple ones, separated by commas. eg: test1@dify.ai,test2@dify.ai
+QUEUE_MONITOR_ALERT_EMAILS=
+# Monitor interval in minutes, default is 30 minutes
+QUEUE_MONITOR_INTERVAL=30
@@ -43,6 +43,7 @@ select = [
    "S307", # suspicious-eval-usage, disallow use of `eval` and `ast.literal_eval`
    "S301", # suspicious-pickle-usage, disallow use of `pickle` and its wrappers.
    "S302", # suspicious-marshal-usage, disallow use of `marshal` module
+    "S311", # suspicious-non-cryptographic-random-usage
 ]

 ignore = [
@@ -4,7 +4,7 @@ FROM python:3.12-slim-bookworm AS base
 WORKDIR /app/api

 # Install uv
-ENV UV_VERSION=0.6.14
+ENV UV_VERSION=0.7.11

 RUN pip install --no-cache-dir uv==${UV_VERSION}

@@ -54,6 +54,7 @@ def initialize_extensions(app: DifyApp):
        ext_otel,
        ext_proxy_fix,
        ext_redis,
+        ext_request_logging,
        ext_sentry,
        ext_set_secretkey,
        ext_storage,
@@ -83,6 +84,7 @@ def initialize_extensions(app: DifyApp):
        ext_blueprints,
        ext_commands,
        ext_otel,
+        ext_request_logging,
    ]
    for ext in extensions:
        short_name = ext.__name__.split(".")[-1]
@@ -6,6 +6,7 @@ from typing import Optional

 import click
 from flask import current_app
+from sqlalchemy import select
 from werkzeug.exceptions import NotFound

 from configs import dify_config
@@ -297,11 +298,11 @@ def migrate_knowledge_vector_database():
    page = 1
    while True:
        try:
-            datasets = (
-                Dataset.query.filter(Dataset.indexing_technique == "high_quality")
-                .order_by(Dataset.created_at.desc())
-                .paginate(page=page, per_page=50)
+            stmt = (
+                select(Dataset).filter(Dataset.indexing_technique == "high_quality").order_by(Dataset.created_at.desc())
            )
+
+            datasets = db.paginate(select=stmt, page=page, per_page=50, max_per_page=50, error_out=False)
        except NotFound:
            break

@@ -551,11 +552,12 @@ def old_metadata_migration():
    page = 1
    while True:
        try:
-            documents = (
-                DatasetDocument.query.filter(DatasetDocument.doc_metadata is not None)
+            stmt = (
+                select(DatasetDocument)
+                .filter(DatasetDocument.doc_metadata.is_not(None))
                .order_by(DatasetDocument.created_at.desc())
-                .paginate(page=page, per_page=50)
            )
+            documents = db.paginate(select=stmt, page=page, per_page=50, max_per_page=50, error_out=False)
        except NotFound:
            break
        if not documents:
@@ -592,11 +594,15 @@ def old_metadata_migration():
                            )
                            db.session.add(dataset_metadata_binding)
                        else:
-                            dataset_metadata_binding = DatasetMetadataBinding.query.filter(
-                                DatasetMetadataBinding.dataset_id == document.dataset_id,
-                                DatasetMetadataBinding.document_id == document.id,
-                                DatasetMetadataBinding.metadata_id == dataset_metadata.id,
-                            ).first()
+                            dataset_metadata_binding = (
+                                db.session.query(DatasetMetadataBinding)  # type: ignore
+                                .filter(
+                                    DatasetMetadataBinding.dataset_id == document.dataset_id,
+                                    DatasetMetadataBinding.document_id == document.id,
+                                    DatasetMetadataBinding.metadata_id == dataset_metadata.id,
+                                )
+                                .first()
+                            )
                            if not dataset_metadata_binding:
                                dataset_metadata_binding = DatasetMetadataBinding(
                                    tenant_id=document.tenant_id,
@@ -840,6 +846,9 @@ def clear_orphaned_file_records(force: bool):
        {"type": "text", "table": "workflow_node_executions", "column": "outputs"},
        {"type": "text", "table": "conversations", "column": "introduction"},
        {"type": "text", "table": "conversations", "column": "system_instruction"},
+        {"type": "text", "table": "accounts", "column": "avatar"},
+        {"type": "text", "table": "apps", "column": "icon"},
+        {"type": "text", "table": "sites", "column": "icon"},
        {"type": "json", "table": "messages", "column": "inputs"},
        {"type": "json", "table": "messages", "column": "message"},
    ]
@@ -17,6 +17,12 @@ class DeploymentConfig(BaseSettings):
        default=False,
    )

+    # Request logging configuration
+    ENABLE_REQUEST_LOGGING: bool = Field(
+        description="Enable request and response body logging",
+        default=False,
+    )
+
    EDITION: str = Field(
        description="Deployment edition of the application (e.g., 'SELF_HOSTED', 'CLOUD')",
        default="SELF_HOSTED",
@@ -74,7 +74,7 @@ class CodeExecutionSandboxConfig(BaseSettings):

    CODE_EXECUTION_ENDPOINT: HttpUrl = Field(
        description="URL endpoint for the code execution service",
-        default="http://sandbox:8194",
+        default=HttpUrl("http://sandbox:8194"),
    )

    CODE_EXECUTION_API_KEY: str = Field(
@@ -145,7 +145,7 @@ class PluginConfig(BaseSettings):

    PLUGIN_DAEMON_URL: HttpUrl = Field(
        description="Plugin API URL",
-        default="http://localhost:5002",
+        default=HttpUrl("http://localhost:5002"),
    )

    PLUGIN_DAEMON_KEY: str = Field(
@@ -188,7 +188,7 @@ class MarketplaceConfig(BaseSettings):

    MARKETPLACE_API_URL: HttpUrl = Field(
        description="Marketplace API URL",
-        default="https://marketplace.dify.ai",
+        default=HttpUrl("https://marketplace.dify.ai"),
    )


@@ -1,8 +1,8 @@
 import os
 from typing import Any, Literal, Optional
-from urllib.parse import quote_plus
+from urllib.parse import parse_qsl, quote_plus

-from pydantic import Field, NonNegativeInt, PositiveFloat, PositiveInt, computed_field
+from pydantic import Field, NonNegativeFloat, NonNegativeInt, PositiveFloat, PositiveInt, computed_field
 from pydantic_settings import BaseSettings

 from .cache.redis_config import RedisConfig
@@ -173,17 +173,31 @@ class DatabaseConfig(BaseSettings):

    RETRIEVAL_SERVICE_EXECUTORS: NonNegativeInt = Field(
        description="Number of processes for the retrieval service, default to CPU cores.",
-        default=os.cpu_count(),
+        default=os.cpu_count() or 1,
    )

-    @computed_field
+    @computed_field  # type: ignore[misc]
+    @property
    def SQLALCHEMY_ENGINE_OPTIONS(self) -> dict[str, Any]:
+        # Parse DB_EXTRAS for 'options'
+        db_extras_dict = dict(parse_qsl(self.DB_EXTRAS))
+        options = db_extras_dict.get("options", "")
+        # Always include timezone
+        timezone_opt = "-c timezone=UTC"
+        if options:
+            # Merge user options and timezone
+            merged_options = f"{options} {timezone_opt}"
+        else:
+            merged_options = timezone_opt
+
+        connect_args = {"options": merged_options}
+
        return {
            "pool_size": self.SQLALCHEMY_POOL_SIZE,
            "max_overflow": self.SQLALCHEMY_MAX_OVERFLOW,
            "pool_recycle": self.SQLALCHEMY_POOL_RECYCLE,
            "pool_pre_ping": self.SQLALCHEMY_POOL_PRE_PING,
-            "connect_args": {"options": "-c timezone=UTC"},
+            "connect_args": connect_args,
        }


@@ -242,6 +256,25 @@ class InternalTestConfig(BaseSettings):
    )


+class DatasetQueueMonitorConfig(BaseSettings):
+    """
+    Configuration settings for Dataset Queue Monitor
+    """
+
+    QUEUE_MONITOR_THRESHOLD: Optional[NonNegativeInt] = Field(
+        description="Threshold for dataset queue monitor",
+        default=200,
+    )
+    QUEUE_MONITOR_ALERT_EMAILS: Optional[str] = Field(
+        description="Emails for dataset queue monitor alert, separated by commas",
+        default=None,
+    )
+    QUEUE_MONITOR_INTERVAL: Optional[NonNegativeFloat] = Field(
+        description="Interval for dataset queue monitor in minutes",
+        default=30,
+    )
+
+
 class MiddlewareConfig(
    # place the configs in alphabet order
    CeleryConfig,
@@ -289,5 +322,6 @@ class MiddlewareConfig(
    BaiduVectorDBConfig,
    OpenGaussConfig,
    TableStoreConfig,
+    DatasetQueueMonitorConfig,
 ):
    pass
@@ -83,3 +83,13 @@ class RedisConfig(BaseSettings):
        description="Password for Redis Clusters authentication (if required)",
        default=None,
    )
+
+    REDIS_SERIALIZATION_PROTOCOL: int = Field(
+        description="Redis serialization protocol (RESP) version",
+        default=3,
+    )
+
+    REDIS_ENABLE_CLIENT_SIDE_CACHE: bool = Field(
+        description="Enable client side cache in redis",
+        default=False,
+    )
@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Literal, Optional

 from pydantic import Field
 from pydantic_settings import BaseSettings
@@ -34,7 +34,7 @@ class S3StorageConfig(BaseSettings):
        default=None,
    )

-    S3_ADDRESS_STYLE: str = Field(
+    S3_ADDRESS_STYLE: Literal["auto", "virtual", "path"] = Field(
        description="S3 addressing style: 'auto', 'path', or 'virtual'",
        default="auto",
    )
@@ -33,6 +33,11 @@ class OpenSearchConfig(BaseSettings):
        default=False,
    )

+    OPENSEARCH_VERIFY_CERTS: bool = Field(
+        description="Whether to verify SSL certificates for HTTPS connections (recommended to set True in production)",
+        default=True,
+    )
+
    OPENSEARCH_AUTH_METHOD: AuthMethod = Field(
        description="Authentication method for OpenSearch connection (default is 'basic')",
        default=AuthMethod.BASIC,
@@ -33,3 +33,8 @@ class QdrantConfig(BaseSettings):
        description="Port number for gRPC connection to Qdrant server (default is 6334)",
        default=6334,
    )
+
+    QDRANT_REPLICATION_FACTOR: PositiveInt = Field(
+        description="Replication factor for Qdrant collections (default is 1)",
+        default=1,
+    )
@@ -27,6 +27,11 @@ class OTelConfig(BaseSettings):
        default="otlp",
    )

+    OTEL_EXPORTER_OTLP_PROTOCOL: str = Field(
+        description="OTLP exporter protocol ('grpc' or 'http')",
+        default="http",
+    )
+
    OTEL_SAMPLING_RATE: float = Field(default=0.1, description="Sampling rate for traces (0.0 to 1.0)")

    OTEL_BATCH_EXPORT_SCHEDULE_DELAY: int = Field(
@@ -9,7 +9,7 @@ class PackagingInfo(BaseSettings):

    CURRENT_VERSION: str = Field(
        description="Dify version",
-        default="1.3.1",
+        default="1.4.2",
    )

    COMMIT_SHA: str = Field(
@@ -60,8 +60,7 @@ class NacosHttpClient:
            sign_str = tenant + "+"
        if group:
            sign_str = sign_str + group + "+"
-        if sign_str:
-            sign_str += ts
+        sign_str += ts  # Directly concatenate ts without conditional checks, because the nacos auth header forced it.
        return sign_str

    def get_access_token(self, force_refresh=False):
@@ -11,10 +11,6 @@ if TYPE_CHECKING:
    from core.workflow.entities.variable_pool import VariablePool


-tenant_id: ContextVar[str] = ContextVar("tenant_id")
-
-workflow_variable_pool: ContextVar["VariablePool"] = ContextVar("workflow_variable_pool")
-
 """
 To avoid race-conditions caused by gunicorn thread recycling, using RecyclableContextVar to replace with
 """
@@ -1,5 +1,7 @@
 from flask_restful import fields

+from libs.helper import AppIconUrlField
+
 parameters__system_parameters = {
    "image_file_size_limit": fields.Integer,
    "video_file_size_limit": fields.Integer,
@@ -22,3 +24,20 @@ parameters_fields = {
    "file_upload": fields.Raw,
    "system_parameters": fields.Nested(parameters__system_parameters),
 }
+
+site_fields = {
+    "title": fields.String,
+    "chat_color_theme": fields.String,
+    "chat_color_theme_inverted": fields.Boolean,
+    "icon_type": fields.String,
+    "icon": fields.String,
+    "icon_background": fields.String,
+    "icon_url": AppIconUrlField,
+    "description": fields.String,
+    "copyright": fields.String,
+    "privacy_policy": fields.String,
+    "custom_disclaimer": fields.String,
+    "default_language": fields.String,
+    "show_workflow_steps": fields.Boolean,
+    "use_icon_as_answer_icon": fields.Boolean,
+}
@@ -208,7 +208,7 @@ class AnnotationBatchImportApi(Resource):
        if len(request.files) > 1:
            raise TooManyFilesError()
        # check file type
-        if not file.filename.endswith(".csv"):
+        if not file.filename or not file.filename.endswith(".csv"):
            raise ValueError("Invalid file type. Only CSV files are allowed")
        return AppAnnotationService.batch_import_app_annotations(app_id, file)

@@ -17,15 +17,13 @@ from controllers.console.wraps import (
 )
 from core.ops.ops_trace_manager import OpsTraceManager
 from extensions.ext_database import db
-from fields.app_fields import (
-    app_detail_fields,
-    app_detail_fields_with_site,
-    app_pagination_fields,
-)
+from fields.app_fields import app_detail_fields, app_detail_fields_with_site, app_pagination_fields
 from libs.login import login_required
 from models import Account, App
 from services.app_dsl_service import AppDslService, ImportMode
 from services.app_service import AppService
+from services.enterprise.enterprise_service import EnterpriseService
+from services.feature_service import FeatureService

 ALLOW_CREATE_APP_MODES = ["chat", "agent-chat", "advanced-chat", "workflow", "completion"]

@@ -75,7 +73,17 @@ class AppListApi(Resource):
        if not app_pagination:
            return {"data": [], "total": 0, "page": 1, "limit": 20, "has_more": False}

-        return marshal(app_pagination, app_pagination_fields)
+        if FeatureService.get_system_features().webapp_auth.enabled:
+            app_ids = [str(app.id) for app in app_pagination.items]
+            res = EnterpriseService.WebAppAuth.batch_get_app_access_mode_by_id(app_ids=app_ids)
+            if len(res) != len(app_ids):
+                raise BadRequest("Invalid app id in webapp auth")
+
+            for app in app_pagination.items:
+                if str(app.id) in res:
+                    app.access_mode = res[str(app.id)].access_mode
+
+        return marshal(app_pagination, app_pagination_fields), 200

    @setup_required
    @login_required
@@ -119,6 +127,10 @@ class AppApi(Resource):

        app_model = app_service.get_app(app_model)

+        if FeatureService.get_system_features().webapp_auth.enabled:
+            app_setting = EnterpriseService.WebAppAuth.get_app_access_mode_by_id(app_id=str(app_model.id))
+            app_model.access_mode = app_setting.access_mode
+
        return app_model

    @setup_required
@@ -81,8 +81,7 @@ class DraftWorkflowApi(Resource):
            parser.add_argument("graph", type=dict, required=True, nullable=False, location="json")
            parser.add_argument("features", type=dict, required=True, nullable=False, location="json")
            parser.add_argument("hash", type=str, required=False, location="json")
-            # TODO: set this to required=True after frontend is updated
-            parser.add_argument("environment_variables", type=list, required=False, location="json")
+            parser.add_argument("environment_variables", type=list, required=True, location="json")
            parser.add_argument("conversation_variables", type=list, required=False, location="json")
            args = parser.parse_args()
        elif "text/plain" in content_type:
@@ -6,12 +6,12 @@ from sqlalchemy.orm import Session
 from controllers.console import api
 from controllers.console.app.wraps import get_app_model
 from controllers.console.wraps import account_initialization_required, setup_required
+from core.workflow.entities.workflow_execution import WorkflowExecutionStatus
 from extensions.ext_database import db
 from fields.workflow_app_log_fields import workflow_app_log_pagination_fields
 from libs.login import login_required
 from models import App
 from models.model import AppMode
-from models.workflow import WorkflowRunStatus
 from services.workflow_app_service import WorkflowAppService


@@ -38,7 +38,7 @@ class WorkflowAppLogApi(Resource):
        parser.add_argument("limit", type=int_range(1, 100), default=20, location="args")
        args = parser.parse_args()

-        args.status = WorkflowRunStatus(args.status) if args.status else None
+        args.status = WorkflowExecutionStatus(args.status) if args.status else None
        if args.created_at__before:
            args.created_at__before = isoparse(args.created_at__before)

@@ -1,3 +1,6 @@
+from typing import cast
+
+from flask_login import current_user
 from flask_restful import Resource, marshal_with, reqparse
 from flask_restful.inputs import int_range

@@ -12,8 +15,7 @@ from fields.workflow_run_fields import (
 )
 from libs.helper import uuid_value
 from libs.login import login_required
-from models import App
-from models.model import AppMode
+from models import Account, App, AppMode, EndUser
 from services.workflow_run_service import WorkflowRunService


@@ -90,7 +92,12 @@ class WorkflowRunNodeExecutionListApi(Resource):
        run_id = str(run_id)

        workflow_run_service = WorkflowRunService()
-        node_executions = workflow_run_service.get_workflow_run_node_executions(app_model=app_model, run_id=run_id)
+        user = cast("Account | EndUser", current_user)
+        node_executions = workflow_run_service.get_workflow_run_node_executions(
+            app_model=app_model,
+            run_id=run_id,
+            user=user,
+        )

        return {"data": node_executions}

@@ -24,7 +24,7 @@ from libs.password import hash_password, valid_password
 from models.account import Account
 from services.account_service import AccountService, TenantService
 from services.errors.account import AccountRegisterError
-from services.errors.workspace import WorkSpaceNotAllowedCreateError
+from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
 from services.feature_service import FeatureService


@@ -168,6 +168,8 @@ class ForgotPasswordResetApi(Resource):
            )
        except WorkSpaceNotAllowedCreateError:
            pass
+        except WorkspacesLimitExceededError:
+            pass
        except AccountRegisterError:
            raise AccountInFreezeError()

@@ -21,6 +21,7 @@ from controllers.console.error import (
    AccountNotFound,
    EmailSendIpLimitError,
    NotAllowedCreateWorkspace,
+    WorkspacesLimitExceeded,
 )
 from controllers.console.wraps import email_password_login_enabled, setup_required
 from events.tenant_event import tenant_was_created
@@ -30,7 +31,7 @@ from models.account import Account
 from services.account_service import AccountService, RegisterService, TenantService
 from services.billing_service import BillingService
 from services.errors.account import AccountRegisterError
-from services.errors.workspace import WorkSpaceNotAllowedCreateError
+from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
 from services.feature_service import FeatureService


@@ -88,10 +89,15 @@ class LoginApi(Resource):
        # SELF_HOSTED only have one workspace
        tenants = TenantService.get_join_tenants(account)
        if len(tenants) == 0:
-            return {
-                "result": "fail",
-                "data": "workspace not found, please contact system admin to invite you to join in a workspace",
-            }
+            system_features = FeatureService.get_system_features()
+
+            if system_features.is_allow_create_workspace and not system_features.license.workspaces.is_available():
+                raise WorkspacesLimitExceeded()
+            else:
+                return {
+                    "result": "fail",
+                    "data": "workspace not found, please contact system admin to invite you to join in a workspace",
+                }

        token_pair = AccountService.login(account=account, ip_address=extract_remote_ip(request))
        AccountService.reset_login_error_rate_limit(args["email"])
@@ -196,15 +202,18 @@ class EmailCodeLoginApi(Resource):
        except AccountRegisterError as are:
            raise AccountInFreezeError()
        if account:
-            tenant = TenantService.get_join_tenants(account)
-            if not tenant:
+            tenants = TenantService.get_join_tenants(account)
+            if not tenants:
+                workspaces = FeatureService.get_system_features().license.workspaces
+                if not workspaces.is_available():
+                    raise WorkspacesLimitExceeded()
                if not FeatureService.get_system_features().is_allow_create_workspace:
                    raise NotAllowedCreateWorkspace()
                else:
-                    tenant = TenantService.create_tenant(f"{account.name}'s Workspace")
-                    TenantService.create_tenant_member(tenant, account, role="owner")
-                    account.current_tenant = tenant
-                    tenant_was_created.send(tenant)
+                    new_tenant = TenantService.create_tenant(f"{account.name}'s Workspace")
+                    TenantService.create_tenant_member(new_tenant, account, role="owner")
+                    account.current_tenant = new_tenant
+                    tenant_was_created.send(new_tenant)

        if account is None:
            try:
@@ -215,6 +224,8 @@ class EmailCodeLoginApi(Resource):
                return NotAllowedCreateWorkspace()
            except AccountRegisterError as are:
                raise AccountInFreezeError()
+            except WorkspacesLimitExceededError:
+                raise WorkspacesLimitExceeded()
        token_pair = AccountService.login(account, ip_address=extract_remote_ip(request))
        AccountService.reset_login_error_rate_limit(args["email"])
        return {"result": "success", "data": token_pair.model_dump()}
@@ -148,15 +148,15 @@ def _generate_account(provider: str, user_info: OAuthUserInfo):
    account = _get_account_by_openid_or_email(provider, user_info)

    if account:
-        tenant = TenantService.get_join_tenants(account)
-        if not tenant:
+        tenants = TenantService.get_join_tenants(account)
+        if not tenants:
            if not FeatureService.get_system_features().is_allow_create_workspace:
                raise WorkSpaceNotAllowedCreateError()
            else:
-                tenant = TenantService.create_tenant(f"{account.name}'s Workspace")
-                TenantService.create_tenant_member(tenant, account, role="owner")
-                account.current_tenant = tenant
-                tenant_was_created.send(tenant)
+                new_tenant = TenantService.create_tenant(f"{account.name}'s Workspace")
+                TenantService.create_tenant_member(new_tenant, account, role="owner")
+                account.current_tenant = new_tenant
+                tenant_was_created.send(new_tenant)

    if not account:
        if not FeatureService.get_system_features().is_allow_register:
@@ -526,17 +526,36 @@ class DatasetIndexingStatusApi(Resource):
        )
        documents_status = []
        for document in documents:
-            completed_segments = DocumentSegment.query.filter(
-                DocumentSegment.completed_at.isnot(None),
-                DocumentSegment.document_id == str(document.id),
-                DocumentSegment.status != "re_segment",
-            ).count()
-            total_segments = DocumentSegment.query.filter(
-                DocumentSegment.document_id == str(document.id), DocumentSegment.status != "re_segment"
-            ).count()
-            document.completed_segments = completed_segments
-            document.total_segments = total_segments
-            documents_status.append(marshal(document, document_status_fields))
+            completed_segments = (
+                db.session.query(DocumentSegment)
+                .filter(
+                    DocumentSegment.completed_at.isnot(None),
+                    DocumentSegment.document_id == str(document.id),
+                    DocumentSegment.status != "re_segment",
+                )
+                .count()
+            )
+            total_segments = (
+                db.session.query(DocumentSegment)
+                .filter(DocumentSegment.document_id == str(document.id), DocumentSegment.status != "re_segment")
+                .count()
+            )
+            # Create a dictionary with document attributes and additional fields
+            document_dict = {
+                "id": document.id,
+                "indexing_status": document.indexing_status,
+                "processing_started_at": document.processing_started_at,
+                "parsing_completed_at": document.parsing_completed_at,
+                "cleaning_completed_at": document.cleaning_completed_at,
+                "splitting_completed_at": document.splitting_completed_at,
+                "completed_at": document.completed_at,
+                "paused_at": document.paused_at,
+                "error": document.error,
+                "stopped_at": document.stopped_at,
+                "completed_segments": completed_segments,
+                "total_segments": total_segments,
+            }
+            documents_status.append(marshal(document_dict, document_status_fields))
        data = {"data": documents_status}
        return data

@@ -6,7 +6,7 @@ from typing import cast
 from flask import request
 from flask_login import current_user
 from flask_restful import Resource, fields, marshal, marshal_with, reqparse
-from sqlalchemy import asc, desc
+from sqlalchemy import asc, desc, select
 from werkzeug.exceptions import Forbidden, NotFound

 import services
@@ -112,7 +112,7 @@ class GetProcessRuleApi(Resource):
        limits = DocumentService.DEFAULT_RULES["limits"]
        if document_id:
            # get the latest process rule
-            document = Document.query.get_or_404(document_id)
+            document = db.get_or_404(Document, document_id)

            dataset = DatasetService.get_dataset(document.dataset_id)

@@ -175,7 +175,7 @@ class DatasetDocumentListApi(Resource):
        except services.errors.account.NoPermissionError as e:
            raise Forbidden(str(e))

-        query = Document.query.filter_by(dataset_id=str(dataset_id), tenant_id=current_user.current_tenant_id)
+        query = select(Document).filter_by(dataset_id=str(dataset_id), tenant_id=current_user.current_tenant_id)

        if search:
            search = f"%{search}%"
@@ -209,18 +209,24 @@ class DatasetDocumentListApi(Resource):
                desc(Document.position),
            )

-        paginated_documents = query.paginate(page=page, per_page=limit, max_per_page=100, error_out=False)
+        paginated_documents = db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)
        documents = paginated_documents.items
        if fetch:
            for document in documents:
-                completed_segments = DocumentSegment.query.filter(
-                    DocumentSegment.completed_at.isnot(None),
-                    DocumentSegment.document_id == str(document.id),
-                    DocumentSegment.status != "re_segment",
-                ).count()
-                total_segments = DocumentSegment.query.filter(
-                    DocumentSegment.document_id == str(document.id), DocumentSegment.status != "re_segment"
-                ).count()
+                completed_segments = (
+                    db.session.query(DocumentSegment)
+                    .filter(
+                        DocumentSegment.completed_at.isnot(None),
+                        DocumentSegment.document_id == str(document.id),
+                        DocumentSegment.status != "re_segment",
+                    )
+                    .count()
+                )
+                total_segments = (
+                    db.session.query(DocumentSegment)
+                    .filter(DocumentSegment.document_id == str(document.id), DocumentSegment.status != "re_segment")
+                    .count()
+                )
                document.completed_segments = completed_segments
                document.total_segments = total_segments
            data = marshal(documents, document_with_segments_fields)
@@ -563,19 +569,36 @@ class DocumentBatchIndexingStatusApi(DocumentResource):
        documents = self.get_batch_documents(dataset_id, batch)
        documents_status = []
        for document in documents:
-            completed_segments = DocumentSegment.query.filter(
-                DocumentSegment.completed_at.isnot(None),
-                DocumentSegment.document_id == str(document.id),
-                DocumentSegment.status != "re_segment",
-            ).count()
-            total_segments = DocumentSegment.query.filter(
-                DocumentSegment.document_id == str(document.id), DocumentSegment.status != "re_segment"
-            ).count()
-            document.completed_segments = completed_segments
-            document.total_segments = total_segments
-            if document.is_paused:
-                document.indexing_status = "paused"
-            documents_status.append(marshal(document, document_status_fields))
+            completed_segments = (
+                db.session.query(DocumentSegment)
+                .filter(
+                    DocumentSegment.completed_at.isnot(None),
+                    DocumentSegment.document_id == str(document.id),
+                    DocumentSegment.status != "re_segment",
+                )
+                .count()
+            )
+            total_segments = (
+                db.session.query(DocumentSegment)
+                .filter(DocumentSegment.document_id == str(document.id), DocumentSegment.status != "re_segment")
+                .count()
+            )
+            # Create a dictionary with document attributes and additional fields
+            document_dict = {
+                "id": document.id,
+                "indexing_status": "paused" if document.is_paused else document.indexing_status,
+                "processing_started_at": document.processing_started_at,
+                "parsing_completed_at": document.parsing_completed_at,
+                "cleaning_completed_at": document.cleaning_completed_at,
+                "splitting_completed_at": document.splitting_completed_at,
+                "completed_at": document.completed_at,
+                "paused_at": document.paused_at,
+                "error": document.error,
+                "stopped_at": document.stopped_at,
+                "completed_segments": completed_segments,
+                "total_segments": total_segments,
+            }
+            documents_status.append(marshal(document_dict, document_status_fields))
        data = {"data": documents_status}
        return data

@@ -589,20 +612,37 @@ class DocumentIndexingStatusApi(DocumentResource):
        document_id = str(document_id)
        document = self.get_document(dataset_id, document_id)

-        completed_segments = DocumentSegment.query.filter(
-            DocumentSegment.completed_at.isnot(None),
-            DocumentSegment.document_id == str(document_id),
-            DocumentSegment.status != "re_segment",
-        ).count()
-        total_segments = DocumentSegment.query.filter(
-            DocumentSegment.document_id == str(document_id), DocumentSegment.status != "re_segment"
-        ).count()
+        completed_segments = (
+            db.session.query(DocumentSegment)
+            .filter(
+                DocumentSegment.completed_at.isnot(None),
+                DocumentSegment.document_id == str(document_id),
+                DocumentSegment.status != "re_segment",
+            )
+            .count()
+        )
+        total_segments = (
+            db.session.query(DocumentSegment)
+            .filter(DocumentSegment.document_id == str(document_id), DocumentSegment.status != "re_segment")
+            .count()
+        )

-        document.completed_segments = completed_segments
-        document.total_segments = total_segments
-        if document.is_paused:
-            document.indexing_status = "paused"
-        return marshal(document, document_status_fields)
+        # Create a dictionary with document attributes and additional fields
+        document_dict = {
+            "id": document.id,
+            "indexing_status": "paused" if document.is_paused else document.indexing_status,
+            "processing_started_at": document.processing_started_at,
+            "parsing_completed_at": document.parsing_completed_at,
+            "cleaning_completed_at": document.cleaning_completed_at,
+            "splitting_completed_at": document.splitting_completed_at,
+            "completed_at": document.completed_at,
+            "paused_at": document.paused_at,
+            "error": document.error,
+            "stopped_at": document.stopped_at,
+            "completed_segments": completed_segments,
+            "total_segments": total_segments,
+        }
+        return marshal(document_dict, document_status_fields)


 class DocumentDetailApi(DocumentResource):
@@ -4,6 +4,7 @@ import pandas as pd
 from flask import request
 from flask_login import current_user
 from flask_restful import Resource, marshal, reqparse
+from sqlalchemy import select
 from werkzeug.exceptions import Forbidden, NotFound

 import services
@@ -26,6 +27,7 @@ from controllers.console.wraps import (
 from core.errors.error import LLMBadRequestError, ProviderTokenNotInitError
 from core.model_manager import ModelManager
 from core.model_runtime.entities.model_entities import ModelType
+from extensions.ext_database import db
 from extensions.ext_redis import redis_client
 from fields.segment_fields import child_chunk_fields, segment_fields
 from libs.login import login_required
@@ -74,9 +76,14 @@ class DatasetDocumentSegmentListApi(Resource):
        hit_count_gte = args["hit_count_gte"]
        keyword = args["keyword"]

-        query = DocumentSegment.query.filter(
-            DocumentSegment.document_id == str(document_id), DocumentSegment.tenant_id == current_user.current_tenant_id
-        ).order_by(DocumentSegment.position.asc())
+        query = (
+            select(DocumentSegment)
+            .filter(
+                DocumentSegment.document_id == str(document_id),
+                DocumentSegment.tenant_id == current_user.current_tenant_id,
+            )
+            .order_by(DocumentSegment.position.asc())
+        )

        if status_list:
            query = query.filter(DocumentSegment.status.in_(status_list))
@@ -93,7 +100,7 @@ class DatasetDocumentSegmentListApi(Resource):
            elif args["enabled"].lower() == "false":
                query = query.filter(DocumentSegment.enabled == False)

-        segments = query.paginate(page=page, per_page=limit, max_per_page=100, error_out=False)
+        segments = db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)

        response = {
            "data": marshal(segments.items, segment_fields),
@@ -276,9 +283,11 @@ class DatasetDocumentSegmentUpdateApi(Resource):
                raise ProviderNotInitializeError(ex.description)
            # check segment
        segment_id = str(segment_id)
-        segment = DocumentSegment.query.filter(
-            DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_user.current_tenant_id
-        ).first()
+        segment = (
+            db.session.query(DocumentSegment)
+            .filter(DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_user.current_tenant_id)
+            .first()
+        )
        if not segment:
            raise NotFound("Segment not found.")
        # The role of the current user in the ta table must be admin, owner, dataset_operator, or editor
@@ -320,9 +329,11 @@ class DatasetDocumentSegmentUpdateApi(Resource):
            raise NotFound("Document not found.")
        # check segment
        segment_id = str(segment_id)
-        segment = DocumentSegment.query.filter(
-            DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_user.current_tenant_id
-        ).first()
+        segment = (
+            db.session.query(DocumentSegment)
+            .filter(DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_user.current_tenant_id)
+            .first()
+        )
        if not segment:
            raise NotFound("Segment not found.")
        # The role of the current user in the ta table must be admin, owner, dataset_operator, or editor
@@ -363,7 +374,7 @@ class DatasetDocumentSegmentBatchImportApi(Resource):
        if len(request.files) > 1:
            raise TooManyFilesError()
        # check file type
-        if not file.filename.endswith(".csv"):
+        if not file.filename or not file.filename.endswith(".csv"):
            raise ValueError("Invalid file type. Only CSV files are allowed")

        try:
@@ -423,9 +434,11 @@ class ChildChunkAddApi(Resource):
            raise NotFound("Document not found.")
        # check segment
        segment_id = str(segment_id)
-        segment = DocumentSegment.query.filter(
-            DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_user.current_tenant_id
-        ).first()
+        segment = (
+            db.session.query(DocumentSegment)
+            .filter(DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_user.current_tenant_id)
+            .first()
+        )
        if not segment:
            raise NotFound("Segment not found.")
        if not current_user.is_dataset_editor:
@@ -478,9 +491,11 @@ class ChildChunkAddApi(Resource):
            raise NotFound("Document not found.")
        # check segment
        segment_id = str(segment_id)
-        segment = DocumentSegment.query.filter(
-            DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_user.current_tenant_id
-        ).first()
+        segment = (
+            db.session.query(DocumentSegment)
+            .filter(DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_user.current_tenant_id)
+            .first()
+        )
        if not segment:
            raise NotFound("Segment not found.")
        parser = reqparse.RequestParser()
@@ -523,9 +538,11 @@ class ChildChunkAddApi(Resource):
            raise NotFound("Document not found.")
            # check segment
        segment_id = str(segment_id)
-        segment = DocumentSegment.query.filter(
-            DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_user.current_tenant_id
-        ).first()
+        segment = (
+            db.session.query(DocumentSegment)
+            .filter(DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_user.current_tenant_id)
+            .first()
+        )
        if not segment:
            raise NotFound("Segment not found.")
        # The role of the current user in the ta table must be admin, owner, dataset_operator, or editor
@@ -567,16 +584,20 @@ class ChildChunkUpdateApi(Resource):
            raise NotFound("Document not found.")
        # check segment
        segment_id = str(segment_id)
-        segment = DocumentSegment.query.filter(
-            DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_user.current_tenant_id
-        ).first()
+        segment = (
+            db.session.query(DocumentSegment)
+            .filter(DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_user.current_tenant_id)
+            .first()
+        )
        if not segment:
            raise NotFound("Segment not found.")
        # check child chunk
        child_chunk_id = str(child_chunk_id)
-        child_chunk = ChildChunk.query.filter(
-            ChildChunk.id == str(child_chunk_id), ChildChunk.tenant_id == current_user.current_tenant_id
-        ).first()
+        child_chunk = (
+            db.session.query(ChildChunk)
+            .filter(ChildChunk.id == str(child_chunk_id), ChildChunk.tenant_id == current_user.current_tenant_id)
+            .first()
+        )
        if not child_chunk:
            raise NotFound("Child chunk not found.")
        # The role of the current user in the ta table must be admin, owner, dataset_operator, or editor
@@ -612,16 +633,20 @@ class ChildChunkUpdateApi(Resource):
            raise NotFound("Document not found.")
            # check segment
        segment_id = str(segment_id)
-        segment = DocumentSegment.query.filter(
-            DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_user.current_tenant_id
-        ).first()
+        segment = (
+            db.session.query(DocumentSegment)
+            .filter(DocumentSegment.id == str(segment_id), DocumentSegment.tenant_id == current_user.current_tenant_id)
+            .first()
+        )
        if not segment:
            raise NotFound("Segment not found.")
        # check child chunk
        child_chunk_id = str(child_chunk_id)
-        child_chunk = ChildChunk.query.filter(
-            ChildChunk.id == str(child_chunk_id), ChildChunk.tenant_id == current_user.current_tenant_id
-        ).first()
+        child_chunk = (
+            db.session.query(ChildChunk)
+            .filter(ChildChunk.id == str(child_chunk_id), ChildChunk.tenant_id == current_user.current_tenant_id)
+            .first()
+        )
        if not child_chunk:
            raise NotFound("Child chunk not found.")
        # The role of the current user in the ta table must be admin, owner, dataset_operator, or editor
@@ -209,6 +209,7 @@ class ExternalKnowledgeHitTestingApi(Resource):
        parser = reqparse.RequestParser()
        parser.add_argument("query", type=str, location="json")
        parser.add_argument("external_retrieval_model", type=dict, required=False, location="json")
+        parser.add_argument("metadata_filtering_conditions", type=dict, required=False, location="json")
        args = parser.parse_args()

        HitTestingService.hit_testing_args_check(args)
@@ -219,6 +220,7 @@ class ExternalKnowledgeHitTestingApi(Resource):
                query=args["query"],
                account=current_user,
                external_retrieval_model=args["external_retrieval_model"],
+                metadata_filtering_conditions=args["metadata_filtering_conditions"],
            )

            return response
@@ -46,6 +46,18 @@ class NotAllowedCreateWorkspace(BaseHTTPException):
    code = 400


+class WorkspaceMembersLimitExceeded(BaseHTTPException):
+    error_code = "limit_exceeded"
+    description = "Unable to add member because the maximum workspace's member limit was exceeded"
+    code = 400
+
+
+class WorkspacesLimitExceeded(BaseHTTPException):
+    error_code = "limit_exceeded"
+    description = "Unable to create workspace because the maximum workspace limit was exceeded"
+    code = 400
+
+
 class AccountBannedError(BaseHTTPException):
    error_code = "account_banned"
    description = "Account is banned."
@@ -23,3 +23,9 @@ class AppSuggestedQuestionsAfterAnswerDisabledError(BaseHTTPException):
    error_code = "app_suggested_questions_after_answer_disabled"
    description = "Function Suggested questions after answer disabled."
    code = 403
+
+
+class AppAccessDeniedError(BaseHTTPException):
+    error_code = "access_denied"
+    description = "App access denied."
+    code = 403
@@ -1,3 +1,4 @@
+import logging
 from datetime import UTC, datetime
 from typing import Any

@@ -15,6 +16,11 @@ from fields.installed_app_fields import installed_app_list_fields
 from libs.login import login_required
 from models import App, InstalledApp, RecommendedApp
 from services.account_service import TenantService
+from services.app_service import AppService
+from services.enterprise.enterprise_service import EnterpriseService
+from services.feature_service import FeatureService
+
+logger = logging.getLogger(__name__)


 class InstalledAppsListApi(Resource):
@@ -48,6 +54,28 @@ class InstalledAppsListApi(Resource):
            for installed_app in installed_apps
            if installed_app.app is not None
        ]
+
+        # filter out apps that user doesn't have access to
+        if FeatureService.get_system_features().webapp_auth.enabled:
+            user_id = current_user.id
+            res = []
+            app_ids = [installed_app["app"].id for installed_app in installed_app_list]
+            webapp_settings = EnterpriseService.WebAppAuth.batch_get_app_access_mode_by_id(app_ids)
+            for installed_app in installed_app_list:
+                webapp_setting = webapp_settings.get(installed_app["app"].id)
+                if not webapp_setting:
+                    continue
+                if webapp_setting.access_mode == "sso_verified":
+                    continue
+                app_code = AppService.get_app_code_by_id(str(installed_app["app"].id))
+                if EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(
+                    user_id=user_id,
+                    app_code=app_code,
+                ):
+                    res.append(installed_app)
+            installed_app_list = res
+            logger.debug(f"installed_app_list: {installed_app_list}, user_id: {user_id}")
+
        installed_app_list.sort(
            key=lambda app: (
                -app["is_pinned"],
@@ -66,7 +94,7 @@ class InstalledAppsListApi(Resource):
        parser.add_argument("app_id", type=str, required=True, help="Invalid app_id")
        args = parser.parse_args()

-        recommended_app = RecommendedApp.query.filter(RecommendedApp.app_id == args["app_id"]).first()
+        recommended_app = db.session.query(RecommendedApp).filter(RecommendedApp.app_id == args["app_id"]).first()
        if recommended_app is None:
            raise NotFound("App not found")

@@ -79,9 +107,11 @@ class InstalledAppsListApi(Resource):
        if not app.is_public:
            raise Forbidden("You can't install a non-public app")

-        installed_app = InstalledApp.query.filter(
-            and_(InstalledApp.app_id == args["app_id"], InstalledApp.tenant_id == current_tenant_id)
-        ).first()
+        installed_app = (
+            db.session.query(InstalledApp)
+            .filter(and_(InstalledApp.app_id == args["app_id"], InstalledApp.tenant_id == current_tenant_id))
+            .first()
+        )

        if installed_app is None:
            # todo: position
@@ -4,10 +4,14 @@ from flask_login import current_user
 from flask_restful import Resource
 from werkzeug.exceptions import NotFound

+from controllers.console.explore.error import AppAccessDeniedError
 from controllers.console.wraps import account_initialization_required
 from extensions.ext_database import db
 from libs.login import login_required
 from models import InstalledApp
+from services.app_service import AppService
+from services.enterprise.enterprise_service import EnterpriseService
+from services.feature_service import FeatureService


 def installed_app_required(view=None):
@@ -48,6 +52,36 @@ def installed_app_required(view=None):
    return decorator


+def user_allowed_to_access_app(view=None):
+    def decorator(view):
+        @wraps(view)
+        def decorated(installed_app: InstalledApp, *args, **kwargs):
+            feature = FeatureService.get_system_features()
+            if feature.webapp_auth.enabled:
+                app_id = installed_app.app_id
+                app_code = AppService.get_app_code_by_id(app_id)
+                res = EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(
+                    user_id=str(current_user.id),
+                    app_code=app_code,
+                )
+                if not res:
+                    raise AppAccessDeniedError()
+
+            return view(installed_app, *args, **kwargs)
+
+        return decorated
+
+    if view:
+        return decorator(view)
+    return decorator
+
+
 class InstalledAppResource(Resource):
    # must be reversed if there are multiple decorators
-    method_decorators = [installed_app_required, account_initialization_required, login_required]
+
+    method_decorators = [
+        user_allowed_to_access_app,
+        installed_app_required,
+        account_initialization_required,
+        login_required,
+    ]
@@ -6,6 +6,7 @@ from flask_restful import Resource, abort, marshal_with, reqparse
 import services
 from configs import dify_config
 from controllers.console import api
+from controllers.console.error import WorkspaceMembersLimitExceeded
 from controllers.console.wraps import (
    account_initialization_required,
    cloud_edition_billing_resource_check,
@@ -17,6 +18,7 @@ from libs.login import login_required
 from models.account import Account, TenantAccountRole
 from services.account_service import RegisterService, TenantService
 from services.errors.account import AccountAlreadyInTenantError
+from services.feature_service import FeatureService


 class MemberListApi(Resource):
@@ -54,6 +56,12 @@ class MemberInviteEmailApi(Resource):
        inviter = current_user
        invitation_results = []
        console_web_url = dify_config.CONSOLE_WEB_URL
+
+        workspace_members = FeatureService.get_features(tenant_id=inviter.current_tenant.id).workspace_members
+
+        if not workspace_members.is_available(len(invitee_emails)):
+            raise WorkspaceMembersLimitExceeded()
+
        for invitee_email in invitee_emails:
            try:
                token = RegisterService.invite_new_member(
@@ -71,7 +79,6 @@ class MemberInviteEmailApi(Resource):
                invitation_results.append(
                    {"status": "success", "email": invitee_email, "url": f"{console_web_url}/signin"}
                )
-                break
            except Exception as e:
                invitation_results.append({"status": "failed", "email": invitee_email, "message": str(e)})

@@ -41,12 +41,16 @@ class PluginListApi(Resource):
    @account_initialization_required
    def get(self):
        tenant_id = current_user.current_tenant_id
+        parser = reqparse.RequestParser()
+        parser.add_argument("page", type=int, required=False, location="args", default=1)
+        parser.add_argument("page_size", type=int, required=False, location="args", default=256)
+        args = parser.parse_args()
        try:
-            plugins = PluginService.list(tenant_id)
+            plugins_with_total = PluginService.list_with_total(tenant_id, args["page"], args["page_size"])
        except PluginDaemonClientSideError as e:
            raise ValueError(e)

-        return jsonable_encoder({"plugins": plugins})
+        return jsonable_encoder({"plugins": plugins_with_total.list, "total": plugins_with_total.total})


 class PluginListLatestVersionsApi(Resource):
@@ -3,6 +3,7 @@ import logging
 from flask import request
 from flask_login import current_user
 from flask_restful import Resource, fields, inputs, marshal, marshal_with, reqparse
+from sqlalchemy import select
 from werkzeug.exceptions import Unauthorized

 import services
@@ -67,16 +68,24 @@ class TenantListApi(Resource):
    @account_initialization_required
    def get(self):
        tenants = TenantService.get_join_tenants(current_user)
+        tenant_dicts = []

        for tenant in tenants:
            features = FeatureService.get_features(tenant.id)
-            if features.billing.enabled:
-                tenant.plan = features.billing.subscription.plan
-            else:
-                tenant.plan = "sandbox"
-            if tenant.id == current_user.current_tenant_id:
-                tenant.current = True  # Set current=True for current tenant
-        return {"workspaces": marshal(tenants, tenants_fields)}, 200
+
+            # Create a dictionary with tenant attributes
+            tenant_dict = {
+                "id": tenant.id,
+                "name": tenant.name,
+                "status": tenant.status,
+                "created_at": tenant.created_at,
+                "plan": features.billing.subscription.plan if features.billing.enabled else "sandbox",
+                "current": tenant.id == current_user.current_tenant_id,
+            }
+
+            tenant_dicts.append(tenant_dict)
+
+        return {"workspaces": marshal(tenant_dicts, tenants_fields)}, 200


 class WorkspaceListApi(Resource):
@@ -88,9 +97,8 @@ class WorkspaceListApi(Resource):
        parser.add_argument("limit", type=inputs.int_range(1, 100), required=False, default=20, location="args")
        args = parser.parse_args()

-        tenants = Tenant.query.order_by(Tenant.created_at.desc()).paginate(
-            page=args["page"], per_page=args["limit"], error_out=False
-        )
+        stmt = select(Tenant).order_by(Tenant.created_at.desc())
+        tenants = db.paginate(select=stmt, page=args["page"], per_page=args["limit"], error_out=False)
        has_more = False

        if tenants.has_next:
@@ -162,7 +170,7 @@ class CustomConfigWorkspaceApi(Resource):
        parser.add_argument("replace_webapp_logo", type=str, location="json")
        args = parser.parse_args()

-        tenant = Tenant.query.filter(Tenant.id == current_user.current_tenant_id).one_or_404()
+        tenant = db.get_or_404(Tenant, current_user.current_tenant_id)

        custom_config_dict = {
            "remove_webapp_brand": args["remove_webapp_brand"],
@@ -226,7 +234,7 @@ class WorkspaceInfoApi(Resource):
        parser.add_argument("name", type=str, required=True, location="json")
        args = parser.parse_args()

-        tenant = Tenant.query.filter(Tenant.id == current_user.current_tenant_id).one_or_404()
+        tenant = db.get_or_404(Tenant, current_user.current_tenant_id)
        tenant.name = args["name"]
        db.session.commit()

@@ -44,6 +44,17 @@ def only_edition_cloud(view):
    return decorated


+def only_edition_enterprise(view):
+    @wraps(view)
+    def decorated(*args, **kwargs):
+        if not dify_config.ENTERPRISE_ENABLED:
+            abort(404)
+
+        return view(*args, **kwargs)
+
+    return decorated
+
+
 def only_edition_self_hosted(view):
    @wraps(view)
    def decorated(*args, **kwargs):
@@ -64,9 +64,24 @@ class PluginUploadFileApi(Resource):

            extension = guess_extension(tool_file.mimetype) or ".bin"
            preview_url = ToolFileManager.sign_file(tool_file_id=tool_file.id, extension=extension)
-            tool_file.mime_type = mimetype
-            tool_file.extension = extension
-            tool_file.preview_url = preview_url
+
+            # Create a dictionary with all the necessary attributes
+            result = {
+                "id": tool_file.id,
+                "user_id": tool_file.user_id,
+                "tenant_id": tool_file.tenant_id,
+                "conversation_id": tool_file.conversation_id,
+                "file_key": tool_file.file_key,
+                "mimetype": tool_file.mimetype,
+                "original_url": tool_file.original_url,
+                "name": tool_file.name,
+                "size": tool_file.size,
+                "mime_type": mimetype,
+                "extension": extension,
+                "preview_url": preview_url,
+            }
+
+            return result, 201
        except services.errors.file.FileTooLargeError as file_too_large_error:
            raise FileTooLargeError(file_too_large_error.description)
        except services.errors.file.UnsupportedFileTypeError:
@@ -5,5 +5,6 @@ from libs.external_api import ExternalApi
 bp = Blueprint("inner_api", __name__, url_prefix="/inner/api")
 api = ExternalApi(bp)

+from . import mail
 from .plugin import plugin
 from .workspace import workspace
@@ -0,0 +1,27 @@
+from flask_restful import (
+    Resource,  # type: ignore
+    reqparse,
+)
+
+from controllers.console.wraps import setup_required
+from controllers.inner_api import api
+from controllers.inner_api.wraps import enterprise_inner_api_only
+from services.enterprise.mail_service import DifyMail, EnterpriseMailService
+
+
+class EnterpriseMail(Resource):
+    @setup_required
+    @enterprise_inner_api_only
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("to", type=str, action="append", required=True)
+        parser.add_argument("subject", type=str, required=True)
+        parser.add_argument("body", type=str, required=True)
+        parser.add_argument("substitutions", type=dict, required=False)
+        args = parser.parse_args()
+
+        EnterpriseMailService.send_mail(DifyMail(**args))
+        return {"message": "success"}, 200
+
+
+api.add_resource(EnterpriseMail, "/enterprise/mail")
@@ -29,7 +29,7 @@ from core.plugin.entities.request import (
    RequestRequestUploadFile,
 )
 from core.tools.entities.tool_entities import ToolProviderType
-from libs.helper import compact_generate_response
+from libs.helper import length_prefixed_response
 from models.account import Account, Tenant
 from models.model import EndUser

@@ -44,7 +44,7 @@ class PluginInvokeLLMApi(Resource):
            response = PluginModelBackwardsInvocation.invoke_llm(user_model.id, tenant_model, payload)
            return PluginModelBackwardsInvocation.convert_to_event_stream(response)

-        return compact_generate_response(generator())
+        return length_prefixed_response(0xF, generator())


 class PluginInvokeTextEmbeddingApi(Resource):
@@ -101,7 +101,7 @@ class PluginInvokeTTSApi(Resource):
            )
            return PluginModelBackwardsInvocation.convert_to_event_stream(response)

-        return compact_generate_response(generator())
+        return length_prefixed_response(0xF, generator())


 class PluginInvokeSpeech2TextApi(Resource):
@@ -162,7 +162,7 @@ class PluginInvokeToolApi(Resource):
                ),
            )

-        return compact_generate_response(generator())
+        return length_prefixed_response(0xF, generator())


 class PluginInvokeParameterExtractorNodeApi(Resource):
@@ -228,7 +228,7 @@ class PluginInvokeAppApi(Resource):
            files=payload.files,
        )

-        return compact_generate_response(PluginAppBackwardsInvocation.convert_to_event_stream(response))
+        return length_prefixed_response(0xF, PluginAppBackwardsInvocation.convert_to_event_stream(response))


 class PluginInvokeEncryptApi(Resource):
@@ -2,12 +2,14 @@ from collections.abc import Callable
 from functools import wraps
 from typing import Optional

-from flask import request
+from flask import current_app, request
+from flask_login import user_logged_in
 from flask_restful import reqparse
 from pydantic import BaseModel
 from sqlalchemy.orm import Session

 from extensions.ext_database import db
+from libs.login import _get_user
 from models.account import Account, Tenant
 from models.model import EndUser
 from services.account_service import AccountService
@@ -30,6 +32,7 @@ def get_user(tenant_id: str, user_id: str | None) -> Account | EndUser:
                    )
                    session.add(user_model)
                    session.commit()
+                    session.refresh(user_model)
            else:
                user_model = AccountService.load_user(user_id)
                if not user_model:
@@ -80,7 +83,12 @@ def get_user_tenant(view: Optional[Callable] = None):
                raise ValueError("tenant not found")

            kwargs["tenant_model"] = tenant_model
-            kwargs["user_model"] = get_user(tenant_id, user_id)
+
+            user = get_user(tenant_id, user_id)
+            kwargs["user_model"] = user
+
+            current_app.login_manager._update_request_context_with_user(user)  # type: ignore
+            user_logged_in.send(current_app._get_current_object(), user=_get_user())  # type: ignore

            return view_func(*args, **kwargs)

@@ -6,6 +6,6 @@ bp = Blueprint("service_api", __name__, url_prefix="/v1")
 api = ExternalApi(bp)

 from . import index
-from .app import annotation, app, audio, completion, conversation, file, message, workflow
+from .app import annotation, app, audio, completion, conversation, file, message, site, workflow
 from .dataset import dataset, document, hit_testing, metadata, segment, upload_file
 from .workspace import models
@@ -3,19 +3,19 @@ from flask_restful import Resource, marshal, marshal_with, reqparse
 from werkzeug.exceptions import Forbidden

 from controllers.service_api import api
-from controllers.service_api.wraps import FetchUserArg, WhereisUserArg, validate_app_token
+from controllers.service_api.wraps import validate_app_token
 from extensions.ext_redis import redis_client
 from fields.annotation_fields import (
    annotation_fields,
 )
 from libs.login import current_user
-from models.model import App, EndUser
+from models.model import App
 from services.annotation_service import AppAnnotationService


 class AnnotationReplyActionApi(Resource):
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON))
-    def post(self, app_model: App, end_user: EndUser, action):
+    @validate_app_token
+    def post(self, app_model: App, action):
        parser = reqparse.RequestParser()
        parser.add_argument("score_threshold", required=True, type=float, location="json")
        parser.add_argument("embedding_provider_name", required=True, type=str, location="json")
@@ -31,8 +31,8 @@ class AnnotationReplyActionApi(Resource):


 class AnnotationReplyActionStatusApi(Resource):
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.QUERY))
-    def get(self, app_model: App, end_user: EndUser, job_id, action):
+    @validate_app_token
+    def get(self, app_model: App, job_id, action):
        job_id = str(job_id)
        app_annotation_job_key = "{}_app_annotation_job_{}".format(action, str(job_id))
        cache_result = redis_client.get(app_annotation_job_key)
@@ -49,8 +49,8 @@ class AnnotationReplyActionStatusApi(Resource):


 class AnnotationListApi(Resource):
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.QUERY))
-    def get(self, app_model: App, end_user: EndUser):
+    @validate_app_token
+    def get(self, app_model: App):
        page = request.args.get("page", default=1, type=int)
        limit = request.args.get("limit", default=20, type=int)
        keyword = request.args.get("keyword", default="", type=str)
@@ -65,9 +65,9 @@ class AnnotationListApi(Resource):
        }
        return response, 200

-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON))
+    @validate_app_token
    @marshal_with(annotation_fields)
-    def post(self, app_model: App, end_user: EndUser):
+    def post(self, app_model: App):
        parser = reqparse.RequestParser()
        parser.add_argument("question", required=True, type=str, location="json")
        parser.add_argument("answer", required=True, type=str, location="json")
@@ -77,9 +77,9 @@ class AnnotationListApi(Resource):


 class AnnotationUpdateDeleteApi(Resource):
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.JSON))
+    @validate_app_token
    @marshal_with(annotation_fields)
-    def put(self, app_model: App, end_user: EndUser, annotation_id):
+    def put(self, app_model: App, annotation_id):
        if not current_user.is_editor:
            raise Forbidden()

@@ -91,8 +91,8 @@ class AnnotationUpdateDeleteApi(Resource):
        annotation = AppAnnotationService.update_app_annotation_directly(args, app_model.id, annotation_id)
        return annotation

-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.QUERY))
-    def delete(self, app_model: App, end_user: EndUser, annotation_id):
+    @validate_app_token
+    def delete(self, app_model: App, annotation_id):
        if not current_user.is_editor:
            raise Forbidden()

@@ -93,6 +93,18 @@ class MessageFeedbackApi(Resource):
        return {"result": "success"}


+class AppGetFeedbacksApi(Resource):
+    @validate_app_token
+    def get(self, app_model: App):
+        """Get All Feedbacks of an app"""
+        parser = reqparse.RequestParser()
+        parser.add_argument("page", type=int, default=1, location="args")
+        parser.add_argument("limit", type=int_range(1, 101), required=False, default=20, location="args")
+        args = parser.parse_args()
+        feedbacks = MessageService.get_all_messages_feedbacks(app_model, page=args["page"], limit=args["limit"])
+        return {"data": feedbacks}
+
+
 class MessageSuggestedApi(Resource):
    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.QUERY, required=True))
    def get(self, app_model: App, end_user: EndUser, message_id):
@@ -119,3 +131,4 @@ class MessageSuggestedApi(Resource):
 api.add_resource(MessageListApi, "/messages")
 api.add_resource(MessageFeedbackApi, "/messages/<uuid:message_id>/feedbacks")
 api.add_resource(MessageSuggestedApi, "/messages/<uuid:message_id>/suggested")
+api.add_resource(AppGetFeedbacksApi, "/app/feedbacks")
@@ -0,0 +1,30 @@
+from flask_restful import Resource, marshal_with
+from werkzeug.exceptions import Forbidden
+
+from controllers.common import fields
+from controllers.service_api import api
+from controllers.service_api.wraps import validate_app_token
+from extensions.ext_database import db
+from models.account import TenantStatus
+from models.model import App, Site
+
+
+class AppSiteApi(Resource):
+    """Resource for app sites."""
+
+    @validate_app_token
+    @marshal_with(fields.site_fields)
+    def get(self, app_model: App):
+        """Retrieve app site info."""
+        site = db.session.query(Site).filter(Site.app_id == app_model.id).first()
+
+        if not site:
+            raise Forbidden()
+
+        if app_model.tenant.status == TenantStatus.ARCHIVE:
+            raise Forbidden()
+
+        return site
+
+
+api.add_resource(AppSiteApi, "/site")
@@ -24,12 +24,13 @@ from core.errors.error import (
    QuotaExceededError,
 )
 from core.model_runtime.errors.invoke import InvokeError
+from core.workflow.entities.workflow_execution import WorkflowExecutionStatus
 from extensions.ext_database import db
 from fields.workflow_app_log_fields import workflow_app_log_pagination_fields
 from libs import helper
 from libs.helper import TimestampField
 from models.model import App, AppMode, EndUser
-from models.workflow import WorkflowRun, WorkflowRunStatus
+from models.workflow import WorkflowRun
 from services.app_generate_service import AppGenerateService
 from services.errors.llm import InvokeRateLimitError
 from services.workflow_app_service import WorkflowAppService
@@ -138,7 +139,7 @@ class WorkflowAppLogApi(Resource):
        parser.add_argument("limit", type=int_range(1, 100), default=20, location="args")
        args = parser.parse_args()

-        args.status = WorkflowRunStatus(args.status) if args.status else None
+        args.status = WorkflowExecutionStatus(args.status) if args.status else None
        if args.created_at__before:
            args.created_at__before = isoparse(args.created_at__before)

@@ -1,19 +1,21 @@
 from flask import request
-from flask_restful import marshal, reqparse
+from flask_restful import marshal, marshal_with, reqparse
 from werkzeug.exceptions import Forbidden, NotFound

 import services.dataset_service
 from controllers.service_api import api
 from controllers.service_api.dataset.error import DatasetInUseError, DatasetNameDuplicateError
-from controllers.service_api.wraps import DatasetApiResource
+from controllers.service_api.wraps import DatasetApiResource, validate_dataset_token
 from core.model_runtime.entities.model_entities import ModelType
 from core.plugin.entities.plugin import ModelProviderID
 from core.provider_manager import ProviderManager
 from fields.dataset_fields import dataset_detail_fields
+from fields.tag_fields import tag_fields
 from libs.login import current_user
 from models.dataset import Dataset, DatasetPermissionEnum
 from services.dataset_service import DatasetPermissionService, DatasetService
 from services.entities.knowledge_entities.knowledge_entities import RetrievalModel
+from services.tag_service import TagService


 def _validate_name(name):
@@ -313,12 +315,142 @@ class DatasetApi(DatasetApiResource):
        try:
            if DatasetService.delete_dataset(dataset_id_str, current_user):
                DatasetPermissionService.clear_partial_member_list(dataset_id_str)
-                return {"result": "success"}, 204
+                return 204
            else:
                raise NotFound("Dataset not found.")
        except services.errors.dataset.DatasetInUseError:
            raise DatasetInUseError()


+class DatasetTagsApi(DatasetApiResource):
+    @validate_dataset_token
+    @marshal_with(tag_fields)
+    def get(self, _, dataset_id):
+        """Get all knowledge type tags."""
+        tags = TagService.get_tags("knowledge", current_user.current_tenant_id)
+
+        return tags, 200
+
+    @validate_dataset_token
+    def post(self, _, dataset_id):
+        """Add a knowledge type tag."""
+        if not (current_user.is_editor or current_user.is_dataset_editor):
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "name",
+            nullable=False,
+            required=True,
+            help="Name must be between 1 to 50 characters.",
+            type=DatasetTagsApi._validate_tag_name,
+        )
+
+        args = parser.parse_args()
+        args["type"] = "knowledge"
+        tag = TagService.save_tags(args)
+
+        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
+
+        return response, 200
+
+    @validate_dataset_token
+    def patch(self, _, dataset_id):
+        if not (current_user.is_editor or current_user.is_dataset_editor):
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "name",
+            nullable=False,
+            required=True,
+            help="Name must be between 1 to 50 characters.",
+            type=DatasetTagsApi._validate_tag_name,
+        )
+        parser.add_argument("tag_id", nullable=False, required=True, help="Id of a tag.", type=str)
+        args = parser.parse_args()
+        args["type"] = "knowledge"
+        tag = TagService.update_tags(args, args.get("tag_id"))
+
+        binding_count = TagService.get_tag_binding_count(args.get("tag_id"))
+
+        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
+
+        return response, 200
+
+    @validate_dataset_token
+    def delete(self, _, dataset_id):
+        """Delete a knowledge type tag."""
+        if not current_user.is_editor:
+            raise Forbidden()
+        parser = reqparse.RequestParser()
+        parser.add_argument("tag_id", nullable=False, required=True, help="Id of a tag.", type=str)
+        args = parser.parse_args()
+        TagService.delete_tag(args.get("tag_id"))
+
+        return 204
+
+    @staticmethod
+    def _validate_tag_name(name):
+        if not name or len(name) < 1 or len(name) > 50:
+            raise ValueError("Name must be between 1 to 50 characters.")
+        return name
+
+
+class DatasetTagBindingApi(DatasetApiResource):
+    @validate_dataset_token
+    def post(self, _, dataset_id):
+        # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
+        if not (current_user.is_editor or current_user.is_dataset_editor):
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument(
+            "tag_ids", type=list, nullable=False, required=True, location="json", help="Tag IDs is required."
+        )
+        parser.add_argument(
+            "target_id", type=str, nullable=False, required=True, location="json", help="Target Dataset ID is required."
+        )
+
+        args = parser.parse_args()
+        args["type"] = "knowledge"
+        TagService.save_tag_binding(args)
+
+        return 204
+
+
+class DatasetTagUnbindingApi(DatasetApiResource):
+    @validate_dataset_token
+    def post(self, _, dataset_id):
+        # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
+        if not (current_user.is_editor or current_user.is_dataset_editor):
+            raise Forbidden()
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("tag_id", type=str, nullable=False, required=True, help="Tag ID is required.")
+        parser.add_argument("target_id", type=str, nullable=False, required=True, help="Target ID is required.")
+
+        args = parser.parse_args()
+        args["type"] = "knowledge"
+        TagService.delete_tag_binding(args)
+
+        return 204
+
+
+class DatasetTagsBindingStatusApi(DatasetApiResource):
+    @validate_dataset_token
+    def get(self, _, *args, **kwargs):
+        """Get all knowledge type tags."""
+        dataset_id = kwargs.get("dataset_id")
+        tags = TagService.get_tags_by_target_id("knowledge", current_user.current_tenant_id, str(dataset_id))
+        tags_list = [{"id": tag.id, "name": tag.name} for tag in tags]
+        response = {"data": tags_list, "total": len(tags)}
+        return response, 200
+
+
 api.add_resource(DatasetListApi, "/datasets")
 api.add_resource(DatasetApi, "/datasets/<uuid:dataset_id>")
+api.add_resource(DatasetTagsApi, "/datasets/tags")
+api.add_resource(DatasetTagBindingApi, "/datasets/tags/binding")
+api.add_resource(DatasetTagUnbindingApi, "/datasets/tags/unbinding")
+api.add_resource(DatasetTagsBindingStatusApi, "/datasets/<uuid:dataset_id>/tags")
@@ -2,10 +2,10 @@ import json

 from flask import request
 from flask_restful import marshal, reqparse
-from sqlalchemy import desc
+from sqlalchemy import desc, select
 from werkzeug.exceptions import NotFound

-import services.dataset_service
+import services
 from controllers.common.errors import FilenameNotExistsError
 from controllers.service_api import api
 from controllers.service_api.app.error import (
@@ -175,8 +175,11 @@ class DocumentAddByFileApi(DatasetApiResource):

        if not dataset:
            raise ValueError("Dataset does not exist.")
-        if not dataset.indexing_technique and not args.get("indexing_technique"):
+
+        indexing_technique = args.get("indexing_technique") or dataset.indexing_technique
+        if not indexing_technique:
            raise ValueError("indexing_technique is required.")
+        args["indexing_technique"] = indexing_technique

        # save file info
        file = request.files["file"]
@@ -206,12 +209,16 @@ class DocumentAddByFileApi(DatasetApiResource):
        knowledge_config = KnowledgeConfig(**args)
        DocumentService.document_create_args_validate(knowledge_config)

+        dataset_process_rule = dataset.latest_process_rule if "process_rule" not in args else None
+        if not knowledge_config.original_document_id and not dataset_process_rule and not knowledge_config.process_rule:
+            raise ValueError("process_rule is required.")
+
        try:
            documents, batch = DocumentService.save_document_with_dataset_id(
                dataset=dataset,
                knowledge_config=knowledge_config,
                account=dataset.created_by_account,
-                dataset_process_rule=dataset.latest_process_rule if "process_rule" not in args else None,
+                dataset_process_rule=dataset_process_rule,
                created_from="api",
            )
        except ProviderTokenNotInitError as ex:
@@ -323,7 +330,7 @@ class DocumentDeleteApi(DatasetApiResource):
        except services.errors.document.DocumentIndexingError:
            raise DocumentIndexingError("Cannot delete document during indexing.")

-        return {"result": "success"}, 204
+        return 204


 class DocumentListApi(DatasetApiResource):
@@ -337,7 +344,7 @@ class DocumentListApi(DatasetApiResource):
        if not dataset:
            raise NotFound("Dataset not found.")

-        query = Document.query.filter_by(dataset_id=str(dataset_id), tenant_id=tenant_id)
+        query = select(Document).filter_by(dataset_id=str(dataset_id), tenant_id=tenant_id)

        if search:
            search = f"%{search}%"
@@ -345,7 +352,7 @@ class DocumentListApi(DatasetApiResource):

        query = query.order_by(desc(Document.created_at), desc(Document.position))

-        paginated_documents = query.paginate(page=page, per_page=limit, max_per_page=100, error_out=False)
+        paginated_documents = db.paginate(select=query, page=page, per_page=limit, max_per_page=100, error_out=False)
        documents = paginated_documents.items

        response = {
@@ -374,19 +381,36 @@ class DocumentIndexingStatusApi(DatasetApiResource):
            raise NotFound("Documents not found.")
        documents_status = []
        for document in documents:
-            completed_segments = DocumentSegment.query.filter(
-                DocumentSegment.completed_at.isnot(None),
-                DocumentSegment.document_id == str(document.id),
-                DocumentSegment.status != "re_segment",
-            ).count()
-            total_segments = DocumentSegment.query.filter(
-                DocumentSegment.document_id == str(document.id), DocumentSegment.status != "re_segment"
-            ).count()
-            document.completed_segments = completed_segments
-            document.total_segments = total_segments
-            if document.is_paused:
-                document.indexing_status = "paused"
-            documents_status.append(marshal(document, document_status_fields))
+            completed_segments = (
+                db.session.query(DocumentSegment)
+                .filter(
+                    DocumentSegment.completed_at.isnot(None),
+                    DocumentSegment.document_id == str(document.id),
+                    DocumentSegment.status != "re_segment",
+                )
+                .count()
+            )
+            total_segments = (
+                db.session.query(DocumentSegment)
+                .filter(DocumentSegment.document_id == str(document.id), DocumentSegment.status != "re_segment")
+                .count()
+            )
+            # Create a dictionary with document attributes and additional fields
+            document_dict = {
+                "id": document.id,
+                "indexing_status": "paused" if document.is_paused else document.indexing_status,
+                "processing_started_at": document.processing_started_at,
+                "parsing_completed_at": document.parsing_completed_at,
+                "cleaning_completed_at": document.cleaning_completed_at,
+                "splitting_completed_at": document.splitting_completed_at,
+                "completed_at": document.completed_at,
+                "paused_at": document.paused_at,
+                "error": document.error,
+                "stopped_at": document.stopped_at,
+                "completed_segments": completed_segments,
+                "total_segments": total_segments,
+            }
+            documents_status.append(marshal(document_dict, document_status_fields))
        data = {"data": documents_status}
        return data

@@ -159,7 +159,7 @@ class DatasetSegmentApi(DatasetApiResource):
        if not segment:
            raise NotFound("Segment not found.")
        SegmentService.delete_segment(segment, document, dataset)
-        return {"result": "success"}, 204
+        return 204

    @cloud_edition_billing_resource_check("vector_space", "dataset")
    def post(self, tenant_id, dataset_id, document_id, segment_id):
@@ -208,6 +208,28 @@ class DatasetSegmentApi(DatasetApiResource):
        )
        return {"data": marshal(updated_segment, segment_fields), "doc_form": document.doc_form}, 200

+    def get(self, tenant_id, dataset_id, document_id, segment_id):
+        # check dataset
+        dataset_id = str(dataset_id)
+        tenant_id = str(tenant_id)
+        dataset = db.session.query(Dataset).filter(Dataset.tenant_id == tenant_id, Dataset.id == dataset_id).first()
+        if not dataset:
+            raise NotFound("Dataset not found.")
+        # check user's model setting
+        DatasetService.check_dataset_model_setting(dataset)
+        # check document
+        document_id = str(document_id)
+        document = DocumentService.get_document(dataset_id, document_id)
+        if not document:
+            raise NotFound("Document not found.")
+        # check segment
+        segment_id = str(segment_id)
+        segment = SegmentService.get_segment_by_id(segment_id=segment_id, tenant_id=current_user.current_tenant_id)
+        if not segment:
+            raise NotFound("Segment not found.")
+
+        return {"data": marshal(segment, segment_fields), "doc_form": document.doc_form}, 200
+

 class ChildChunkApi(DatasetApiResource):
    """Resource for child chunks."""
@@ -344,7 +366,7 @@ class DatasetChildChunkApi(DatasetApiResource):
        except ChildChunkDeleteIndexServiceError as e:
            raise ChildChunkDeleteIndexError(str(e))

-        return {"result": "success"}, 204
+        return 204

    @cloud_edition_billing_resource_check("vector_space", "dataset")
    @cloud_edition_billing_knowledge_limit_check("add_segment", "dataset")
@@ -99,7 +99,12 @@ def validate_app_token(view: Optional[Callable] = None, *, fetch_user_arg: Optio
                if user_id:
                    user_id = str(user_id)

-                kwargs["end_user"] = create_or_update_end_user_for_user_id(app_model, user_id)
+                end_user = create_or_update_end_user_for_user_id(app_model, user_id)
+                kwargs["end_user"] = end_user
+
+                # Set EndUser as current logged-in user for flask_login.current_user
+                current_app.login_manager._update_request_context_with_user(end_user)  # type: ignore
+                user_logged_in.send(current_app._get_current_object(), user=end_user)  # type: ignore

            return view_func(*args, **kwargs)

@@ -15,4 +15,17 @@ api.add_resource(FileApi, "/files/upload")
 api.add_resource(RemoteFileInfoApi, "/remote-files/<path:url>")
 api.add_resource(RemoteFileUploadApi, "/remote-files/upload")

-from . import app, audio, completion, conversation, feature, message, passport, saved_message, site, workflow
+from . import (
+    app,
+    audio,
+    completion,
+    conversation,
+    feature,
+    forgot_password,
+    login,
+    message,
+    passport,
+    saved_message,
+    site,
+    workflow,
+)
@@ -1,12 +1,17 @@
-from flask_restful import marshal_with
+from flask import request
+from flask_restful import Resource, marshal_with, reqparse

 from controllers.common import fields
 from controllers.web import api
 from controllers.web.error import AppUnavailableError
 from controllers.web.wraps import WebApiResource
 from core.app.app_config.common.parameters_mapping import get_parameters_from_feature_dict
+from libs.passport import PassportService
 from models.model import App, AppMode
 from services.app_service import AppService
+from services.enterprise.enterprise_service import EnterpriseService
+from services.feature_service import FeatureService
+from services.webapp_auth_service import WebAppAuthService


 class AppParameterApi(WebApiResource):
@@ -40,5 +45,69 @@ class AppMeta(WebApiResource):
        return AppService().get_app_meta(app_model)


+class AppAccessMode(Resource):
+    def get(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("appId", type=str, required=False, location="args")
+        parser.add_argument("appCode", type=str, required=False, location="args")
+        args = parser.parse_args()
+
+        features = FeatureService.get_system_features()
+        if not features.webapp_auth.enabled:
+            return {"accessMode": "public"}
+
+        app_id = args.get("appId")
+        if args.get("appCode"):
+            app_code = args["appCode"]
+            app_id = AppService.get_app_id_by_code(app_code)
+
+        if not app_id:
+            raise ValueError("appId or appCode must be provided")
+
+        res = EnterpriseService.WebAppAuth.get_app_access_mode_by_id(app_id)
+
+        return {"accessMode": res.access_mode}
+
+
+class AppWebAuthPermission(Resource):
+    def get(self):
+        user_id = "visitor"
+        try:
+            auth_header = request.headers.get("Authorization")
+            if auth_header is None:
+                raise
+            if " " not in auth_header:
+                raise
+
+            auth_scheme, tk = auth_header.split(None, 1)
+            auth_scheme = auth_scheme.lower()
+            if auth_scheme != "bearer":
+                raise
+
+            decoded = PassportService().verify(tk)
+            user_id = decoded.get("user_id", "visitor")
+        except Exception as e:
+            pass
+
+        features = FeatureService.get_system_features()
+        if not features.webapp_auth.enabled:
+            return {"result": True}
+
+        parser = reqparse.RequestParser()
+        parser.add_argument("appId", type=str, required=True, location="args")
+        args = parser.parse_args()
+
+        app_id = args["appId"]
+        app_code = AppService.get_app_code_by_id(app_id)
+
+        res = True
+        if WebAppAuthService.is_app_require_permission_check(app_id=app_id):
+            res = EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(str(user_id), app_code)
+        return {"result": res}
+
+
 api.add_resource(AppParameterApi, "/parameters")
 api.add_resource(AppMeta, "/meta")
+# webapp auth apis
+api.add_resource(AppAccessMode, "/webapp/access-mode")
+api.add_resource(AppWebAuthPermission, "/webapp/permission")
@@ -121,9 +121,15 @@ class UnsupportedFileTypeError(BaseHTTPException):
    code = 415


-class WebSSOAuthRequiredError(BaseHTTPException):
+class WebAppAuthRequiredError(BaseHTTPException):
    error_code = "web_sso_auth_required"
-    description = "Web SSO authentication required."
+    description = "Web app authentication required."
+    code = 401
+
+
+class WebAppAuthAccessDeniedError(BaseHTTPException):
+    error_code = "web_app_access_denied"
+    description = "You do not have permission to access this web app."
    code = 401


@@ -0,0 +1,147 @@
+import base64
+import secrets
+
+from flask import request
+from flask_restful import Resource, reqparse
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from controllers.console.auth.error import (
+    EmailCodeError,
+    EmailPasswordResetLimitError,
+    InvalidEmailError,
+    InvalidTokenError,
+    PasswordMismatchError,
+)
+from controllers.console.error import AccountNotFound, EmailSendIpLimitError
+from controllers.console.wraps import email_password_login_enabled, only_edition_enterprise, setup_required
+from controllers.web import api
+from extensions.ext_database import db
+from libs.helper import email, extract_remote_ip
+from libs.password import hash_password, valid_password
+from models.account import Account
+from services.account_service import AccountService
+
+
+class ForgotPasswordSendEmailApi(Resource):
+    @only_edition_enterprise
+    @setup_required
+    @email_password_login_enabled
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("email", type=email, required=True, location="json")
+        parser.add_argument("language", type=str, required=False, location="json")
+        args = parser.parse_args()
+
+        ip_address = extract_remote_ip(request)
+        if AccountService.is_email_send_ip_limit(ip_address):
+            raise EmailSendIpLimitError()
+
+        if args["language"] is not None and args["language"] == "zh-Hans":
+            language = "zh-Hans"
+        else:
+            language = "en-US"
+
+        with Session(db.engine) as session:
+            account = session.execute(select(Account).filter_by(email=args["email"])).scalar_one_or_none()
+        token = None
+        if account is None:
+            raise AccountNotFound()
+        else:
+            token = AccountService.send_reset_password_email(account=account, email=args["email"], language=language)
+
+        return {"result": "success", "data": token}
+
+
+class ForgotPasswordCheckApi(Resource):
+    @only_edition_enterprise
+    @setup_required
+    @email_password_login_enabled
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("email", type=str, required=True, location="json")
+        parser.add_argument("code", type=str, required=True, location="json")
+        parser.add_argument("token", type=str, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+
+        user_email = args["email"]
+
+        is_forgot_password_error_rate_limit = AccountService.is_forgot_password_error_rate_limit(args["email"])
+        if is_forgot_password_error_rate_limit:
+            raise EmailPasswordResetLimitError()
+
+        token_data = AccountService.get_reset_password_data(args["token"])
+        if token_data is None:
+            raise InvalidTokenError()
+
+        if user_email != token_data.get("email"):
+            raise InvalidEmailError()
+
+        if args["code"] != token_data.get("code"):
+            AccountService.add_forgot_password_error_rate_limit(args["email"])
+            raise EmailCodeError()
+
+        # Verified, revoke the first token
+        AccountService.revoke_reset_password_token(args["token"])
+
+        # Refresh token data by generating a new token
+        _, new_token = AccountService.generate_reset_password_token(
+            user_email, code=args["code"], additional_data={"phase": "reset"}
+        )
+
+        AccountService.reset_forgot_password_error_rate_limit(args["email"])
+        return {"is_valid": True, "email": token_data.get("email"), "token": new_token}
+
+
+class ForgotPasswordResetApi(Resource):
+    @only_edition_enterprise
+    @setup_required
+    @email_password_login_enabled
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("token", type=str, required=True, nullable=False, location="json")
+        parser.add_argument("new_password", type=valid_password, required=True, nullable=False, location="json")
+        parser.add_argument("password_confirm", type=valid_password, required=True, nullable=False, location="json")
+        args = parser.parse_args()
+
+        # Validate passwords match
+        if args["new_password"] != args["password_confirm"]:
+            raise PasswordMismatchError()
+
+        # Validate token and get reset data
+        reset_data = AccountService.get_reset_password_data(args["token"])
+        if not reset_data:
+            raise InvalidTokenError()
+        # Must use token in reset phase
+        if reset_data.get("phase", "") != "reset":
+            raise InvalidTokenError()
+
+        # Revoke token to prevent reuse
+        AccountService.revoke_reset_password_token(args["token"])
+
+        # Generate secure salt and hash password
+        salt = secrets.token_bytes(16)
+        password_hashed = hash_password(args["new_password"], salt)
+
+        email = reset_data.get("email", "")
+
+        with Session(db.engine) as session:
+            account = session.execute(select(Account).filter_by(email=email)).scalar_one_or_none()
+
+            if account:
+                self._update_existing_account(account, password_hashed, salt, session)
+            else:
+                raise AccountNotFound()
+
+        return {"result": "success"}
+
+    def _update_existing_account(self, account, password_hashed, salt, session):
+        # Update existing account credentials
+        account.password = base64.b64encode(password_hashed).decode()
+        account.password_salt = base64.b64encode(salt).decode()
+        session.commit()
+
+
+api.add_resource(ForgotPasswordSendEmailApi, "/forgot-password")
+api.add_resource(ForgotPasswordCheckApi, "/forgot-password/validity")
+api.add_resource(ForgotPasswordResetApi, "/forgot-password/resets")
@@ -0,0 +1,108 @@
+from flask_restful import Resource, reqparse
+from jwt import InvalidTokenError  # type: ignore
+
+import services
+from controllers.console.auth.error import EmailCodeError, EmailOrPasswordMismatchError, InvalidEmailError
+from controllers.console.error import AccountBannedError, AccountNotFound
+from controllers.console.wraps import only_edition_enterprise, setup_required
+from controllers.web import api
+from libs.helper import email
+from libs.password import valid_password
+from services.account_service import AccountService
+from services.webapp_auth_service import WebAppAuthService
+
+
+class LoginApi(Resource):
+    """Resource for web app email/password login."""
+
+    @setup_required
+    @only_edition_enterprise
+    def post(self):
+        """Authenticate user and login."""
+        parser = reqparse.RequestParser()
+        parser.add_argument("email", type=email, required=True, location="json")
+        parser.add_argument("password", type=valid_password, required=True, location="json")
+        args = parser.parse_args()
+
+        try:
+            account = WebAppAuthService.authenticate(args["email"], args["password"])
+        except services.errors.account.AccountLoginError:
+            raise AccountBannedError()
+        except services.errors.account.AccountPasswordError:
+            raise EmailOrPasswordMismatchError()
+        except services.errors.account.AccountNotFoundError:
+            raise AccountNotFound()
+
+        token = WebAppAuthService.login(account=account)
+        return {"result": "success", "data": {"access_token": token}}
+
+
+# class LogoutApi(Resource):
+#     @setup_required
+#     def get(self):
+#         account = cast(Account, flask_login.current_user)
+#         if isinstance(account, flask_login.AnonymousUserMixin):
+#             return {"result": "success"}
+#         flask_login.logout_user()
+#         return {"result": "success"}
+
+
+class EmailCodeLoginSendEmailApi(Resource):
+    @setup_required
+    @only_edition_enterprise
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("email", type=email, required=True, location="json")
+        parser.add_argument("language", type=str, required=False, location="json")
+        args = parser.parse_args()
+
+        if args["language"] is not None and args["language"] == "zh-Hans":
+            language = "zh-Hans"
+        else:
+            language = "en-US"
+
+        account = WebAppAuthService.get_user_through_email(args["email"])
+        if account is None:
+            raise AccountNotFound()
+        else:
+            token = WebAppAuthService.send_email_code_login_email(account=account, language=language)
+
+        return {"result": "success", "data": token}
+
+
+class EmailCodeLoginApi(Resource):
+    @setup_required
+    @only_edition_enterprise
+    def post(self):
+        parser = reqparse.RequestParser()
+        parser.add_argument("email", type=str, required=True, location="json")
+        parser.add_argument("code", type=str, required=True, location="json")
+        parser.add_argument("token", type=str, required=True, location="json")
+        args = parser.parse_args()
+
+        user_email = args["email"]
+
+        token_data = WebAppAuthService.get_email_code_login_data(args["token"])
+        if token_data is None:
+            raise InvalidTokenError()
+
+        if token_data["email"] != args["email"]:
+            raise InvalidEmailError()
+
+        if token_data["code"] != args["code"]:
+            raise EmailCodeError()
+
+        WebAppAuthService.revoke_email_code_login_token(args["token"])
+        account = WebAppAuthService.get_user_through_email(user_email)
+        if not account:
+            raise AccountNotFound()
+
+        token = WebAppAuthService.login(account=account)
+        AccountService.reset_login_error_rate_limit(args["email"])
+        return {"result": "success", "data": {"access_token": token}}
+
+
+api.add_resource(LoginApi, "/login")
+# api.add_resource(LogoutApi, "/logout")
+api.add_resource(EmailCodeLoginSendEmailApi, "/email-code-login")
+api.add_resource(EmailCodeLoginApi, "/email-code-login/validity")
@@ -1,16 +1,19 @@
 import uuid
+from datetime import UTC, datetime, timedelta

 from flask import request
 from flask_restful import Resource
 from werkzeug.exceptions import NotFound, Unauthorized

+from configs import dify_config
 from controllers.web import api
-from controllers.web.error import WebSSOAuthRequiredError
+from controllers.web.error import WebAppAuthRequiredError
 from extensions.ext_database import db
 from libs.passport import PassportService
 from models.model import App, EndUser, Site
 from services.enterprise.enterprise_service import EnterpriseService
 from services.feature_service import FeatureService
+from services.webapp_auth_service import WebAppAuthService, WebAppAuthType


 class PassportResource(Resource):
@@ -20,14 +23,23 @@ class PassportResource(Resource):
        system_features = FeatureService.get_system_features()
        app_code = request.headers.get("X-App-Code")
        user_id = request.args.get("user_id")
+        web_app_access_token = request.args.get("web_app_access_token")

        if app_code is None:
            raise Unauthorized("X-App-Code header is missing.")

-        if system_features.sso_enforced_for_web:
-            app_web_sso_enabled = EnterpriseService.get_app_web_sso_enabled(app_code).get("enabled", False)
-            if app_web_sso_enabled:
-                raise WebSSOAuthRequiredError()
+        # exchange token for enterprise logined web user
+        enterprise_user_decoded = decode_enterprise_webapp_user_id(web_app_access_token)
+        if enterprise_user_decoded:
+            # a web user has already logged in, exchange a token for this app without redirecting to the login page
+            return exchange_token_for_existing_web_user(
+                app_code=app_code, enterprise_user_decoded=enterprise_user_decoded
+            )
+
+        if system_features.webapp_auth.enabled:
+            app_settings = EnterpriseService.WebAppAuth.get_app_access_mode_by_code(app_code=app_code)
+            if not app_settings or not app_settings.access_mode == "public":
+                raise WebAppAuthRequiredError()

        # get site from db and check if it is normal
        site = db.session.query(Site).filter(Site.code == app_code, Site.status == "normal").first()
@@ -84,6 +96,128 @@ class PassportResource(Resource):
 api.add_resource(PassportResource, "/passport")


+def decode_enterprise_webapp_user_id(jwt_token: str | None):
+    """
+    Decode the enterprise user session from the Authorization header.
+    """
+    if not jwt_token:
+        return None
+
+    decoded = PassportService().verify(jwt_token)
+    source = decoded.get("token_source")
+    if not source or source != "webapp_login_token":
+        raise Unauthorized("Invalid token source. Expected 'webapp_login_token'.")
+    return decoded
+
+
+def exchange_token_for_existing_web_user(app_code: str, enterprise_user_decoded: dict):
+    """
+    Exchange a token for an existing web user session.
+    """
+    user_id = enterprise_user_decoded.get("user_id")
+    end_user_id = enterprise_user_decoded.get("end_user_id")
+    session_id = enterprise_user_decoded.get("session_id")
+    user_auth_type = enterprise_user_decoded.get("auth_type")
+    if not user_auth_type:
+        raise Unauthorized("Missing auth_type in the token.")
+
+    site = db.session.query(Site).filter(Site.code == app_code, Site.status == "normal").first()
+    if not site:
+        raise NotFound()
+
+    app_model = db.session.query(App).filter(App.id == site.app_id).first()
+    if not app_model or app_model.status != "normal" or not app_model.enable_site:
+        raise NotFound()
+
+    app_auth_type = WebAppAuthService.get_app_auth_type(app_code=app_code)
+
+    if app_auth_type == WebAppAuthType.PUBLIC:
+        return _exchange_for_public_app_token(app_model, site, enterprise_user_decoded)
+    elif app_auth_type == WebAppAuthType.EXTERNAL and user_auth_type != "external":
+        raise WebAppAuthRequiredError("Please login as external user.")
+    elif app_auth_type == WebAppAuthType.INTERNAL and user_auth_type != "internal":
+        raise WebAppAuthRequiredError("Please login as internal user.")
+
+    end_user = None
+    if end_user_id:
+        end_user = db.session.query(EndUser).filter(EndUser.id == end_user_id).first()
+    if session_id:
+        end_user = (
+            db.session.query(EndUser)
+            .filter(
+                EndUser.session_id == session_id,
+                EndUser.tenant_id == app_model.tenant_id,
+                EndUser.app_id == app_model.id,
+            )
+            .first()
+        )
+    if not end_user:
+        if not session_id:
+            raise NotFound("Missing session_id for existing web user.")
+        end_user = EndUser(
+            tenant_id=app_model.tenant_id,
+            app_id=app_model.id,
+            type="browser",
+            is_anonymous=True,
+            session_id=session_id,
+        )
+        db.session.add(end_user)
+        db.session.commit()
+    exp_dt = datetime.now(UTC) + timedelta(hours=dify_config.ACCESS_TOKEN_EXPIRE_MINUTES * 24)
+    exp = int(exp_dt.timestamp())
+    payload = {
+        "iss": site.id,
+        "sub": "Web API Passport",
+        "app_id": site.app_id,
+        "app_code": site.code,
+        "user_id": user_id,
+        "end_user_id": end_user.id,
+        "auth_type": user_auth_type,
+        "granted_at": int(datetime.now(UTC).timestamp()),
+        "token_source": "webapp",
+        "exp": exp,
+    }
+    token: str = PassportService().issue(payload)
+    return {
+        "access_token": token,
+    }
+
+
+def _exchange_for_public_app_token(app_model, site, token_decoded):
+    user_id = token_decoded.get("user_id")
+    end_user = None
+    if user_id:
+        end_user = (
+            db.session.query(EndUser).filter(EndUser.app_id == app_model.id, EndUser.session_id == user_id).first()
+        )
+
+    if not end_user:
+        end_user = EndUser(
+            tenant_id=app_model.tenant_id,
+            app_id=app_model.id,
+            type="browser",
+            is_anonymous=True,
+            session_id=generate_session_id(),
+        )
+
+        db.session.add(end_user)
+        db.session.commit()
+
+    payload = {
+        "iss": site.app_id,
+        "sub": "Web API Passport",
+        "app_id": site.app_id,
+        "app_code": site.code,
+        "end_user_id": end_user.id,
+    }
+
+    tk = PassportService().issue(payload)
+
+    return {
+        "access_token": tk,
+    }
+
+
 def generate_session_id():
    """
    Generate a unique session ID.
@@ -1,15 +1,17 @@
+from datetime import UTC, datetime
 from functools import wraps

 from flask import request
 from flask_restful import Resource
 from werkzeug.exceptions import BadRequest, NotFound, Unauthorized

-from controllers.web.error import WebSSOAuthRequiredError
+from controllers.web.error import WebAppAuthAccessDeniedError, WebAppAuthRequiredError
 from extensions.ext_database import db
 from libs.passport import PassportService
 from models.model import App, EndUser, Site
-from services.enterprise.enterprise_service import EnterpriseService
+from services.enterprise.enterprise_service import EnterpriseService, WebAppSettings
 from services.feature_service import FeatureService
+from services.webapp_auth_service import WebAppAuthService


 def validate_jwt_token(view=None):
@@ -29,7 +31,7 @@ def validate_jwt_token(view=None):

 def decode_jwt_token():
    system_features = FeatureService.get_system_features()
-    app_code = request.headers.get("X-App-Code")
+    app_code = str(request.headers.get("X-App-Code"))
    try:
        auth_header = request.headers.get("Authorization")
        if auth_header is None:
@@ -45,7 +47,8 @@ def decode_jwt_token():
            raise Unauthorized("Invalid Authorization header format. Expected 'Bearer <api-key>' format.")
        decoded = PassportService().verify(tk)
        app_code = decoded.get("app_code")
-        app_model = db.session.query(App).filter(App.id == decoded["app_id"]).first()
+        app_id = decoded.get("app_id")
+        app_model = db.session.query(App).filter(App.id == app_id).first()
        site = db.session.query(Site).filter(Site.code == app_code).first()
        if not app_model:
            raise NotFound()
@@ -53,39 +56,90 @@ def decode_jwt_token():
            raise BadRequest("Site URL is no longer valid.")
        if app_model.enable_site is False:
            raise BadRequest("Site is disabled.")
-        end_user = db.session.query(EndUser).filter(EndUser.id == decoded["end_user_id"]).first()
+        end_user_id = decoded.get("end_user_id")
+        end_user = db.session.query(EndUser).filter(EndUser.id == end_user_id).first()
        if not end_user:
            raise NotFound()

-        _validate_web_sso_token(decoded, system_features, app_code)
+        # for enterprise webapp auth
+        app_web_auth_enabled = False
+        webapp_settings = None
+        if system_features.webapp_auth.enabled:
+            webapp_settings = EnterpriseService.WebAppAuth.get_app_access_mode_by_code(app_code=app_code)
+            if not webapp_settings:
+                raise NotFound("Web app settings not found.")
+            app_web_auth_enabled = webapp_settings.access_mode != "public"
+
+        _validate_webapp_token(decoded, app_web_auth_enabled, system_features.webapp_auth.enabled)
+        _validate_user_accessibility(
+            decoded, app_code, app_web_auth_enabled, system_features.webapp_auth.enabled, webapp_settings
+        )

        return app_model, end_user
    except Unauthorized as e:
-        if system_features.sso_enforced_for_web:
-            app_web_sso_enabled = EnterpriseService.get_app_web_sso_enabled(app_code).get("enabled", False)
-            if app_web_sso_enabled:
-                raise WebSSOAuthRequiredError()
+        if system_features.webapp_auth.enabled:
+            if not app_code:
+                raise Unauthorized("Please re-login to access the web app.")
+            app_web_auth_enabled = (
+                EnterpriseService.WebAppAuth.get_app_access_mode_by_code(app_code=str(app_code)).access_mode != "public"
+            )
+            if app_web_auth_enabled:
+                raise WebAppAuthRequiredError()

        raise Unauthorized(e.description)


-def _validate_web_sso_token(decoded, system_features, app_code):
-    app_web_sso_enabled = False
-
-    # Check if SSO is enforced for web, and if the token source is not SSO, raise an error and redirect to SSO login
-    if system_features.sso_enforced_for_web:
-        app_web_sso_enabled = EnterpriseService.get_app_web_sso_enabled(app_code).get("enabled", False)
-        if app_web_sso_enabled:
-            source = decoded.get("token_source")
-            if not source or source != "sso":
-                raise WebSSOAuthRequiredError()
-
-    # Check if SSO is not enforced for web, and if the token source is SSO,
-    # raise an error and redirect to normal passport login
-    if not system_features.sso_enforced_for_web or not app_web_sso_enabled:
+def _validate_webapp_token(decoded, app_web_auth_enabled: bool, system_webapp_auth_enabled: bool):
+    # Check if authentication is enforced for web app, and if the token source is not webapp,
+    # raise an error and redirect to login
+    if system_webapp_auth_enabled and app_web_auth_enabled:
        source = decoded.get("token_source")
-        if source and source == "sso":
-            raise Unauthorized("sso token expired.")
+        if not source or source != "webapp":
+            raise WebAppAuthRequiredError()
+
+    # Check if authentication is not enforced for web, and if the token source is webapp,
+    # raise an error and redirect to normal passport login
+    if not system_webapp_auth_enabled or not app_web_auth_enabled:
+        source = decoded.get("token_source")
+        if source and source == "webapp":
+            raise Unauthorized("webapp token expired.")
+
+
+def _validate_user_accessibility(
+    decoded,
+    app_code,
+    app_web_auth_enabled: bool,
+    system_webapp_auth_enabled: bool,
+    webapp_settings: WebAppSettings | None,
+):
+    if system_webapp_auth_enabled and app_web_auth_enabled:
+        # Check if the user is allowed to access the web app
+        user_id = decoded.get("user_id")
+        if not user_id:
+            raise WebAppAuthRequiredError()
+
+        if not webapp_settings:
+            raise WebAppAuthRequiredError("Web app settings not found.")
+
+        if WebAppAuthService.is_app_require_permission_check(access_mode=webapp_settings.access_mode):
+            if not EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(user_id, app_code=app_code):
+                raise WebAppAuthAccessDeniedError()
+
+        auth_type = decoded.get("auth_type")
+        granted_at = decoded.get("granted_at")
+        if not auth_type:
+            raise WebAppAuthAccessDeniedError("Missing auth_type in the token.")
+        if not granted_at:
+            raise WebAppAuthAccessDeniedError("Missing granted_at in the token.")
+        # check if sso has been updated
+        if auth_type == "external":
+            last_update_time = EnterpriseService.get_app_sso_settings_last_update_time()
+            if granted_at and datetime.fromtimestamp(granted_at, tz=UTC) < last_update_time:
+                raise WebAppAuthAccessDeniedError("SSO settings have been updated. Please re-login.")
+        elif auth_type == "internal":
+            last_update_time = EnterpriseService.get_workspace_sso_settings_last_update_time()
+            if granted_at and datetime.fromtimestamp(granted_at, tz=UTC) < last_update_time:
+                raise WebAppAuthAccessDeniedError("SSO settings have been updated. Please re-login.")


 class WebApiResource(Resource):
@@ -91,6 +91,8 @@ class BaseAgentRunner(AppRunner):
            return_resource=app_config.additional_features.show_retrieve_source,
            invoke_from=application_generate_entity.invoke_from,
            hit_callback=hit_callback,
+            user_id=user_id,
+            inputs=cast(dict, application_generate_entity.inputs),
        )
        # get how many agent thoughts have been created
        self.agent_thought_count = (
@@ -63,19 +63,12 @@ class CotAgentRunner(BaseAgentRunner, ABC):
        self._instruction = self._fill_in_inputs_from_external_data_tools(instruction, inputs)

        iteration_step = 1
-        max_iteration_steps = min(app_config.agent.max_iteration if app_config.agent else 5, 5) + 1
+        max_iteration_steps = min(app_config.agent.max_iteration, 99) + 1

        # convert tools into ModelRuntime Tool format
        tool_instances, prompt_messages_tools = self._init_prompt_tools()
        self._prompt_messages_tools = prompt_messages_tools

-        # fix metadata filter not work
-        if app_config.dataset is not None:
-            metadata_filtering_conditions = app_config.dataset.retrieve_config.metadata_filtering_conditions
-            for key, dataset_retriever_tool in tool_instances.items():
-                if hasattr(dataset_retriever_tool, "retrieval_tool"):
-                    dataset_retriever_tool.retrieval_tool.metadata_filtering_conditions = metadata_filtering_conditions
-
        function_call_state = True
        llm_usage: dict[str, Optional[LLMUsage]] = {"usage": None}
        final_answer = ""
@@ -87,6 +80,7 @@ class CotAgentRunner(BaseAgentRunner, ABC):
                llm_usage = final_llm_usage_dict["usage"]
                llm_usage.prompt_tokens += usage.prompt_tokens
                llm_usage.completion_tokens += usage.completion_tokens
+                llm_usage.total_tokens += usage.total_tokens
                llm_usage.prompt_price += usage.prompt_price
                llm_usage.completion_price += usage.completion_price
                llm_usage.total_price += usage.total_price
@@ -82,7 +82,7 @@ class AgentEntity(BaseModel):
    strategy: Strategy
    prompt: Optional[AgentPromptEntity] = None
    tools: Optional[list[AgentToolEntity]] = None
-    max_iteration: int = 5
+    max_iteration: int = 10


 class AgentInvokeMessage(ToolInvokeMessage):
@@ -45,17 +45,10 @@ class FunctionCallAgentRunner(BaseAgentRunner):
        # convert tools into ModelRuntime Tool format
        tool_instances, prompt_messages_tools = self._init_prompt_tools()

-        # fix metadata filter not work
-        if app_config.dataset is not None:
-            metadata_filtering_conditions = app_config.dataset.retrieve_config.metadata_filtering_conditions
-            for key, dataset_retriever_tool in tool_instances.items():
-                if hasattr(dataset_retriever_tool, "retrieval_tool"):
-                    dataset_retriever_tool.retrieval_tool.metadata_filtering_conditions = metadata_filtering_conditions
-
        assert app_config.agent

        iteration_step = 1
-        max_iteration_steps = min(app_config.agent.max_iteration, 5) + 1
+        max_iteration_steps = min(app_config.agent.max_iteration, 99) + 1

        # continue to run until there is not any tool call
        function_call_state = True
@@ -72,6 +65,7 @@ class FunctionCallAgentRunner(BaseAgentRunner):
                llm_usage = final_llm_usage_dict["usage"]
                llm_usage.prompt_tokens += usage.prompt_tokens
                llm_usage.completion_tokens += usage.completion_tokens
+                llm_usage.total_tokens += usage.total_tokens
                llm_usage.prompt_price += usage.prompt_price
                llm_usage.completion_price += usage.completion_price
                llm_usage.total_price += usage.total_price
@@ -75,7 +75,7 @@ class AgentConfigManager:
                    strategy=strategy,
                    prompt=agent_prompt_entity,
                    tools=agent_tools,
-                    max_iteration=agent_dict.get("max_iteration", 5),
+                    max_iteration=agent_dict.get("max_iteration", 10),
                )

        return None
@@ -70,7 +70,7 @@ class ModelConfigConverter:
        if not model_mode:
            model_mode = LLMMode.CHAT.value
            if model_schema and model_schema.model_properties.get(ModelPropertyKey.MODE):
-                model_mode = LLMMode.value_of(model_schema.model_properties[ModelPropertyKey.MODE]).value
+                model_mode = LLMMode(model_schema.model_properties[ModelPropertyKey.MODE]).value

        if not model_schema:
            raise ValueError(f"Model {model_name} not exist.")
@@ -109,6 +109,7 @@ class VariableEntity(BaseModel):
    description: str = ""
    type: VariableEntityType
    required: bool = False
+    hide: bool = False
    max_length: Optional[int] = None
    options: Sequence[str] = Field(default_factory=list)
    allowed_file_types: Sequence[FileType] = Field(default_factory=list)
@@ -5,7 +5,7 @@ import uuid
 from collections.abc import Generator, Mapping
 from typing import Any, Literal, Optional, Union, overload

-from flask import Flask, current_app
+from flask import Flask, copy_current_request_context, current_app, has_request_context
 from pydantic import ValidationError
 from sqlalchemy.orm import sessionmaker

@@ -26,12 +26,13 @@ from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.prompt.utils.get_thread_messages_length import get_thread_messages_length
 from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
-from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from core.repositories.sqlalchemy_workflow_execution_repository import SQLAlchemyWorkflowExecutionRepository
+from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
+from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from extensions.ext_database import db
 from factories import file_factory
-from models.account import Account
-from models.model import App, Conversation, EndUser, Message
-from models.workflow import Workflow
+from models import Account, App, Conversation, EndUser, Message, Workflow, WorkflowNodeExecutionTriggeredFrom
+from models.enums import WorkflowRunTriggeredFrom
 from services.conversation_service import ConversationService
 from services.errors.message import MessageNotExistsError

@@ -157,16 +158,30 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            trace_manager=trace_manager,
            workflow_run_id=workflow_run_id,
        )
-        contexts.tenant_id.set(application_generate_entity.app_config.tenant_id)
        contexts.plugin_tool_providers.set({})
        contexts.plugin_tool_providers_lock.set(threading.Lock())

-        # Create workflow node execution repository
+        # Create repositories
+        #
+        # Create session factory
        session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
+        # Create workflow execution(aka workflow run) repository
+        if invoke_from == InvokeFrom.DEBUGGER:
+            workflow_triggered_from = WorkflowRunTriggeredFrom.DEBUGGING
+        else:
+            workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
+        workflow_execution_repository = SQLAlchemyWorkflowExecutionRepository(
+            session_factory=session_factory,
+            user=user,
+            app_id=application_generate_entity.app_config.app_id,
+            triggered_from=workflow_triggered_from,
+        )
+        # Create workflow node execution repository
        workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
            session_factory=session_factory,
-            tenant_id=application_generate_entity.app_config.tenant_id,
+            user=user,
            app_id=application_generate_entity.app_config.app_id,
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
        )

        return self._generate(
@@ -174,6 +189,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            user=user,
            invoke_from=invoke_from,
            application_generate_entity=application_generate_entity,
+            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
            conversation=conversation,
            stream=streaming,
@@ -223,16 +239,26 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
                node_id=node_id, inputs=args["inputs"]
            ),
        )
-        contexts.tenant_id.set(application_generate_entity.app_config.tenant_id)
        contexts.plugin_tool_providers.set({})
        contexts.plugin_tool_providers_lock.set(threading.Lock())

-        # Create workflow node execution repository
+        # Create repositories
+        #
+        # Create session factory
        session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
+        # Create workflow execution(aka workflow run) repository
+        workflow_execution_repository = SQLAlchemyWorkflowExecutionRepository(
+            session_factory=session_factory,
+            user=user,
+            app_id=application_generate_entity.app_config.app_id,
+            triggered_from=WorkflowRunTriggeredFrom.DEBUGGING,
+        )
+        # Create workflow node execution repository
        workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
            session_factory=session_factory,
-            tenant_id=application_generate_entity.app_config.tenant_id,
+            user=user,
            app_id=application_generate_entity.app_config.app_id,
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
        )

        return self._generate(
@@ -240,6 +266,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            user=user,
            invoke_from=InvokeFrom.DEBUGGER,
            application_generate_entity=application_generate_entity,
+            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
            conversation=None,
            stream=streaming,
@@ -287,16 +314,26 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            extras={"auto_generate_conversation_name": False},
            single_loop_run=AdvancedChatAppGenerateEntity.SingleLoopRunEntity(node_id=node_id, inputs=args["inputs"]),
        )
-        contexts.tenant_id.set(application_generate_entity.app_config.tenant_id)
        contexts.plugin_tool_providers.set({})
        contexts.plugin_tool_providers_lock.set(threading.Lock())

-        # Create workflow node execution repository
+        # Create repositories
+        #
+        # Create session factory
        session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
+        # Create workflow execution(aka workflow run) repository
+        workflow_execution_repository = SQLAlchemyWorkflowExecutionRepository(
+            session_factory=session_factory,
+            user=user,
+            app_id=application_generate_entity.app_config.app_id,
+            triggered_from=WorkflowRunTriggeredFrom.DEBUGGING,
+        )
+        # Create workflow node execution repository
        workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
            session_factory=session_factory,
-            tenant_id=application_generate_entity.app_config.tenant_id,
+            user=user,
            app_id=application_generate_entity.app_config.app_id,
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
        )

        return self._generate(
@@ -304,6 +341,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            user=user,
            invoke_from=InvokeFrom.DEBUGGER,
            application_generate_entity=application_generate_entity,
+            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
            conversation=None,
            stream=streaming,
@@ -316,6 +354,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
        user: Union[Account, EndUser],
        invoke_from: InvokeFrom,
        application_generate_entity: AdvancedChatAppGenerateEntity,
+        workflow_execution_repository: WorkflowExecutionRepository,
        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
        conversation: Optional[Conversation] = None,
        stream: bool = True,
@@ -357,18 +396,23 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            message_id=message.id,
        )

-        # new thread
-        worker_thread = threading.Thread(
-            target=self._generate_worker,
-            kwargs={
-                "flask_app": current_app._get_current_object(),  # type: ignore
-                "application_generate_entity": application_generate_entity,
-                "queue_manager": queue_manager,
-                "conversation_id": conversation.id,
-                "message_id": message.id,
-                "context": contextvars.copy_context(),
-            },
-        )
+        # new thread with request context and contextvars
+        context = contextvars.copy_context()
+
+        @copy_current_request_context
+        def worker_with_context():
+            # Run the worker within the copied context
+            return context.run(
+                self._generate_worker,
+                flask_app=current_app._get_current_object(),  # type: ignore
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
+                conversation_id=conversation.id,
+                message_id=message.id,
+                context=context,
+            )
+
+        worker_thread = threading.Thread(target=worker_with_context)

        worker_thread.start()

@@ -380,6 +424,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            conversation=conversation,
            message=message,
            user=user,
+            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
            stream=stream,
        )
@@ -406,8 +451,22 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
        """
        for var, val in context.items():
            var.set(val)
+
+        # FIXME(-LAN-): Save current user before entering new app context
+        from flask import g
+
+        saved_user = None
+        if has_request_context() and hasattr(g, "_login_user"):
+            saved_user = g._login_user
+
        with flask_app.app_context():
            try:
+                # Restore user in new app context
+                if saved_user is not None:
+                    from flask import g
+
+                    g._login_user = saved_user
+
                # get conversation and message
                conversation = self._get_conversation(conversation_id)
                message = self._get_message(message_id)
@@ -452,6 +511,7 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
        conversation: Conversation,
        message: Message,
        user: Union[Account, EndUser],
+        workflow_execution_repository: WorkflowExecutionRepository,
        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
        stream: bool = False,
    ) -> Union[ChatbotAppBlockingResponse, Generator[ChatbotAppStreamResponse, None, None]]:
@@ -475,9 +535,10 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            conversation=conversation,
            message=message,
            user=user,
-            stream=stream,
            dialogue_count=self._dialogue_count,
+            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
+            stream=stream,
        )

        try:
@@ -140,7 +140,7 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
                SystemVariableKey.DIALOGUE_COUNT: self._dialogue_count,
                SystemVariableKey.APP_ID: app_config.app_id,
                SystemVariableKey.WORKFLOW_ID: app_config.workflow_id,
-                SystemVariableKey.WORKFLOW_RUN_ID: self.application_generate_entity.workflow_run_id,
+                SystemVariableKey.WORKFLOW_EXECUTION_ID: self.application_generate_entity.workflow_run_id,
            }

            # init variable pool
@@ -1,4 +1,3 @@
-import json
 import logging
 import time
 from collections.abc import Generator, Mapping
@@ -10,6 +9,7 @@ from sqlalchemy.orm import Session

 from constants.tts_auto_play_timeout import TTS_AUTO_PLAY_TIMEOUT, TTS_AUTO_PLAY_YIELD_CPU_TIME
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
+from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
 from core.app.entities.app_invoke_entities import (
    AdvancedChatAppGenerateEntity,
    InvokeFrom,
@@ -56,25 +56,23 @@ from core.app.entities.task_entities import (
    WorkflowTaskState,
 )
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
-from core.app.task_pipeline.message_cycle_manage import MessageCycleManage
+from core.app.task_pipeline.message_cycle_manager import MessageCycleManager
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.model_runtime.entities.llm_entities import LLMUsage
-from core.model_runtime.utils.encoders import jsonable_encoder
 from core.ops.ops_trace_manager import TraceQueueManager
+from core.workflow.entities.workflow_execution import WorkflowExecutionStatus, WorkflowType
 from core.workflow.enums import SystemVariableKey
 from core.workflow.graph_engine.entities.graph_runtime_state import GraphRuntimeState
 from core.workflow.nodes import NodeType
-from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from core.workflow.workflow_cycle_manager import WorkflowCycleManager
+from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
+from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
+from core.workflow.workflow_cycle_manager import CycleManagerWorkflowInfo, WorkflowCycleManager
 from events.message_event import message_was_created
 from extensions.ext_database import db
 from models import Conversation, EndUser, Message, MessageFile
 from models.account import Account
-from models.enums import CreatedByRole
-from models.workflow import (
-    Workflow,
-    WorkflowRunStatus,
-)
+from models.enums import CreatorUserRole
+from models.workflow import Workflow

 logger = logging.getLogger(__name__)

@@ -94,6 +92,7 @@ class AdvancedChatAppGenerateTaskPipeline:
        user: Union[Account, EndUser],
        stream: bool,
        dialogue_count: int,
+        workflow_execution_repository: WorkflowExecutionRepository,
        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
    ) -> None:
        self._base_task_pipeline = BasedGenerateTaskPipeline(
@@ -105,11 +104,11 @@ class AdvancedChatAppGenerateTaskPipeline:
        if isinstance(user, EndUser):
            self._user_id = user.id
            user_session_id = user.session_id
-            self._created_by_role = CreatedByRole.END_USER
+            self._created_by_role = CreatorUserRole.END_USER
        elif isinstance(user, Account):
            self._user_id = user.id
            user_session_id = user.id
-            self._created_by_role = CreatedByRole.ACCOUNT
+            self._created_by_role = CreatorUserRole.ACCOUNT
        else:
            raise NotImplementedError(f"User type not supported: {type(user)}")

@@ -123,13 +122,24 @@ class AdvancedChatAppGenerateTaskPipeline:
                SystemVariableKey.DIALOGUE_COUNT: dialogue_count,
                SystemVariableKey.APP_ID: application_generate_entity.app_config.app_id,
                SystemVariableKey.WORKFLOW_ID: workflow.id,
-                SystemVariableKey.WORKFLOW_RUN_ID: application_generate_entity.workflow_run_id,
+                SystemVariableKey.WORKFLOW_EXECUTION_ID: application_generate_entity.workflow_run_id,
            },
+            workflow_info=CycleManagerWorkflowInfo(
+                workflow_id=workflow.id,
+                workflow_type=WorkflowType(workflow.type),
+                version=workflow.version,
+                graph_data=workflow.graph_dict,
+            ),
+            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
        )

+        self._workflow_response_converter = WorkflowResponseConverter(
+            application_generate_entity=application_generate_entity,
+        )
+
        self._task_state = WorkflowTaskState()
-        self._message_cycle_manager = MessageCycleManage(
+        self._message_cycle_manager = MessageCycleManager(
            application_generate_entity=application_generate_entity, task_state=self._task_state
        )

@@ -150,7 +160,7 @@ class AdvancedChatAppGenerateTaskPipeline:
        :return:
        """
        # start generate conversation name thread
-        self._conversation_name_generate_thread = self._message_cycle_manager._generate_conversation_name(
+        self._conversation_name_generate_thread = self._message_cycle_manager.generate_conversation_name(
            conversation_id=self._conversation_id, query=self._application_generate_entity.query
        )

@@ -294,19 +304,15 @@ class AdvancedChatAppGenerateTaskPipeline:

                with Session(db.engine, expire_on_commit=False) as session:
                    # init workflow run
-                    workflow_run = self._workflow_cycle_manager._handle_workflow_run_start(
-                        session=session,
-                        workflow_id=self._workflow_id,
-                        user_id=self._user_id,
-                        created_by_role=self._created_by_role,
-                    )
-                    self._workflow_run_id = workflow_run.id
+                    workflow_execution = self._workflow_cycle_manager.handle_workflow_run_start()
+                    self._workflow_run_id = workflow_execution.id_
                    message = self._get_message(session=session)
                    if not message:
                        raise ValueError(f"Message not found: {self._message_id}")
-                    message.workflow_run_id = workflow_run.id
-                    workflow_start_resp = self._workflow_cycle_manager._workflow_start_to_stream_response(
-                        session=session, task_id=self._application_generate_entity.task_id, workflow_run=workflow_run
+                    message.workflow_run_id = workflow_execution.id_
+                    workflow_start_resp = self._workflow_response_converter.workflow_start_to_stream_response(
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_execution=workflow_execution,
                    )
                    session.commit()

@@ -319,13 +325,10 @@ class AdvancedChatAppGenerateTaskPipeline:
                    raise ValueError("workflow run not initialized.")

                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._get_workflow_run(
-                        session=session, workflow_run_id=self._workflow_run_id
+                    workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_retried(
+                        workflow_execution_id=self._workflow_run_id, event=event
                    )
-                    workflow_node_execution = self._workflow_cycle_manager._handle_workflow_node_execution_retried(
-                        workflow_run=workflow_run, event=event
-                    )
-                    node_retry_resp = self._workflow_cycle_manager._workflow_node_retry_to_stream_response(
+                    node_retry_resp = self._workflow_response_converter.workflow_node_retry_to_stream_response(
                        event=event,
                        task_id=self._application_generate_entity.task_id,
                        workflow_node_execution=workflow_node_execution,
@@ -338,20 +341,15 @@ class AdvancedChatAppGenerateTaskPipeline:
                if not self._workflow_run_id:
                    raise ValueError("workflow run not initialized.")

-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._get_workflow_run(
-                        session=session, workflow_run_id=self._workflow_run_id
-                    )
-                    workflow_node_execution = self._workflow_cycle_manager._handle_node_execution_start(
-                        workflow_run=workflow_run, event=event
-                    )
+                workflow_node_execution = self._workflow_cycle_manager.handle_node_execution_start(
+                    workflow_execution_id=self._workflow_run_id, event=event
+                )

-                    node_start_resp = self._workflow_cycle_manager._workflow_node_start_to_stream_response(
-                        event=event,
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_node_execution=workflow_node_execution,
-                    )
-                    session.commit()
+                node_start_resp = self._workflow_response_converter.workflow_node_start_to_stream_response(
+                    event=event,
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_node_execution=workflow_node_execution,
+                )

                if node_start_resp:
                    yield node_start_resp
@@ -359,15 +357,15 @@ class AdvancedChatAppGenerateTaskPipeline:
                # Record files if it's an answer node or end node
                if event.node_type in [NodeType.ANSWER, NodeType.END]:
                    self._recorded_files.extend(
-                        self._workflow_cycle_manager._fetch_files_from_node_outputs(event.outputs or {})
+                        self._workflow_response_converter.fetch_files_from_node_outputs(event.outputs or {})
                    )

                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_node_execution = self._workflow_cycle_manager._handle_workflow_node_execution_success(
+                    workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_success(
                        event=event
                    )

-                    node_finish_resp = self._workflow_cycle_manager._workflow_node_finish_to_stream_response(
+                    node_finish_resp = self._workflow_response_converter.workflow_node_finish_to_stream_response(
                        event=event,
                        task_id=self._application_generate_entity.task_id,
                        workflow_node_execution=workflow_node_execution,
@@ -383,11 +381,11 @@ class AdvancedChatAppGenerateTaskPipeline:
                | QueueNodeInLoopFailedEvent
                | QueueNodeExceptionEvent,
            ):
-                workflow_node_execution = self._workflow_cycle_manager._handle_workflow_node_execution_failed(
+                workflow_node_execution = self._workflow_cycle_manager.handle_workflow_node_execution_failed(
                    event=event
                )

-                node_finish_resp = self._workflow_cycle_manager._workflow_node_finish_to_stream_response(
+                node_finish_resp = self._workflow_response_converter.workflow_node_finish_to_stream_response(
                    event=event,
                    task_id=self._application_generate_entity.task_id,
                    workflow_node_execution=workflow_node_execution,
@@ -399,132 +397,92 @@ class AdvancedChatAppGenerateTaskPipeline:
                if not self._workflow_run_id:
                    raise ValueError("workflow run not initialized.")

-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._get_workflow_run(
-                        session=session, workflow_run_id=self._workflow_run_id
-                    )
-                    parallel_start_resp = (
-                        self._workflow_cycle_manager._workflow_parallel_branch_start_to_stream_response(
-                            session=session,
-                            task_id=self._application_generate_entity.task_id,
-                            workflow_run=workflow_run,
-                            event=event,
-                        )
+                parallel_start_resp = (
+                    self._workflow_response_converter.workflow_parallel_branch_start_to_stream_response(
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_execution_id=self._workflow_run_id,
+                        event=event,
                    )
+                )

                yield parallel_start_resp
            elif isinstance(event, QueueParallelBranchRunSucceededEvent | QueueParallelBranchRunFailedEvent):
                if not self._workflow_run_id:
                    raise ValueError("workflow run not initialized.")

-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._get_workflow_run(
-                        session=session, workflow_run_id=self._workflow_run_id
-                    )
-                    parallel_finish_resp = (
-                        self._workflow_cycle_manager._workflow_parallel_branch_finished_to_stream_response(
-                            session=session,
-                            task_id=self._application_generate_entity.task_id,
-                            workflow_run=workflow_run,
-                            event=event,
-                        )
+                parallel_finish_resp = (
+                    self._workflow_response_converter.workflow_parallel_branch_finished_to_stream_response(
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_execution_id=self._workflow_run_id,
+                        event=event,
                    )
+                )

                yield parallel_finish_resp
            elif isinstance(event, QueueIterationStartEvent):
                if not self._workflow_run_id:
                    raise ValueError("workflow run not initialized.")

-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._get_workflow_run(
-                        session=session, workflow_run_id=self._workflow_run_id
-                    )
-                    iter_start_resp = self._workflow_cycle_manager._workflow_iteration_start_to_stream_response(
-                        session=session,
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_run=workflow_run,
-                        event=event,
-                    )
+                iter_start_resp = self._workflow_response_converter.workflow_iteration_start_to_stream_response(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_execution_id=self._workflow_run_id,
+                    event=event,
+                )

                yield iter_start_resp
            elif isinstance(event, QueueIterationNextEvent):
                if not self._workflow_run_id:
                    raise ValueError("workflow run not initialized.")

-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._get_workflow_run(
-                        session=session, workflow_run_id=self._workflow_run_id
-                    )
-                    iter_next_resp = self._workflow_cycle_manager._workflow_iteration_next_to_stream_response(
-                        session=session,
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_run=workflow_run,
-                        event=event,
-                    )
+                iter_next_resp = self._workflow_response_converter.workflow_iteration_next_to_stream_response(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_execution_id=self._workflow_run_id,
+                    event=event,
+                )

                yield iter_next_resp
            elif isinstance(event, QueueIterationCompletedEvent):
                if not self._workflow_run_id:
                    raise ValueError("workflow run not initialized.")

-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._get_workflow_run(
-                        session=session, workflow_run_id=self._workflow_run_id
-                    )
-                    iter_finish_resp = self._workflow_cycle_manager._workflow_iteration_completed_to_stream_response(
-                        session=session,
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_run=workflow_run,
-                        event=event,
-                    )
+                iter_finish_resp = self._workflow_response_converter.workflow_iteration_completed_to_stream_response(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_execution_id=self._workflow_run_id,
+                    event=event,
+                )

                yield iter_finish_resp
            elif isinstance(event, QueueLoopStartEvent):
                if not self._workflow_run_id:
                    raise ValueError("workflow run not initialized.")

-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._get_workflow_run(
-                        session=session, workflow_run_id=self._workflow_run_id
-                    )
-                    loop_start_resp = self._workflow_cycle_manager._workflow_loop_start_to_stream_response(
-                        session=session,
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_run=workflow_run,
-                        event=event,
-                    )
+                loop_start_resp = self._workflow_response_converter.workflow_loop_start_to_stream_response(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_execution_id=self._workflow_run_id,
+                    event=event,
+                )

                yield loop_start_resp
            elif isinstance(event, QueueLoopNextEvent):
                if not self._workflow_run_id:
                    raise ValueError("workflow run not initialized.")

-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._get_workflow_run(
-                        session=session, workflow_run_id=self._workflow_run_id
-                    )
-                    loop_next_resp = self._workflow_cycle_manager._workflow_loop_next_to_stream_response(
-                        session=session,
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_run=workflow_run,
-                        event=event,
-                    )
+                loop_next_resp = self._workflow_response_converter.workflow_loop_next_to_stream_response(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_execution_id=self._workflow_run_id,
+                    event=event,
+                )

                yield loop_next_resp
            elif isinstance(event, QueueLoopCompletedEvent):
                if not self._workflow_run_id:
                    raise ValueError("workflow run not initialized.")

-                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._get_workflow_run(
-                        session=session, workflow_run_id=self._workflow_run_id
-                    )
-                    loop_finish_resp = self._workflow_cycle_manager._workflow_loop_completed_to_stream_response(
-                        session=session,
-                        task_id=self._application_generate_entity.task_id,
-                        workflow_run=workflow_run,
-                        event=event,
-                    )
+                loop_finish_resp = self._workflow_response_converter.workflow_loop_completed_to_stream_response(
+                    task_id=self._application_generate_entity.task_id,
+                    workflow_execution_id=self._workflow_run_id,
+                    event=event,
+                )

                yield loop_finish_resp
            elif isinstance(event, QueueWorkflowSucceededEvent):
@@ -535,10 +493,8 @@ class AdvancedChatAppGenerateTaskPipeline:
                    raise ValueError("workflow run not initialized.")

                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._handle_workflow_run_success(
-                        session=session,
+                    workflow_execution = self._workflow_cycle_manager.handle_workflow_run_success(
                        workflow_run_id=self._workflow_run_id,
-                        start_at=graph_runtime_state.start_at,
                        total_tokens=graph_runtime_state.total_tokens,
                        total_steps=graph_runtime_state.node_run_steps,
                        outputs=event.outputs,
@@ -546,10 +502,11 @@ class AdvancedChatAppGenerateTaskPipeline:
                        trace_manager=trace_manager,
                    )

-                    workflow_finish_resp = self._workflow_cycle_manager._workflow_finish_to_stream_response(
-                        session=session, task_id=self._application_generate_entity.task_id, workflow_run=workflow_run
+                    workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
+                        session=session,
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_execution=workflow_execution,
                    )
-                    session.commit()

                yield workflow_finish_resp
                self._base_task_pipeline._queue_manager.publish(
@@ -562,10 +519,8 @@ class AdvancedChatAppGenerateTaskPipeline:
                    raise ValueError("graph runtime state not initialized.")

                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._handle_workflow_run_partial_success(
-                        session=session,
+                    workflow_execution = self._workflow_cycle_manager.handle_workflow_run_partial_success(
                        workflow_run_id=self._workflow_run_id,
-                        start_at=graph_runtime_state.start_at,
                        total_tokens=graph_runtime_state.total_tokens,
                        total_steps=graph_runtime_state.node_run_steps,
                        outputs=event.outputs,
@@ -573,10 +528,11 @@ class AdvancedChatAppGenerateTaskPipeline:
                        conversation_id=None,
                        trace_manager=trace_manager,
                    )
-                    workflow_finish_resp = self._workflow_cycle_manager._workflow_finish_to_stream_response(
-                        session=session, task_id=self._application_generate_entity.task_id, workflow_run=workflow_run
+                    workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
+                        session=session,
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_execution=workflow_execution,
                    )
-                    session.commit()

                yield workflow_finish_resp
                self._base_task_pipeline._queue_manager.publish(
@@ -589,26 +545,25 @@ class AdvancedChatAppGenerateTaskPipeline:
                    raise ValueError("graph runtime state not initialized.")

                with Session(db.engine, expire_on_commit=False) as session:
-                    workflow_run = self._workflow_cycle_manager._handle_workflow_run_failed(
-                        session=session,
+                    workflow_execution = self._workflow_cycle_manager.handle_workflow_run_failed(
                        workflow_run_id=self._workflow_run_id,
-                        start_at=graph_runtime_state.start_at,
                        total_tokens=graph_runtime_state.total_tokens,
                        total_steps=graph_runtime_state.node_run_steps,
-                        status=WorkflowRunStatus.FAILED,
-                        error=event.error,
+                        status=WorkflowExecutionStatus.FAILED,
+                        error_message=event.error,
                        conversation_id=self._conversation_id,
                        trace_manager=trace_manager,
                        exceptions_count=event.exceptions_count,
                    )
-                    workflow_finish_resp = self._workflow_cycle_manager._workflow_finish_to_stream_response(
-                        session=session, task_id=self._application_generate_entity.task_id, workflow_run=workflow_run
+                    workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
+                        session=session,
+                        task_id=self._application_generate_entity.task_id,
+                        workflow_execution=workflow_execution,
                    )
-                    err_event = QueueErrorEvent(error=ValueError(f"Run failed: {workflow_run.error}"))
+                    err_event = QueueErrorEvent(error=ValueError(f"Run failed: {workflow_execution.error_message}"))
                    err = self._base_task_pipeline._handle_error(
                        event=err_event, session=session, message_id=self._message_id
                    )
-                    session.commit()

                yield workflow_finish_resp
                yield self._base_task_pipeline._error_to_stream_response(err)
@@ -616,21 +571,19 @@ class AdvancedChatAppGenerateTaskPipeline:
            elif isinstance(event, QueueStopEvent):
                if self._workflow_run_id and graph_runtime_state:
                    with Session(db.engine, expire_on_commit=False) as session:
-                        workflow_run = self._workflow_cycle_manager._handle_workflow_run_failed(
-                            session=session,
+                        workflow_execution = self._workflow_cycle_manager.handle_workflow_run_failed(
                            workflow_run_id=self._workflow_run_id,
-                            start_at=graph_runtime_state.start_at,
                            total_tokens=graph_runtime_state.total_tokens,
                            total_steps=graph_runtime_state.node_run_steps,
-                            status=WorkflowRunStatus.STOPPED,
-                            error=event.get_stop_reason(),
+                            status=WorkflowExecutionStatus.STOPPED,
+                            error_message=event.get_stop_reason(),
                            conversation_id=self._conversation_id,
                            trace_manager=trace_manager,
                        )
-                        workflow_finish_resp = self._workflow_cycle_manager._workflow_finish_to_stream_response(
+                        workflow_finish_resp = self._workflow_response_converter.workflow_finish_to_stream_response(
                            session=session,
                            task_id=self._application_generate_entity.task_id,
-                            workflow_run=workflow_run,
+                            workflow_execution=workflow_execution,
                        )
                        # Save message
                        self._save_message(session=session, graph_runtime_state=graph_runtime_state)
@@ -650,22 +603,18 @@ class AdvancedChatAppGenerateTaskPipeline:
                yield self._message_end_to_stream_response()
                break
            elif isinstance(event, QueueRetrieverResourcesEvent):
-                self._message_cycle_manager._handle_retriever_resources(event)
+                self._message_cycle_manager.handle_retriever_resources(event)

                with Session(db.engine, expire_on_commit=False) as session:
                    message = self._get_message(session=session)
-                    message.message_metadata = (
-                        json.dumps(jsonable_encoder(self._task_state.metadata)) if self._task_state.metadata else None
-                    )
+                    message.message_metadata = self._task_state.metadata.model_dump_json()
                    session.commit()
            elif isinstance(event, QueueAnnotationReplyEvent):
-                self._message_cycle_manager._handle_annotation_reply(event)
+                self._message_cycle_manager.handle_annotation_reply(event)

                with Session(db.engine, expire_on_commit=False) as session:
                    message = self._get_message(session=session)
-                    message.message_metadata = (
-                        json.dumps(jsonable_encoder(self._task_state.metadata)) if self._task_state.metadata else None
-                    )
+                    message.message_metadata = self._task_state.metadata.model_dump_json()
                    session.commit()
            elif isinstance(event, QueueTextChunkEvent):
                delta_text = event.text
@@ -682,12 +631,12 @@ class AdvancedChatAppGenerateTaskPipeline:
                    tts_publisher.publish(queue_message)

                self._task_state.answer += delta_text
-                yield self._message_cycle_manager._message_to_stream_response(
+                yield self._message_cycle_manager.message_to_stream_response(
                    answer=delta_text, message_id=self._message_id, from_variable_selector=event.from_variable_selector
                )
            elif isinstance(event, QueueMessageReplaceEvent):
                # published by moderation
-                yield self._message_cycle_manager._message_replace_to_stream_response(
+                yield self._message_cycle_manager.message_replace_to_stream_response(
                    answer=event.text, reason=event.reason
                )
            elif isinstance(event, QueueAdvancedChatMessageEndEvent):
@@ -699,7 +648,7 @@ class AdvancedChatAppGenerateTaskPipeline:
                )
                if output_moderation_answer:
                    self._task_state.answer = output_moderation_answer
-                    yield self._message_cycle_manager._message_replace_to_stream_response(
+                    yield self._message_cycle_manager.message_replace_to_stream_response(
                        answer=output_moderation_answer,
                        reason=QueueMessageReplaceEvent.MessageReplaceReason.OUTPUT_MODERATION,
                    )
@@ -711,7 +660,7 @@ class AdvancedChatAppGenerateTaskPipeline:

                yield self._message_end_to_stream_response()
            elif isinstance(event, QueueAgentLogEvent):
-                yield self._workflow_cycle_manager._handle_agent_log(
+                yield self._workflow_response_converter.handle_agent_log(
                    task_id=self._application_generate_entity.task_id, event=event
                )
            else:
@@ -728,9 +677,7 @@ class AdvancedChatAppGenerateTaskPipeline:
        message = self._get_message(session=session)
        message.answer = self._task_state.answer
        message.provider_response_latency = time.perf_counter() - self._base_task_pipeline._start_at
-        message.message_metadata = (
-            json.dumps(jsonable_encoder(self._task_state.metadata)) if self._task_state.metadata else None
-        )
+        message.message_metadata = self._task_state.metadata.model_dump_json()
        message_files = [
            MessageFile(
                message_id=message.id,
@@ -739,9 +686,9 @@ class AdvancedChatAppGenerateTaskPipeline:
                url=file["remote_url"],
                belongs_to="assistant",
                upload_file_id=file["related_id"],
-                created_by_role=CreatedByRole.ACCOUNT
+                created_by_role=CreatorUserRole.ACCOUNT
                if message.invoke_from in {InvokeFrom.EXPLORE, InvokeFrom.DEBUGGER}
-                else CreatedByRole.END_USER,
+                else CreatorUserRole.END_USER,
                created_by=message.from_account_id or message.from_end_user_id or "",
            )
            for file in self._recorded_files
@@ -758,9 +705,9 @@ class AdvancedChatAppGenerateTaskPipeline:
            message.answer_price_unit = usage.completion_price_unit
            message.total_price = usage.total_price
            message.currency = usage.currency
-            self._task_state.metadata["usage"] = jsonable_encoder(usage)
+            self._task_state.metadata.usage = usage
        else:
-            self._task_state.metadata["usage"] = jsonable_encoder(LLMUsage.empty_usage())
+            self._task_state.metadata.usage = LLMUsage.empty_usage()
        message_was_created.send(
            message,
            application_generate_entity=self._application_generate_entity,
@@ -771,18 +718,16 @@ class AdvancedChatAppGenerateTaskPipeline:
        Message end to stream response.
        :return:
        """
-        extras = {}
-        if self._task_state.metadata:
-            extras["metadata"] = self._task_state.metadata.copy()
+        extras = self._task_state.metadata.model_dump()

-            if "annotation_reply" in extras["metadata"]:
-                del extras["metadata"]["annotation_reply"]
+        if self._task_state.metadata.annotation_reply:
+            del extras["annotation_reply"]

        return MessageEndStreamResponse(
            task_id=self._application_generate_entity.task_id,
            id=self._message_id,
            files=self._recorded_files,
-            metadata=extras.get("metadata", {}),
+            metadata=extras,
        )

    def _handle_output_moderation_chunk(self, text: str) -> bool:
@@ -5,7 +5,7 @@ import uuid
 from collections.abc import Generator, Mapping
 from typing import Any, Literal, Union, overload

-from flask import Flask, current_app
+from flask import Flask, copy_current_request_context, current_app, has_request_context
 from pydantic import ValidationError

 from configs import dify_config
@@ -179,18 +179,23 @@ class AgentChatAppGenerator(MessageBasedAppGenerator):
            message_id=message.id,
        )

-        # new thread
-        worker_thread = threading.Thread(
-            target=self._generate_worker,
-            kwargs={
-                "flask_app": current_app._get_current_object(),  # type: ignore
-                "context": contextvars.copy_context(),
-                "application_generate_entity": application_generate_entity,
-                "queue_manager": queue_manager,
-                "conversation_id": conversation.id,
-                "message_id": message.id,
-            },
-        )
+        # new thread with request context and contextvars
+        context = contextvars.copy_context()
+
+        @copy_current_request_context
+        def worker_with_context():
+            # Run the worker within the copied context
+            return context.run(
+                self._generate_worker,
+                flask_app=current_app._get_current_object(),  # type: ignore
+                context=context,
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
+                conversation_id=conversation.id,
+                message_id=message.id,
+            )
+
+        worker_thread = threading.Thread(target=worker_with_context)

        worker_thread.start()

@@ -227,8 +232,21 @@ class AgentChatAppGenerator(MessageBasedAppGenerator):
        for var, val in context.items():
            var.set(val)

+        # FIXME(-LAN-): Save current user before entering new app context
+        from flask import g
+
+        saved_user = None
+        if has_request_context() and hasattr(g, "_login_user"):
+            saved_user = g._login_user
+
        with flask_app.app_context():
            try:
+                # Restore user in new app context
+                if saved_user is not None:
+                    from flask import g
+
+                    g._login_user = saved_user
+
                # get conversation and message
                conversation = self._get_conversation(conversation_id)
                message = self._get_message(message_id)
@@ -4,7 +4,7 @@ import uuid
 from collections.abc import Generator, Mapping
 from typing import Any, Literal, Union, overload

-from flask import Flask, current_app
+from flask import Flask, copy_current_request_context, current_app
 from pydantic import ValidationError

 from configs import dify_config
@@ -170,17 +170,18 @@ class ChatAppGenerator(MessageBasedAppGenerator):
            message_id=message.id,
        )

-        # new thread
-        worker_thread = threading.Thread(
-            target=self._generate_worker,
-            kwargs={
-                "flask_app": current_app._get_current_object(),  # type: ignore
-                "application_generate_entity": application_generate_entity,
-                "queue_manager": queue_manager,
-                "conversation_id": conversation.id,
-                "message_id": message.id,
-            },
-        )
+        # new thread with request context
+        @copy_current_request_context
+        def worker_with_context():
+            return self._generate_worker(
+                flask_app=current_app._get_current_object(),  # type: ignore
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
+                conversation_id=conversation.id,
+                message_id=message.id,
+            )
+
+        worker_thread = threading.Thread(target=worker_with_context)

        worker_thread.start()

@@ -0,0 +1,561 @@
+import time
+from collections.abc import Mapping, Sequence
+from datetime import UTC, datetime
+from typing import Any, Optional, Union, cast
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from core.app.entities.app_invoke_entities import AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity
+from core.app.entities.queue_entities import (
+    QueueAgentLogEvent,
+    QueueIterationCompletedEvent,
+    QueueIterationNextEvent,
+    QueueIterationStartEvent,
+    QueueLoopCompletedEvent,
+    QueueLoopNextEvent,
+    QueueLoopStartEvent,
+    QueueNodeExceptionEvent,
+    QueueNodeFailedEvent,
+    QueueNodeInIterationFailedEvent,
+    QueueNodeInLoopFailedEvent,
+    QueueNodeRetryEvent,
+    QueueNodeStartedEvent,
+    QueueNodeSucceededEvent,
+    QueueParallelBranchRunFailedEvent,
+    QueueParallelBranchRunStartedEvent,
+    QueueParallelBranchRunSucceededEvent,
+)
+from core.app.entities.task_entities import (
+    AgentLogStreamResponse,
+    IterationNodeCompletedStreamResponse,
+    IterationNodeNextStreamResponse,
+    IterationNodeStartStreamResponse,
+    LoopNodeCompletedStreamResponse,
+    LoopNodeNextStreamResponse,
+    LoopNodeStartStreamResponse,
+    NodeFinishStreamResponse,
+    NodeRetryStreamResponse,
+    NodeStartStreamResponse,
+    ParallelBranchFinishedStreamResponse,
+    ParallelBranchStartStreamResponse,
+    WorkflowFinishStreamResponse,
+    WorkflowStartStreamResponse,
+)
+from core.file import FILE_MODEL_IDENTITY, File
+from core.tools.tool_manager import ToolManager
+from core.workflow.entities.workflow_execution import WorkflowExecution
+from core.workflow.entities.workflow_node_execution import WorkflowNodeExecution, WorkflowNodeExecutionStatus
+from core.workflow.nodes import NodeType
+from core.workflow.nodes.tool.entities import ToolNodeData
+from models import (
+    Account,
+    CreatorUserRole,
+    EndUser,
+    WorkflowRun,
+)
+
+
+class WorkflowResponseConverter:
+    def __init__(
+        self,
+        *,
+        application_generate_entity: Union[AdvancedChatAppGenerateEntity, WorkflowAppGenerateEntity],
+    ) -> None:
+        self._application_generate_entity = application_generate_entity
+
+    def workflow_start_to_stream_response(
+        self,
+        *,
+        task_id: str,
+        workflow_execution: WorkflowExecution,
+    ) -> WorkflowStartStreamResponse:
+        return WorkflowStartStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_execution.id_,
+            data=WorkflowStartStreamResponse.Data(
+                id=workflow_execution.id_,
+                workflow_id=workflow_execution.workflow_id,
+                inputs=workflow_execution.inputs,
+                created_at=int(workflow_execution.started_at.timestamp()),
+            ),
+        )
+
+    def workflow_finish_to_stream_response(
+        self,
+        *,
+        session: Session,
+        task_id: str,
+        workflow_execution: WorkflowExecution,
+    ) -> WorkflowFinishStreamResponse:
+        created_by = None
+        workflow_run = session.scalar(select(WorkflowRun).where(WorkflowRun.id == workflow_execution.id_))
+        assert workflow_run is not None
+        if workflow_run.created_by_role == CreatorUserRole.ACCOUNT:
+            stmt = select(Account).where(Account.id == workflow_run.created_by)
+            account = session.scalar(stmt)
+            if account:
+                created_by = {
+                    "id": account.id,
+                    "name": account.name,
+                    "email": account.email,
+                }
+        elif workflow_run.created_by_role == CreatorUserRole.END_USER:
+            stmt = select(EndUser).where(EndUser.id == workflow_run.created_by)
+            end_user = session.scalar(stmt)
+            if end_user:
+                created_by = {
+                    "id": end_user.id,
+                    "user": end_user.session_id,
+                }
+        else:
+            raise NotImplementedError(f"unknown created_by_role: {workflow_run.created_by_role}")
+
+        # Handle the case where finished_at is None by using current time as default
+        finished_at_timestamp = (
+            int(workflow_execution.finished_at.timestamp())
+            if workflow_execution.finished_at
+            else int(datetime.now(UTC).timestamp())
+        )
+
+        return WorkflowFinishStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_execution.id_,
+            data=WorkflowFinishStreamResponse.Data(
+                id=workflow_execution.id_,
+                workflow_id=workflow_execution.workflow_id,
+                status=workflow_execution.status,
+                outputs=workflow_execution.outputs,
+                error=workflow_execution.error_message,
+                elapsed_time=workflow_execution.elapsed_time,
+                total_tokens=workflow_execution.total_tokens,
+                total_steps=workflow_execution.total_steps,
+                created_by=created_by,
+                created_at=int(workflow_execution.started_at.timestamp()),
+                finished_at=finished_at_timestamp,
+                files=self.fetch_files_from_node_outputs(workflow_execution.outputs),
+                exceptions_count=workflow_execution.exceptions_count,
+            ),
+        )
+
+    def workflow_node_start_to_stream_response(
+        self,
+        *,
+        event: QueueNodeStartedEvent,
+        task_id: str,
+        workflow_node_execution: WorkflowNodeExecution,
+    ) -> Optional[NodeStartStreamResponse]:
+        if workflow_node_execution.node_type in {NodeType.ITERATION, NodeType.LOOP}:
+            return None
+        if not workflow_node_execution.workflow_execution_id:
+            return None
+
+        response = NodeStartStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_node_execution.workflow_execution_id,
+            data=NodeStartStreamResponse.Data(
+                id=workflow_node_execution.id,
+                node_id=workflow_node_execution.node_id,
+                node_type=workflow_node_execution.node_type,
+                title=workflow_node_execution.title,
+                index=workflow_node_execution.index,
+                predecessor_node_id=workflow_node_execution.predecessor_node_id,
+                inputs=workflow_node_execution.inputs,
+                created_at=int(workflow_node_execution.created_at.timestamp()),
+                parallel_id=event.parallel_id,
+                parallel_start_node_id=event.parallel_start_node_id,
+                parent_parallel_id=event.parent_parallel_id,
+                parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                iteration_id=event.in_iteration_id,
+                loop_id=event.in_loop_id,
+                parallel_run_id=event.parallel_mode_run_id,
+                agent_strategy=event.agent_strategy,
+            ),
+        )
+
+        # extras logic
+        if event.node_type == NodeType.TOOL:
+            node_data = cast(ToolNodeData, event.node_data)
+            response.data.extras["icon"] = ToolManager.get_tool_icon(
+                tenant_id=self._application_generate_entity.app_config.tenant_id,
+                provider_type=node_data.provider_type,
+                provider_id=node_data.provider_id,
+            )
+
+        return response
+
+    def workflow_node_finish_to_stream_response(
+        self,
+        *,
+        event: QueueNodeSucceededEvent
+        | QueueNodeFailedEvent
+        | QueueNodeInIterationFailedEvent
+        | QueueNodeInLoopFailedEvent
+        | QueueNodeExceptionEvent,
+        task_id: str,
+        workflow_node_execution: WorkflowNodeExecution,
+    ) -> Optional[NodeFinishStreamResponse]:
+        if workflow_node_execution.node_type in {NodeType.ITERATION, NodeType.LOOP}:
+            return None
+        if not workflow_node_execution.workflow_execution_id:
+            return None
+        if not workflow_node_execution.finished_at:
+            return None
+
+        return NodeFinishStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_node_execution.workflow_execution_id,
+            data=NodeFinishStreamResponse.Data(
+                id=workflow_node_execution.id,
+                node_id=workflow_node_execution.node_id,
+                node_type=workflow_node_execution.node_type,
+                index=workflow_node_execution.index,
+                title=workflow_node_execution.title,
+                predecessor_node_id=workflow_node_execution.predecessor_node_id,
+                inputs=workflow_node_execution.inputs,
+                process_data=workflow_node_execution.process_data,
+                outputs=workflow_node_execution.outputs,
+                status=workflow_node_execution.status,
+                error=workflow_node_execution.error,
+                elapsed_time=workflow_node_execution.elapsed_time,
+                execution_metadata=workflow_node_execution.metadata,
+                created_at=int(workflow_node_execution.created_at.timestamp()),
+                finished_at=int(workflow_node_execution.finished_at.timestamp()),
+                files=self.fetch_files_from_node_outputs(workflow_node_execution.outputs or {}),
+                parallel_id=event.parallel_id,
+                parallel_start_node_id=event.parallel_start_node_id,
+                parent_parallel_id=event.parent_parallel_id,
+                parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                iteration_id=event.in_iteration_id,
+                loop_id=event.in_loop_id,
+            ),
+        )
+
+    def workflow_node_retry_to_stream_response(
+        self,
+        *,
+        event: QueueNodeRetryEvent,
+        task_id: str,
+        workflow_node_execution: WorkflowNodeExecution,
+    ) -> Optional[Union[NodeRetryStreamResponse, NodeFinishStreamResponse]]:
+        if workflow_node_execution.node_type in {NodeType.ITERATION, NodeType.LOOP}:
+            return None
+        if not workflow_node_execution.workflow_execution_id:
+            return None
+        if not workflow_node_execution.finished_at:
+            return None
+
+        return NodeRetryStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_node_execution.workflow_execution_id,
+            data=NodeRetryStreamResponse.Data(
+                id=workflow_node_execution.id,
+                node_id=workflow_node_execution.node_id,
+                node_type=workflow_node_execution.node_type,
+                index=workflow_node_execution.index,
+                title=workflow_node_execution.title,
+                predecessor_node_id=workflow_node_execution.predecessor_node_id,
+                inputs=workflow_node_execution.inputs,
+                process_data=workflow_node_execution.process_data,
+                outputs=workflow_node_execution.outputs,
+                status=workflow_node_execution.status,
+                error=workflow_node_execution.error,
+                elapsed_time=workflow_node_execution.elapsed_time,
+                execution_metadata=workflow_node_execution.metadata,
+                created_at=int(workflow_node_execution.created_at.timestamp()),
+                finished_at=int(workflow_node_execution.finished_at.timestamp()),
+                files=self.fetch_files_from_node_outputs(workflow_node_execution.outputs or {}),
+                parallel_id=event.parallel_id,
+                parallel_start_node_id=event.parallel_start_node_id,
+                parent_parallel_id=event.parent_parallel_id,
+                parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                iteration_id=event.in_iteration_id,
+                loop_id=event.in_loop_id,
+                retry_index=event.retry_index,
+            ),
+        )
+
+    def workflow_parallel_branch_start_to_stream_response(
+        self,
+        *,
+        task_id: str,
+        workflow_execution_id: str,
+        event: QueueParallelBranchRunStartedEvent,
+    ) -> ParallelBranchStartStreamResponse:
+        return ParallelBranchStartStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_execution_id,
+            data=ParallelBranchStartStreamResponse.Data(
+                parallel_id=event.parallel_id,
+                parallel_branch_id=event.parallel_start_node_id,
+                parent_parallel_id=event.parent_parallel_id,
+                parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                iteration_id=event.in_iteration_id,
+                loop_id=event.in_loop_id,
+                created_at=int(time.time()),
+            ),
+        )
+
+    def workflow_parallel_branch_finished_to_stream_response(
+        self,
+        *,
+        task_id: str,
+        workflow_execution_id: str,
+        event: QueueParallelBranchRunSucceededEvent | QueueParallelBranchRunFailedEvent,
+    ) -> ParallelBranchFinishedStreamResponse:
+        return ParallelBranchFinishedStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_execution_id,
+            data=ParallelBranchFinishedStreamResponse.Data(
+                parallel_id=event.parallel_id,
+                parallel_branch_id=event.parallel_start_node_id,
+                parent_parallel_id=event.parent_parallel_id,
+                parent_parallel_start_node_id=event.parent_parallel_start_node_id,
+                iteration_id=event.in_iteration_id,
+                loop_id=event.in_loop_id,
+                status="succeeded" if isinstance(event, QueueParallelBranchRunSucceededEvent) else "failed",
+                error=event.error if isinstance(event, QueueParallelBranchRunFailedEvent) else None,
+                created_at=int(time.time()),
+            ),
+        )
+
+    def workflow_iteration_start_to_stream_response(
+        self,
+        *,
+        task_id: str,
+        workflow_execution_id: str,
+        event: QueueIterationStartEvent,
+    ) -> IterationNodeStartStreamResponse:
+        return IterationNodeStartStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_execution_id,
+            data=IterationNodeStartStreamResponse.Data(
+                id=event.node_id,
+                node_id=event.node_id,
+                node_type=event.node_type.value,
+                title=event.node_data.title,
+                created_at=int(time.time()),
+                extras={},
+                inputs=event.inputs or {},
+                metadata=event.metadata or {},
+                parallel_id=event.parallel_id,
+                parallel_start_node_id=event.parallel_start_node_id,
+            ),
+        )
+
+    def workflow_iteration_next_to_stream_response(
+        self,
+        *,
+        task_id: str,
+        workflow_execution_id: str,
+        event: QueueIterationNextEvent,
+    ) -> IterationNodeNextStreamResponse:
+        return IterationNodeNextStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_execution_id,
+            data=IterationNodeNextStreamResponse.Data(
+                id=event.node_id,
+                node_id=event.node_id,
+                node_type=event.node_type.value,
+                title=event.node_data.title,
+                index=event.index,
+                pre_iteration_output=event.output,
+                created_at=int(time.time()),
+                extras={},
+                parallel_id=event.parallel_id,
+                parallel_start_node_id=event.parallel_start_node_id,
+                parallel_mode_run_id=event.parallel_mode_run_id,
+                duration=event.duration,
+            ),
+        )
+
+    def workflow_iteration_completed_to_stream_response(
+        self,
+        *,
+        task_id: str,
+        workflow_execution_id: str,
+        event: QueueIterationCompletedEvent,
+    ) -> IterationNodeCompletedStreamResponse:
+        return IterationNodeCompletedStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_execution_id,
+            data=IterationNodeCompletedStreamResponse.Data(
+                id=event.node_id,
+                node_id=event.node_id,
+                node_type=event.node_type.value,
+                title=event.node_data.title,
+                outputs=event.outputs,
+                created_at=int(time.time()),
+                extras={},
+                inputs=event.inputs or {},
+                status=WorkflowNodeExecutionStatus.SUCCEEDED
+                if event.error is None
+                else WorkflowNodeExecutionStatus.FAILED,
+                error=None,
+                elapsed_time=(datetime.now(UTC).replace(tzinfo=None) - event.start_at).total_seconds(),
+                total_tokens=event.metadata.get("total_tokens", 0) if event.metadata else 0,
+                execution_metadata=event.metadata,
+                finished_at=int(time.time()),
+                steps=event.steps,
+                parallel_id=event.parallel_id,
+                parallel_start_node_id=event.parallel_start_node_id,
+            ),
+        )
+
+    def workflow_loop_start_to_stream_response(
+        self, *, task_id: str, workflow_execution_id: str, event: QueueLoopStartEvent
+    ) -> LoopNodeStartStreamResponse:
+        return LoopNodeStartStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_execution_id,
+            data=LoopNodeStartStreamResponse.Data(
+                id=event.node_id,
+                node_id=event.node_id,
+                node_type=event.node_type.value,
+                title=event.node_data.title,
+                created_at=int(time.time()),
+                extras={},
+                inputs=event.inputs or {},
+                metadata=event.metadata or {},
+                parallel_id=event.parallel_id,
+                parallel_start_node_id=event.parallel_start_node_id,
+            ),
+        )
+
+    def workflow_loop_next_to_stream_response(
+        self,
+        *,
+        task_id: str,
+        workflow_execution_id: str,
+        event: QueueLoopNextEvent,
+    ) -> LoopNodeNextStreamResponse:
+        return LoopNodeNextStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_execution_id,
+            data=LoopNodeNextStreamResponse.Data(
+                id=event.node_id,
+                node_id=event.node_id,
+                node_type=event.node_type.value,
+                title=event.node_data.title,
+                index=event.index,
+                pre_loop_output=event.output,
+                created_at=int(time.time()),
+                extras={},
+                parallel_id=event.parallel_id,
+                parallel_start_node_id=event.parallel_start_node_id,
+                parallel_mode_run_id=event.parallel_mode_run_id,
+                duration=event.duration,
+            ),
+        )
+
+    def workflow_loop_completed_to_stream_response(
+        self,
+        *,
+        task_id: str,
+        workflow_execution_id: str,
+        event: QueueLoopCompletedEvent,
+    ) -> LoopNodeCompletedStreamResponse:
+        return LoopNodeCompletedStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_execution_id,
+            data=LoopNodeCompletedStreamResponse.Data(
+                id=event.node_id,
+                node_id=event.node_id,
+                node_type=event.node_type.value,
+                title=event.node_data.title,
+                outputs=event.outputs,
+                created_at=int(time.time()),
+                extras={},
+                inputs=event.inputs or {},
+                status=WorkflowNodeExecutionStatus.SUCCEEDED
+                if event.error is None
+                else WorkflowNodeExecutionStatus.FAILED,
+                error=None,
+                elapsed_time=(datetime.now(UTC).replace(tzinfo=None) - event.start_at).total_seconds(),
+                total_tokens=event.metadata.get("total_tokens", 0) if event.metadata else 0,
+                execution_metadata=event.metadata,
+                finished_at=int(time.time()),
+                steps=event.steps,
+                parallel_id=event.parallel_id,
+                parallel_start_node_id=event.parallel_start_node_id,
+            ),
+        )
+
+    def fetch_files_from_node_outputs(self, outputs_dict: Mapping[str, Any] | None) -> Sequence[Mapping[str, Any]]:
+        """
+        Fetch files from node outputs
+        :param outputs_dict: node outputs dict
+        :return:
+        """
+        if not outputs_dict:
+            return []
+
+        files = [self._fetch_files_from_variable_value(output_value) for output_value in outputs_dict.values()]
+        # Remove None
+        files = [file for file in files if file]
+        # Flatten list
+        # Flatten the list of sequences into a single list of mappings
+        flattened_files = [file for sublist in files if sublist for file in sublist]
+
+        # Convert to tuple to match Sequence type
+        return tuple(flattened_files)
+
+    def _fetch_files_from_variable_value(self, value: Union[dict, list]) -> Sequence[Mapping[str, Any]]:
+        """
+        Fetch files from variable value
+        :param value: variable value
+        :return:
+        """
+        if not value:
+            return []
+
+        files = []
+        if isinstance(value, list):
+            for item in value:
+                file = self._get_file_var_from_value(item)
+                if file:
+                    files.append(file)
+        elif isinstance(value, dict):
+            file = self._get_file_var_from_value(value)
+            if file:
+                files.append(file)
+
+        return files
+
+    def _get_file_var_from_value(self, value: Union[dict, list]) -> Mapping[str, Any] | None:
+        """
+        Get file var from value
+        :param value: variable value
+        :return:
+        """
+        if not value:
+            return None
+
+        if isinstance(value, dict) and value.get("dify_model_identity") == FILE_MODEL_IDENTITY:
+            return value
+        elif isinstance(value, File):
+            return value.to_dict()
+
+        return None
+
+    def handle_agent_log(self, task_id: str, event: QueueAgentLogEvent) -> AgentLogStreamResponse:
+        """
+        Handle agent log
+        :param task_id: task id
+        :param event: agent log event
+        :return:
+        """
+        return AgentLogStreamResponse(
+            task_id=task_id,
+            data=AgentLogStreamResponse.Data(
+                node_execution_id=event.node_execution_id,
+                id=event.id,
+                parent_id=event.parent_id,
+                label=event.label,
+                error=event.error,
+                status=event.status,
+                data=event.data,
+                metadata=event.metadata,
+                node_id=event.node_id,
+            ),
+        )
@@ -4,7 +4,7 @@ import uuid
 from collections.abc import Generator, Mapping
 from typing import Any, Literal, Union, overload

-from flask import Flask, current_app
+from flask import Flask, copy_current_request_context, current_app
 from pydantic import ValidationError

 from configs import dify_config
@@ -151,16 +151,17 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
            message_id=message.id,
        )

-        # new thread
-        worker_thread = threading.Thread(
-            target=self._generate_worker,
-            kwargs={
-                "flask_app": current_app._get_current_object(),  # type: ignore
-                "application_generate_entity": application_generate_entity,
-                "queue_manager": queue_manager,
-                "message_id": message.id,
-            },
-        )
+        # new thread with request context
+        @copy_current_request_context
+        def worker_with_context():
+            return self._generate_worker(
+                flask_app=current_app._get_current_object(),  # type: ignore
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
+                message_id=message.id,
+            )
+
+        worker_thread = threading.Thread(target=worker_with_context)

        worker_thread.start()

@@ -313,16 +314,17 @@ class CompletionAppGenerator(MessageBasedAppGenerator):
            message_id=message.id,
        )

-        # new thread
-        worker_thread = threading.Thread(
-            target=self._generate_worker,
-            kwargs={
-                "flask_app": current_app._get_current_object(),  # type: ignore
-                "application_generate_entity": application_generate_entity,
-                "queue_manager": queue_manager,
-                "message_id": message.id,
-            },
-        )
+        # new thread with request context
+        @copy_current_request_context
+        def worker_with_context():
+            return self._generate_worker(
+                flask_app=current_app._get_current_object(),  # type: ignore
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
+                message_id=message.id,
+            )
+
+        worker_thread = threading.Thread(target=worker_with_context)

        worker_thread.start()

@@ -25,7 +25,7 @@ from core.app.task_pipeline.easy_ui_based_generate_task_pipeline import EasyUIBa
 from core.prompt.utils.prompt_template_parser import PromptTemplateParser
 from extensions.ext_database import db
 from models import Account
-from models.enums import CreatedByRole
+from models.enums import CreatorUserRole
 from models.model import App, AppMode, AppModelConfig, Conversation, EndUser, Message, MessageFile
 from services.errors.app_model_config import AppModelConfigBrokenError
 from services.errors.conversation import ConversationNotExistsError
@@ -223,7 +223,7 @@ class MessageBasedAppGenerator(BaseAppGenerator):
                belongs_to="user",
                url=file.remote_url,
                upload_file_id=file.related_id,
-                created_by_role=(CreatedByRole.ACCOUNT if account_id else CreatedByRole.END_USER),
+                created_by_role=(CreatorUserRole.ACCOUNT if account_id else CreatorUserRole.END_USER),
                created_by=account_id or end_user_id or "",
            )
            db.session.add(message_file)
@@ -5,7 +5,7 @@ import uuid
 from collections.abc import Generator, Mapping, Sequence
 from typing import Any, Literal, Optional, Union, overload

-from flask import Flask, current_app
+from flask import Flask, copy_current_request_context, current_app, has_request_context
 from pydantic import ValidationError
 from sqlalchemy.orm import sessionmaker

@@ -18,16 +18,19 @@ from core.app.apps.workflow.app_config_manager import WorkflowAppConfigManager
 from core.app.apps.workflow.app_queue_manager import WorkflowAppQueueManager
 from core.app.apps.workflow.app_runner import WorkflowAppRunner
 from core.app.apps.workflow.generate_response_converter import WorkflowAppGenerateResponseConverter
+from core.app.apps.workflow.generate_task_pipeline import WorkflowAppGenerateTaskPipeline
 from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
 from core.app.entities.task_entities import WorkflowAppBlockingResponse, WorkflowAppStreamResponse
 from core.model_runtime.errors.invoke import InvokeAuthorizationError
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
-from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from core.workflow.workflow_app_generate_task_pipeline import WorkflowAppGenerateTaskPipeline
+from core.repositories.sqlalchemy_workflow_execution_repository import SQLAlchemyWorkflowExecutionRepository
+from core.workflow.repositories.workflow_execution_repository import WorkflowExecutionRepository
+from core.workflow.repositories.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from extensions.ext_database import db
 from factories import file_factory
-from models import Account, App, EndUser, Workflow
+from models import Account, App, EndUser, Workflow, WorkflowNodeExecutionTriggeredFrom
+from models.enums import WorkflowRunTriggeredFrom

 logger = logging.getLogger(__name__)

@@ -129,19 +132,33 @@ class WorkflowAppGenerator(BaseAppGenerator):
            invoke_from=invoke_from,
            call_depth=call_depth,
            trace_manager=trace_manager,
-            workflow_run_id=workflow_run_id,
+            workflow_execution_id=workflow_run_id,
        )

-        contexts.tenant_id.set(application_generate_entity.app_config.tenant_id)
        contexts.plugin_tool_providers.set({})
        contexts.plugin_tool_providers_lock.set(threading.Lock())

-        # Create workflow node execution repository
+        # Create repositories
+        #
+        # Create session factory
        session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
+        # Create workflow execution(aka workflow run) repository
+        if invoke_from == InvokeFrom.DEBUGGER:
+            workflow_triggered_from = WorkflowRunTriggeredFrom.DEBUGGING
+        else:
+            workflow_triggered_from = WorkflowRunTriggeredFrom.APP_RUN
+        workflow_execution_repository = SQLAlchemyWorkflowExecutionRepository(
+            session_factory=session_factory,
+            user=user,
+            app_id=application_generate_entity.app_config.app_id,
+            triggered_from=workflow_triggered_from,
+        )
+        # Create workflow node execution repository
        workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
            session_factory=session_factory,
-            tenant_id=application_generate_entity.app_config.tenant_id,
+            user=user,
            app_id=application_generate_entity.app_config.app_id,
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
        )

        return self._generate(
@@ -150,6 +167,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
            user=user,
            application_generate_entity=application_generate_entity,
            invoke_from=invoke_from,
+            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
            streaming=streaming,
            workflow_thread_pool_id=workflow_thread_pool_id,
@@ -163,6 +181,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
        user: Union[Account, EndUser],
        application_generate_entity: WorkflowAppGenerateEntity,
        invoke_from: InvokeFrom,
+        workflow_execution_repository: WorkflowExecutionRepository,
        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
        streaming: bool = True,
        workflow_thread_pool_id: Optional[str] = None,
@@ -187,17 +206,22 @@ class WorkflowAppGenerator(BaseAppGenerator):
            app_mode=app_model.mode,
        )

-        # new thread
-        worker_thread = threading.Thread(
-            target=self._generate_worker,
-            kwargs={
-                "flask_app": current_app._get_current_object(),  # type: ignore
-                "application_generate_entity": application_generate_entity,
-                "queue_manager": queue_manager,
-                "context": contextvars.copy_context(),
-                "workflow_thread_pool_id": workflow_thread_pool_id,
-            },
-        )
+        # new thread with request context and contextvars
+        context = contextvars.copy_context()
+
+        @copy_current_request_context
+        def worker_with_context():
+            # Run the worker within the copied context
+            return context.run(
+                self._generate_worker,
+                flask_app=current_app._get_current_object(),  # type: ignore
+                application_generate_entity=application_generate_entity,
+                queue_manager=queue_manager,
+                context=context,
+                workflow_thread_pool_id=workflow_thread_pool_id,
+            )
+
+        worker_thread = threading.Thread(target=worker_with_context)

        worker_thread.start()

@@ -207,6 +231,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
            workflow=workflow,
            queue_manager=queue_manager,
            user=user,
+            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
            stream=streaming,
        )
@@ -254,18 +279,30 @@ class WorkflowAppGenerator(BaseAppGenerator):
            single_iteration_run=WorkflowAppGenerateEntity.SingleIterationRunEntity(
                node_id=node_id, inputs=args["inputs"]
            ),
-            workflow_run_id=str(uuid.uuid4()),
+            workflow_execution_id=str(uuid.uuid4()),
        )
-        contexts.tenant_id.set(application_generate_entity.app_config.tenant_id)
        contexts.plugin_tool_providers.set({})
        contexts.plugin_tool_providers_lock.set(threading.Lock())

+        # Create repositories
+        #
+        # Create session factory
+        session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
+        # Create workflow execution(aka workflow run) repository
+        workflow_execution_repository = SQLAlchemyWorkflowExecutionRepository(
+            session_factory=session_factory,
+            user=user,
+            app_id=application_generate_entity.app_config.app_id,
+            triggered_from=WorkflowRunTriggeredFrom.DEBUGGING,
+        )
        # Create workflow node execution repository
        session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
+
        workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
            session_factory=session_factory,
-            tenant_id=application_generate_entity.app_config.tenant_id,
+            user=user,
            app_id=application_generate_entity.app_config.app_id,
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
        )

        return self._generate(
@@ -274,6 +311,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
            user=user,
            invoke_from=InvokeFrom.DEBUGGER,
            application_generate_entity=application_generate_entity,
+            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
            streaming=streaming,
        )
@@ -317,18 +355,30 @@ class WorkflowAppGenerator(BaseAppGenerator):
            invoke_from=InvokeFrom.DEBUGGER,
            extras={"auto_generate_conversation_name": False},
            single_loop_run=WorkflowAppGenerateEntity.SingleLoopRunEntity(node_id=node_id, inputs=args["inputs"]),
-            workflow_run_id=str(uuid.uuid4()),
+            workflow_execution_id=str(uuid.uuid4()),
        )
-        contexts.tenant_id.set(application_generate_entity.app_config.tenant_id)
        contexts.plugin_tool_providers.set({})
        contexts.plugin_tool_providers_lock.set(threading.Lock())

+        # Create repositories
+        #
+        # Create session factory
+        session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
+        # Create workflow execution(aka workflow run) repository
+        workflow_execution_repository = SQLAlchemyWorkflowExecutionRepository(
+            session_factory=session_factory,
+            user=user,
+            app_id=application_generate_entity.app_config.app_id,
+            triggered_from=WorkflowRunTriggeredFrom.DEBUGGING,
+        )
        # Create workflow node execution repository
        session_factory = sessionmaker(bind=db.engine, expire_on_commit=False)
+
        workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
            session_factory=session_factory,
-            tenant_id=application_generate_entity.app_config.tenant_id,
+            user=user,
            app_id=application_generate_entity.app_config.app_id,
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.SINGLE_STEP,
        )

        return self._generate(
@@ -337,6 +387,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
            user=user,
            invoke_from=InvokeFrom.DEBUGGER,
            application_generate_entity=application_generate_entity,
+            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
            streaming=streaming,
        )
@@ -359,8 +410,22 @@ class WorkflowAppGenerator(BaseAppGenerator):
        """
        for var, val in context.items():
            var.set(val)
+
+        # FIXME(-LAN-): Save current user before entering new app context
+        from flask import g
+
+        saved_user = None
+        if has_request_context() and hasattr(g, "_login_user"):
+            saved_user = g._login_user
+
        with flask_app.app_context():
            try:
+                # Restore user in new app context
+                if saved_user is not None:
+                    from flask import g
+
+                    g._login_user = saved_user
+
                # workflow app
                runner = WorkflowAppRunner(
                    application_generate_entity=application_generate_entity,
@@ -394,6 +459,7 @@ class WorkflowAppGenerator(BaseAppGenerator):
        workflow: Workflow,
        queue_manager: AppQueueManager,
        user: Union[Account, EndUser],
+        workflow_execution_repository: WorkflowExecutionRepository,
        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
        stream: bool = False,
    ) -> Union[WorkflowAppBlockingResponse, Generator[WorkflowAppStreamResponse, None, None]]:
@@ -413,8 +479,9 @@ class WorkflowAppGenerator(BaseAppGenerator):
            workflow=workflow,
            queue_manager=queue_manager,
            user=user,
-            stream=stream,
+            workflow_execution_repository=workflow_execution_repository,
            workflow_node_execution_repository=workflow_node_execution_repository,
+            stream=stream,
        )

        try:
@@ -95,7 +95,7 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
                SystemVariableKey.USER_ID: user_id,
                SystemVariableKey.APP_ID: app_config.app_id,
                SystemVariableKey.WORKFLOW_ID: app_config.workflow_id,
-                SystemVariableKey.WORKFLOW_RUN_ID: self.application_generate_entity.workflow_run_id,
+                SystemVariableKey.WORKFLOW_EXECUTION_ID: self.application_generate_entity.workflow_execution_id,
            }

            variable_pool = VariablePool(
--- a/Show More
+++ b/Show More