chore(*): Bump version to 1.4.1 (#20275 )

Signed-off-by: -LAN- <laipz8200@outlook.com>
chore(remove_app_and_related_data_task): Revert _delete_app_workflow_node_executions (#20278 )
2025-05-27 15:32:59 +08:00 · 2025-05-27 14:58:08 +08:00 · 2025-05-27 14:46:04 +08:00 · 2025-05-27 14:33:11 +08:00 · 2025-05-27 14:17:32 +08:00 · 2025-05-27 14:04:22 +08:00
751 changed files with 23154 additions and 7364 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
@@ -7,6 +7,7 @@ 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

@@ -139,6 +139,7 @@ jobs:
      - name: Checkout code
        uses: actions/checkout@v4
        with:
+          fetch-depth: 0
          persist-credentials: false

      - name: Check changed files
@@ -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
@@ -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,
@@ -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,6 +1,6 @@
 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_settings import BaseSettings
@@ -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,
        }


@@ -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.1",
    )

    COMMIT_SHA: str = Field(
@@ -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,
+}
@@ -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:
@@ -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


@@ -119,6 +119,9 @@ class ForgotPasswordResetApi(Resource):
        if not reset_data:
            raise InvalidTokenError()
        # Must use token in reset phase
+        if reset_data.get("phase", "") != "reset":
+            raise InvalidTokenError()
+        # Must use token in reset phase
        if reset_data.get("phase", "") != "reset":
            raise InvalidTokenError()

@@ -168,6 +171,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
@@ -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,21 @@ 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 = []
+            for installed_app in installed_app_list:
+                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 +87,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 +100,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()

@@ -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")
@@ -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,7 +3,7 @@ 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,
@@ -14,7 +14,7 @@ from services.annotation_service import AppAnnotationService


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


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


 class AnnotationListApi(Resource):
-    @validate_app_token(fetch_user_arg=FetchUserArg(fetch_from=WhereisUserArg.QUERY))
+    @validate_app_token
    def get(self, app_model: App, end_user: EndUser):
        page = request.args.get("page", default=1, type=int)
        limit = request.args.get("limit", default=20, type=int)
@@ -65,7 +65,7 @@ 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):
        parser = reqparse.RequestParser()
@@ -77,7 +77,7 @@ 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):
        if not current_user.is_editor:
@@ -91,7 +91,7 @@ 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))
+    @validate_app_token
    def delete(self, app_model: App, end_user: EndUser, 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")
@@ -313,7 +313,7 @@ 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:
@@ -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 (
@@ -323,7 +323,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 +337,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 +345,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 +374,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):
@@ -344,7 +344,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)

@@ -1,12 +1,15 @@
-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


 class AppParameterApi(WebApiResource):
@@ -40,5 +43,51 @@ 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=True, location="args")
+        args = parser.parse_args()
+
+        app_id = args["appId"]
+        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
+
+        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 = 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,120 @@
+from flask import request
+from flask_restful import Resource, reqparse
+from jwt import InvalidTokenError  # type: ignore
+from werkzeug.exceptions import BadRequest
+
+import services
+from controllers.console.auth.error import EmailCodeError, EmailOrPasswordMismatchError, InvalidEmailError
+from controllers.console.error import AccountBannedError, AccountNotFound
+from controllers.console.wraps import setup_required
+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."""
+
+    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()
+
+        app_code = request.headers.get("X-App-Code")
+        if app_code is None:
+            raise BadRequest("X-App-Code header is missing.")
+
+        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()
+
+        WebAppAuthService._validate_user_accessibility(account=account, app_code=app_code)
+
+        end_user = WebAppAuthService.create_end_user(email=args["email"], app_code=app_code)
+
+        token = WebAppAuthService.login(account=account, app_code=app_code, end_user_id=end_user.id)
+        return {"result": "success", "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
+    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
+    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"]
+        app_code = request.headers.get("X-App-Code")
+        if app_code is None:
+            raise BadRequest("X-App-Code header is missing.")
+
+        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()
+
+        WebAppAuthService._validate_user_accessibility(account=account, app_code=app_code)
+
+        end_user = WebAppAuthService.create_end_user(email=user_email, app_code=app_code)
+
+        token = WebAppAuthService.login(account=account, app_code=app_code, end_user_id=end_user.id)
+        AccountService.reset_login_error_rate_limit(args["email"])
+        return {"result": "success", "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")
@@ -5,7 +5,7 @@ from flask_restful import Resource
 from werkzeug.exceptions import NotFound, Unauthorized

 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
@@ -24,10 +24,10 @@ class PassportResource(Resource):
        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()
+        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()
@@ -4,7 +4,7 @@ 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
@@ -29,7 +29,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:
@@ -57,35 +57,53 @@ def decode_jwt_token():
        if not end_user:
            raise NotFound()

-        _validate_web_sso_token(decoded, system_features, app_code)
+        # for enterprise webapp auth
+        app_web_auth_enabled = False
+        if system_features.webapp_auth.enabled:
+            app_web_auth_enabled = (
+                EnterpriseService.WebAppAuth.get_app_access_mode_by_code(app_code=app_code).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)

        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:
+            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):
+    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 EnterpriseService.WebAppAuth.is_user_allowed_to_access_webapp(user_id, app_code=app_code):
+            raise WebAppAuthAccessDeniedError()


 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 = (
@@ -69,13 +69,6 @@ class CotAgentRunner(BaseAgentRunner, ABC):
        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
@@ -45,13 +45,6 @@ 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
@@ -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
@@ -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.repositories.sqlalchemy_workflow_execution_repository import SQLAlchemyWorkflowExecutionRepository
+from core.workflow.repository.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repository.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)
+
+        # 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:
@@ -10,6 +10,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,
@@ -64,13 +65,14 @@ from core.ops.ops_trace_manager import TraceQueueManager
 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_execution_repository import WorkflowExecutionRepository
 from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.workflow_cycle_manager import 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.enums import CreatorUserRole
 from models.workflow import (
    Workflow,
    WorkflowRunStatus,
@@ -94,6 +96,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 +108,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)}")

@@ -125,9 +128,14 @@ class AdvancedChatAppGenerateTaskPipeline:
                SystemVariableKey.WORKFLOW_ID: workflow.id,
                SystemVariableKey.WORKFLOW_RUN_ID: application_generate_entity.workflow_run_id,
            },
+            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(
            application_generate_entity=application_generate_entity, task_state=self._task_state
@@ -294,19 +302,18 @@ 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(
+                    workflow_execution = 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
+                    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 +326,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 +342,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 +358,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 +382,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 +398,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 +494,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 +503,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 +520,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 +529,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 +546,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,
+                        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 +572,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(),
+                            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)
@@ -711,7 +665,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:
@@ -739,9 +693,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
@@ -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)

+        # 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,564 @@
+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.node_execution_entities import NodeExecution
+from core.workflow.entities.workflow_execution_entities import WorkflowExecution
+from core.workflow.nodes import NodeType
+from core.workflow.nodes.tool.entities import ToolNodeData
+from models import (
+    Account,
+    CreatorUserRole,
+    EndUser,
+    WorkflowNodeExecutionStatus,
+    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,
+                sequence_number=workflow_execution.sequence_number,
+                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,
+                sequence_number=workflow_execution.sequence_number,
+                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: NodeExecution,
+    ) -> Optional[NodeStartStreamResponse]:
+        if workflow_node_execution.node_type in {NodeType.ITERATION, NodeType.LOOP}:
+            return None
+        if not workflow_node_execution.workflow_run_id:
+            return None
+
+        response = NodeStartStreamResponse(
+            task_id=task_id,
+            workflow_run_id=workflow_node_execution.workflow_run_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: NodeExecution,
+    ) -> Optional[NodeFinishStreamResponse]:
+        if workflow_node_execution.node_type in {NodeType.ITERATION, NodeType.LOOP}:
+            return None
+        if not workflow_node_execution.workflow_run_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_run_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: NodeExecution,
+    ) -> Optional[Union[NodeRetryStreamResponse, NodeFinishStreamResponse]]:
+        if workflow_node_execution.node_type in {NodeType.ITERATION, NodeType.LOOP}:
+            return None
+        if not workflow_node_execution.workflow_run_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_run_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.repositories.sqlalchemy_workflow_execution_repository import SQLAlchemyWorkflowExecutionRepository
+from core.workflow.repository.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
-from core.workflow.workflow_app_generate_task_pipeline import WorkflowAppGenerateTaskPipeline
 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__)

@@ -132,16 +135,30 @@ class WorkflowAppGenerator(BaseAppGenerator):
            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(
@@ -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,
        )
@@ -256,16 +281,28 @@ class WorkflowAppGenerator(BaseAppGenerator):
            ),
            workflow_run_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,
        )
@@ -319,16 +357,28 @@ class WorkflowAppGenerator(BaseAppGenerator):
            single_loop_run=WorkflowAppGenerateEntity.SingleLoopRunEntity(node_id=node_id, inputs=args["inputs"]),
            workflow_run_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)
+
+        # 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:
@@ -3,10 +3,12 @@ import time
 from collections.abc import Generator
 from typing import Optional, Union

+from sqlalchemy import select
 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
+from core.app.apps.common.workflow_response_converter import WorkflowResponseConverter
 from core.app.entities.app_invoke_entities import (
    InvokeFrom,
    WorkflowAppGenerateEntity,
@@ -53,12 +55,14 @@ from core.app.entities.task_entities import (
 from core.app.task_pipeline.based_generate_task_pipeline import BasedGenerateTaskPipeline
 from core.base.tts import AppGeneratorTTSPublisher, AudioTrunk
 from core.ops.ops_trace_manager import TraceQueueManager
+from core.workflow.entities.workflow_execution_entities import WorkflowExecution
 from core.workflow.enums import SystemVariableKey
+from core.workflow.repository.workflow_execution_repository import WorkflowExecutionRepository
 from core.workflow.repository.workflow_node_execution_repository import WorkflowNodeExecutionRepository
 from core.workflow.workflow_cycle_manager import WorkflowCycleManager
 from extensions.ext_database import db
 from models.account import Account
-from models.enums import CreatedByRole
+from models.enums import CreatorUserRole
 from models.model import EndUser
 from models.workflow import (
    Workflow,
@@ -83,6 +87,7 @@ class WorkflowAppGenerateTaskPipeline:
        queue_manager: AppQueueManager,
        user: Union[Account, EndUser],
        stream: bool,
+        workflow_execution_repository: WorkflowExecutionRepository,
        workflow_node_execution_repository: WorkflowNodeExecutionRepository,
    ) -> None:
        self._base_task_pipeline = BasedGenerateTaskPipeline(
@@ -94,11 +99,11 @@ class WorkflowAppGenerateTaskPipeline:
        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 ValueError(f"Invalid user type: {type(user)}")

@@ -111,9 +116,14 @@ class WorkflowAppGenerateTaskPipeline:
                SystemVariableKey.WORKFLOW_ID: workflow.id,
                SystemVariableKey.WORKFLOW_RUN_ID: application_generate_entity.workflow_run_id,
            },
+            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._application_generate_entity = application_generate_entity
        self._workflow_id = workflow.id
        self._workflow_features_dict = workflow.features_dict
@@ -258,17 +268,15 @@ class WorkflowAppGenerateTaskPipeline:

                with Session(db.engine, expire_on_commit=False) as session:
                    # init workflow run
-                    workflow_run = self._workflow_cycle_manager._handle_workflow_run_start(
+                    workflow_execution = 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
-                    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
+                    self._workflow_run_id = workflow_execution.id
+                    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()

                yield start_resp
            elif isinstance(
@@ -278,13 +286,11 @@ class WorkflowAppGenerateTaskPipeline:
                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_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
-                    )
-                    response = self._workflow_cycle_manager._workflow_node_retry_to_stream_response(
+                    response = 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,
@@ -297,27 +303,22 @@ class WorkflowAppGenerateTaskPipeline:
                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
-                    )
-                    node_start_response = 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()
+                workflow_node_execution = self._workflow_cycle_manager.handle_node_execution_start(
+                    workflow_execution_id=self._workflow_run_id, event=event
+                )
+                node_start_response = 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_response:
                    yield node_start_response
            elif isinstance(event, QueueNodeSucceededEvent):
-                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_success_response = self._workflow_cycle_manager._workflow_node_finish_to_stream_response(
+                node_success_response = 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,
@@ -332,10 +333,10 @@ class WorkflowAppGenerateTaskPipeline:
                | 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_failed_response = self._workflow_cycle_manager._workflow_node_finish_to_stream_response(
+                node_failed_response = 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,
@@ -348,18 +349,13 @@ class WorkflowAppGenerateTaskPipeline:
                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

@@ -367,18 +363,13 @@ class WorkflowAppGenerateTaskPipeline:
                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

@@ -386,16 +377,11 @@ class WorkflowAppGenerateTaskPipeline:
                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

@@ -403,16 +389,11 @@ class WorkflowAppGenerateTaskPipeline:
                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

@@ -420,16 +401,11 @@ class WorkflowAppGenerateTaskPipeline:
                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

@@ -437,16 +413,11 @@ class WorkflowAppGenerateTaskPipeline:
                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

@@ -454,16 +425,11 @@ class WorkflowAppGenerateTaskPipeline:
                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

@@ -471,16 +437,11 @@ class WorkflowAppGenerateTaskPipeline:
                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

@@ -491,10 +452,8 @@ class WorkflowAppGenerateTaskPipeline:
                    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_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,
@@ -503,12 +462,12 @@ class WorkflowAppGenerateTaskPipeline:
                    )

                    # save workflow app log
-                    self._save_workflow_app_log(session=session, workflow_run=workflow_run)
+                    self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)

-                    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,
                    )
                    session.commit()

@@ -520,10 +479,8 @@ class WorkflowAppGenerateTaskPipeline:
                    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,
@@ -533,10 +490,12 @@ class WorkflowAppGenerateTaskPipeline:
                    )

                    # save workflow app log
-                    self._save_workflow_app_log(session=session, workflow_run=workflow_run)
+                    self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)

-                    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()

@@ -548,26 +507,28 @@ class WorkflowAppGenerateTaskPipeline:
                    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
                        if isinstance(event, QueueWorkflowFailedEvent)
                        else WorkflowRunStatus.STOPPED,
-                        error=event.error if isinstance(event, QueueWorkflowFailedEvent) else event.get_stop_reason(),
+                        error_message=event.error
+                        if isinstance(event, QueueWorkflowFailedEvent)
+                        else event.get_stop_reason(),
                        conversation_id=None,
                        trace_manager=trace_manager,
                        exceptions_count=event.exceptions_count if isinstance(event, QueueWorkflowFailedEvent) else 0,
                    )

                    # save workflow app log
-                    self._save_workflow_app_log(session=session, workflow_run=workflow_run)
+                    self._save_workflow_app_log(session=session, workflow_execution=workflow_execution)

-                    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()

@@ -586,7 +547,7 @@ class WorkflowAppGenerateTaskPipeline:
                    delta_text, from_variable_selector=event.from_variable_selector
                )
            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:
@@ -595,11 +556,9 @@ class WorkflowAppGenerateTaskPipeline:
        if tts_publisher:
            tts_publisher.publish(None)

-    def _save_workflow_app_log(self, *, session: Session, workflow_run: WorkflowRun) -> None:
-        """
-        Save workflow app log.
-        :return:
-        """
+    def _save_workflow_app_log(self, *, session: Session, workflow_execution: WorkflowExecution) -> None:
+        workflow_run = session.scalar(select(WorkflowRun).where(WorkflowRun.id == workflow_execution.id))
+        assert workflow_run is not None
        invoke_from = self._application_generate_entity.invoke_from
        if invoke_from == InvokeFrom.SERVICE_API:
            created_from = WorkflowAppLogCreatedFrom.SERVICE_API
@@ -6,7 +6,7 @@ from pydantic import BaseModel, ConfigDict

 from core.model_runtime.entities.llm_entities import LLMResult
 from core.model_runtime.utils.encoders import jsonable_encoder
-from core.workflow.entities.node_entities import AgentNodeStrategyInit
+from core.workflow.entities.node_entities import AgentNodeStrategyInit, NodeRunMetadataKey
 from models.workflow import WorkflowNodeExecutionStatus


@@ -190,7 +190,7 @@ class WorkflowStartStreamResponse(StreamResponse):
        id: str
        workflow_id: str
        sequence_number: int
-        inputs: dict
+        inputs: Mapping[str, Any]
        created_at: int

    event: StreamEvent = StreamEvent.WORKFLOW_STARTED
@@ -212,7 +212,7 @@ class WorkflowFinishStreamResponse(StreamResponse):
        workflow_id: str
        sequence_number: int
        status: str
-        outputs: Optional[dict] = None
+        outputs: Optional[Mapping[str, Any]] = None
        error: Optional[str] = None
        elapsed_time: float
        total_tokens: int
@@ -244,7 +244,7 @@ class NodeStartStreamResponse(StreamResponse):
        title: str
        index: int
        predecessor_node_id: Optional[str] = None
-        inputs: Optional[dict] = None
+        inputs: Optional[Mapping[str, Any]] = None
        created_at: int
        extras: dict = {}
        parallel_id: Optional[str] = None
@@ -301,13 +301,13 @@ class NodeFinishStreamResponse(StreamResponse):
        title: str
        index: int
        predecessor_node_id: Optional[str] = None
-        inputs: Optional[dict] = None
-        process_data: Optional[dict] = None
-        outputs: Optional[dict] = None
+        inputs: Optional[Mapping[str, Any]] = None
+        process_data: Optional[Mapping[str, Any]] = None
+        outputs: Optional[Mapping[str, Any]] = None
        status: str
        error: Optional[str] = None
        elapsed_time: float
-        execution_metadata: Optional[dict] = None
+        execution_metadata: Optional[Mapping[NodeRunMetadataKey, Any]] = None
        created_at: int
        finished_at: int
        files: Optional[Sequence[Mapping[str, Any]]] = []
@@ -370,13 +370,13 @@ class NodeRetryStreamResponse(StreamResponse):
        title: str
        index: int
        predecessor_node_id: Optional[str] = None
-        inputs: Optional[dict] = None
-        process_data: Optional[dict] = None
-        outputs: Optional[dict] = None
+        inputs: Optional[Mapping[str, Any]] = None
+        process_data: Optional[Mapping[str, Any]] = None
+        outputs: Optional[Mapping[str, Any]] = None
        status: str
        error: Optional[str] = None
        elapsed_time: float
-        execution_metadata: Optional[dict] = None
+        execution_metadata: Optional[Mapping[NodeRunMetadataKey, Any]] = None
        created_at: int
        finished_at: int
        files: Optional[Sequence[Mapping[str, Any]]] = []
@@ -788,7 +788,7 @@ class WorkflowAppBlockingResponse(AppBlockingResponse):
        id: str
        workflow_id: str
        status: str
-        outputs: Optional[dict] = None
+        outputs: Optional[Mapping[str, Any]] = None
        error: Optional[str] = None
        elapsed_time: float
        total_tokens: int
@@ -1,3 +1,5 @@
+import logging
+
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.entities.queue_entities import QueueRetrieverResourcesEvent
@@ -7,6 +9,8 @@ from extensions.ext_database import db
 from models.dataset import ChildChunk, DatasetQuery, DocumentSegment
 from models.dataset import Document as DatasetDocument

+_logger = logging.getLogger(__name__)
+

 class DatasetIndexToolCallbackHandler:
    """Callback handler for dataset tool."""
@@ -42,18 +46,31 @@ class DatasetIndexToolCallbackHandler:
        """Handle tool end."""
        for document in documents:
            if document.metadata is not None:
-                dataset_document = DatasetDocument.query.filter(
-                    DatasetDocument.id == document.metadata["document_id"]
-                ).first()
+                document_id = document.metadata["document_id"]
+                dataset_document = db.session.query(DatasetDocument).filter(DatasetDocument.id == document_id).first()
+                if not dataset_document:
+                    _logger.warning(
+                        "Expected DatasetDocument record to exist, but none was found, document_id=%s",
+                        document_id,
+                    )
+                    continue
                if dataset_document.doc_form == IndexType.PARENT_CHILD_INDEX:
-                    child_chunk = ChildChunk.query.filter(
-                        ChildChunk.index_node_id == document.metadata["doc_id"],
-                        ChildChunk.dataset_id == dataset_document.dataset_id,
-                        ChildChunk.document_id == dataset_document.id,
-                    ).first()
+                    child_chunk = (
+                        db.session.query(ChildChunk)
+                        .filter(
+                            ChildChunk.index_node_id == document.metadata["doc_id"],
+                            ChildChunk.dataset_id == dataset_document.dataset_id,
+                            ChildChunk.document_id == dataset_document.id,
+                        )
+                        .first()
+                    )
                    if child_chunk:
-                        segment = DocumentSegment.query.filter(DocumentSegment.id == child_chunk.segment_id).update(
-                            {DocumentSegment.hit_count: DocumentSegment.hit_count + 1}, synchronize_session=False
+                        segment = (
+                            db.session.query(DocumentSegment)
+                            .filter(DocumentSegment.id == child_chunk.segment_id)
+                            .update(
+                                {DocumentSegment.hit_count: DocumentSegment.hit_count + 1}, synchronize_session=False
+                            )
                        )
                else:
                    query = db.session.query(DocumentSegment).filter(
@@ -754,7 +754,7 @@ class ProviderConfiguration(BaseModel):
        :param only_active: return active model only
        :return:
        """
-        provider_models = self.get_provider_models(model_type, only_active)
+        provider_models = self.get_provider_models(model_type, only_active, model)

        for provider_model in provider_models:
            if provider_model.model == model:
@@ -763,12 +763,13 @@ class ProviderConfiguration(BaseModel):
        return None

    def get_provider_models(
-        self, model_type: Optional[ModelType] = None, only_active: bool = False
+        self, model_type: Optional[ModelType] = None, only_active: bool = False, model: Optional[str] = None
    ) -> list[ModelWithProviderEntity]:
        """
        Get provider models.
        :param model_type: model type
        :param only_active: only active models
+        :param model: model name
        :return:
        """
        model_provider_factory = ModelProviderFactory(self.tenant_id)
@@ -791,7 +792,10 @@ class ProviderConfiguration(BaseModel):
            )
        else:
            provider_models = self._get_custom_provider_models(
-                model_types=model_types, provider_schema=provider_schema, model_setting_map=model_setting_map
+                model_types=model_types,
+                provider_schema=provider_schema,
+                model_setting_map=model_setting_map,
+                model=model,
            )

        if only_active:
@@ -897,37 +901,36 @@ class ProviderConfiguration(BaseModel):
                            )
                        except Exception as ex:
                            logger.warning(f"get custom model schema failed, {ex}")
+                            continue

-                            if not custom_model_schema:
-                                continue
+                        if not custom_model_schema:
+                            continue

-                            if custom_model_schema.model_type not in model_types:
-                                continue
+                        if custom_model_schema.model_type not in model_types:
+                            continue

-                            status = ModelStatus.ACTIVE
-                            if (
-                                custom_model_schema.model_type in model_setting_map
-                                and custom_model_schema.model in model_setting_map[custom_model_schema.model_type]
-                            ):
-                                model_setting = model_setting_map[custom_model_schema.model_type][
-                                    custom_model_schema.model
-                                ]
-                                if model_setting.enabled is False:
-                                    status = ModelStatus.DISABLED
+                        status = ModelStatus.ACTIVE
+                        if (
+                            custom_model_schema.model_type in model_setting_map
+                            and custom_model_schema.model in model_setting_map[custom_model_schema.model_type]
+                        ):
+                            model_setting = model_setting_map[custom_model_schema.model_type][custom_model_schema.model]
+                            if model_setting.enabled is False:
+                                status = ModelStatus.DISABLED

-                            provider_models.append(
-                                ModelWithProviderEntity(
-                                    model=custom_model_schema.model,
-                                    label=custom_model_schema.label,
-                                    model_type=custom_model_schema.model_type,
-                                    features=custom_model_schema.features,
-                                    fetch_from=FetchFrom.PREDEFINED_MODEL,
-                                    model_properties=custom_model_schema.model_properties,
-                                    deprecated=custom_model_schema.deprecated,
-                                    provider=SimpleModelProviderEntity(self.provider),
-                                    status=status,
-                                )
+                        provider_models.append(
+                            ModelWithProviderEntity(
+                                model=custom_model_schema.model,
+                                label=custom_model_schema.label,
+                                model_type=custom_model_schema.model_type,
+                                features=custom_model_schema.features,
+                                fetch_from=FetchFrom.PREDEFINED_MODEL,
+                                model_properties=custom_model_schema.model_properties,
+                                deprecated=custom_model_schema.deprecated,
+                                provider=SimpleModelProviderEntity(self.provider),
+                                status=status,
                            )
+                        )

            # if llm name not in restricted llm list, remove it
            restrict_model_names = [rm.model for rm in restrict_models]
@@ -944,6 +947,7 @@ class ProviderConfiguration(BaseModel):
        model_types: Sequence[ModelType],
        provider_schema: ProviderEntity,
        model_setting_map: dict[ModelType, dict[str, ModelSettings]],
+        model: Optional[str] = None,
    ) -> list[ModelWithProviderEntity]:
        """
        Get custom provider models.
@@ -996,7 +1000,8 @@ class ProviderConfiguration(BaseModel):
        for model_configuration in self.custom_configuration.models:
            if model_configuration.model_type not in model_types:
                continue
-
+            if model and model != model_configuration.model:
+                continue
            try:
                custom_model_schema = self.get_model_schema(
                    model_type=model_configuration.model_type,
@@ -51,7 +51,7 @@ class IndexingRunner:
        for dataset_document in dataset_documents:
            try:
                # get dataset
-                dataset = Dataset.query.filter_by(id=dataset_document.dataset_id).first()
+                dataset = db.session.query(Dataset).filter_by(id=dataset_document.dataset_id).first()

                if not dataset:
                    raise ValueError("no dataset found")
@@ -103,15 +103,17 @@ class IndexingRunner:
        """Run the indexing process when the index_status is splitting."""
        try:
            # get dataset
-            dataset = Dataset.query.filter_by(id=dataset_document.dataset_id).first()
+            dataset = db.session.query(Dataset).filter_by(id=dataset_document.dataset_id).first()

            if not dataset:
                raise ValueError("no dataset found")

            # get exist document_segment list and delete
-            document_segments = DocumentSegment.query.filter_by(
-                dataset_id=dataset.id, document_id=dataset_document.id
-            ).all()
+            document_segments = (
+                db.session.query(DocumentSegment)
+                .filter_by(dataset_id=dataset.id, document_id=dataset_document.id)
+                .all()
+            )

            for document_segment in document_segments:
                db.session.delete(document_segment)
@@ -162,15 +164,17 @@ class IndexingRunner:
        """Run the indexing process when the index_status is indexing."""
        try:
            # get dataset
-            dataset = Dataset.query.filter_by(id=dataset_document.dataset_id).first()
+            dataset = db.session.query(Dataset).filter_by(id=dataset_document.dataset_id).first()

            if not dataset:
                raise ValueError("no dataset found")

            # get exist document_segment list and delete
-            document_segments = DocumentSegment.query.filter_by(
-                dataset_id=dataset.id, document_id=dataset_document.id
-            ).all()
+            document_segments = (
+                db.session.query(DocumentSegment)
+                .filter_by(dataset_id=dataset.id, document_id=dataset_document.id)
+                .all()
+            )

            documents = []
            if document_segments:
@@ -254,7 +258,7 @@ class IndexingRunner:

        embedding_model_instance = None
        if dataset_id:
-            dataset = Dataset.query.filter_by(id=dataset_id).first()
+            dataset = db.session.query(Dataset).filter_by(id=dataset_id).first()
            if not dataset:
                raise ValueError("Dataset not found.")
            if dataset.indexing_technique == "high_quality" or indexing_technique == "high_quality":
@@ -587,7 +591,7 @@ class IndexingRunner:
    @staticmethod
    def _process_keyword_index(flask_app, dataset_id, document_id, documents):
        with flask_app.app_context():
-            dataset = Dataset.query.filter_by(id=dataset_id).first()
+            dataset = db.session.query(Dataset).filter_by(id=dataset_id).first()
            if not dataset:
                raise ValueError("no dataset found")
            keyword = Keyword(dataset)
@@ -656,10 +660,10 @@ class IndexingRunner:
        """
        Update the document indexing status.
        """
-        count = DatasetDocument.query.filter_by(id=document_id, is_paused=True).count()
+        count = db.session.query(DatasetDocument).filter_by(id=document_id, is_paused=True).count()
        if count > 0:
            raise DocumentIsPausedError()
-        document = DatasetDocument.query.filter_by(id=document_id).first()
+        document = db.session.query(DatasetDocument).filter_by(id=document_id).first()
        if not document:
            raise DocumentIsDeletedPausedError()

@@ -668,7 +672,7 @@ class IndexingRunner:
        if extra_update_params:
            update_params.update(extra_update_params)

-        DatasetDocument.query.filter_by(id=document_id).update(update_params)
+        db.session.query(DatasetDocument).filter_by(id=document_id).update(update_params)
        db.session.commit()

    @staticmethod
@@ -676,7 +680,7 @@ class IndexingRunner:
        """
        Update the document segment by document id.
        """
-        DocumentSegment.query.filter_by(document_id=dataset_document_id).update(update_params)
+        db.session.query(DocumentSegment).filter_by(document_id=dataset_document_id).update(update_params)
        db.session.commit()

    def _transform(
@@ -51,15 +51,19 @@ class LLMGenerator:
            response = cast(
                LLMResult,
                model_instance.invoke_llm(
-                    prompt_messages=list(prompts), model_parameters={"max_tokens": 100, "temperature": 1}, stream=False
+                    prompt_messages=list(prompts), model_parameters={"max_tokens": 500, "temperature": 1}, stream=False
                ),
            )
        answer = cast(str, response.message.content)
        cleaned_answer = re.sub(r"^.*(\{.*\}).*$", r"\1", answer, flags=re.DOTALL)
        if cleaned_answer is None:
            return ""
-        result_dict = json.loads(cleaned_answer)
-        answer = result_dict["Your Output"]
+        try:
+            result_dict = json.loads(cleaned_answer)
+            answer = result_dict["Your Output"]
+        except json.JSONDecodeError as e:
+            logging.exception("Failed to generate name after answer, use query instead")
+            answer = query
        name = answer.strip()

        if len(name) > 75:
@@ -1,9 +1,9 @@
-from enum import Enum
+from enum import StrEnum

 from pydantic import BaseModel, ValidationInfo, field_validator


-class TracingProviderEnum(Enum):
+class TracingProviderEnum(StrEnum):
    LANGFUSE = "langfuse"
    LANGSMITH = "langsmith"
    OPIK = "opik"
@@ -1,3 +1,4 @@
+from collections.abc import Mapping
 from datetime import datetime
 from enum import StrEnum
 from typing import Any, Optional, Union
@@ -155,10 +156,10 @@ class LangfuseSpan(BaseModel):
        description="The status message of the span. Additional field for context of the event. E.g. the error "
        "message of an error event.",
    )
-    input: Optional[Union[str, dict[str, Any], list, None]] = Field(
+    input: Optional[Union[str, Mapping[str, Any], list, None]] = Field(
        default=None, description="The input of the span. Can be any JSON object."
    )
-    output: Optional[Union[str, dict[str, Any], list, None]] = Field(
+    output: Optional[Union[str, Mapping[str, Any], list, None]] = Field(
        default=None, description="The output of the span. Can be any JSON object."
    )
    version: Optional[str] = Field(
@@ -1,11 +1,10 @@
-import json
 import logging
 import os
 from datetime import datetime, timedelta
 from typing import Optional

 from langfuse import Langfuse  # type: ignore
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session, sessionmaker

 from core.ops.base_trace_instance import BaseTraceInstance
 from core.ops.entities.config_entity import LangfuseConfig
@@ -30,8 +29,9 @@ from core.ops.langfuse_trace.entities.langfuse_trace_entity import (
 )
 from core.ops.utils import filter_none_values
 from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
+from core.workflow.nodes.enums import NodeType
 from extensions.ext_database import db
-from models.model import EndUser
+from models import Account, App, EndUser, WorkflowNodeExecutionTriggeredFrom

 logger = logging.getLogger(__name__)

@@ -113,8 +113,29 @@ class LangFuseDataTrace(BaseTraceInstance):

        # through workflow_run_id get all_nodes_execution using repository
        session_factory = sessionmaker(bind=db.engine)
+        # Find the app's creator account
+        with Session(db.engine, expire_on_commit=False) as session:
+            # Get the app to find its creator
+            app_id = trace_info.metadata.get("app_id")
+            if not app_id:
+                raise ValueError("No app_id found in trace_info metadata")
+
+            app = session.query(App).filter(App.id == app_id).first()
+            if not app:
+                raise ValueError(f"App with id {app_id} not found")
+
+            if not app.created_by:
+                raise ValueError(f"App with id {app_id} has no creator (created_by is None)")
+
+            service_account = session.query(Account).filter(Account.id == app.created_by).first()
+            if not service_account:
+                raise ValueError(f"Creator account with id {app.created_by} not found for app {app_id}")
+
        workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
-            session_factory=session_factory, tenant_id=trace_info.tenant_id
+            session_factory=session_factory,
+            user=service_account,
+            app_id=trace_info.metadata.get("app_id"),
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
        )

        # Get all executions for this workflow run
@@ -124,23 +145,22 @@ class LangFuseDataTrace(BaseTraceInstance):

        for node_execution in workflow_node_executions:
            node_execution_id = node_execution.id
-            tenant_id = node_execution.tenant_id
-            app_id = node_execution.app_id
+            tenant_id = trace_info.tenant_id  # Use from trace_info instead
+            app_id = trace_info.metadata.get("app_id")  # Use from trace_info instead
            node_name = node_execution.title
            node_type = node_execution.node_type
            status = node_execution.status
-            if node_type == "llm":
-                inputs = (
-                    json.loads(node_execution.process_data).get("prompts", {}) if node_execution.process_data else {}
-                )
+            if node_type == NodeType.LLM:
+                inputs = node_execution.process_data.get("prompts", {}) if node_execution.process_data else {}
            else:
-                inputs = json.loads(node_execution.inputs) if node_execution.inputs else {}
-            outputs = json.loads(node_execution.outputs) if node_execution.outputs else {}
+                inputs = node_execution.inputs if node_execution.inputs else {}
+            outputs = node_execution.outputs if node_execution.outputs else {}
            created_at = node_execution.created_at or datetime.now()
            elapsed_time = node_execution.elapsed_time
            finished_at = created_at + timedelta(seconds=elapsed_time)

-            metadata = json.loads(node_execution.execution_metadata) if node_execution.execution_metadata else {}
+            execution_metadata = node_execution.metadata if node_execution.metadata else {}
+            metadata = {str(k): v for k, v in execution_metadata.items()}
            metadata.update(
                {
                    "workflow_run_id": trace_info.workflow_run_id,
@@ -152,7 +172,7 @@ class LangFuseDataTrace(BaseTraceInstance):
                    "status": status,
                }
            )
-            process_data = json.loads(node_execution.process_data) if node_execution.process_data else {}
+            process_data = node_execution.process_data if node_execution.process_data else {}
            model_provider = process_data.get("model_provider", None)
            model_name = process_data.get("model_name", None)
            if model_provider is not None and model_name is not None:
@@ -1,3 +1,4 @@
+from collections.abc import Mapping
 from datetime import datetime
 from enum import StrEnum
 from typing import Any, Optional, Union
@@ -30,8 +31,8 @@ class LangSmithMultiModel(BaseModel):

 class LangSmithRunModel(LangSmithTokenUsage, LangSmithMultiModel):
    name: Optional[str] = Field(..., description="Name of the run")
-    inputs: Optional[Union[str, dict[str, Any], list, None]] = Field(None, description="Inputs of the run")
-    outputs: Optional[Union[str, dict[str, Any], list, None]] = Field(None, description="Outputs of the run")
+    inputs: Optional[Union[str, Mapping[str, Any], list, None]] = Field(None, description="Inputs of the run")
+    outputs: Optional[Union[str, Mapping[str, Any], list, None]] = Field(None, description="Outputs of the run")
    run_type: LangSmithRunType = Field(..., description="Type of the run")
    start_time: Optional[datetime | str] = Field(None, description="Start time of the run")
    end_time: Optional[datetime | str] = Field(None, description="End time of the run")
@@ -1,4 +1,3 @@
-import json
 import logging
 import os
 import uuid
@@ -7,7 +6,7 @@ from typing import Optional, cast

 from langsmith import Client
 from langsmith.schemas import RunBase
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session, sessionmaker

 from core.ops.base_trace_instance import BaseTraceInstance
 from core.ops.entities.config_entity import LangSmithConfig
@@ -29,8 +28,10 @@ from core.ops.langsmith_trace.entities.langsmith_trace_entity import (
 )
 from core.ops.utils import filter_none_values, generate_dotted_order
 from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
+from core.workflow.entities.node_entities import NodeRunMetadataKey
+from core.workflow.nodes.enums import NodeType
 from extensions.ext_database import db
-from models.model import EndUser, MessageFile
+from models import Account, App, EndUser, MessageFile, WorkflowNodeExecutionTriggeredFrom

 logger = logging.getLogger(__name__)

@@ -137,8 +138,29 @@ class LangSmithDataTrace(BaseTraceInstance):

        # through workflow_run_id get all_nodes_execution using repository
        session_factory = sessionmaker(bind=db.engine)
+        # Find the app's creator account
+        with Session(db.engine, expire_on_commit=False) as session:
+            # Get the app to find its creator
+            app_id = trace_info.metadata.get("app_id")
+            if not app_id:
+                raise ValueError("No app_id found in trace_info metadata")
+
+            app = session.query(App).filter(App.id == app_id).first()
+            if not app:
+                raise ValueError(f"App with id {app_id} not found")
+
+            if not app.created_by:
+                raise ValueError(f"App with id {app_id} has no creator (created_by is None)")
+
+            service_account = session.query(Account).filter(Account.id == app.created_by).first()
+            if not service_account:
+                raise ValueError(f"Creator account with id {app.created_by} not found for app {app_id}")
+
        workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
-            session_factory=session_factory, tenant_id=trace_info.tenant_id, app_id=trace_info.metadata.get("app_id")
+            session_factory=session_factory,
+            user=service_account,
+            app_id=trace_info.metadata.get("app_id"),
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
        )

        # Get all executions for this workflow run
@@ -148,27 +170,23 @@ class LangSmithDataTrace(BaseTraceInstance):

        for node_execution in workflow_node_executions:
            node_execution_id = node_execution.id
-            tenant_id = node_execution.tenant_id
-            app_id = node_execution.app_id
+            tenant_id = trace_info.tenant_id  # Use from trace_info instead
+            app_id = trace_info.metadata.get("app_id")  # Use from trace_info instead
            node_name = node_execution.title
            node_type = node_execution.node_type
            status = node_execution.status
-            if node_type == "llm":
-                inputs = (
-                    json.loads(node_execution.process_data).get("prompts", {}) if node_execution.process_data else {}
-                )
+            if node_type == NodeType.LLM:
+                inputs = node_execution.process_data.get("prompts", {}) if node_execution.process_data else {}
            else:
-                inputs = json.loads(node_execution.inputs) if node_execution.inputs else {}
-            outputs = json.loads(node_execution.outputs) if node_execution.outputs else {}
+                inputs = node_execution.inputs if node_execution.inputs else {}
+            outputs = node_execution.outputs if node_execution.outputs else {}
            created_at = node_execution.created_at or datetime.now()
            elapsed_time = node_execution.elapsed_time
            finished_at = created_at + timedelta(seconds=elapsed_time)

-            execution_metadata = (
-                json.loads(node_execution.execution_metadata) if node_execution.execution_metadata else {}
-            )
-            node_total_tokens = execution_metadata.get("total_tokens", 0)
-            metadata = execution_metadata.copy()
+            execution_metadata = node_execution.metadata if node_execution.metadata else {}
+            node_total_tokens = execution_metadata.get(NodeRunMetadataKey.TOTAL_TOKENS) or 0
+            metadata = {str(key): value for key, value in execution_metadata.items()}
            metadata.update(
                {
                    "workflow_run_id": trace_info.workflow_run_id,
@@ -181,7 +199,7 @@ class LangSmithDataTrace(BaseTraceInstance):
                }
            )

-            process_data = json.loads(node_execution.process_data) if node_execution.process_data else {}
+            process_data = node_execution.process_data if node_execution.process_data else {}

            if process_data and process_data.get("model_mode") == "chat":
                run_type = LangSmithRunType.llm
@@ -191,7 +209,7 @@ class LangSmithDataTrace(BaseTraceInstance):
                        "ls_model_name": process_data.get("model_name", ""),
                    }
                )
-            elif node_type == "knowledge-retrieval":
+            elif node_type == NodeType.KNOWLEDGE_RETRIEVAL:
                run_type = LangSmithRunType.retriever
            else:
                run_type = LangSmithRunType.tool
@@ -1,4 +1,3 @@
-import json
 import logging
 import os
 import uuid
@@ -7,7 +6,7 @@ from typing import Optional, cast

 from opik import Opik, Trace
 from opik.id_helpers import uuid4_to_uuid7
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import Session, sessionmaker

 from core.ops.base_trace_instance import BaseTraceInstance
 from core.ops.entities.config_entity import OpikConfig
@@ -23,8 +22,10 @@ from core.ops.entities.trace_entity import (
    WorkflowTraceInfo,
 )
 from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
+from core.workflow.entities.node_entities import NodeRunMetadataKey
+from core.workflow.nodes.enums import NodeType
 from extensions.ext_database import db
-from models.model import EndUser, MessageFile
+from models import Account, App, EndUser, MessageFile, WorkflowNodeExecutionTriggeredFrom

 logger = logging.getLogger(__name__)

@@ -114,6 +115,7 @@ class OpikDataTrace(BaseTraceInstance):
                "metadata": workflow_metadata,
                "input": wrap_dict("input", trace_info.workflow_run_inputs),
                "output": wrap_dict("output", trace_info.workflow_run_outputs),
+                "thread_id": trace_info.conversation_id,
                "tags": ["message", "workflow"],
                "project_name": self.project,
            }
@@ -143,6 +145,7 @@ class OpikDataTrace(BaseTraceInstance):
                "metadata": workflow_metadata,
                "input": wrap_dict("input", trace_info.workflow_run_inputs),
                "output": wrap_dict("output", trace_info.workflow_run_outputs),
+                "thread_id": trace_info.conversation_id,
                "tags": ["workflow"],
                "project_name": self.project,
            }
@@ -150,8 +153,29 @@ class OpikDataTrace(BaseTraceInstance):

        # through workflow_run_id get all_nodes_execution using repository
        session_factory = sessionmaker(bind=db.engine)
+        # Find the app's creator account
+        with Session(db.engine, expire_on_commit=False) as session:
+            # Get the app to find its creator
+            app_id = trace_info.metadata.get("app_id")
+            if not app_id:
+                raise ValueError("No app_id found in trace_info metadata")
+
+            app = session.query(App).filter(App.id == app_id).first()
+            if not app:
+                raise ValueError(f"App with id {app_id} not found")
+
+            if not app.created_by:
+                raise ValueError(f"App with id {app_id} has no creator (created_by is None)")
+
+            service_account = session.query(Account).filter(Account.id == app.created_by).first()
+            if not service_account:
+                raise ValueError(f"Creator account with id {app.created_by} not found for app {app_id}")
+
        workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
-            session_factory=session_factory, tenant_id=trace_info.tenant_id, app_id=trace_info.metadata.get("app_id")
+            session_factory=session_factory,
+            user=service_account,
+            app_id=trace_info.metadata.get("app_id"),
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
        )

        # Get all executions for this workflow run
@@ -161,26 +185,22 @@ class OpikDataTrace(BaseTraceInstance):

        for node_execution in workflow_node_executions:
            node_execution_id = node_execution.id
-            tenant_id = node_execution.tenant_id
-            app_id = node_execution.app_id
+            tenant_id = trace_info.tenant_id  # Use from trace_info instead
+            app_id = trace_info.metadata.get("app_id")  # Use from trace_info instead
            node_name = node_execution.title
            node_type = node_execution.node_type
            status = node_execution.status
-            if node_type == "llm":
-                inputs = (
-                    json.loads(node_execution.process_data).get("prompts", {}) if node_execution.process_data else {}
-                )
+            if node_type == NodeType.LLM:
+                inputs = node_execution.process_data.get("prompts", {}) if node_execution.process_data else {}
            else:
-                inputs = json.loads(node_execution.inputs) if node_execution.inputs else {}
-            outputs = json.loads(node_execution.outputs) if node_execution.outputs else {}
+                inputs = node_execution.inputs if node_execution.inputs else {}
+            outputs = node_execution.outputs if node_execution.outputs else {}
            created_at = node_execution.created_at or datetime.now()
            elapsed_time = node_execution.elapsed_time
            finished_at = created_at + timedelta(seconds=elapsed_time)

-            execution_metadata = (
-                json.loads(node_execution.execution_metadata) if node_execution.execution_metadata else {}
-            )
-            metadata = execution_metadata.copy()
+            execution_metadata = node_execution.metadata if node_execution.metadata else {}
+            metadata = {str(k): v for k, v in execution_metadata.items()}
            metadata.update(
                {
                    "workflow_run_id": trace_info.workflow_run_id,
@@ -193,7 +213,7 @@ class OpikDataTrace(BaseTraceInstance):
                }
            )

-            process_data = json.loads(node_execution.process_data) if node_execution.process_data else {}
+            process_data = node_execution.process_data if node_execution.process_data else {}

            provider = None
            model = None
@@ -226,7 +246,7 @@ class OpikDataTrace(BaseTraceInstance):
            parent_span_id = trace_info.workflow_app_log_id or trace_info.workflow_run_id

            if not total_tokens:
-                total_tokens = execution_metadata.get("total_tokens", 0)
+                total_tokens = execution_metadata.get(NodeRunMetadataKey.TOTAL_TOKENS) or 0

            span_data = {
                "trace_id": opik_trace_id,
@@ -288,6 +308,7 @@ class OpikDataTrace(BaseTraceInstance):
            "metadata": wrap_metadata(metadata),
            "input": trace_info.inputs,
            "output": message_data.answer,
+            "thread_id": message_data.conversation_id,
            "tags": ["message", str(trace_info.conversation_mode)],
            "project_name": self.project,
        }
@@ -402,6 +423,7 @@ class OpikDataTrace(BaseTraceInstance):
            "metadata": wrap_metadata(trace_info.metadata),
            "input": trace_info.inputs,
            "output": trace_info.outputs,
+            "thread_id": trace_info.conversation_id,
            "tags": ["generate_name"],
            "project_name": self.project,
        }
@@ -16,11 +16,7 @@ from sqlalchemy.orm import Session
 from core.helper.encrypter import decrypt_token, encrypt_token, obfuscated_token
 from core.ops.entities.config_entity import (
    OPS_FILE_PATH,
-    LangfuseConfig,
-    LangSmithConfig,
-    OpikConfig,
    TracingProviderEnum,
-    WeaveConfig,
 )
 from core.ops.entities.trace_entity import (
    DatasetRetrievalTraceInfo,
@@ -33,11 +29,8 @@ from core.ops.entities.trace_entity import (
    TraceTaskName,
    WorkflowTraceInfo,
 )
-from core.ops.langfuse_trace.langfuse_trace import LangFuseDataTrace
-from core.ops.langsmith_trace.langsmith_trace import LangSmithDataTrace
-from core.ops.opik_trace.opik_trace import OpikDataTrace
 from core.ops.utils import get_message_data
-from core.ops.weave_trace.weave_trace import WeaveDataTrace
+from core.workflow.entities.workflow_execution_entities import WorkflowExecution
 from extensions.ext_database import db
 from extensions.ext_storage import storage
 from models.model import App, AppModelConfig, Conversation, Message, MessageFile, TraceAppConfig
@@ -45,36 +38,58 @@ from models.workflow import WorkflowAppLog, WorkflowRun
 from tasks.ops_trace_task import process_trace_tasks


-def build_opik_trace_instance(config: OpikConfig):
-    return OpikDataTrace(config)
+class OpsTraceProviderConfigMap(dict[str, dict[str, Any]]):
+    def __getitem__(self, provider: str) -> dict[str, Any]:
+        match provider:
+            case TracingProviderEnum.LANGFUSE:
+                from core.ops.entities.config_entity import LangfuseConfig
+                from core.ops.langfuse_trace.langfuse_trace import LangFuseDataTrace
+
+                return {
+                    "config_class": LangfuseConfig,
+                    "secret_keys": ["public_key", "secret_key"],
+                    "other_keys": ["host", "project_key"],
+                    "trace_instance": LangFuseDataTrace,
+                }
+
+            case TracingProviderEnum.LANGSMITH:
+                from core.ops.entities.config_entity import LangSmithConfig
+                from core.ops.langsmith_trace.langsmith_trace import LangSmithDataTrace
+
+                return {
+                    "config_class": LangSmithConfig,
+                    "secret_keys": ["api_key"],
+                    "other_keys": ["project", "endpoint"],
+                    "trace_instance": LangSmithDataTrace,
+                }
+
+            case TracingProviderEnum.OPIK:
+                from core.ops.entities.config_entity import OpikConfig
+                from core.ops.opik_trace.opik_trace import OpikDataTrace
+
+                return {
+                    "config_class": OpikConfig,
+                    "secret_keys": ["api_key"],
+                    "other_keys": ["project", "url", "workspace"],
+                    "trace_instance": OpikDataTrace,
+                }
+
+            case TracingProviderEnum.WEAVE:
+                from core.ops.entities.config_entity import WeaveConfig
+                from core.ops.weave_trace.weave_trace import WeaveDataTrace
+
+                return {
+                    "config_class": WeaveConfig,
+                    "secret_keys": ["api_key"],
+                    "other_keys": ["project", "entity", "endpoint"],
+                    "trace_instance": WeaveDataTrace,
+                }
+
+            case _:
+                raise KeyError(f"Unsupported tracing provider: {provider}")


-provider_config_map: dict[str, dict[str, Any]] = {
-    TracingProviderEnum.LANGFUSE.value: {
-        "config_class": LangfuseConfig,
-        "secret_keys": ["public_key", "secret_key"],
-        "other_keys": ["host", "project_key"],
-        "trace_instance": LangFuseDataTrace,
-    },
-    TracingProviderEnum.LANGSMITH.value: {
-        "config_class": LangSmithConfig,
-        "secret_keys": ["api_key"],
-        "other_keys": ["project", "endpoint"],
-        "trace_instance": LangSmithDataTrace,
-    },
-    TracingProviderEnum.OPIK.value: {
-        "config_class": OpikConfig,
-        "secret_keys": ["api_key"],
-        "other_keys": ["project", "url", "workspace"],
-        "trace_instance": lambda config: build_opik_trace_instance(config),
-    },
-    TracingProviderEnum.WEAVE.value: {
-        "config_class": WeaveConfig,
-        "secret_keys": ["api_key"],
-        "other_keys": ["project", "entity", "endpoint"],
-        "trace_instance": WeaveDataTrace,
-    },
-}
+provider_config_map: dict[str, dict[str, Any]] = OpsTraceProviderConfigMap()


 class OpsTraceManager:
@@ -220,7 +235,11 @@ class OpsTraceManager:
            return None

        tracing_provider = app_ops_trace_config.get("tracing_provider")
-        if tracing_provider is None or tracing_provider not in provider_config_map:
+        if tracing_provider is None:
+            return None
+        try:
+            provider_config_map[tracing_provider]
+        except KeyError:
            return None

        # decrypt_token
@@ -273,8 +292,14 @@ class OpsTraceManager:
        :return:
        """
        # auth check
-        if tracing_provider not in provider_config_map and tracing_provider is not None:
-            raise ValueError(f"Invalid tracing provider: {tracing_provider}")
+        if enabled == True:
+            try:
+                provider_config_map[tracing_provider]
+            except KeyError:
+                raise ValueError(f"Invalid tracing provider: {tracing_provider}")
+        else:
+            if tracing_provider is not None:
+                raise ValueError(f"Invalid tracing provider: {tracing_provider}")

        app_config: Optional[App] = db.session.query(App).filter(App.id == app_id).first()
        if not app_config:
@@ -353,7 +378,7 @@ class TraceTask:
        self,
        trace_type: Any,
        message_id: Optional[str] = None,
-        workflow_run: Optional[WorkflowRun] = None,
+        workflow_execution: Optional[WorkflowExecution] = None,
        conversation_id: Optional[str] = None,
        user_id: Optional[str] = None,
        timer: Optional[Any] = None,
@@ -361,7 +386,7 @@ class TraceTask:
    ):
        self.trace_type = trace_type
        self.message_id = message_id
-        self.workflow_run_id = workflow_run.id if workflow_run else None
+        self.workflow_run_id = workflow_execution.id if workflow_execution else None
        self.conversation_id = conversation_id
        self.user_id = user_id
        self.timer = timer
@@ -1,3 +1,4 @@
+from collections.abc import Mapping
 from typing import Any, Optional, Union

 from pydantic import BaseModel, Field, field_validator
@@ -19,8 +20,8 @@ class WeaveMultiModel(BaseModel):
 class WeaveTraceModel(WeaveTokenUsage, WeaveMultiModel):
    id: str = Field(..., description="ID of the trace")
    op: str = Field(..., description="Name of the operation")
-    inputs: Optional[Union[str, dict[str, Any], list, None]] = Field(None, description="Inputs of the trace")
-    outputs: Optional[Union[str, dict[str, Any], list, None]] = Field(None, description="Outputs of the trace")
+    inputs: Optional[Union[str, Mapping[str, Any], list, None]] = Field(None, description="Inputs of the trace")
+    outputs: Optional[Union[str, Mapping[str, Any], list, None]] = Field(None, description="Outputs of the trace")
    attributes: Optional[Union[str, dict[str, Any], list, None]] = Field(
        None, description="Metadata and attributes associated with trace"
    )
@@ -1,4 +1,3 @@
-import json
 import logging
 import os
 import uuid
@@ -7,6 +6,7 @@ from typing import Any, Optional, cast

 import wandb
 import weave
+from sqlalchemy.orm import Session, sessionmaker

 from core.ops.base_trace_instance import BaseTraceInstance
 from core.ops.entities.config_entity import WeaveConfig
@@ -22,9 +22,11 @@ from core.ops.entities.trace_entity import (
    WorkflowTraceInfo,
 )
 from core.ops.weave_trace.entities.weave_trace_entity import WeaveTraceModel
+from core.repositories import SQLAlchemyWorkflowNodeExecutionRepository
+from core.workflow.entities.node_entities import NodeRunMetadataKey
+from core.workflow.nodes.enums import NodeType
 from extensions.ext_database import db
-from models.model import EndUser, MessageFile
-from models.workflow import WorkflowNodeExecution
+from models import Account, App, EndUser, MessageFile, WorkflowNodeExecutionTriggeredFrom

 logger = logging.getLogger(__name__)

@@ -128,58 +130,57 @@ class WeaveDataTrace(BaseTraceInstance):

        self.start_call(workflow_run, parent_run_id=trace_info.message_id)

-        # through workflow_run_id get all_nodes_execution
-        workflow_nodes_execution_id_records = (
-            db.session.query(WorkflowNodeExecution.id)
-            .filter(WorkflowNodeExecution.workflow_run_id == trace_info.workflow_run_id)
-            .all()
+        # through workflow_run_id get all_nodes_execution using repository
+        session_factory = sessionmaker(bind=db.engine)
+        # Find the app's creator account
+        with Session(db.engine, expire_on_commit=False) as session:
+            # Get the app to find its creator
+            app_id = trace_info.metadata.get("app_id")
+            if not app_id:
+                raise ValueError("No app_id found in trace_info metadata")
+
+            app = session.query(App).filter(App.id == app_id).first()
+            if not app:
+                raise ValueError(f"App with id {app_id} not found")
+
+            if not app.created_by:
+                raise ValueError(f"App with id {app_id} has no creator (created_by is None)")
+
+            service_account = session.query(Account).filter(Account.id == app.created_by).first()
+            if not service_account:
+                raise ValueError(f"Creator account with id {app.created_by} not found for app {app_id}")
+
+        workflow_node_execution_repository = SQLAlchemyWorkflowNodeExecutionRepository(
+            session_factory=session_factory,
+            user=service_account,
+            app_id=trace_info.metadata.get("app_id"),
+            triggered_from=WorkflowNodeExecutionTriggeredFrom.WORKFLOW_RUN,
        )

-        for node_execution_id_record in workflow_nodes_execution_id_records:
-            node_execution = (
-                db.session.query(
-                    WorkflowNodeExecution.id,
-                    WorkflowNodeExecution.tenant_id,
-                    WorkflowNodeExecution.app_id,
-                    WorkflowNodeExecution.title,
-                    WorkflowNodeExecution.node_type,
-                    WorkflowNodeExecution.status,
-                    WorkflowNodeExecution.inputs,
-                    WorkflowNodeExecution.outputs,
-                    WorkflowNodeExecution.created_at,
-                    WorkflowNodeExecution.elapsed_time,
-                    WorkflowNodeExecution.process_data,
-                    WorkflowNodeExecution.execution_metadata,
-                )
-                .filter(WorkflowNodeExecution.id == node_execution_id_record.id)
-                .first()
-            )
-
-            if not node_execution:
-                continue
+        # Get all executions for this workflow run
+        workflow_node_executions = workflow_node_execution_repository.get_by_workflow_run(
+            workflow_run_id=trace_info.workflow_run_id
+        )

+        for node_execution in workflow_node_executions:
            node_execution_id = node_execution.id
-            tenant_id = node_execution.tenant_id
-            app_id = node_execution.app_id
+            tenant_id = trace_info.tenant_id  # Use from trace_info instead
+            app_id = trace_info.metadata.get("app_id")  # Use from trace_info instead
            node_name = node_execution.title
            node_type = node_execution.node_type
            status = node_execution.status
-            if node_type == "llm":
-                inputs = (
-                    json.loads(node_execution.process_data).get("prompts", {}) if node_execution.process_data else {}
-                )
+            if node_type == NodeType.LLM:
+                inputs = node_execution.process_data.get("prompts", {}) if node_execution.process_data else {}
            else:
-                inputs = json.loads(node_execution.inputs) if node_execution.inputs else {}
-            outputs = json.loads(node_execution.outputs) if node_execution.outputs else {}
+                inputs = node_execution.inputs if node_execution.inputs else {}
+            outputs = node_execution.outputs if node_execution.outputs else {}
            created_at = node_execution.created_at or datetime.now()
            elapsed_time = node_execution.elapsed_time
            finished_at = created_at + timedelta(seconds=elapsed_time)

-            execution_metadata = (
-                json.loads(node_execution.execution_metadata) if node_execution.execution_metadata else {}
-            )
-            node_total_tokens = execution_metadata.get("total_tokens", 0)
-            attributes = execution_metadata.copy()
+            execution_metadata = node_execution.metadata if node_execution.metadata else {}
+            node_total_tokens = execution_metadata.get(NodeRunMetadataKey.TOTAL_TOKENS) or 0
+            attributes = {str(k): v for k, v in execution_metadata.items()}
            attributes.update(
                {
                    "workflow_run_id": trace_info.workflow_run_id,
@@ -192,7 +193,7 @@ class WeaveDataTrace(BaseTraceInstance):
                }
            )

-            process_data = json.loads(node_execution.process_data) if node_execution.process_data else {}
+            process_data = node_execution.process_data if node_execution.process_data else {}
            if process_data and process_data.get("model_mode") == "chat":
                attributes.update(
                    {
@@ -64,9 +64,9 @@ class PluginNodeBackwardsInvocation(BaseBackwardsInvocation):
        )

        return {
-            "inputs": execution.inputs_dict,
-            "outputs": execution.outputs_dict,
-            "process_data": execution.process_data_dict,
+            "inputs": execution.inputs,
+            "outputs": execution.outputs,
+            "process_data": execution.process_data,
        }

    @classmethod
@@ -113,7 +113,7 @@ class PluginNodeBackwardsInvocation(BaseBackwardsInvocation):
        )

        return {
-            "inputs": execution.inputs_dict,
-            "outputs": execution.outputs_dict,
-            "process_data": execution.process_data_dict,
+            "inputs": execution.inputs,
+            "outputs": execution.outputs,
+            "process_data": execution.process_data,
        }
@@ -0,0 +1,6 @@
+class EndpointSetupFailedError(ValueError):
+    """
+    Endpoint setup failed error
+    """
+
+    pass
@@ -24,7 +24,7 @@ class EndpointProviderDeclaration(BaseModel):
    """

    settings: list[ProviderConfig] = Field(default_factory=list)
-    endpoints: Optional[list[EndpointDeclaration]] = Field(default_factory=list)
+    endpoints: Optional[list[EndpointDeclaration]] = Field(default_factory=list[EndpointDeclaration])


 class EndpointEntity(BasePluginEntity):
@@ -52,7 +52,7 @@ class PluginResourceRequirements(BaseModel):
        model: Optional[Model] = Field(default=None)
        node: Optional[Node] = Field(default=None)
        endpoint: Optional[Endpoint] = Field(default=None)
-        storage: Storage = Field(default=None)
+        storage: Optional[Storage] = Field(default=None)

    permission: Optional[Permission] = Field(default=None)

@@ -66,9 +66,9 @@ class PluginCategory(enum.StrEnum):

 class PluginDeclaration(BaseModel):
    class Plugins(BaseModel):
-        tools: Optional[list[str]] = Field(default_factory=list)
-        models: Optional[list[str]] = Field(default_factory=list)
-        endpoints: Optional[list[str]] = Field(default_factory=list)
+        tools: Optional[list[str]] = Field(default_factory=list[str])
+        models: Optional[list[str]] = Field(default_factory=list[str])
+        endpoints: Optional[list[str]] = Field(default_factory=list[str])

    class Meta(BaseModel):
        minimum_dify_version: Optional[str] = Field(default=None, pattern=r"^\d{1,4}(\.\d{1,4}){1,3}(-\w{1,16})?$")
@@ -84,6 +84,7 @@ class PluginDeclaration(BaseModel):
    resource: PluginResourceRequirements
    plugins: Plugins
    tags: list[str] = Field(default_factory=list)
+    repo: Optional[str] = Field(default=None)
    verified: bool = Field(default=False)
    tool: Optional[ToolProviderEntity] = None
    model: Optional[ProviderEntity] = None
@@ -9,7 +9,7 @@ from core.agent.plugin_entities import AgentProviderEntityWithPlugin
 from core.model_runtime.entities.model_entities import AIModelEntity
 from core.model_runtime.entities.provider_entities import ProviderEntity
 from core.plugin.entities.base import BasePluginEntity
-from core.plugin.entities.plugin import PluginDeclaration
+from core.plugin.entities.plugin import PluginDeclaration, PluginEntity
 from core.tools.entities.common_entities import I18nObject
 from core.tools.entities.tool_entities import ToolProviderEntityWithPlugin

@@ -167,3 +167,8 @@ class PluginOAuthAuthorizationUrlResponse(BaseModel):

 class PluginOAuthCredentialsResponse(BaseModel):
    credentials: Mapping[str, Any] = Field(description="The credentials of the OAuth.")
+
+
+class PluginListResponse(BaseModel):
+    list: list[PluginEntity]
+    total: int
@@ -55,8 +55,8 @@ class RequestInvokeLLM(BaseRequestInvokeModel):
    mode: str
    completion_params: dict[str, Any] = Field(default_factory=dict)
    prompt_messages: list[PromptMessage] = Field(default_factory=list)
-    tools: Optional[list[PromptMessageTool]] = Field(default_factory=list)
-    stop: Optional[list[str]] = Field(default_factory=list)
+    tools: Optional[list[PromptMessageTool]] = Field(default_factory=list[PromptMessageTool])
+    stop: Optional[list[str]] = Field(default_factory=list[str])
    stream: Optional[bool] = False

    model_config = ConfigDict(protected_namespaces=())
@@ -17,6 +17,7 @@ from core.model_runtime.errors.invoke import (
    InvokeServerUnavailableError,
 )
 from core.model_runtime.errors.validate import CredentialsValidateFailedError
+from core.plugin.endpoint.exc import EndpointSetupFailedError
 from core.plugin.entities.plugin_daemon import PluginDaemonBasicResponse, PluginDaemonError, PluginDaemonInnerError
 from core.plugin.impl.exc import (
    PluginDaemonBadRequestError,
@@ -219,6 +220,8 @@ class BasePluginClient:
                        raise InvokeServerUnavailableError(description=args.get("description"))
                    case CredentialsValidateFailedError.__name__:
                        raise CredentialsValidateFailedError(error_object.get("message"))
+                    case EndpointSetupFailedError.__name__:
+                        raise EndpointSetupFailedError(error_object.get("message"))
                    case _:
                        raise PluginInvokeError(description=message)
            case PluginDaemonInternalServerError.__name__:
--- a/Show More
+++ b/Show More