Update docker/docker-compose.yaml

2026-07-28 23:48:27 +00:00 · 2026-07-28 23:34:50 +00:00 · 2026-07-28 21:50:33 +00:00 · 2026-07-28 21:31:41 +00:00 · 2026-07-28 21:27:27 +00:00 · 2026-07-28 21:23:54 +00:00
603 changed files with 33036 additions and 21497 deletions
@@ -29,13 +29,13 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: ${{ matrix.python-version }}
@@ -88,13 +88,13 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: ${{ matrix.python-version }}
@@ -139,13 +139,13 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: "3.12"
@@ -20,7 +20,7 @@ jobs:
        run: echo "autofix.ci updates pull request branches, not merge group refs."

      - if: github.event_name != 'merge_group'
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1

      - name: Check Docker Compose inputs
        if: github.event_name != 'merge_group'
@@ -84,12 +84,12 @@ jobs:
            dify-agent/pyproject.toml
            dify-agent/uv.lock
      - if: github.event_name != 'merge_group'
-        uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
+        uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
        with:
          python-version: "3.11"

      - if: github.event_name != 'merge_group'
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0

      - name: Generate Docker Compose
        if: github.event_name != 'merge_group' && steps.docker-compose-changes.outputs.any_changed == 'true'
@@ -97,7 +97,7 @@ jobs:
          echo "PLATFORM_PAIR=${platform//\//-}" >> $GITHUB_ENV

      - name: Login to Docker Hub
-        uses: docker/login-action@af1e73f918a031802d376d3c8bbc3fe56130a9b0 # v4.4.0
+        uses: docker/login-action@abd2ef45e78c5afb21d64d4ca52ee8550d9572c7 # v4.5.1
        with:
          username: ${{ env.DOCKERHUB_USER }}
          password: ${{ env.DOCKERHUB_TOKEN }}
@@ -199,7 +199,7 @@ jobs:
          merge-multiple: true

      - name: Login to Docker Hub
-        uses: docker/login-action@af1e73f918a031802d376d3c8bbc3fe56130a9b0 # v4.4.0
+        uses: docker/login-action@abd2ef45e78c5afb21d64d4ca52ee8550d9572c7 # v4.5.1
        with:
          username: ${{ env.DOCKERHUB_USER }}
          password: ${{ env.DOCKERHUB_TOKEN }}
@@ -79,7 +79,7 @@ jobs:
      ws2_app_id:                 ${{ steps.out.outputs.DIFY_E2E_WS2_APP_ID }}

    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
        with:
          ref: ${{ inputs.cli_ref || github.ref }}
          persist-credentials: false
@@ -123,7 +123,7 @@ jobs:
        shell: bash

    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
        with:
          ref: ${{ inputs.cli_ref || github.ref }}
          persist-credentials: false
@@ -170,7 +170,7 @@ jobs:
        shell: bash

    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
        with:
          ref: ${{ inputs.cli_ref || github.ref }}
          persist-credentials: false
@@ -233,7 +233,7 @@ jobs:
        shell: bash

    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
        with:
          ref: ${{ inputs.cli_ref || github.ref }}
          persist-credentials: false
@@ -295,7 +295,7 @@ jobs:
        shell: bash

    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
        with:
          ref: ${{ inputs.cli_ref || github.ref }}
          persist-credentials: false
@@ -351,7 +351,7 @@ jobs:
        shell: bash

    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v4
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v4
        with:
          ref: ${{ inputs.cli_ref || github.ref }}
          persist-credentials: false
@@ -23,7 +23,7 @@ jobs:
        working-directory: ./cli
    steps:
      - name: Checkout
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false
          fetch-depth: 0
@@ -35,7 +35,7 @@ jobs:
      dify_tag: ${{ steps.resolve.outputs.dify_tag }}
    steps:
      - name: Checkout
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -98,7 +98,7 @@ jobs:
      DIFY_TAG: ${{ needs.validate.outputs.dify_tag }}
    steps:
      - name: Checkout
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false
          fetch-depth: 1
@@ -24,7 +24,7 @@ jobs:
        shell: bash
    steps:
      - name: Checkout cli ref
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          ref: ${{ inputs.cli_ref || github.ref }}
          persist-credentials: false
@@ -30,7 +30,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -13,13 +13,13 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: "3.12"
@@ -63,13 +63,13 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: "3.12"
@@ -1,6 +1,7 @@
 name: Deploy Knowledge

 permissions:
+  actions: read
  contents: read

 on:
@@ -18,6 +19,48 @@ jobs:
      github.event.workflow_run.conclusion == 'success' &&
      github.event.workflow_run.head_branch == 'deploy/konwledge'
    steps:
+      - name: Wait for KnowledgeFS CI
+        uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0
+        timeout-minutes: 35
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          script: |
+            const workflowId = "knowledge-fs-ci.yml";
+            const headBranch = context.payload.workflow_run.head_branch;
+            const headSha = context.payload.workflow_run.head_sha;
+            const deadline = Date.now() + 30 * 60 * 1000;
+            const pollIntervalMs = 15 * 1000;
+
+            while (Date.now() < deadline) {
+              const { data } = await github.rest.actions.listWorkflowRuns({
+                owner: context.repo.owner,
+                repo: context.repo.repo,
+                workflow_id: workflowId,
+                branch: headBranch,
+                event: "push",
+                head_sha: headSha,
+                per_page: 10,
+              });
+              const run = data.workflow_runs[0];
+
+              if (!run) {
+                core.info(`Waiting for ${workflowId} to start for ${headSha}.`);
+              } else if (run.status !== "completed") {
+                core.info(`Waiting for ${run.html_url}; current status is ${run.status}.`);
+              } else if (run.conclusion !== "success") {
+                throw new Error(
+                  `${workflowId} did not succeed for ${headSha}: ${run.conclusion} (${run.html_url})`,
+                );
+              } else {
+                core.info(`KnowledgeFS CI succeeded for ${headSha}: ${run.html_url}`);
+                return;
+              }
+
+              await new Promise((resolve) => setTimeout(resolve, pollIntervalMs));
+            }
+
+            throw new Error(`Timed out waiting for ${workflowId} to succeed for ${headSha}.`);
+
      - name: Deploy to server
        uses: appleboy/ssh-action@0ff4204d59e8e51228ff73bce53f80d53301dee2 # v1.2.5
        with:
@@ -24,7 +24,7 @@ jobs:
    name: Require cherry-pick provenance
    runs-on: depot-ubuntu-24.04
    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0

@@ -9,6 +9,6 @@ jobs:
      pull-requests: write
    runs-on: depot-ubuntu-24.04
    steps:
-      - uses: actions/labeler@b8dd2d9be0f68b860e7dae5dae7d772984eacd6d # v6.2.0
+      - uses: actions/labeler@bf12e9b00b37c5c0ca2b87b79b2daf7891dbda13 # v7.0.0
        with:
          sync-labels: true
@@ -47,7 +47,7 @@ jobs:
      migration-changed: ${{ steps.changes.outputs.migration }}
      sandbox-runtime-changed: ${{ steps.changes.outputs.sandbox-runtime }}
    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
      - uses: dorny/paths-filter@7b450fff21473bca461d4b92ce414b9d0420d706 # v4.0.2
        id: changes
        with:
@@ -18,7 +18,7 @@ jobs:
    outputs:
      external-e2e-changed: ${{ steps.changes.outputs.external_e2e }}
    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
      - uses: dorny/paths-filter@7b450fff21473bca461d4b92ce414b9d0420d706 # v4.0.2
        id: changes
        with:
@@ -17,12 +17,12 @@ jobs:
      pull-requests: write
    steps:
      - name: Checkout PR branch
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0

      - name: Setup Python & UV
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true

@@ -21,10 +21,10 @@ jobs:
    if: ${{ github.event.workflow_run.conclusion == 'success' && github.event.workflow_run.pull_requests[0].head.repo.full_name != github.repository }}
    steps:
      - name: Checkout default branch (trusted code)
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1

      - name: Setup Python & UV
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true

@@ -17,12 +17,12 @@ jobs:
      pull-requests: write
    steps:
      - name: Checkout PR branch
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0

      - name: Setup Python & UV
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true

@@ -21,7 +21,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false
@@ -45,7 +45,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false
@@ -72,7 +72,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false
@@ -23,7 +23,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false
          fetch-depth: 0
@@ -45,7 +45,7 @@ jobs:

      - name: Setup UV and Python
        if: steps.changed-files.outputs.any_changed == 'true'
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: false
          python-version: "3.12"
@@ -93,7 +93,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -144,7 +144,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -186,7 +186,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          persist-credentials: false
@@ -24,7 +24,7 @@ jobs:
        working-directory: sdks/nodejs-client

    steps:
-      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -40,7 +40,7 @@ jobs:

    steps:
      - name: Checkout repository
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0
          token: ${{ secrets.GITHUB_TOKEN }}
@@ -158,7 +158,7 @@ jobs:

      - name: Run Claude Code for Translation Sync
        if: steps.context.outputs.CHANGED_FILES != ''
-        uses: anthropics/claude-code-action@af0559ee4f514d1ef21826982bed13f7edc3c35e # v1.0.178
+        uses: anthropics/claude-code-action@be7b93b1907a4abad570368f3c74b6fe3807510b # v1.0.183
        with:
          anthropic_api_key: ${{ secrets.ANTHROPIC_API_KEY }}
          github_token: ${{ secrets.GITHUB_TOKEN }}
@@ -21,7 +21,7 @@ jobs:

    steps:
      - name: Checkout repository
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          fetch-depth: 0

@@ -24,7 +24,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -36,7 +36,7 @@ jobs:
          remove_tool_cache: true

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: ${{ matrix.python-version }}
@@ -21,7 +21,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -33,7 +33,7 @@ jobs:
          remove_tool_cache: true

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: ${{ matrix.python-version }}
@@ -26,7 +26,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -34,7 +34,7 @@ jobs:
        uses: ./.github/actions/setup-web

      - name: Setup UV and Python
-        uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
+        uses: astral-sh/setup-uv@c771a70e6277c0a99b617c7a806ffedaca235ff9 # v9.0.0
        with:
          enable-cache: true
          python-version: "3.12"
@@ -63,6 +63,7 @@ jobs:
          E2E_ADMIN_PASSWORD: E2eAdmin12345
          E2E_FORCE_WEB_BUILD: "1"
          E2E_INIT_PASSWORD: E2eInit12345
+          E2E_START_AGENT_BACKEND: "1"
        run: vp run e2e:full

      - name: Preserve Chromium E2E report and logs
@@ -138,21 +139,6 @@ jobs:
            exit 1
          fi

-          teardown_external_runtime() {
-            local run_status=$?
-            trap - EXIT
-            if ! vp run e2e:middleware:down; then
-              echo "::error title=E2E teardown failed::External runtime middleware did not shut down cleanly."
-              if [[ "$run_status" -eq 0 ]]; then
-                run_status=1
-              fi
-            fi
-            exit "$run_status"
-          }
-
-          trap teardown_external_runtime EXIT
-          vp run e2e:middleware:up
-          vp run e2e:post-merge:prepare
          vp run e2e:post-merge

      - name: Upload Cucumber report
@@ -29,7 +29,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -62,7 +62,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -100,7 +100,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -132,7 +132,7 @@ jobs:

    steps:
      - name: Checkout code
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          persist-credentials: false

@@ -666,6 +666,7 @@ PLUGIN_REMOTE_INSTALL_PORT=5003
 PLUGIN_REMOTE_INSTALL_HOST=localhost
 PLUGIN_MAX_PACKAGE_SIZE=15728640
 PLUGIN_MODEL_SCHEMA_CACHE_TTL=3600
+PLUGIN_MODEL_PROVIDERS_CACHE_ENABLED=true
 PLUGIN_MODEL_PROVIDERS_CACHE_TTL=86400
 # Comma-separated marketplace plugin IDs whose latest versions are installed for newly registered users.
 # Example: langgenius/openai,langgenius/gemini
@@ -677,6 +678,8 @@ INNER_API_KEY_FOR_PLUGIN=QaHbTe77CtuXmsfyhR7+vRjI/+XbV1AaFy691iy+kGDv2Jvy0/eAh8Y

 # Dify Agent backend
 AGENT_BACKEND_BASE_URL=http://localhost:5050
+# Bearer token sent to the Agent backend /runs API. Must match DIFY_AGENT_API_TOKEN on the server side.
+AGENT_BACKEND_API_TOKEN=dify-agent-run-token-for-dev-only
 AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS=30
 AGENT_BACKEND_STREAM_MAX_RECONNECTS=3
 AGENT_BACKEND_RUN_TIMEOUT_SECONDS=1200
@@ -729,7 +732,6 @@ OTEL_MAX_EXPORT_BATCH_SIZE=512
 OTEL_METRIC_EXPORT_INTERVAL=60000
 OTEL_BATCH_EXPORT_TIMEOUT=10000
 OTEL_METRIC_EXPORT_TIMEOUT=30000
-
 # Prevent Clickjacking
 ALLOW_EMBED=false

@@ -99,9 +99,9 @@ ENV VIRTUAL_ENV=/app/api/.venv
 COPY --from=packages --chown=dify:dify ${VIRTUAL_ENV} ${VIRTUAL_ENV}
 ENV PATH="${VIRTUAL_ENV}/bin:${PATH}"

-# Download nltk data
 RUN mkdir -p /usr/local/share/nltk_data \
-    && NLTK_DATA=/usr/local/share/nltk_data python -c "import nltk; nltk.download('punkt'); nltk.download('averaged_perceptron_tagger'); nltk.download('stopwords')" \
+    && NLTK_DATA=/usr/local/share/nltk_data python -m nltk.downloader punkt_tab averaged_perceptron_tagger_eng stopwords \
+    && NLTK_DATA=/usr/local/share/nltk_data python -c "import nltk; nltk.data.find('tokenizers/punkt_tab'); nltk.data.find('taggers/averaged_perceptron_tagger_eng'); nltk.data.find('corpora/stopwords')" \
    && chmod -R 755 /usr/local/share/nltk_data

 ENV TIKTOKEN_CACHE_DIR=/app/api/.tiktoken_cache
@@ -1,10 +1,13 @@
 import logging
 import time
+from collections.abc import Callable
+from typing import NamedTuple

 import socketio
 from flask import request
 from opentelemetry.trace import get_current_span
 from opentelemetry.trace.span import INVALID_SPAN_ID, INVALID_TRACE_ID
+from werkzeug.exceptions import Forbidden, HTTPException, ServiceUnavailable

 from configs import dify_config
 from contexts.wrapper import RecyclableContextVar
@@ -42,6 +45,53 @@ _CONSOLE_EXEMPT_PREFIXES = (
    "/console/api/activate/check",
 )

+_WEBAPP_EXEMPT_PREFIXES = ("/api/system-features",)
+
+_INVALID_LICENSE_STATUSES = (LicenseStatus.INACTIVE, LicenseStatus.EXPIRED, LicenseStatus.LOST)
+
+
+def _session_surface_error(license_status: LicenseStatus | None) -> HTTPException:
+    if license_status is None:
+        return UnauthorizedAndForceLogout("Unable to verify enterprise license. Please contact your administrator.")
+    return UnauthorizedAndForceLogout(f"Enterprise license is {license_status}. Please contact your administrator.")
+
+
+def _bearer_surface_error(license_status: LicenseStatus | None) -> HTTPException:
+    """Token-authed: forcing a logout is meaningless and license state must not leak."""
+    return Forbidden(description="license_required")
+
+
+def _retryable_surface_error(license_status: LicenseStatus | None) -> HTTPException:
+    """Webhook senders retry on 5xx but treat 4xx as permanent, disabling the subscription."""
+    return ServiceUnavailable(description="license_required")
+
+
+class _LicenseGatedSurface(NamedTuple):
+    prefix: str
+    exempt_prefixes: tuple[str, ...]
+    build_error: Callable[[LicenseStatus | None], HTTPException]
+
+
+# /files (plugin-daemon data plane), /inner/api (enterprise control plane) and /health
+# stay ungated: blocking them breaks workflow execution or license recovery itself.
+_LICENSE_GATED_SURFACES = (
+    _LicenseGatedSurface("/console/api/", _CONSOLE_EXEMPT_PREFIXES, _session_surface_error),
+    _LicenseGatedSurface("/api/", _WEBAPP_EXEMPT_PREFIXES, _session_surface_error),
+    _LicenseGatedSurface("/v1", (), _bearer_surface_error),
+    _LicenseGatedSurface("/mcp", (), _bearer_surface_error),
+    _LicenseGatedSurface("/triggers", (), _retryable_surface_error),
+)
+
+
+def _match_license_gated_surface(path: str) -> _LicenseGatedSurface | None:
+    for surface in _LICENSE_GATED_SURFACES:
+        if not path.startswith(surface.prefix):
+            continue
+        if any(path.startswith(exempt) for exempt in surface.exempt_prefixes):
+            return None
+        return surface
+    return None
+

 # ----------------------------
 # Application Factory Function
@@ -62,38 +112,17 @@ def create_flask_app_with_configs() -> DifyApp:
        init_request_context()
        RecyclableContextVar.increment_thread_recycles()

-        # Enterprise license validation for API endpoints (both console and webapp)
-        # When license expires, block all API access except bootstrap endpoints needed
-        # for the frontend to load the license expiration page without infinite reloads.
        if dify_config.ENTERPRISE_ENABLED:
-            is_console_api = request.path.startswith("/console/api/")
-            is_webapp_api = request.path.startswith("/api/")
+            surface = _match_license_gated_surface(request.path)
+            if surface is not None:
+                try:
+                    license_status = EnterpriseService.get_cached_license_status()
+                except Exception:
+                    logger.exception("Failed to check enterprise license status")
+                    license_status = None

-            if is_console_api or is_webapp_api:
-                if is_console_api:
-                    is_exempt = any(request.path.startswith(p) for p in _CONSOLE_EXEMPT_PREFIXES)
-                else:  # webapp API
-                    is_exempt = request.path.startswith("/api/system-features")
-
-                if not is_exempt:
-                    try:
-                        # Check license status (cached — see EnterpriseService for TTL details)
-                        license_status = EnterpriseService.get_cached_license_status()
-                        if license_status in (LicenseStatus.INACTIVE, LicenseStatus.EXPIRED, LicenseStatus.LOST):
-                            raise UnauthorizedAndForceLogout(
-                                f"Enterprise license is {license_status}. Please contact your administrator."
-                            )
-                        if license_status is None:
-                            raise UnauthorizedAndForceLogout(
-                                "Unable to verify enterprise license. Please contact your administrator."
-                            )
-                    except UnauthorizedAndForceLogout:
-                        raise
-                    except Exception:
-                        logger.exception("Failed to check enterprise license status")
-                        raise UnauthorizedAndForceLogout(
-                            "Unable to verify enterprise license. Please contact your administrator."
-                        )
+                if license_status is None or license_status in _INVALID_LICENSE_STATUSES:
+                    raise surface.build_error(license_status)

    # add after request hook for injecting trace headers from OpenTelemetry span context
    # Only adds headers when OTEL is enabled and has valid context
@@ -5,8 +5,6 @@ API adapters: request building from Dify product concepts, a thin client wrapper
 event adaptation for future workflow integration, and deterministic fakes.
 """

-from dify_agent.protocol import RuntimeLayerSpec, extract_runtime_layer_specs
-
 from clients.agent_backend.client import AgentBackendRunClient, DifyAgentBackendRunClient
 from clients.agent_backend.errors import (
    AgentBackendError,
@@ -47,11 +45,6 @@ from clients.agent_backend.request_builder import (
    AgentBackendWorkflowNodeRunInput,
    redact_for_agent_backend_log,
 )
-from clients.agent_backend.session_cleanup import (
-    AgentBackendSessionCleanupPayload,
-    AgentBackendSessionCleanupResult,
-    cleanup_agent_backend_session,
-)

 __all__ = [
    "AGENT_SOUL_PROMPT_LAYER_ID",
@@ -80,8 +73,6 @@ __all__ = [
    "AgentBackendRunRequestBuilder",
    "AgentBackendRunStartedInternalEvent",
    "AgentBackendRunSucceededInternalEvent",
-    "AgentBackendSessionCleanupPayload",
-    "AgentBackendSessionCleanupResult",
    "AgentBackendStreamError",
    "AgentBackendStreamInternalEvent",
    "AgentBackendTransportError",
@@ -90,9 +81,6 @@ __all__ = [
    "DifyAgentBackendRunClient",
    "FakeAgentBackendRunClient",
    "FakeAgentBackendScenario",
-    "RuntimeLayerSpec",
-    "cleanup_agent_backend_session",
    "create_agent_backend_run_client",
-    "extract_runtime_layer_specs",
    "redact_for_agent_backend_log",
 ]
@@ -11,6 +11,7 @@ from clients.agent_backend.fake_client import FakeAgentBackendRunClient, FakeAge
 def create_agent_backend_run_client(
    *,
    base_url: str | None = None,
+    api_token: str | None = None,
    use_fake: bool = False,
    fake_scenario: str | FakeAgentBackendScenario = FakeAgentBackendScenario.SUCCESS,
    stream_read_timeout_seconds: float = 30,
@@ -22,8 +23,11 @@ def create_agent_backend_run_client(
        return FakeAgentBackendRunClient(scenario=FakeAgentBackendScenario(fake_scenario))
    if base_url is None:
        raise ValueError("base_url is required when creating a real Agent backend client")
+    headers: dict[str, str] = {}
+    if api_token:
+        headers["Authorization"] = f"Bearer {api_token}"
    return DifyAgentBackendRunClient(
-        Client(base_url=base_url, stream_timeout=stream_read_timeout_seconds),
+        Client(base_url=base_url, stream_timeout=stream_read_timeout_seconds, headers=headers),
        stream_max_reconnects=stream_max_reconnects,
        stream_timeout_seconds=stream_run_timeout_seconds,
    )
@@ -16,7 +16,6 @@ from collections.abc import Mapping
 from typing import ClassVar, Literal

 from agenton.compositor import CompositorSessionSnapshot
-from agenton.compositor.schemas import LayerSessionSnapshot
 from agenton.layers import ExitIntent
 from agenton_collections.layers.plain import PLAIN_PROMPT_LAYER_TYPE_ID, PromptLayerConfig
 from agenton_collections.layers.pydantic_ai import PYDANTIC_AI_HISTORY_LAYER_TYPE_ID
@@ -37,6 +36,7 @@ from dify_agent.layers.execution_context import (
 )
 from dify_agent.layers.knowledge import DIFY_KNOWLEDGE_BASE_LAYER_TYPE_ID, DifyKnowledgeBaseLayerConfig
 from dify_agent.layers.output import DIFY_OUTPUT_LAYER_TYPE_ID, DifyOutputLayerConfig
+from dify_agent.layers.runtime import DIFY_RUNTIME_LAYER_TYPE_ID, DifyRuntimeLayerConfig
 from dify_agent.layers.shell import DIFY_SHELL_LAYER_TYPE_ID, DifyShellLayerConfig
 from dify_agent.protocol import (
    DIFY_AGENT_HISTORY_LAYER_ID,
@@ -47,7 +47,6 @@ from dify_agent.protocol import (
    LayerExitSignals,
    RunComposition,
    RunLayerSpec,
-    RuntimeLayerSpec,
 )
 from pydantic import BaseModel, ConfigDict, Field, JsonValue, field_validator

@@ -56,6 +55,7 @@ WORKFLOW_NODE_JOB_PROMPT_LAYER_ID = "workflow_node_job_prompt"
 WORKFLOW_USER_PROMPT_LAYER_ID = "workflow_user_prompt"
 AGENT_APP_USER_PROMPT_LAYER_ID = "agent_app_user_prompt"
 DIFY_EXECUTION_CONTEXT_LAYER_ID = "execution_context"
+DIFY_RUNTIME_LAYER_ID = "runtime"
 DIFY_CONFIG_LAYER_ID = "config"
 DIFY_DRIVE_LAYER_ID = "drive"
 DIFY_PLUGIN_TOOLS_LAYER_ID = "tools"
@@ -66,25 +66,11 @@ DIFY_SHELL_LAYER_ID = "shell"
 type AgentConfigVersionKind = Literal["snapshot", "draft", "build_draft"]


-def _filter_snapshot_to_specs(
-    snapshot: CompositorSessionSnapshot,
-    specs: list[RuntimeLayerSpec],
-) -> CompositorSessionSnapshot:
-    """Keep only snapshot layers whose names appear in the cleanup spec list.
-
-    The agenton compositor rejects a snapshot whose layer-name sequence does
-    not match the active composition exactly. Cleanup-replay drops plugin
-    layers, so we must drop the matching snapshot entries here.
-    """
-    kept_names = {spec.name for spec in specs}
-    filtered_layers: list[LayerSessionSnapshot] = [layer for layer in snapshot.layers if layer.name in kept_names]
-    if len(filtered_layers) == len(snapshot.layers):
-        return snapshot
-    return CompositorSessionSnapshot(schema_version=snapshot.schema_version, layers=filtered_layers)
-
-
 def _shell_layer_deps() -> dict[str, str]:
-    return {"execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID}
+    return {
+        "execution_context": DIFY_EXECUTION_CONTEXT_LAYER_ID,
+        "runtime": DIFY_RUNTIME_LAYER_ID,
+    }


 def _drive_layer_deps() -> dict[str, str]:
@@ -214,6 +200,7 @@ class AgentBackendWorkflowNodeRunInput(BaseModel):

    model: AgentBackendModelConfig
    execution_context: DifyExecutionContextLayerConfig
+    backend_binding_ref: str = Field(min_length=1)
    workflow_node_job_prompt: str
    user_prompt: str
    agent_soul_prompt: str | None = None
@@ -231,8 +218,8 @@ class AgentBackendWorkflowNodeRunInput(BaseModel):
    # the Agent Soul configures human involvement; a deferred call ends the run and
    # the workflow pauses via the existing HITL form mechanism (ENG-635).
    ask_human_config: DifyAskHumanLayerConfig | None = None
-    # Inject the sandboxed shell layer (dify.shell). Requires the agent backend
-    # to be wired with a shellctl entrypoint; see configs AGENT_SHELL_ENABLED.
+    # Inject the sandboxed shell graph. Requires a deployment-selected runtime
+    # backend plus the product-resolved persistent Binding.
    include_shell: bool = False
    shell_config: DifyShellLayerConfig | None = None
    session_snapshot: CompositorSessionSnapshot | None = None
@@ -240,7 +227,6 @@ class AgentBackendWorkflowNodeRunInput(BaseModel):
    # (ENG-638). Keyed by the original deferred tool_call_id.
    deferred_tool_results: DeferredToolResultsPayload | None = None
    include_history: bool = True
-    suspend_on_exit: bool = True
    metadata: dict[str, JsonValue] = Field(default_factory=dict)

    model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid", arbitrary_types_allowed=True)
@@ -264,6 +250,7 @@ class AgentBackendAgentAppRunInput(BaseModel):

    model: AgentBackendModelConfig
    execution_context: DifyExecutionContextLayerConfig
+    backend_binding_ref: str = Field(min_length=1)
    user_prompt: str
    agent_soul_prompt: str | None = None
    agent_config_version_kind: AgentConfigVersionKind = "snapshot"
@@ -279,8 +266,8 @@ class AgentBackendAgentAppRunInput(BaseModel):
    # Human-in-the-loop ask_human deferred tool (dify.ask_human). Present only when
    # the Agent Soul configures human involvement (ENG-635).
    ask_human_config: DifyAskHumanLayerConfig | None = None
-    # Inject the sandboxed shell layer (dify.shell). Requires the agent backend
-    # to be wired with a shellctl entrypoint; see configs AGENT_SHELL_ENABLED.
+    # Inject the sandboxed shell graph. Requires a deployment-selected runtime
+    # backend plus the product-resolved persistent Binding.
    include_shell: bool = False
    shell_config: DifyShellLayerConfig | None = None
    session_snapshot: CompositorSessionSnapshot | None = None
@@ -288,7 +275,6 @@ class AgentBackendAgentAppRunInput(BaseModel):
    # (ENG-638). Keyed by the original deferred tool_call_id.
    deferred_tool_results: DeferredToolResultsPayload | None = None
    include_history: bool = True
-    suspend_on_exit: bool = True
    metadata: dict[str, JsonValue] = Field(default_factory=dict)

    model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid", arbitrary_types_allowed=True)
@@ -350,6 +336,14 @@ class AgentBackendRunRequestBuilder:
            run_input.include_shell or run_input.config_layer_config is not None or run_input.drive_config is not None
        )
        if include_shell:
+            layers.append(
+                RunLayerSpec(
+                    name=DIFY_RUNTIME_LAYER_ID,
+                    type=DIFY_RUNTIME_LAYER_TYPE_ID,
+                    metadata=run_input.metadata,
+                    config=DifyRuntimeLayerConfig(backend_binding_ref=run_input.backend_binding_ref),
+                )
+            )
            # Sandboxed bash workspace (dify.shell). It enters before config/drive
            # so eager pulls materialize content in the same filesystem used by
            # model commands.
@@ -481,53 +475,7 @@ class AgentBackendRunRequestBuilder:
            metadata=run_input.metadata,
            session_snapshot=run_input.session_snapshot,
            deferred_tool_results=run_input.deferred_tool_results,
-            on_exit=LayerExitSignals(
-                default=ExitIntent.SUSPEND if run_input.suspend_on_exit else ExitIntent.DELETE,
-            ),
-        )
-
-    def build_cleanup_request(
-        self,
-        *,
-        session_snapshot: CompositorSessionSnapshot,
-        runtime_layer_specs: list[RuntimeLayerSpec],
-        idempotency_key: str | None = None,
-        metadata: dict[str, JsonValue] | None = None,
-    ) -> CreateRunRequest:
-        """Build a lifecycle-only cleanup request that replays the prior layers.
-
-        The agenton compositor enforces that the session snapshot's layer names
-        match the active composition in order, so cleanup must replay the same
-        non-plugin layer graph that produced the snapshot. Plugin layers
-        (``dify.plugin.llm``, ``dify.plugin.tools``) are excluded from both the
-        composition and the snapshot before submission because their configs
-        may carry credentials or runtime-only declarations that are not
-        persisted between runs.
-        """
-        if not runtime_layer_specs:
-            raise ValueError(
-                "build_cleanup_request requires runtime_layer_specs; an empty "
-                "composition would fail the agent backend's snapshot validation."
-            )
-        request_metadata = dict(metadata or {})
-        request_metadata["agent_backend_lifecycle"] = "session_cleanup"
-        layers = [
-            RunLayerSpec(
-                name=spec.name,
-                type=spec.type,
-                deps=dict(spec.deps),
-                metadata=dict(spec.metadata),
-                config=spec.config,
-            )
-            for spec in runtime_layer_specs
-        ]
-        filtered_snapshot = _filter_snapshot_to_specs(session_snapshot, runtime_layer_specs)
-        return CreateRunRequest(
-            composition=RunComposition(layers=layers),
-            idempotency_key=idempotency_key,
-            metadata=request_metadata,
-            session_snapshot=filtered_snapshot,
-            on_exit=LayerExitSignals(default=ExitIntent.DELETE),
+            on_exit=LayerExitSignals(default=ExitIntent.SUSPEND),
        )

    def build_for_workflow_node(self, run_input: AgentBackendWorkflowNodeRunInput) -> CreateRunRequest:
@@ -580,6 +528,14 @@ class AgentBackendRunRequestBuilder:
            run_input.include_shell or run_input.config_layer_config is not None or run_input.drive_config is not None
        )
        if include_shell:
+            layers.append(
+                RunLayerSpec(
+                    name=DIFY_RUNTIME_LAYER_ID,
+                    type=DIFY_RUNTIME_LAYER_TYPE_ID,
+                    metadata=run_input.metadata,
+                    config=DifyRuntimeLayerConfig(backend_binding_ref=run_input.backend_binding_ref),
+                )
+            )
            # Sandboxed bash workspace (dify.shell). It enters before drive so
            # drive can materialize mentioned targets with `dify-agent drive pull`
            # in the same shell-visible filesystem used by model commands.
@@ -713,9 +669,7 @@ class AgentBackendRunRequestBuilder:
            metadata=run_input.metadata,
            session_snapshot=run_input.session_snapshot,
            deferred_tool_results=run_input.deferred_tool_results,
-            on_exit=LayerExitSignals(
-                default=ExitIntent.SUSPEND if run_input.suspend_on_exit else ExitIntent.DELETE,
-            ),
+            on_exit=LayerExitSignals(default=ExitIntent.SUSPEND),
        )


@@ -1,100 +0,0 @@
-"""Shared API-side helper for Agent backend lifecycle-only session cleanup.
-
-Product code owns local row retirement and background-task dispatch. This module
-only adapts persisted cleanup inputs into the public ``dify-agent`` run
-protocol, performs the synchronous ``create_run + wait_run`` loop used by Celery
-workers, and reports whether the backend cleanup succeeded, was skipped, or
-failed.
-"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import ClassVar, Literal
-
-from agenton.compositor import CompositorSessionSnapshot
-from dify_agent.protocol import RuntimeLayerSpec
-from pydantic import BaseModel, ConfigDict, Field, JsonValue
-
-from clients.agent_backend.client import AgentBackendRunClient
-from clients.agent_backend.errors import AgentBackendError
-from clients.agent_backend.request_builder import AgentBackendRunRequestBuilder
-
-
-class AgentBackendSessionCleanupPayload(BaseModel):
-    """Serialized cleanup inputs preserved across API and Celery boundaries."""
-
-    session_snapshot: CompositorSessionSnapshot | None = None
-    runtime_layer_specs: list[RuntimeLayerSpec] = Field(default_factory=list)
-    idempotency_key: str | None = None
-    metadata: dict[str, JsonValue] = Field(default_factory=dict)
-    timeout_seconds: float = 30.0
-
-    model_config: ClassVar[ConfigDict] = ConfigDict(extra="forbid")
-
-
-@dataclass(frozen=True, slots=True)
-class AgentBackendSessionCleanupResult:
-    """Terminal outcome of one backend cleanup attempt."""
-
-    status: Literal["succeeded", "skipped", "failed"]
-    reason: str | None = None
-    cleanup_run_id: str | None = None
-
-    @classmethod
-    def succeeded(cls, cleanup_run_id: str) -> AgentBackendSessionCleanupResult:
-        return cls(status="succeeded", cleanup_run_id=cleanup_run_id)
-
-    @classmethod
-    def skipped(cls, reason: str) -> AgentBackendSessionCleanupResult:
-        return cls(status="skipped", reason=reason)
-
-    @classmethod
-    def failed(cls, reason: str, cleanup_run_id: str | None = None) -> AgentBackendSessionCleanupResult:
-        return cls(status="failed", reason=reason, cleanup_run_id=cleanup_run_id)
-
-
-def cleanup_agent_backend_session(
-    *,
-    payload: AgentBackendSessionCleanupPayload,
-    client: AgentBackendRunClient | None,
-    request_builder: AgentBackendRunRequestBuilder | None = None,
-) -> AgentBackendSessionCleanupResult:
-    """Run lifecycle-only cleanup against the Agent backend and report status."""
-    if client is None:
-        return AgentBackendSessionCleanupResult.skipped("no_agent_backend_client")
-    if payload.session_snapshot is None:
-        return AgentBackendSessionCleanupResult.skipped("missing_session_snapshot")
-    if not payload.runtime_layer_specs:
-        return AgentBackendSessionCleanupResult.skipped("missing_runtime_layer_specs")
-
-    builder = request_builder or AgentBackendRunRequestBuilder()
-    request = builder.build_cleanup_request(
-        session_snapshot=payload.session_snapshot,
-        runtime_layer_specs=payload.runtime_layer_specs,
-        idempotency_key=payload.idempotency_key,
-        metadata=payload.metadata,
-    )
-
-    try:
-        response = client.create_run(request)
-    except AgentBackendError as exc:
-        return AgentBackendSessionCleanupResult.failed(str(exc))
-
-    try:
-        status_response = client.wait_run(response.run_id, timeout_seconds=payload.timeout_seconds)
-    except AgentBackendError as exc:
-        return AgentBackendSessionCleanupResult.failed(str(exc), cleanup_run_id=response.run_id)
-
-    if status_response.status != "succeeded":
-        reason = status_response.error or f"cleanup run ended with status {status_response.status}"
-        return AgentBackendSessionCleanupResult.failed(reason, cleanup_run_id=response.run_id)
-
-    return AgentBackendSessionCleanupResult.succeeded(response.run_id)
-
-
-__all__ = [
-    "AgentBackendSessionCleanupPayload",
-    "AgentBackendSessionCleanupResult",
-    "cleanup_agent_backend_session",
-]
@@ -12,6 +12,11 @@ class AgentBackendConfig(BaseSettings):
        default=None,
    )

+    AGENT_BACKEND_API_TOKEN: str | None = Field(
+        description="Bearer token for authenticating with the Agent backend /runs API.",
+        default=None,
+    )
+
    AGENT_BACKEND_USE_FAKE: bool = Field(
        description="Use the deterministic in-process fake Agent backend client.",
        default=False,
@@ -39,9 +44,8 @@ class AgentBackendConfig(BaseSettings):

    AGENT_SHELL_ENABLED: bool = Field(
        description=(
-            "Inject the dify.shell layer (sandboxed bash workspace) into Agent runs. "
-            "Requires the agent backend to be wired with a shellctl entrypoint before "
-            "shell-using Agent runs are executed."
+            "Inject the Home, Workspace, Sandbox, and Shell runtime layers into Agent runs. "
+            "Requires Dify Agent to have a deployment-selected runtime backend."
        ),
        default=True,
    )
@@ -266,6 +266,12 @@ class PluginConfig(BaseSettings):
        default=60 * 60,
    )

+    PLUGIN_MODEL_PROVIDERS_CACHE_ENABLED: bool = Field(
+        description="Whether tenant plugin model providers are cached in Redis. Disable when plugins are installed "
+        "by a system other than this one, which cannot invalidate the cache when a tenant's plugins change.",
+        default=True,
+    )
+
    PLUGIN_MODEL_PROVIDERS_CACHE_TTL: PositiveInt = Field(
        description="TTL in seconds for caching tenant plugin model providers in Redis",
        default=60 * 60 * 24,
@@ -816,6 +822,41 @@ class UpdateConfig(BaseSettings):
    )


+class CommunityTelemetryConfig(BaseSettings):
+    """
+    Configuration for anonymous self-hosted community telemetry.
+    """
+
+    DISABLE_TELEMETRY: bool = Field(
+        description="Disable anonymous community telemetry",
+        default=False,
+    )
+    DO_NOT_TRACK: bool = Field(
+        description="Respect the standard do-not-track opt-out signal for telemetry",
+        default=False,
+    )
+    TELEMETRY_ENDPOINT: str = Field(
+        description="Endpoint for anonymous community telemetry events",
+        default="https://otel.dify.ai/v1/events",
+    )
+    TELEMETRY_FALLBACK_ENDPOINT: str = Field(
+        description="Fallback endpoint for anonymous community telemetry events",
+        default="https://otel.dify.cn/v1/events",
+    )
+    TELEMETRY_TIMEOUT_SECONDS: PositiveInt = Field(
+        description="HTTP timeout in seconds for anonymous community telemetry requests",
+        default=3,
+    )
+    TELEMETRY_HEARTBEAT_INTERVAL_MINUTES: PositiveInt = Field(
+        description="Celery beat interval in minutes for checking whether heartbeat telemetry is due",
+        default=30,
+    )
+    CI: bool = Field(
+        description="Whether the process is running in CI; telemetry is skipped when true",
+        default=False,
+    )
+
+
 class WorkflowVariableTruncationConfig(BaseSettings):
    WORKFLOW_VARIABLE_TRUNCATION_MAX_SIZE: PositiveInt = Field(
        # 1000 KiB
@@ -1599,6 +1640,7 @@ class FeatureConfig(
    TenantIsolatedTaskQueueConfig,
    ToolConfig,
    UpdateConfig,
+    CommunityTelemetryConfig,
    WorkflowConfig,
    WorkflowNodeExecutionConfig,
    WorkspaceConfig,
@@ -52,7 +52,7 @@ def with_session[T, **P, R](
                        session.commit()
                        return result
                    except Exception:
-                        session.rollback()  # noqa: no-new-controller-sqlalchemy decorator owns transaction rollback
+                        session.rollback()  # guard-ignore: no-new-controller-sqlalchemy -- decorator owns rollback
                        raise

            with session_factory.create_session() as session:
@@ -62,7 +62,7 @@ from libs.datetime_utils import parse_time_range
 from libs.helper import dump_response
 from libs.login import login_required
 from models import Account
-from models.agent import Agent, AgentConfigDraftType, AgentStatus
+from models.agent import Agent, AgentStatus
 from models.agent_config_entities import AgentSoulConfig
 from models.enums import ApiTokenType
 from models.model import ApiToken, App, IconType
@@ -257,7 +257,6 @@ class AgentAppDetailWithSite(GenericAppDetailWithSite):
    debug_conversation_has_messages: bool = False
    debug_conversation_message_count: int = 0
    role: str | None = None
-    active_config_is_published: bool = False


 class AgentDebugConversationRefreshResponse(BaseModel):
@@ -266,13 +265,6 @@ class AgentDebugConversationRefreshResponse(BaseModel):
    debug_conversation_message_count: int = 0


-class AgentDebugConversationRefreshPayload(BaseModel):
-    draft_type: AgentConfigDraftType = Field(
-        default=AgentConfigDraftType.DEBUG_BUILD,
-        description="Agent draft surface whose conversation should be refreshed",
-    )
-
-
 class AgentPublishPayload(BaseModel):
    version_note: str | None = Field(default=None, description="Optional note for this published Agent version")

@@ -316,7 +308,6 @@ register_schema_models(
    AgentAppCopyPayload,
    AgentPublishPayload,
    AgentBuildDraftCheckoutPayload,
-    AgentDebugConversationRefreshPayload,
    ComposerSavePayload,
    AgentApiStatusPayload,
    AgentInviteOptionsQuery,
@@ -396,11 +387,10 @@ def _serialize_agent_app_detail(
    payload["backing_app_id"] = roster_service.runtime_backing_app_id(agent)
    payload["hidden_app_backed"] = bool(agent.backing_app_id and agent.backing_app_id != agent.app_id)
    payload["id"] = agent.id
-    debug_conversation_id = roster_service.get_or_create_agent_app_debug_conversation_id(
+    debug_conversation_id = roster_service.get_or_create_build_conversation(
        tenant_id=app_model.tenant_id,
        agent_id=agent.id,
        account_id=current_user.id,
-        draft_type=AgentConfigDraftType.DEBUG_BUILD,
        commit=False,
    )
    message_count = roster_service.count_agent_app_debug_conversation_messages(
@@ -410,10 +400,6 @@ def _serialize_agent_app_detail(
    payload["debug_conversation_has_messages"] = message_count > 0
    payload["debug_conversation_message_count"] = message_count
    payload["role"] = agent.role or ""
-    payload["active_config_is_published"] = roster_service.active_config_is_published(
-        tenant_id=app_model.tenant_id,
-        agent=agent,
-    )
    return payload


@@ -444,11 +430,10 @@ def _serialize_agent_app_pagination(session: Session, app_pagination, *, tenant_
        tenant_id=tenant_id,
        agent_ids=[agent.id for agent in agents_by_app_id.values()],
    )
-    debug_conversation_ids_by_agent_id = roster_service.load_or_create_agent_app_debug_conversation_ids_by_agent_id(
+    debug_conversation_ids_by_agent_id = roster_service.load_or_create_build_conversation_ids_by_agent_id(
        tenant_id=tenant_id,
        agents=list(agents_by_app_id.values()),
        account_id=current_user.id,
-        draft_type=AgentConfigDraftType.DEBUG_BUILD,
    )
    payload = AgentAppPagination.model_validate(
        app_pagination,
@@ -683,16 +668,6 @@ class AgentAppApi(Resource):

@console_ns.route("/agent/<uuid:agent_id>/debug-conversation/refresh")
 class AgentDebugConversationRefreshApi(Resource):
-    @console_ns.expect(console_ns.models[AgentDebugConversationRefreshPayload.__name__])
-    @console_ns.doc(
-        params={
-            "payload": {
-                "in": "body",
-                "required": False,
-                "schema": {"$ref": f"#/components/schemas/{AgentDebugConversationRefreshPayload.__name__}"},
-            }
-        }
-    )
    @console_ns.response(
        200,
        "Agent debug conversation refreshed",
@@ -707,12 +682,10 @@ class AgentDebugConversationRefreshApi(Resource):
    @with_current_tenant_id
    @with_session
    def post(self, session: Session, tenant_id: str, current_user: Account, agent_id: UUID):
-        args = AgentDebugConversationRefreshPayload.model_validate(request.get_json(silent=True) or {})
-        debug_conversation_id = _agent_roster_service(session).refresh_agent_app_debug_conversation_id(
+        debug_conversation_id = _agent_roster_service(session).reset_build_conversation(
            tenant_id=tenant_id,
            agent_id=str(agent_id),
            account_id=current_user.id,
-            draft_type=args.draft_type,
        )
        return AgentDebugConversationRefreshResponse(
            debug_conversation_id=debug_conversation_id,
@@ -756,7 +729,7 @@ class AgentBuildDraftCheckoutApi(Resource):
    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user
    @with_current_tenant_id
-    @with_session
+    @with_session(write=False)
    def post(self, session: Session, tenant_id: str, current_user: Account, agent_id: UUID):
        args = AgentBuildDraftCheckoutPayload.model_validate(console_ns.payload or {})
        return AgentComposerService.checkout_agent_app_build_draft(
@@ -813,7 +786,7 @@ class AgentBuildDraftApi(Resource):
    @edit_permission_required
    @with_current_user
    @with_current_tenant_id
-    @with_session
+    @with_session(write=False)
    def delete(self, session: Session, tenant_id: str, current_user: Account, agent_id: UUID):
        return AgentComposerService.discard_agent_app_build_draft(
            session=session,
@@ -833,7 +806,7 @@ class AgentBuildDraftApplyApi(Resource):
    @rbac_permission_required(RBACResourceScope.WORKSPACE, RBACPermission.AGENT_MANAGE, resource_required=False)
    @with_current_user
    @with_current_tenant_id
-    @with_session
+    @with_session(write=False)
    def post(self, session: Session, tenant_id: str, current_user: Account, agent_id: UUID):
        return AgentComposerService.apply_agent_app_build_draft(
            session=session,
@@ -1,7 +1,7 @@
 """Console routes for Agent App and workflow Agent sandbox file access.

-The API keeps product-facing locators (conversation or workflow node identity)
-on this public boundary and proxies list/read/upload to the agent backend's new
+The API accepts product-facing Conversation, Build Draft, or Workflow Node
+Execution locators and proxies list/read/upload to the agent backend's
 ``/sandbox`` contract.
 """

@@ -26,10 +26,16 @@ from controllers.common.session import with_session
 from controllers.console import console_ns
 from controllers.console.agent.app_helpers import resolve_agent_runtime_app_model
 from controllers.console.app.wraps import get_app_model
-from controllers.console.wraps import account_initialization_required, setup_required, with_current_tenant_id
+from controllers.console.wraps import (
+    account_initialization_required,
+    setup_required,
+    with_current_tenant_id,
+    with_current_user,
+)
 from extensions.ext_database import db
 from fields.base import ResponseModel
 from libs.login import login_required
+from models import Account
 from models.model import App, AppMode
 from services.agent_app_sandbox_service import (
    AgentAppSandboxService,
@@ -37,52 +43,43 @@ from services.agent_app_sandbox_service import (
    WorkflowAgentSandboxService,
 )

-_NODE_EXECUTION_ID_DESCRIPTION = (
-    "Optional workflow node execution ID. When omitted, the latest active session for the node is used."
-)
-

 class AgentSandboxListQuery(BaseModel):
-    conversation_id: str = Field(min_length=1, description="Agent App conversation ID")
+    caller_type: Literal["conversation", "build_draft"]
+    caller_id: str = Field(min_length=1, description="Agent App caller ID")
    path: str = Field(default=".", description="Directory path relative to the sandbox workspace")


 class AgentSandboxInfoQuery(BaseModel):
-    conversation_id: str = Field(min_length=1, description="Agent App conversation ID")
+    caller_type: Literal["conversation", "build_draft"]
+    caller_id: str = Field(min_length=1, description="Agent App caller ID")


 class AgentSandboxFileQuery(BaseModel):
-    conversation_id: str = Field(min_length=1, description="Agent App conversation ID")
+    caller_type: Literal["conversation", "build_draft"]
+    caller_id: str = Field(min_length=1, description="Agent App caller ID")
    path: str = Field(min_length=1, description="File path relative to the sandbox workspace")


 class AgentSandboxUploadPayload(BaseModel):
-    conversation_id: str = Field(min_length=1, description="Agent App conversation ID")
+    caller_type: Literal["conversation", "build_draft"]
+    caller_id: str = Field(min_length=1, description="Agent App caller ID")
    path: str = Field(min_length=1, description="File path relative to the sandbox workspace")


 class WorkflowAgentSandboxListQuery(BaseModel):
+    node_execution_id: str = Field(min_length=1, description="Workflow node execution ID")
    path: str = Field(default=".", description="Directory path relative to the sandbox workspace")
-    node_execution_id: str | None = Field(
-        default=None,
-        description=_NODE_EXECUTION_ID_DESCRIPTION,
-    )


 class WorkflowAgentSandboxFileQuery(BaseModel):
+    node_execution_id: str = Field(min_length=1, description="Workflow node execution ID")
    path: str = Field(min_length=1, description="File path relative to the sandbox workspace")
-    node_execution_id: str | None = Field(
-        default=None,
-        description=_NODE_EXECUTION_ID_DESCRIPTION,
-    )


 class WorkflowAgentSandboxUploadPayload(BaseModel):
+    node_execution_id: str = Field(min_length=1, description="Workflow node execution ID")
    path: str = Field(min_length=1, description="File path relative to the sandbox workspace")
-    node_execution_id: str | None = Field(
-        default=None,
-        description=_NODE_EXECUTION_ID_DESCRIPTION,
-    )


 class SandboxFileEntryResponse(ResponseModel):
@@ -99,7 +96,6 @@ class SandboxListResponse(ResponseModel):


 class SandboxInfoResponse(ResponseModel):
-    session_id: str
    workspace_cwd: str


@@ -155,15 +151,19 @@ class AgentAppSandboxInfoResource(Resource):
    @login_required
    @account_initialization_required
    @with_current_tenant_id
+    @with_current_user
    @with_session(write=False)
-    def get(self, session: Session, tenant_id: str, agent_id: UUID):
+    def get(self, session: Session, current_user: Account, tenant_id: str, agent_id: UUID):
        app_model = resolve_agent_runtime_app_model(session=session, tenant_id=tenant_id, agent_id=agent_id)
        query = query_params_from_request(AgentSandboxInfoQuery)
        try:
            result = AgentAppSandboxService().get_info(
                tenant_id=tenant_id,
                app_id=app_model.id,
-                conversation_id=query.conversation_id,
+                agent_id=str(agent_id),
+                caller_type=query.caller_type,
+                caller_id=query.caller_id,
+                account_id=current_user.id,
            )
        except Exception as exc:
            return _handle(exc)
@@ -180,15 +180,19 @@ class AgentAppSandboxListResource(Resource):
    @login_required
    @account_initialization_required
    @with_current_tenant_id
+    @with_current_user
    @with_session(write=False)
-    def get(self, session: Session, tenant_id: str, agent_id: UUID):
+    def get(self, session: Session, current_user: Account, tenant_id: str, agent_id: UUID):
        app_model = resolve_agent_runtime_app_model(session=session, tenant_id=tenant_id, agent_id=agent_id)
        query = query_params_from_request(AgentSandboxListQuery)
        try:
            result = AgentAppSandboxService().list_files(
                tenant_id=tenant_id,
                app_id=app_model.id,
-                conversation_id=query.conversation_id,
+                agent_id=str(agent_id),
+                caller_type=query.caller_type,
+                caller_id=query.caller_id,
+                account_id=current_user.id,
                path=query.path,
            )
        except Exception as exc:
@@ -206,15 +210,19 @@ class AgentAppSandboxReadResource(Resource):
    @login_required
    @account_initialization_required
    @with_current_tenant_id
+    @with_current_user
    @with_session(write=False)
-    def get(self, session: Session, tenant_id: str, agent_id: UUID):
+    def get(self, session: Session, current_user: Account, tenant_id: str, agent_id: UUID):
        app_model = resolve_agent_runtime_app_model(session=session, tenant_id=tenant_id, agent_id=agent_id)
        query = query_params_from_request(AgentSandboxFileQuery)
        try:
            result = AgentAppSandboxService().read_file(
                tenant_id=tenant_id,
                app_id=app_model.id,
-                conversation_id=query.conversation_id,
+                agent_id=str(agent_id),
+                caller_type=query.caller_type,
+                caller_id=query.caller_id,
+                account_id=current_user.id,
                path=query.path,
            )
        except Exception as exc:
@@ -232,15 +240,19 @@ class AgentAppSandboxUploadResource(Resource):
    @login_required
    @account_initialization_required
    @with_current_tenant_id
+    @with_current_user
    @with_session(write=False)
-    def post(self, session: Session, tenant_id: str, agent_id: UUID):
+    def post(self, session: Session, current_user: Account, tenant_id: str, agent_id: UUID):
        app_model = resolve_agent_runtime_app_model(session=session, tenant_id=tenant_id, agent_id=agent_id)
        payload = AgentSandboxUploadPayload.model_validate(request.get_json(silent=True) or {})
        try:
            result = AgentAppSandboxService().upload_file(
                tenant_id=tenant_id,
                app_id=app_model.id,
-                conversation_id=payload.conversation_id,
+                agent_id=str(agent_id),
+                caller_type=payload.caller_type,
+                caller_id=payload.caller_id,
+                account_id=current_user.id,
                path=payload.path,
            )
        except Exception as exc:
@@ -58,6 +58,7 @@ from services.app_service import (
    AppResponseView,
    AppService,
    CreateAppParams,
+    RecentAppMode,
    StarredAppListParams,
 )
 from services.enterprise import rbac_service as enterprise_rbac_service
@@ -139,6 +140,10 @@ class AppListBaseQuery(BaseModel):
            raise ValueError("Invalid UUID format in creator_ids.") from exc


+class RecentAppListQuery(BaseModel):
+    limit: int = Field(default=8, ge=1, le=8, description="Number of recently modified apps to return (1-8)")
+
+
 class AppListQuery(AppListBaseQuery):
    pass

@@ -411,6 +416,33 @@ class AppPartial(AppResponseModel):
        return to_timestamp(value)


+class RecentAppResponse(ResponseModel):
+    id: str
+    name: str
+    icon_type: IconType | None = None
+    icon: str | None = None
+    icon_background: str | None = None
+    mode: RecentAppMode
+    author_name: str | None = None
+    updated_at: int
+    permission_keys: list[str] = Field(default_factory=list)
+    maintainer: str | None = None
+
+    @computed_field(return_type=str | None)  # type: ignore[prop-decorator]
+    @property
+    def icon_url(self) -> str | None:
+        return build_icon_url(self.icon_type, self.icon)
+
+    @field_validator("updated_at", mode="before")
+    @classmethod
+    def _normalize_timestamp(cls, value: datetime | int) -> int:
+        return to_timestamp(value)
+
+
+class RecentAppListResponse(ResponseModel):
+    data: list[RecentAppResponse]
+
+
 class AppDetail(AppResponseModel):
    id: str
    name: str
@@ -575,6 +607,8 @@ register_schema_models(
 register_response_schema_models(
    console_ns,
    AppPartial,
+    RecentAppResponse,
+    RecentAppListResponse,
    AppDetailWithSite,
    AppPagination,
 )
@@ -699,6 +733,49 @@ class AppListApi(Resource):
        return app_detail.model_dump(mode="json"), 201


+@console_ns.route("/apps/recent")
+class RecentAppListApi(Resource):
+    @console_ns.doc("list_recent_apps")
+    @console_ns.doc(description="Get recently modified apps for the home Continue Work section")
+    @console_ns.doc(params=query_params_from_model(RecentAppListQuery))
+    @console_ns.response(200, "Success", console_ns.models[RecentAppListResponse.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @enterprise_license_required
+    @with_session(write=False)
+    @with_current_user_id
+    @with_current_tenant_id
+    def get(self, current_tenant_id: str, current_user_id: str, session: Session):
+        """Return the lightweight app cards needed by the Explore home page."""
+        args = query_params_from_request(RecentAppListQuery)
+        params = AppListParams(limit=args.limit)
+
+        permissions = enterprise_rbac_service.RBACService.MyPermissions.get(
+            current_tenant_id,
+            current_user_id,
+            session=session,
+        )
+        if dify_config.RBAC_ENABLED:
+            access_filter = resolve_app_access_filter(
+                current_tenant_id,
+                current_user_id,
+                session=session,
+                permissions=permissions,
+            )
+            access_filter.apply_to_params(params)
+
+        recent_apps = AppService().get_recent_apps(current_user_id, current_tenant_id, params, session)
+        permission_keys_map = permissions.app.permission_keys_by_resource_ids([app.id for app in recent_apps])
+        response_items = [
+            RecentAppResponse.model_validate(app, from_attributes=True).model_copy(
+                update={"permission_keys": permission_keys_map.get(app.id, [])}
+            )
+            for app in recent_apps
+        ]
+        return dump_response(RecentAppListResponse, {"data": response_items}), 200
+
+
@console_ns.route("/apps/starred")
 class StarredAppListApi(Resource):
    @console_ns.doc("list_starred_apps")
@@ -36,7 +36,7 @@ from controllers.console.wraps import (
    with_current_user_id,
 )
 from controllers.web.error import InvokeRateLimitError as InvokeRateLimitHttpError
-from core.app.entities.app_invoke_entities import AGENT_RUNTIME_EXIT_INTENT_ARG, InvokeFrom
+from core.app.entities.app_invoke_entities import InvokeFrom
 from core.app.features.rate_limiting.rate_limit import RateLimitGenerator
 from core.errors.error import (
    ModelCurrentlyNotSupportError,
@@ -353,12 +353,7 @@ def _resolve_current_user_agent_debug_conversation_id(
    draft_type: AgentConfigDraftType,
    start_new: bool = False,
 ) -> str:
-    """Resolve or rotate the current editor's conversation within one draft surface.
-
-    ``start_new`` rotates the scoped mapping through ``AgentRosterService`` so
-    the old runtime session is retired before the new conversation is used.
-    Continuations and Build chat keep resolving the existing mapping.
-    """
+    """Resolve the current editor's Build or Preview conversation."""

    roster_service = AgentRosterService(session)
    resolved_agent_id = agent_id
@@ -368,17 +363,26 @@ def _resolve_current_user_agent_debug_conversation_id(
            raise AgentNotFoundError()
        resolved_agent_id = agent.id

-    resolve_conversation = (
-        roster_service.refresh_agent_app_debug_conversation_id
-        if start_new
-        else roster_service.get_or_create_agent_app_debug_conversation_id
-    )
-    return resolve_conversation(
+    if draft_type == AgentConfigDraftType.DEBUG_BUILD:
+        return roster_service.get_or_create_build_conversation(
+            tenant_id=current_tenant_id,
+            agent_id=resolved_agent_id,
+            account_id=current_user.id,
+        )
+    if start_new:
+        return roster_service.rotate_preview_conversation(
+            tenant_id=current_tenant_id,
+            agent_id=resolved_agent_id,
+            account_id=current_user.id,
+        )
+    conversation_id = roster_service.get_current_preview_conversation(
        tenant_id=current_tenant_id,
        agent_id=resolved_agent_id,
        account_id=current_user.id,
-        draft_type=draft_type,
    )
+    if conversation_id is None:
+        raise NotFound("Conversation Not Exists.")
+    return conversation_id


 def _create_chat_message(
@@ -450,7 +454,6 @@ def _create_build_chat_finalization_message(
        "draft_type": "debug_build",
        "conversation_id": debug_conversation_id,
        "auto_generate_name": False,
-        AGENT_RUNTIME_EXIT_INTENT_ARG: "delete",
    }
    external_trace_id = get_external_trace_id(request)
    if external_trace_id:
@@ -76,11 +76,13 @@ from models import Account, App
 from models.model import AppMode
 from models.workflow import Workflow
 from repositories.workflow_collaboration_repository import WORKFLOW_ONLINE_USERS_PREFIX
+from services.agent.retirement_service import WorkflowAgentRetirementService
 from services.app_generate_service import AppGenerateService
 from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError, WorkflowNotFoundError
 from services.errors.llm import InvokeRateLimitError
 from services.workflow_ref_service import WorkflowRefService
 from services.workflow_service import DraftWorkflowDeletionError, WorkflowInUseError, WorkflowService
+from tasks.collect_agent_resources_task import enqueue_agent_resource_collection

 logger = logging.getLogger(__name__)

@@ -317,7 +319,7 @@ class _WorkflowResponseSource:
        self._session = session

    def __getattr__(self, name: str) -> object:
-        return getattr(self._workflow, name)  # noqa: no-new-getattr response adapter delegates model fields
+        return getattr(self._workflow, name)  # guard-ignore: no-new-getattr -- delegates model fields

    @property
    def created_by_account(self) -> Account | None:
@@ -1245,7 +1247,7 @@ class PublishedWorkflowApi(Resource):

        workflow_service = WorkflowService()
        with sessionmaker(db.engine).begin() as session:
-            workflow = workflow_service.publish_workflow(
+            workflow, retirement_candidates = workflow_service.publish_workflow(
                session=session,
                app_model=app_model,
                account=current_user,
@@ -1262,6 +1264,16 @@ class PublishedWorkflowApi(Resource):

            workflow_created_at = TimestampField().format(workflow.created_at)

+        binding_ids, home_snapshot_ids = WorkflowAgentRetirementService.retire_unowned(
+            tenant_id=app_model.tenant_id,
+            agent_ids=retirement_candidates,
+            account_id=current_user.id,
+        )
+        enqueue_agent_resource_collection(
+            tenant_id=app_model.tenant_id,
+            binding_ids=binding_ids,
+            home_snapshot_ids=home_snapshot_ids,
+        )
        return {
            "result": "success",
            "created_at": workflow_created_at,
@@ -218,7 +218,7 @@ class _DatasetQueryResponseSource:
        return self.query.get_queries(session=self.session)

    def __getattr__(self, name: str) -> Any:
-        return getattr(self.query, name)  # noqa: no-new-getattr response adapter delegates model fields
+        return getattr(self.query, name)  # guard-ignore: no-new-getattr -- delegates model fields


 class DatasetQueryListResponse(ResponseModel):
@@ -257,7 +257,7 @@ class _RelatedAppResponseSource:
        return self.app.mode_compatible_with_agent_with_session(session=self.session)

    def __getattr__(self, name: str) -> Any:
-        return getattr(self.app, name)  # noqa: no-new-getattr response adapter delegates model fields
+        return getattr(self.app, name)  # guard-ignore: no-new-getattr -- delegates model fields


 class RelatedAppListResponse(ResponseModel):
@@ -106,7 +106,7 @@ class ExternalKnowledgeApiResponseSource:
        return self.external_knowledge_api.get_dataset_bindings(session=self.session)

    def __getattr__(self, name: str) -> Any:
-        return getattr(self.external_knowledge_api, name)  # noqa: no-new-getattr response adapter delegates model fields
+        return getattr(self.external_knowledge_api, name)  # guard-ignore: no-new-getattr -- delegates model fields


 def external_knowledge_api_response(
@@ -404,7 +404,7 @@ class TrialWorkflowResponseSource:
        return self.workflow.get_tool_published(session=self.session)

    def __getattr__(self, name: str) -> Any:
-        return getattr(self.workflow, name)  # noqa: no-new-getattr response adapter delegates model fields
+        return getattr(self.workflow, name)  # guard-ignore: no-new-getattr -- delegates model fields


 register_schema_models(
@@ -56,10 +56,12 @@ from libs.helper import TimestampField
 from libs.login import current_account_with_tenant, login_required
 from models import Account
 from models.snippet import CustomizedSnippet
+from services.agent.retirement_service import WorkflowAgentRetirementService
 from services.agent.workflow_publish_service import WorkflowAgentPublishService
 from services.errors.app import IsDraftWorkflowError, WorkflowHashNotEqualError, WorkflowNotFoundError
 from services.snippet_generate_service import SnippetGenerateService
 from services.snippet_service import SnippetService
+from tasks.collect_agent_resources_task import enqueue_agent_resource_collection

 logger = logging.getLogger(__name__)

@@ -295,8 +297,9 @@ class SnippetPublishedWorkflowApi(Resource):

        with Session(db.engine) as session:
            snippet = session.merge(snippet)
+            tenant_id = snippet.tenant_id
            try:
-                workflow = snippet_service.publish_workflow(
+                workflow, retirement_candidates = snippet_service.publish_workflow(
                    session=session,
                    snippet=snippet,
                    account=current_user,
@@ -306,6 +309,16 @@ class SnippetPublishedWorkflowApi(Resource):
            except ValueError as e:
                return {"message": str(e)}, 400

+        binding_ids, home_snapshot_ids = WorkflowAgentRetirementService.retire_unowned(
+            tenant_id=tenant_id,
+            agent_ids=retirement_candidates,
+            account_id=current_user.id,
+        )
+        enqueue_agent_resource_collection(
+            tenant_id=tenant_id,
+            binding_ids=binding_ids,
+            home_snapshot_ids=home_snapshot_ids,
+        )
        return {
            "result": "success",
            "created_at": workflow_created_at,
@@ -24,7 +24,7 @@ from extensions.ext_database import db
 from fields.base import ResponseModel
 from libs.helper import RateLimiter, dump_response, extract_remote_ip, to_timestamp
 from models.account import TenantStatus
-from models.model import App, Site
+from models.model import App, AppMode, Site
 from repositories.factory import DifyAPIRepositoryFactory
 from services.feature_service import FeatureService
 from services.human_input_file_upload_service import HumanInputFileUploadService
@@ -207,6 +207,7 @@ class HumanInputFormApi(Resource):
                site=WebAppSiteResponse.from_app_site(
                    tenant=tenant,
                    app_model=app_model,
+                    mode=AppMode.value_of(app_model.mode),
                    site=site,
                    end_user_id=None,
                    features=features,
@@ -14,7 +14,7 @@ from extensions.storage.storage_type import StorageType
 from fields.base import ResponseModel
 from libs.helper import build_icon_url
 from models.account import Tenant, TenantStatus
-from models.model import App, EndUser, IconType, Site
+from models.model import App, AppMode, EndUser, IconType, Site
 from services.feature_service import FeatureModel, FeatureService
 from services.file_service import FileService

@@ -67,6 +67,7 @@ class WebAppCustomConfigResponse(ResponseModel):

 class WebAppSiteResponse(ResponseModel):
    app_id: str
+    mode: AppMode
    end_user_id: str | None = None
    enable_site: bool
    site: WebSiteResponse
@@ -83,6 +84,7 @@ class WebAppSiteResponse(ResponseModel):
        *,
        tenant: Tenant,
        app_model: App,
+        mode: AppMode,
        site: Site,
        end_user_id: str | None,
        features: FeatureModel,
@@ -109,6 +111,7 @@ class WebAppSiteResponse(ResponseModel):

        return cls(
            app_id=app_model.id,
+            mode=mode,
            end_user_id=end_user_id,
            enable_site=app_model.enable_site,
            site=site_response,
@@ -167,6 +170,7 @@ class AppSiteApi(WebApiResource):
        return WebAppSiteResponse.from_app_site(
            tenant=tenant,
            app_model=app_model,
+            mode=AppMode.value_of(app_model.mode_compatible_with_agent_with_session(session=db.session())),
            site=site,
            end_user_id=end_user.id,
            features=features,
@@ -616,23 +616,34 @@ class AdvancedChatAppGenerator(MessageBasedAppGenerator):
            message_snapshot = MessageSnapshot.from_message(message)
            session.close()

-            # return response or stream generator
-            response = self._handle_advanced_chat_response(
-                application_generate_entity=application_generate_entity,
-                workflow=workflow_snapshot,
-                queue_manager=queue_manager,
-                conversation=conversation_snapshot,
-                message=message_snapshot,
-                user=user,
-                stream=stream,
-                draft_var_saver_factory=self._get_draft_var_saver_factory(
-                    invoke_from,
-                    account=user,
-                    tenant_id=application_generate_entity.app_config.tenant_id,
-                ),
-            )
+            try:
+                response = self._handle_advanced_chat_response(
+                    application_generate_entity=application_generate_entity,
+                    workflow=workflow_snapshot,
+                    queue_manager=queue_manager,
+                    conversation=conversation_snapshot,
+                    message=message_snapshot,
+                    user=user,
+                    stream=stream,
+                    draft_var_saver_factory=self._get_draft_var_saver_factory(
+                        invoke_from,
+                        account=user,
+                        tenant_id=application_generate_entity.app_config.tenant_id,
+                    ),
+                )
+                converted_response = AdvancedChatAppGenerateResponseConverter.convert(
+                    response=response,
+                    invoke_from=invoke_from,
+                )
+            except BaseException:
+                self._join_worker_thread(worker_thread)
+                raise

-            return AdvancedChatAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+            if isinstance(converted_response, Generator):
+                return self._wrap_stream_with_worker_thread_join(converted_response, worker_thread)
+
+            self._join_worker_thread(worker_thread)
+            return converted_response

    def _generate_worker(
        self,
@@ -16,6 +16,7 @@ from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
 from core.app.entities.app_invoke_entities import (
    AdvancedChatAppGenerateEntity,
    AppGenerateEntity,
+    DifyRunContext,
    InvokeFrom,
 )
 from core.app.entities.queue_entities import (
@@ -31,7 +32,7 @@ from core.moderation.base import ModerationError
 from core.moderation.input_moderation import InputModeration
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from core.workflow.node_factory import get_default_root_node_id
-from core.workflow.nodes.agent_v2.session_cleanup_layer import build_workflow_agent_session_cleanup_layer
+from core.workflow.nodes.agent_v2.workspace_retirement_layer import build_workflow_agent_workspace_retirement_layer
 from core.workflow.system_variables import (
    build_bootstrap_variables,
    build_system_variables,
@@ -267,7 +268,18 @@ class AdvancedChatAppRunner(WorkflowBasedAppRunner):
        )

        workflow_entry.graph_engine.layer(persistence_layer)
-        workflow_entry.graph_engine.layer(build_workflow_agent_session_cleanup_layer())
+        workflow_entry.graph_engine.layer(
+            build_workflow_agent_workspace_retirement_layer(
+                dify_run_context=DifyRunContext(
+                    tenant_id=self._workflow.tenant_id,
+                    app_id=self._workflow.app_id,
+                    user_id=self.application_generate_entity.user_id,
+                    user_from=user_from,
+                    invoke_from=invoke_from,
+                    trace_session_id=self.application_generate_entity.extras.get("trace_session_id"),
+                )
+            )
+        )
        conversation_variable_layer = ConversationVariablePersistenceLayer(
            ConversationVariableUpdater(session_factory.get_session_maker())
        )
@@ -33,15 +33,13 @@ from core.app.apps.agent_app.app_runner import AgentAppRunner
 from core.app.apps.agent_app.errors import AgentAppGeneratorError, AgentAppNotPublishedError
 from core.app.apps.agent_app.generate_response_converter import AgentAppGenerateResponseConverter
 from core.app.apps.agent_app.runtime_request_builder import AgentAppRuntimeRequestBuilder
-from core.app.apps.agent_app.session_store import AgentAppRuntimeSessionStore
+from core.app.apps.agent_app.session_store import AgentAppWorkspaceStore
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.exc import GenerateTaskStoppedError
 from core.app.apps.message_based_app_generator import MessageBasedAppGenerator
 from core.app.apps.message_based_app_queue_manager import MessageBasedAppQueueManager
 from core.app.entities.app_invoke_entities import (
-    AGENT_RUNTIME_EXIT_INTENT_ARG,
    AgentAppGenerateEntity,
-    AgentRuntimeExitIntent,
    DifyRunContext,
    InvokeFrom,
    UserFrom,
@@ -51,19 +49,24 @@ from core.db.session_factory import session_factory
 from core.ops.ops_trace_manager import TraceQueueManager
 from core.workflow.file_reference import build_file_reference, is_canonical_file_reference
 from extensions.ext_database import db
-from models import Account, App, AppModelConfig, EndUser, Message, MessageAnnotation
+from models import Account, App, AppModelConfig, Conversation, EndUser, Message, MessageAnnotation
 from models.agent import (
    APP_BACKED_AGENT_SOURCES,
    Agent,
    AgentConfigDraft,
    AgentConfigDraftType,
    AgentConfigSnapshot,
+    AgentConfigVersionKind,
    AgentScope,
    AgentSource,
    AgentStatus,
+    AgentWorkingResourceStatus,
+    AgentWorkspaceBinding,
+    AgentWorkspaceOwnerType,
 )
 from models.agent_config_entities import AgentSoulConfig
 from models.model import load_annotation_reply_config
+from services.agent.workspace_service import AgentWorkspaceService, WorkspaceOwnerScope
 from services.conversation_service import ConversationService

 logger = logging.getLogger(__name__)
@@ -150,19 +153,6 @@ class AgentAppGenerator(MessageBasedAppGenerator):
        inputs = args["inputs"]
        prompt_file_mappings = args.get("files") or []

-        # Resolve the bound roster Agent + its current Agent Soul snapshot.
-        agent, agent_config_id, agent_config_version_kind, agent_soul = self._resolve_agent(
-            app_model,
-            invoke_from=invoke_from,
-            draft_type=args.get("draft_type"),
-            user=user,
-            session=session,
-        )
-        runtime_session_snapshot_id = self._runtime_session_snapshot_id(
-            invoke_from=invoke_from,
-            snapshot_id=agent_config_id,
-        )
-
        conversation = None
        conversation_id = args.get("conversation_id")
        if conversation_id:
@@ -170,6 +160,21 @@ class AgentAppGenerator(MessageBasedAppGenerator):
                app_model=app_model, conversation_id=conversation_id, user=user, session=session
            )

+        # New conversations use the current Agent generation. Existing
+        # conversations use the immutable generation named by their Binding.
+        agent, agent_config_id, agent_config_version_kind, agent_soul = self._resolve_agent(
+            app_model,
+            invoke_from=invoke_from,
+            draft_type=args.get("draft_type"),
+            user=user,
+            session=session,
+            conversation=conversation,
+        )
+        session_scope_config_version_id = self._session_scope_config_version_id(
+            invoke_from=invoke_from,
+            config_version_id=agent_config_id,
+        )
+
        # Build the EasyUI-shaped config from the Agent Soul so the chat pipeline
        # can persist usage; the answer itself comes from the agent backend.
        app_model_config = (
@@ -186,8 +191,6 @@ class AgentAppGenerator(MessageBasedAppGenerator):
        model_conf = ModelConfigConverter.convert(app_config)

        trace_manager = TraceQueueManager(app_model.id, user.id if isinstance(user, Account) else user.session_id)
-        agent_runtime_exit_intent = self._resolve_agent_runtime_exit_intent(args)
-
        application_generate_entity = AgentAppGenerateEntity(
            task_id=str(uuid.uuid4()),
            app_config=app_config,
@@ -215,8 +218,7 @@ class AgentAppGenerator(MessageBasedAppGenerator):
            agent_id=agent.id,
            agent_config_snapshot_id=agent_config_id,
            agent_config_version_kind=agent_config_version_kind,
-            agent_runtime_session_snapshot_id=runtime_session_snapshot_id,
-            agent_runtime_exit_intent=agent_runtime_exit_intent,
+            agent_session_scope_config_version_id=session_scope_config_version_id,
        )

        conversation, message = self._init_generate_records(
@@ -265,6 +267,7 @@ class AgentAppGenerator(MessageBasedAppGenerator):
        app_model: App,
        user: Account | EndUser,
        conversation_id: str,
+        form_id: str,
        invoke_from: InvokeFrom,
        session: Session,
    ) -> None:
@@ -279,14 +282,21 @@ class AgentAppGenerator(MessageBasedAppGenerator):
        conversation = ConversationService.get_conversation(
            app_model=app_model, conversation_id=conversation_id, user=user, session=session
        )
+        draft_type, draft_id = self._resolve_resume_draft(
+            app_model=app_model,
+            conversation=conversation,
+            user=user,
+            form_id=form_id,
+            session=session,
+        )
        agent, agent_config_id, agent_config_version_kind, agent_soul = self._resolve_agent(
            app_model,
            invoke_from=invoke_from,
-            draft_type=self._resume_draft_type(
-                app_model=app_model, conversation=conversation, user=user, session=session
-            ),
+            draft_type=draft_type,
+            draft_id=draft_id,
            user=user,
            session=session,
+            conversation=conversation,
        )

        app_model_config = (
@@ -384,30 +394,37 @@ class AgentAppGenerator(MessageBasedAppGenerator):
        )

    @staticmethod
-    def _resume_draft_type(
-        *, app_model: App, conversation: Any, user: Account | EndUser, session: Session
-    ) -> str | None:
+    def _resolve_resume_draft(
+        *,
+        app_model: App,
+        conversation: Any,
+        user: Account | EndUser,
+        form_id: str,
+        session: Session,
+    ) -> tuple[str | None, str | None]:
        if conversation.invoke_from != InvokeFrom.DEBUGGER:
-            return None
-        active_session = AgentAppRuntimeSessionStore().load_active_session_for_conversation(
-            tenant_id=app_model.tenant_id,
-            app_id=app_model.id,
-            conversation_id=conversation.id,
-        )
-        snapshot_id = active_session.scope.agent_config_snapshot_id if active_session is not None else None
-        if snapshot_id and isinstance(user, Account):
-            draft = session.scalar(
-                select(AgentConfigDraft).where(
-                    AgentConfigDraft.tenant_id == app_model.tenant_id,
-                    AgentConfigDraft.id == snapshot_id,
-                )
+            return None, None
+        if not isinstance(user, Account):
+            return AgentConfigDraftType.DRAFT.value, None
+
+        build_draft = session.scalar(
+            select(AgentConfigDraft)
+            .join(
+                AgentWorkspaceBinding,
+                AgentWorkspaceBinding.id == AgentConfigDraft.agent_workspace_binding_id,
            )
-            if draft is not None:
-                if draft.draft_type == AgentConfigDraftType.DEBUG_BUILD and draft.account_id == user.id:
-                    return AgentConfigDraftType.DEBUG_BUILD.value
-                if draft.draft_type == AgentConfigDraftType.DRAFT and draft.account_id is None:
-                    return AgentConfigDraftType.DRAFT.value
-        return AgentConfigDraftType.DRAFT.value
+            .where(
+                AgentConfigDraft.tenant_id == app_model.tenant_id,
+                AgentConfigDraft.draft_type == AgentConfigDraftType.DEBUG_BUILD,
+                AgentConfigDraft.account_id == user.id,
+                AgentWorkspaceBinding.tenant_id == app_model.tenant_id,
+                AgentWorkspaceBinding.status == AgentWorkingResourceStatus.ACTIVE,
+                AgentWorkspaceBinding.pending_form_id == form_id,
+            )
+        )
+        if build_draft is not None:
+            return AgentConfigDraftType.DEBUG_BUILD.value, build_draft.id
+        return AgentConfigDraftType.DRAFT.value, None

    def _generate_worker(
        self,
@@ -482,7 +499,7 @@ class AgentAppGenerator(MessageBasedAppGenerator):
                    invoke_from=application_generate_entity.invoke_from,
                )
                with session_factory.create_session() as session:
-                    _, _, agent_soul = self._resolve_agent_by_id(
+                    agent, config_version, agent_soul = self._resolve_agent_by_id(
                        tenant_id=app_config.tenant_id,
                        agent_id=application_generate_entity.agent_id,
                        snapshot_id=application_generate_entity.agent_config_snapshot_id,
@@ -496,13 +513,18 @@ class AgentAppGenerator(MessageBasedAppGenerator):
                    agent_config_snapshot_id=application_generate_entity.agent_config_snapshot_id,
                    agent_config_version_kind=application_generate_entity.agent_config_version_kind,
                    agent_soul=agent_soul,
+                    home_snapshot_id=config_version.home_snapshot_id,
                    conversation_id=conversation.id,
                    query=query,
                    message_id=message.id,
                    model_name=application_generate_entity.model_conf.model,
                    queue_manager=queue_manager,
-                    session_scope_snapshot_id=application_generate_entity.agent_runtime_session_snapshot_id,
-                    agent_runtime_exit_intent=application_generate_entity.agent_runtime_exit_intent,
+                    session_scope_snapshot_id=application_generate_entity.agent_session_scope_config_version_id,
+                    build_draft_id=(
+                        application_generate_entity.agent_config_snapshot_id
+                        if application_generate_entity.agent_config_version_kind == AgentConfigVersionKind.BUILD_DRAFT
+                        else None
+                    ),
                )
            except GenerateTaskStoppedError:
                pass
@@ -519,18 +541,6 @@ class AgentAppGenerator(MessageBasedAppGenerator):
            raise AgentAppGeneratorError("query is required")
        return query.replace("\x00", "")

-    @staticmethod
-    def _resolve_agent_runtime_exit_intent(args: Mapping[str, Any]) -> AgentRuntimeExitIntent:
-        """Resolve API-internal runtime exit policy from controller-owned args.
-
-        Only the private controller-injected "delete" value changes behavior.
-        Normal chat and resume flows default/fallback to "suspend" so public
-        payloads and invalid internal values preserve existing semantics.
-        """
-        if args.get(AGENT_RUNTIME_EXIT_INTENT_ARG) == "delete":
-            return "delete"
-        return "suspend"
-
    @staticmethod
    def _build_runner(dify_context: DifyRunContext) -> AgentAppRunner:
        credentials_provider, _ = build_dify_model_access(dify_context)
@@ -538,6 +548,7 @@ class AgentAppGenerator(MessageBasedAppGenerator):
            request_builder=AgentAppRuntimeRequestBuilder(credentials_provider=credentials_provider),
            agent_backend_client=create_agent_backend_run_client(
                base_url=dify_config.AGENT_BACKEND_BASE_URL,
+                api_token=dify_config.AGENT_BACKEND_API_TOKEN,
                use_fake=dify_config.AGENT_BACKEND_USE_FAKE,
                fake_scenario=dify_config.AGENT_BACKEND_FAKE_SCENARIO,
                stream_read_timeout_seconds=dify_config.AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS,
@@ -545,7 +556,7 @@ class AgentAppGenerator(MessageBasedAppGenerator):
                stream_run_timeout_seconds=dify_config.AGENT_BACKEND_RUN_TIMEOUT_SECONDS,
            ),
            event_adapter=AgentBackendRunEventAdapter(),
-            session_store=AgentAppRuntimeSessionStore(),
+            session_store=AgentAppWorkspaceStore(),
            text_delta_debounce_seconds=dify_config.AGENT_APP_TEXT_DELTA_DEBOUNCE_SECONDS,
        )

@@ -609,8 +620,10 @@ class AgentAppGenerator(MessageBasedAppGenerator):
        *,
        invoke_from: InvokeFrom,
        draft_type: Any,
+        draft_id: str | None = None,
        user: Account | EndUser,
        session: Session,
+        conversation: Conversation | None = None,
    ) -> tuple[Agent, str, Literal["snapshot", "draft", "build_draft"], AgentSoulConfig]:
        agent = session.scalar(
            select(Agent)
@@ -642,6 +655,7 @@ class AgentAppGenerator(MessageBasedAppGenerator):
                tenant_id=app_model.tenant_id,
                agent=agent,
                draft_type=draft_type,
+                draft_id=draft_id,
                account_id=user.id if isinstance(user, Account) else None,
                session=session,
            )
@@ -654,28 +668,82 @@ class AgentAppGenerator(MessageBasedAppGenerator):
        # Public runtime must keep serving the active snapshot even when unpublished draft edits exist.
        if not agent.active_config_snapshot_id:
            raise AgentAppNotPublishedError("Agent has not been published")
+        conversation_binding = self._resolve_conversation_binding(
+            session=session,
+            tenant_id=app_model.tenant_id,
+            app_id=app_model.id,
+            agent_id=agent.id,
+            conversation=conversation,
+        )
+        snapshot_id = (
+            conversation_binding.agent_config_version_id
+            if conversation_binding is not None
+            else agent.active_config_snapshot_id
+        )
        _, snapshot, agent_soul = self._resolve_agent_by_id(
            tenant_id=app_model.tenant_id,
            agent_id=agent.id,
-            snapshot_id=agent.active_config_snapshot_id,
+            snapshot_id=snapshot_id,
            session=session,
        )
+        if conversation_binding is not None:
+            AgentWorkspaceService.validate_binding_generation(
+                conversation_binding,
+                base_home_snapshot_id=snapshot.home_snapshot_id,
+                agent_config_version_id=snapshot.id,
+                agent_config_version_kind=AgentConfigVersionKind.SNAPSHOT,
+            )
        return agent, snapshot.id, "snapshot", agent_soul

    @staticmethod
-    def _runtime_session_snapshot_id(*, invoke_from: InvokeFrom, snapshot_id: str) -> str | None:
-        """Return the session scope snapshot id for Agent App runtime state.
+    def _resolve_conversation_binding(
+        *,
+        session: Session,
+        tenant_id: str,
+        app_id: str,
+        agent_id: str,
+        conversation: Conversation | None,
+    ) -> AgentWorkspaceBinding | None:
+        """Resolve the exact participant generation owned by an existing conversation."""
+
+        if conversation is None or conversation.agent_workspace_binding_id is None:
+            return None
+        binding = AgentWorkspaceService.get_active_binding(
+            session=session,
+            tenant_id=tenant_id,
+            binding_id=conversation.agent_workspace_binding_id,
+            expected_owner_scope=WorkspaceOwnerScope(
+                tenant_id=tenant_id,
+                app_id=app_id,
+                owner_type=AgentWorkspaceOwnerType.CONVERSATION,
+                owner_id=conversation.id,
+            ),
+        )
+        if binding is None or binding.agent_id != agent_id:
+            raise AgentAppGeneratorError("Conversation participant Binding is unavailable")
+        return binding
+
+    @staticmethod
+    def _session_scope_config_version_id(*, invoke_from: InvokeFrom, config_version_id: str) -> str | None:
+        """Return the config version id that scopes Agent App session reuse.

        Console preview/debug chat uses a stable Agent draft row id; build mode
        uses the current user's build-draft row id. Published/web/API runs use
-        immutable published snapshot ids. This keeps runtime session continuity
+        immutable published snapshot ids. This keeps Workspace Binding continuity
        inside one editable surface without mixing draft/build/published state.
        """
-        return snapshot_id
+        del invoke_from
+        return config_version_id

    @staticmethod
    def _resolve_debug_draft(
-        *, tenant_id: str, agent: Agent, draft_type: Any, account_id: str | None, session: Session
+        *,
+        tenant_id: str,
+        agent: Agent,
+        draft_type: Any,
+        account_id: str | None,
+        session: Session,
+        draft_id: str | None = None,
    ) -> AgentConfigDraft:
        effective_draft_type = (
            AgentConfigDraftType.DEBUG_BUILD
@@ -699,6 +767,8 @@ class AgentAppGenerator(MessageBasedAppGenerator):
            AgentConfigDraft.draft_type == AgentConfigDraftType.DEBUG_BUILD,
            AgentConfigDraft.account_id == account_id,
        )
+        if draft_id is not None:
+            stmt = stmt.where(AgentConfigDraft.id == draft_id)
        draft = session.scalar(stmt.order_by(AgentConfigDraft.updated_at.desc()).limit(1))
        if draft is not None:
            return draft
@@ -3,8 +3,7 @@
 Unlike the legacy ``AgentChatAppRunner`` (which runs an in-process ReAct loop),
 this runner delegates to the Agent backend, consumes the streamed event flow,
 republishes the assistant answer through the existing EasyUI chat task
-pipeline, and then either saves or retires the conversation-owned runtime
-session depending on the turn's exit policy.
+pipeline, and saves the latest Agenton snapshot on the persistent Binding.
 """

 from __future__ import annotations
@@ -31,22 +30,20 @@ from clients.agent_backend import (
    AgentBackendRunFailedInternalEvent,
    AgentBackendRunSucceededInternalEvent,
    AgentBackendStreamInternalEvent,
-    extract_runtime_layer_specs,
 )
-from clients.agent_backend.session_cleanup import AgentBackendSessionCleanupPayload
 from core.app.apps.agent_app.runtime_request_builder import (
    AgentAppRuntimeBuildContext,
    AgentAppRuntimeRequest,
    AgentAppRuntimeRequestBuilder,
 )
 from core.app.apps.agent_app.session_store import (
-    AgentAppRuntimeSessionStore,
    AgentAppSessionScope,
+    AgentAppWorkspaceStore,
    StoredAgentAppSession,
 )
 from core.app.apps.base_app_queue_manager import AppQueueManager, PublishFrom
 from core.app.apps.exc import GenerateTaskStoppedError
-from core.app.entities.app_invoke_entities import AgentRuntimeExitIntent, DifyRunContext
+from core.app.entities.app_invoke_entities import DifyRunContext
 from core.app.entities.queue_entities import (
    QueueAgentMessageEvent,
    QueueAgentThoughtEvent,
@@ -67,10 +64,10 @@ from graphon.model_runtime.errors.invoke import (
    InvokeRateLimitError,
    InvokeServerUnavailableError,
 )
+from models.agent import AgentConfigVersionKind
 from models.agent_config_entities import AgentSoulConfig
 from models.enums import CreatorUserRole
 from models.model import MessageAgentThought
-from tasks.agent_backend_session_cleanup_task import cleanup_conversation_agent_runtime_session

 logger = logging.getLogger(__name__)

@@ -620,7 +617,7 @@ class AgentAppRunner:
        request_builder: AgentAppRuntimeRequestBuilder,
        agent_backend_client: AgentBackendRunClient,
        event_adapter: AgentBackendRunEventAdapter,
-        session_store: AgentAppRuntimeSessionStore,
+        session_store: AgentAppWorkspaceStore,
        text_delta_debounce_seconds: float,
    ) -> None:
        self._request_builder = request_builder
@@ -637,37 +634,41 @@ class AgentAppRunner:
        agent_config_snapshot_id: str,
        agent_config_version_kind: Literal["snapshot", "draft", "build_draft"] = "snapshot",
        agent_soul: AgentSoulConfig,
+        home_snapshot_id: str,
        conversation_id: str,
        query: str,
        message_id: str,
        model_name: str,
        queue_manager: AppQueueManager,
        session_scope_snapshot_id: str | None | _DefaultSessionScopeSnapshotId = _DEFAULT_SESSION_SCOPE_SNAPSHOT_ID,
-        agent_runtime_exit_intent: AgentRuntimeExitIntent = "suspend",
+        build_draft_id: str | None = None,
    ) -> None:
-        preserve_session = agent_runtime_exit_intent == "suspend"
        scope = self._build_session_scope(
            dify_context=dify_context,
            agent_id=agent_id,
            agent_config_snapshot_id=agent_config_snapshot_id,
+            home_snapshot_id=home_snapshot_id,
            conversation_id=conversation_id,
            session_scope_snapshot_id=session_scope_snapshot_id,
+            agent_config_version_kind=AgentConfigVersionKind(agent_config_version_kind),
+            build_draft_id=build_draft_id,
        )
        # ENG-638: if a prior turn paused on ask_human and the form is now answered,
        # resume by threading the human's reply into this run as deferred_tool_results.
-        stored = self._session_store.load_active_session(scope)
+        stored = self._session_store.load_or_create(scope)
        runtime = self._build_runtime(
            dify_context=dify_context,
            agent_id=agent_id,
            agent_config_snapshot_id=agent_config_snapshot_id,
            agent_config_version_kind=agent_config_version_kind,
            agent_soul=agent_soul,
+            binding_id=stored.binding_id,
+            backend_binding_ref=stored.backend_binding_ref,
            conversation_id=conversation_id,
            query=query,
            idempotency_key=message_id,
            stored=stored,
            message_id=message_id,
-            suspend_on_exit=preserve_session,
        )

        create_response = self._agent_backend_client.create_run(runtime.request)
@@ -681,9 +682,6 @@ class AgentAppRunner:
        )

        if isinstance(terminal, AgentBackendDeferredToolCallInternalEvent):
-            if not preserve_session:
-                self._mark_session_cleaned(scope=scope, backend_run_id=terminal.run_id)
-                raise AgentBackendError("Agent App finalization cannot pause for human input.")
            # ENG-635: the agent asked a human. End this turn with the question and
            # a conversation-owned HITL form; a form submission resumes the run.
            self._pause_for_ask_human(
@@ -703,8 +701,8 @@ class AgentAppRunner:
        if not isinstance(terminal, AgentBackendRunSucceededInternalEvent):
            if isinstance(terminal, AgentBackendRunFailedInternalEvent):
                reason = terminal.reason
-                if reason == "sandbox_expired":
-                    raise AgentBackendError("The agent session sandbox has expired. Please start a new conversation.")
+                if reason == "binding_lost":
+                    raise AgentBackendError("The retained agent working environment is no longer available.")
                raise _agent_backend_failure_to_exception(terminal)
            raise AgentBackendError("Agent backend run did not complete successfully.")

@@ -719,38 +717,18 @@ class AgentAppRunner:
                message_id,
                exc_info=True,
            )
-        if preserve_session:
-            superseded_sessions = self._load_superseded_sessions(scope=scope)
-            self._publish_terminal_answer(
-                queue_manager=queue_manager,
-                model_name=model_name,
-                answer=answer,
-                query=query,
-                usage=_llm_usage_from_agent_backend(terminal.usage),
-            )
-            session_saved = self._save_session(
-                scope=scope,
-                backend_run_id=terminal.run_id,
-                snapshot=terminal.session_snapshot,
-                runtime_layer_specs=extract_runtime_layer_specs(runtime.request.composition),
-            )
-            if session_saved:
-                self._cleanup_superseded_sessions(superseded_sessions)
-        else:
-            # The backend has already accepted a terminal success with
-            # delete-on-exit semantics. Local publish/persistence errors must
-            # not keep the API-side session row active, and cleanup failures
-            # must not replace the original publish/error outcome.
-            try:
-                self._publish_terminal_answer(
-                    queue_manager=queue_manager,
-                    model_name=model_name,
-                    answer=answer,
-                    query=query,
-                    usage=_llm_usage_from_agent_backend(terminal.usage),
-                )
-            finally:
-                self._mark_session_cleaned(scope=scope, backend_run_id=terminal.run_id)
+        self._publish_terminal_answer(
+            queue_manager=queue_manager,
+            model_name=model_name,
+            answer=answer,
+            query=query,
+            usage=_llm_usage_from_agent_backend(terminal.usage),
+        )
+        self._save_session(
+            scope=scope,
+            binding_id=runtime.binding_id,
+            snapshot=terminal.session_snapshot,
+        )

    def _build_session_scope(
        self,
@@ -758,8 +736,11 @@ class AgentAppRunner:
        dify_context: DifyRunContext,
        agent_id: str,
        agent_config_snapshot_id: str,
+        home_snapshot_id: str,
        conversation_id: str,
        session_scope_snapshot_id: str | None | _DefaultSessionScopeSnapshotId,
+        agent_config_version_kind: AgentConfigVersionKind,
+        build_draft_id: str | None = None,
    ) -> AgentAppSessionScope:
        if isinstance(session_scope_snapshot_id, _DefaultSessionScopeSnapshotId):
            effective_session_scope_snapshot_id: str | None = agent_config_snapshot_id
@@ -770,7 +751,10 @@ class AgentAppRunner:
            app_id=dify_context.app_id,
            conversation_id=conversation_id,
            agent_id=agent_id,
-            agent_config_snapshot_id=effective_session_scope_snapshot_id,
+            agent_config_snapshot_id=effective_session_scope_snapshot_id or agent_config_snapshot_id,
+            home_snapshot_id=home_snapshot_id,
+            agent_config_version_kind=agent_config_version_kind,
+            build_draft_id=build_draft_id,
        )

    def _build_runtime(
@@ -781,14 +765,15 @@ class AgentAppRunner:
        agent_config_snapshot_id: str,
        agent_config_version_kind: Literal["snapshot", "draft", "build_draft"],
        agent_soul: AgentSoulConfig,
+        binding_id: str,
+        backend_binding_ref: str,
        conversation_id: str,
        query: str,
        idempotency_key: str,
-        stored: StoredAgentAppSession | None,
+        stored: StoredAgentAppSession,
        message_id: str | None,
-        suspend_on_exit: bool,
    ) -> AgentAppRuntimeRequest:
-        session_snapshot = stored.session_snapshot if stored is not None else None
+        session_snapshot = stored.session_snapshot
        deferred_tool_results = (
            self._resolve_pending_ask_human(stored=stored, dify_context=dify_context, message_id=message_id)
            if message_id is not None
@@ -804,9 +789,10 @@ class AgentAppRunner:
                conversation_id=conversation_id,
                user_query=query,
                idempotency_key=idempotency_key,
+                binding_id=binding_id,
+                backend_binding_ref=backend_binding_ref,
                session_snapshot=session_snapshot,
                deferred_tool_results=deferred_tool_results,
-                suspend_on_exit=suspend_on_exit,
            )
        )

@@ -843,9 +829,8 @@ class AgentAppRunner:
        # second run with the human's answer (ENG-637/638 columns, conversation owner).
        self._save_session(
            scope=scope,
-            backend_run_id=terminal.run_id,
+            binding_id=runtime.binding_id,
            snapshot=terminal.session_snapshot,
-            runtime_layer_specs=extract_runtime_layer_specs(runtime.request.composition),
            pending_form_id=created.form_id,
            pending_tool_call_id=terminal.deferred_tool_call.tool_call_id,
        )
@@ -862,12 +847,12 @@ class AgentAppRunner:
    def _resolve_pending_ask_human(
        self,
        *,
-        stored: StoredAgentAppSession | None,
+        stored: StoredAgentAppSession,
        dify_context: DifyRunContext,
        message_id: str,
    ) -> DeferredToolResultsPayload | None:
        """Build deferred_tool_results when a pending ask_human form is answered."""
-        if stored is None or stored.pending_form_id is None or stored.pending_tool_call_id is None:
+        if stored.pending_form_id is None or stored.pending_tool_call_id is None:
            return None
        outcome = resolve_ask_human_form(
            form_id=stored.pending_form_id,
@@ -1036,18 +1021,16 @@ class AgentAppRunner:
        self,
        *,
        scope: AgentAppSessionScope,
-        backend_run_id: str,
+        binding_id: str,
        snapshot: Any,
-        runtime_layer_specs: Any,
        pending_form_id: str | None = None,
        pending_tool_call_id: str | None = None,
    ) -> bool:
        try:
            self._session_store.save_active_snapshot(
                scope=scope,
-                backend_run_id=backend_run_id,
+                binding_id=binding_id,
                snapshot=snapshot,
-                runtime_layer_specs=runtime_layer_specs,
                pending_form_id=pending_form_id,
                pending_tool_call_id=pending_tool_call_id,
            )
@@ -1064,87 +1047,6 @@ class AgentAppRunner:
            )
            return False

-    def _load_superseded_sessions(self, *, scope: AgentAppSessionScope) -> list[StoredAgentAppSession]:
-        try:
-            stored_sessions = self._session_store.list_active_sessions_for_conversation(
-                tenant_id=scope.tenant_id,
-                app_id=scope.app_id,
-                conversation_id=scope.conversation_id,
-            )
-        except Exception:
-            logger.warning(
-                "Failed to load existing Agent App conversation sessions before snapshot save: "
-                "tenant_id=%s app_id=%s conversation_id=%s agent_id=%s",
-                scope.tenant_id,
-                scope.app_id,
-                scope.conversation_id,
-                scope.agent_id,
-                exc_info=True,
-            )
-            return []
-
-        return [stored for stored in stored_sessions if stored.scope != scope]
-
-    def _cleanup_superseded_sessions(self, stored_sessions: list[StoredAgentAppSession]) -> None:
-        for stored_session in stored_sessions:
-            try:
-                if stored_session.runtime_layer_specs:
-                    payload = AgentBackendSessionCleanupPayload(
-                        session_snapshot=stored_session.session_snapshot,
-                        runtime_layer_specs=stored_session.runtime_layer_specs,
-                        idempotency_key=(
-                            f"{stored_session.scope.tenant_id}:{stored_session.scope.app_id}:"
-                            f"{stored_session.scope.conversation_id}:{stored_session.scope.agent_id}:"
-                            f"{stored_session.scope.agent_config_snapshot_id or 'no-config'}:"
-                            f"superseded-session-cleanup:{stored_session.backend_run_id or 'no-run'}"
-                        ),
-                        metadata={
-                            "tenant_id": stored_session.scope.tenant_id,
-                            "app_id": stored_session.scope.app_id,
-                            "conversation_id": stored_session.scope.conversation_id,
-                            "agent_id": stored_session.scope.agent_id,
-                            "agent_config_snapshot_id": stored_session.scope.agent_config_snapshot_id,
-                            "previous_agent_backend_run_id": stored_session.backend_run_id,
-                        },
-                    )
-                    cleanup_conversation_agent_runtime_session.delay(payload.model_dump(mode="json"))
-            except Exception:
-                logger.warning(
-                    "Failed to enqueue Agent backend cleanup for superseded Agent App session: "
-                    "tenant_id=%s app_id=%s conversation_id=%s agent_id=%s backend_run_id=%s",
-                    stored_session.scope.tenant_id,
-                    stored_session.scope.app_id,
-                    stored_session.scope.conversation_id,
-                    stored_session.scope.agent_id,
-                    stored_session.backend_run_id,
-                    exc_info=True,
-                )
-
-    def _mark_session_cleaned(
-        self,
-        *,
-        scope: AgentAppSessionScope,
-        backend_run_id: str,
-    ) -> None:
-        """Best-effort delete-on-exit cleanup for the API-side session row.
-
-        Once the Agent backend reaches a terminal event, cleanup persistence
-        must not replace the original publish/error outcome for that turn.
-        """
-        try:
-            self._session_store.mark_cleaned(scope=scope, backend_run_id=backend_run_id)
-        except Exception:
-            logger.warning(
-                "Failed to retire Agent App conversation session after delete-on-exit: "
-                "tenant_id=%s app_id=%s conversation_id=%s agent_id=%s backend_run_id=%s",
-                scope.tenant_id,
-                scope.app_id,
-                scope.conversation_id,
-                scope.agent_id,
-                backend_run_id,
-                exc_info=True,
-            )
-
    @staticmethod
    def _terminal_output_to_answer(output: JsonValue) -> str:
        """Normalize the backend's terminal output to assistant text.
@@ -70,11 +70,12 @@ class AgentAppRuntimeBuildContext:
    conversation_id: str
    user_query: str
    idempotency_key: str
+    binding_id: str
+    backend_binding_ref: str
    agent_config_version_kind: Literal["snapshot", "draft", "build_draft"] = "snapshot"
    session_snapshot: CompositorSessionSnapshot | None = None
    # ENG-638: set when resuming a chat turn after a submitted ask_human form.
    deferred_tool_results: DeferredToolResultsPayload | None = None
-    suspend_on_exit: bool = True


@dataclass(frozen=True, slots=True)
@@ -82,6 +83,7 @@ class AgentAppRuntimeRequest:
    request: CreateRunRequest
    redacted_request: dict[str, Any]
    metadata: dict[str, Any]
+    binding_id: str


 class AgentAppRuntimeRequestBuilder:
@@ -160,6 +162,7 @@ class AgentAppRuntimeRequestBuilder:
                    invoke_from=cast(DifyExecutionContextInvokeFrom, context.dify_context.invoke_from.value),
                    agent_mode="agent_app",
                ),
+                backend_binding_ref=context.backend_binding_ref,
                # ENG-616: expand slash-menu mention tokens to canonical names so
                # no frontend-internal {{#…#}} marker ever reaches the model.
                agent_soul_prompt=expand_prompt_mentions(agent_soul.prompt.system_prompt, soul_prompt_resolver).strip()
@@ -175,13 +178,17 @@ class AgentAppRuntimeRequestBuilder:
                shell_config=build_shell_layer_config(agent_soul),
                session_snapshot=context.session_snapshot,
                deferred_tool_results=context.deferred_tool_results,
-                suspend_on_exit=context.suspend_on_exit,
                idempotency_key=context.idempotency_key,
                metadata=metadata,
            )
        )
        redacted = cast(dict[str, Any], redact_for_agent_backend_log(request))
-        return AgentAppRuntimeRequest(request=request, redacted_request=redacted, metadata=metadata)
+        return AgentAppRuntimeRequest(
+            request=request,
+            redacted_request=redacted,
+            metadata=metadata,
+            binding_id=context.binding_id,
+        )

    def _build_tool_layers(
        self,
@@ -1,255 +1,176 @@
-"""Conversation-keyed Agent backend session store for the Agent App type.
-
-Shares the unified ``agent_runtime_sessions`` table with the workflow Agent
-Node store, but owns rows with ``owner_type = conversation``: one Agent App
-conversation maps to one Agent session, so multi-turn chat re-enters the same
-``session_snapshot``. Cross-conversation memory (PRD Global / Per app) is a
-phase-2 concern and not modeled here.
-"""
+"""Persist and resolve the exact participant owned by an Agent App caller."""

 from __future__ import annotations

-from dataclasses import dataclass, field
+from dataclasses import dataclass

 from agenton.compositor import CompositorSessionSnapshot
-from dify_agent.protocol import RuntimeLayerSpec
-from pydantic import TypeAdapter
 from sqlalchemy import select
+from sqlalchemy.orm import Session

 from core.db.session_factory import session_factory
-from libs.datetime_utils import naive_utc_now
 from models.agent import (
-    AgentRuntimeSession,
-    AgentRuntimeSessionOwnerType,
-    AgentRuntimeSessionStatus,
+    AgentConfigDraft,
+    AgentConfigDraftType,
+    AgentConfigVersionKind,
+    AgentWorkspaceBinding,
+    AgentWorkspaceOwnerType,
+)
+from models.model import App, Conversation
+from services.agent.workspace_service import (
+    AgentWorkspaceNotFoundError,
+    AgentWorkspaceService,
+    WorkspaceOwnerScope,
 )
-
-_RUNTIME_LAYER_SPECS_ADAPTER: TypeAdapter[list[RuntimeLayerSpec]] = TypeAdapter(list[RuntimeLayerSpec])
-
-
-def _serialize_runtime_layer_specs(specs: list[RuntimeLayerSpec]) -> str:
-    return _RUNTIME_LAYER_SPECS_ADAPTER.dump_json(specs).decode()
-
-
-def _deserialize_runtime_layer_specs(value: str | None) -> list[RuntimeLayerSpec]:
-    if not value:
-        return []
-    return _RUNTIME_LAYER_SPECS_ADAPTER.validate_json(value)


@dataclass(frozen=True, slots=True)
 class AgentAppSessionScope:
-    """Identity of one Agent App conversation session."""
-
    tenant_id: str
    app_id: str
    conversation_id: str
    agent_id: str
-    agent_config_snapshot_id: str | None
+    agent_config_snapshot_id: str
+    home_snapshot_id: str
+    agent_config_version_kind: AgentConfigVersionKind = AgentConfigVersionKind.SNAPSHOT
+    build_draft_id: str | None = None
+
+    @property
+    def workspace_owner(self) -> WorkspaceOwnerScope:
+        owner_type = (
+            AgentWorkspaceOwnerType.BUILD_DRAFT if self.build_draft_id else AgentWorkspaceOwnerType.CONVERSATION
+        )
+        return WorkspaceOwnerScope(
+            tenant_id=self.tenant_id,
+            app_id=self.app_id,
+            owner_type=owner_type,
+            owner_id=self.build_draft_id or self.conversation_id,
+        )


@dataclass(frozen=True, slots=True)
 class StoredAgentAppSession:
-    """Persisted Agent App conversation session with reusable runtime specs."""
-
    scope: AgentAppSessionScope
-    session_snapshot: CompositorSessionSnapshot
-    backend_run_id: str | None
-    runtime_layer_specs: list[RuntimeLayerSpec] = field(default_factory=list)
-    # ENG-635: set while the conversation turn is paused on a dify.ask_human
-    # deferred call, awaiting a HITL form submission.
+    binding_id: str
+    workspace_id: str
+    backend_binding_ref: str
+    session_snapshot: CompositorSessionSnapshot | None
    pending_form_id: str | None = None
    pending_tool_call_id: str | None = None


-class AgentAppRuntimeSessionStore:
-    """Persists Agent backend session snapshots for Agent App conversations."""
+class AgentAppWorkspaceStore:
+    """Resolve Agent App sessions through a caller-owned Binding pointer."""

-    def load_active_snapshot(self, scope: AgentAppSessionScope) -> CompositorSessionSnapshot | None:
-        stored = self.load_active_session(scope)
-        return stored.session_snapshot if stored is not None else None
-
-    def load_active_session(self, scope: AgentAppSessionScope) -> StoredAgentAppSession | None:
+    def load_or_create(self, scope: AgentAppSessionScope) -> StoredAgentAppSession:
        with session_factory.create_session() as session:
-            row = session.scalar(self._active_stmt(scope))
-            if row is None:
-                return None
-            return StoredAgentAppSession(
-                scope=scope,
-                session_snapshot=CompositorSessionSnapshot.model_validate_json(row.session_snapshot),
-                backend_run_id=row.backend_run_id,
-                runtime_layer_specs=_deserialize_runtime_layer_specs(row.composition_layer_specs),
-                pending_form_id=row.pending_form_id,
-                pending_tool_call_id=row.pending_tool_call_id,
-            )
-
-    def load_active_session_for_conversation(
-        self, *, tenant_id: str, app_id: str, conversation_id: str
-    ) -> StoredAgentAppSession | None:
-        """Load the latest ACTIVE session for one conversation-level sandbox lookup.
-
-        Sandbox inspection only knows the product locator
-        ``tenant_id + app_id + conversation_id``; it does not know which
-        ``agent_id`` or Agent Soul snapshot produced the active shell session.
-        This method therefore resolves the newest ACTIVE conversation-owned row
-        for that conversation and returns both the resumable snapshot and the
-        persisted non-sensitive runtime layer specs needed to build a
-        ``SandboxLocator``.
-        """
-        stmt = (
-            select(AgentRuntimeSession)
-            .where(
-                AgentRuntimeSession.owner_type == AgentRuntimeSessionOwnerType.CONVERSATION,
-                AgentRuntimeSession.tenant_id == tenant_id,
-                AgentRuntimeSession.app_id == app_id,
-                AgentRuntimeSession.conversation_id == conversation_id,
-                AgentRuntimeSession.status == AgentRuntimeSessionStatus.ACTIVE,
-            )
-            .order_by(AgentRuntimeSession.updated_at.desc())
-        )
-        with session_factory.create_session() as session:
-            row = session.scalar(stmt)
-            if row is None:
-                return None
-            return StoredAgentAppSession(
-                scope=AgentAppSessionScope(
-                    tenant_id=row.tenant_id,
-                    app_id=row.app_id,
-                    conversation_id=row.conversation_id or "",
-                    agent_id=row.agent_id,
-                    agent_config_snapshot_id=row.agent_config_snapshot_id or "",
-                ),
-                session_snapshot=CompositorSessionSnapshot.model_validate_json(row.session_snapshot),
-                backend_run_id=row.backend_run_id,
-                runtime_layer_specs=_deserialize_runtime_layer_specs(row.composition_layer_specs),
-            )
-
-    def list_active_sessions_for_conversation(
-        self, *, tenant_id: str, app_id: str, conversation_id: str
-    ) -> list[StoredAgentAppSession]:
-        """List all ACTIVE conversation-owned sessions for lifecycle cleanup."""
-        stmt = (
-            select(AgentRuntimeSession)
-            .where(
-                AgentRuntimeSession.owner_type == AgentRuntimeSessionOwnerType.CONVERSATION,
-                AgentRuntimeSession.tenant_id == tenant_id,
-                AgentRuntimeSession.app_id == app_id,
-                AgentRuntimeSession.conversation_id == conversation_id,
-                AgentRuntimeSession.status == AgentRuntimeSessionStatus.ACTIVE,
-            )
-            .order_by(AgentRuntimeSession.updated_at.desc())
-        )
-        with session_factory.create_session() as session:
-            rows = session.scalars(stmt).all()
-            return [
-                StoredAgentAppSession(
-                    scope=AgentAppSessionScope(
-                        tenant_id=row.tenant_id,
-                        app_id=row.app_id,
-                        conversation_id=row.conversation_id or "",
-                        agent_id=row.agent_id,
-                        agent_config_snapshot_id=row.agent_config_snapshot_id,
-                    ),
-                    session_snapshot=CompositorSessionSnapshot.model_validate_json(row.session_snapshot),
-                    backend_run_id=row.backend_run_id,
-                    runtime_layer_specs=_deserialize_runtime_layer_specs(row.composition_layer_specs),
-                    pending_form_id=row.pending_form_id,
-                    pending_tool_call_id=row.pending_tool_call_id,
+            caller = self._load_caller(session=session, scope=scope)
+            binding_id = caller.agent_workspace_binding_id
+            if binding_id is None:
+                binding = AgentWorkspaceService.create_binding(
+                    session=session,
+                    scope=scope.workspace_owner,
+                    agent_id=scope.agent_id,
+                    base_home_snapshot_id=scope.home_snapshot_id,
+                    agent_config_version_id=scope.agent_config_snapshot_id,
+                    agent_config_version_kind=scope.agent_config_version_kind,
                )
-                for row in rows
-            ]
+                caller.agent_workspace_binding_id = binding.id
+                session.commit()
+            else:
+                binding = self._get_binding(session=session, scope=scope, binding_id=binding_id)
+            return self._stored(scope, binding)
+
+    @staticmethod
+    def _load_caller(*, session: Session, scope: AgentAppSessionScope) -> Conversation | AgentConfigDraft:
+        if scope.build_draft_id is not None:
+            if scope.agent_config_version_kind != AgentConfigVersionKind.BUILD_DRAFT:
+                raise AgentWorkspaceNotFoundError("Build Draft caller requires build_draft generation")
+            draft = session.scalar(
+                select(AgentConfigDraft).where(
+                    AgentConfigDraft.id == scope.build_draft_id,
+                    AgentConfigDraft.tenant_id == scope.tenant_id,
+                    AgentConfigDraft.agent_id == scope.agent_id,
+                    AgentConfigDraft.draft_type == AgentConfigDraftType.DEBUG_BUILD,
+                )
+            )
+            if draft is None:
+                raise AgentWorkspaceNotFoundError("Build Draft caller is unavailable")
+            return draft
+        if scope.agent_config_version_kind == AgentConfigVersionKind.BUILD_DRAFT:
+            raise AgentWorkspaceNotFoundError("Build Draft caller ID is required")
+        conversation = session.scalar(
+            select(Conversation)
+            .join(App, App.id == Conversation.app_id)
+            .where(
+                App.tenant_id == scope.tenant_id,
+                Conversation.id == scope.conversation_id,
+                Conversation.app_id == scope.app_id,
+                Conversation.is_deleted.is_(False),
+            )
+        )
+        if conversation is None:
+            raise AgentWorkspaceNotFoundError("Conversation caller is unavailable")
+        return conversation
+
+    @staticmethod
+    def _get_binding(
+        *,
+        session: Session,
+        scope: AgentAppSessionScope,
+        binding_id: str,
+    ) -> AgentWorkspaceBinding:
+        binding = AgentWorkspaceService.get_active_binding(
+            session=session,
+            tenant_id=scope.tenant_id,
+            binding_id=binding_id,
+            expected_owner_scope=scope.workspace_owner,
+        )
+        if binding is None or binding.agent_id != scope.agent_id:
+            raise AgentWorkspaceNotFoundError("Caller participant Binding is unavailable")
+        AgentWorkspaceService.validate_binding_generation(
+            binding,
+            base_home_snapshot_id=scope.home_snapshot_id,
+            agent_config_version_id=scope.agent_config_snapshot_id,
+            agent_config_version_kind=scope.agent_config_version_kind,
+        )
+        return binding

    def save_active_snapshot(
        self,
        *,
        scope: AgentAppSessionScope,
-        backend_run_id: str,
+        binding_id: str,
        snapshot: CompositorSessionSnapshot | None,
-        runtime_layer_specs: list[RuntimeLayerSpec],
        pending_form_id: str | None = None,
        pending_tool_call_id: str | None = None,
    ) -> None:
-        """Persist the current conversation snapshot and enforce one ACTIVE row.
-
-        Agent App chat treats one conversation as one resumable runtime shell.
-        Saving the latest snapshot therefore upserts the scoped row back to
-        ACTIVE and retires any other ACTIVE conversation-owned rows for the
-        same ``tenant_id + app_id + conversation_id`` so later lookups see a
-        single active session.
-        """
        if snapshot is None:
            return
-        snapshot_json = snapshot.model_dump_json()
-        runtime_layer_specs_json = _serialize_runtime_layer_specs(runtime_layer_specs)
-        with session_factory.create_session() as session:
-            row = session.scalar(self._scope_stmt(scope))
-            if row is None:
-                row = AgentRuntimeSession(
-                    tenant_id=scope.tenant_id,
-                    app_id=scope.app_id,
-                    owner_type=AgentRuntimeSessionOwnerType.CONVERSATION,
-                    agent_id=scope.agent_id,
-                    agent_config_snapshot_id=scope.agent_config_snapshot_id,
-                    conversation_id=scope.conversation_id,
-                    backend_run_id=backend_run_id,
-                    session_snapshot=snapshot_json,
-                    composition_layer_specs=runtime_layer_specs_json,
-                    status=AgentRuntimeSessionStatus.ACTIVE,
-                    pending_form_id=pending_form_id,
-                    pending_tool_call_id=pending_tool_call_id,
-                )
-                session.add(row)
-            else:
-                row.backend_run_id = backend_run_id
-                row.session_snapshot = snapshot_json
-                row.composition_layer_specs = runtime_layer_specs_json
-                row.status = AgentRuntimeSessionStatus.ACTIVE
-                row.cleaned_at = None
-                # Set (or clear, when omitted) the ask_human pause correlation.
-                row.pending_form_id = pending_form_id
-                row.pending_tool_call_id = pending_tool_call_id
-            session.flush()
-            other_rows = session.scalars(
-                select(AgentRuntimeSession).where(
-                    AgentRuntimeSession.owner_type == AgentRuntimeSessionOwnerType.CONVERSATION,
-                    AgentRuntimeSession.tenant_id == scope.tenant_id,
-                    AgentRuntimeSession.app_id == scope.app_id,
-                    AgentRuntimeSession.conversation_id == scope.conversation_id,
-                    AgentRuntimeSession.status == AgentRuntimeSessionStatus.ACTIVE,
-                    AgentRuntimeSession.id != row.id,
-                )
-            ).all()
-            for other_row in other_rows:
-                other_row.status = AgentRuntimeSessionStatus.CLEANED
-                other_row.cleaned_at = naive_utc_now()
-            session.commit()
-
-    def mark_cleaned(self, *, scope: AgentAppSessionScope, backend_run_id: str | None = None) -> None:
-        with session_factory.create_session() as session:
-            row = session.scalar(self._active_stmt(scope))
-            if row is None:
-                return
-            if backend_run_id is not None:
-                row.backend_run_id = backend_run_id
-            row.status = AgentRuntimeSessionStatus.CLEANED
-            row.cleaned_at = naive_utc_now()
-            session.commit()
+        AgentWorkspaceService.save_binding_session_snapshot(
+            tenant_id=scope.tenant_id,
+            binding_id=binding_id,
+            session_snapshot=snapshot.model_dump_json(),
+            pending_form_id=pending_form_id,
+            pending_tool_call_id=pending_tool_call_id,
+        )

    @staticmethod
-    def _scope_stmt(scope: AgentAppSessionScope):
-        stmt = select(AgentRuntimeSession).where(
-            AgentRuntimeSession.owner_type == AgentRuntimeSessionOwnerType.CONVERSATION,
-            AgentRuntimeSession.tenant_id == scope.tenant_id,
-            AgentRuntimeSession.conversation_id == scope.conversation_id,
-            AgentRuntimeSession.agent_id == scope.agent_id,
+    def _stored(scope: AgentAppSessionScope, binding: AgentWorkspaceBinding) -> StoredAgentAppSession:
+        snapshot = (
+            CompositorSessionSnapshot.model_validate_json(binding.session_snapshot)
+            if binding.session_snapshot
+            else None
+        )
+        return StoredAgentAppSession(
+            scope=scope,
+            binding_id=binding.id,
+            workspace_id=binding.workspace_id,
+            backend_binding_ref=binding.backend_binding_ref,
+            session_snapshot=snapshot,
+            pending_form_id=binding.pending_form_id,
+            pending_tool_call_id=binding.pending_tool_call_id,
        )
-        if scope.agent_config_snapshot_id is None:
-            return stmt.where(AgentRuntimeSession.agent_config_snapshot_id.is_(None))
-        return stmt.where(AgentRuntimeSession.agent_config_snapshot_id == scope.agent_config_snapshot_id)
-
-    @classmethod
-    def _active_stmt(cls, scope: AgentAppSessionScope):
-        return cls._scope_stmt(scope).where(AgentRuntimeSession.status == AgentRuntimeSessionStatus.ACTIVE)


-__all__ = ["AgentAppRuntimeSessionStore", "AgentAppSessionScope", "StoredAgentAppSession"]
+__all__ = ["AgentAppSessionScope", "AgentAppWorkspaceStore", "StoredAgentAppSession"]
@@ -1,3 +1,5 @@
+import logging
+import threading
 from collections.abc import Generator, Mapping, Sequence
 from contextlib import AbstractContextManager, nullcontext
 from typing import TYPE_CHECKING, Any, Union, final
@@ -23,6 +25,10 @@ from services.workflow_draft_variable_service import DraftVariableSaver as Draft
 if TYPE_CHECKING:
    from graphon.variables.input_entities import VariableEntity

+logger = logging.getLogger(__name__)
+
+_WORKER_THREAD_JOIN_TIMEOUT_SECONDS = 300
+

@final
 class _DebuggerDraftVariableSaver:
@@ -64,6 +70,29 @@ class _DebuggerDraftVariableSaver:
 class BaseAppGenerator:
    _file_access_controller: DatabaseFileAccessController = DatabaseFileAccessController()

+    @staticmethod
+    def _join_worker_thread(worker_thread: threading.Thread) -> None:
+        # Bound the wait so a leaked app worker cannot occupy an execution slot indefinitely.
+        worker_thread.join(timeout=_WORKER_THREAD_JOIN_TIMEOUT_SECONDS)
+        if worker_thread.is_alive():
+            logger.warning(
+                "Possible app worker thread leak: thread_name=%s timeout_seconds=%s; "
+                "continuing without waiting further to avoid occupying an execution slot indefinitely",
+                worker_thread.name,
+                _WORKER_THREAD_JOIN_TIMEOUT_SECONDS,
+            )
+
+    @staticmethod
+    def _wrap_stream_with_worker_thread_join[ResponseT](
+        response_stream: Generator[ResponseT, None, None],
+        worker_thread: threading.Thread,
+    ) -> Generator[ResponseT, None, None]:
+        """Keep the producer owned by the response stream until both finish."""
+        try:
+            yield from response_stream
+        finally:
+            BaseAppGenerator._join_worker_thread(worker_thread)
+
    @staticmethod
    def _bind_file_access_scope(
        *,
@@ -351,17 +351,28 @@ class PipelineGenerator(BaseAppGenerator):
                user,
                tenant_id=pipeline.tenant_id,
            )
-            # return response or stream generator
-            response = self._handle_response(
-                application_generate_entity=application_generate_entity,
-                workflow=workflow,
-                queue_manager=queue_manager,
-                user=user,
-                stream=streaming,
-                draft_var_saver_factory=draft_var_saver_factory,
-            )
+            try:
+                response = self._handle_response(
+                    application_generate_entity=application_generate_entity,
+                    workflow=workflow,
+                    queue_manager=queue_manager,
+                    user=user,
+                    stream=streaming,
+                    draft_var_saver_factory=draft_var_saver_factory,
+                )
+                converted_response = WorkflowAppGenerateResponseConverter.convert(
+                    response=response,
+                    invoke_from=invoke_from,
+                )
+            except BaseException:
+                self._join_worker_thread(worker_thread)
+                raise

-            return WorkflowAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+            if isinstance(converted_response, Generator):
+                return self._wrap_stream_with_worker_thread_join(converted_response, worker_thread)
+
+            self._join_worker_thread(worker_thread)
+            return converted_response

    def single_iteration_generate(
        self,
@@ -405,17 +405,28 @@ class WorkflowAppGenerator(BaseAppGenerator):
                tenant_id=app_model.tenant_id,
            )

-            # return response or stream generator
-            response = self._handle_response(
-                application_generate_entity=application_generate_entity,
-                workflow=workflow,
-                queue_manager=queue_manager,
-                user=user,
-                draft_var_saver_factory=draft_var_saver_factory,
-                stream=streaming,
-            )
+            try:
+                response = self._handle_response(
+                    application_generate_entity=application_generate_entity,
+                    workflow=workflow,
+                    queue_manager=queue_manager,
+                    user=user,
+                    draft_var_saver_factory=draft_var_saver_factory,
+                    stream=streaming,
+                )
+                converted_response = WorkflowAppGenerateResponseConverter.convert(
+                    response=response,
+                    invoke_from=invoke_from,
+                )
+            except BaseException:
+                self._join_worker_thread(worker_thread)
+                raise

-            return WorkflowAppGenerateResponseConverter.convert(response=response, invoke_from=invoke_from)
+            if isinstance(converted_response, Generator):
+                return self._wrap_stream_with_worker_thread_join(converted_response, worker_thread)
+
+            self._join_worker_thread(worker_thread)
+            return converted_response

    def single_iteration_generate(
        self,
@@ -10,11 +10,11 @@ from core.app.apps.workflow.command_channels import (
    CombinedCommandChannel,
 )
 from core.app.apps.workflow_app_runner import WorkflowBasedAppRunner
-from core.app.entities.app_invoke_entities import InvokeFrom, WorkflowAppGenerateEntity
+from core.app.entities.app_invoke_entities import DifyRunContext, InvokeFrom, WorkflowAppGenerateEntity
 from core.app.workflow.layers.persistence import PersistenceWorkflowInfo, WorkflowPersistenceLayer
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
 from core.workflow.node_factory import get_default_root_node_id
-from core.workflow.nodes.agent_v2.session_cleanup_layer import build_workflow_agent_session_cleanup_layer
+from core.workflow.nodes.agent_v2.workspace_retirement_layer import build_workflow_agent_workspace_retirement_layer
 from core.workflow.snippet_start import get_compatible_start_aliases
 from core.workflow.system_variables import build_bootstrap_variables, build_system_variables
 from core.workflow.variable_pool_initializer import add_node_inputs_to_pool, add_variables_to_pool
@@ -197,7 +197,18 @@ class WorkflowAppRunner(WorkflowBasedAppRunner):
        )

        workflow_entry.graph_engine.layer(persistence_layer)
-        workflow_entry.graph_engine.layer(build_workflow_agent_session_cleanup_layer())
+        workflow_entry.graph_engine.layer(
+            build_workflow_agent_workspace_retirement_layer(
+                dify_run_context=DifyRunContext(
+                    tenant_id=self._workflow.tenant_id,
+                    app_id=self._workflow.app_id,
+                    user_id=self.application_generate_entity.user_id,
+                    user_from=user_from,
+                    invoke_from=invoke_from,
+                    trace_session_id=self.application_generate_entity.extras.get("trace_session_id"),
+                )
+            )
+        )
        for layer in self._graph_engine_layers:
            workflow_entry.graph_engine.layer(layer)

@@ -15,8 +15,6 @@ if TYPE_CHECKING:


 DIFY_RUN_CONTEXT_KEY = "_dify"
-AGENT_RUNTIME_EXIT_INTENT_ARG = "_agent_runtime_exit_intent"
-type AgentRuntimeExitIntent = Literal["suspend", "delete"]


 class UserFrom(StrEnum):
@@ -227,12 +225,8 @@ class AgentAppGenerateEntity(ChatAppGenerateEntity):
    backend should read from: immutable snapshot, shared draft, or per-user
    build draft.

-    ``agent_runtime_session_snapshot_id`` carries the runtime session scope
-    used to resume or suspend within the same editable config surface.
-
-    ``agent_runtime_exit_intent`` is API-internal lifecycle policy for the
-    Agent backend session after this turn finishes. Normal chat/resume turns
-    suspend on exit; build-chat finalization deletes the backend runtime.
+    ``agent_session_scope_config_version_id`` identifies the draft or immutable
+    config version whose Workspace Binding should be reused for this session.

    ``prompt_file_mappings`` preserves the raw request ``files`` array for the
    Agent backend prompt. These references are appended to the backend prompt
@@ -242,8 +236,7 @@ class AgentAppGenerateEntity(ChatAppGenerateEntity):
    agent_id: str
    agent_config_snapshot_id: str
    agent_config_version_kind: Literal["snapshot", "draft", "build_draft"] = "snapshot"
-    agent_runtime_session_snapshot_id: str | None = None
-    agent_runtime_exit_intent: AgentRuntimeExitIntent = "suspend"
+    agent_session_scope_config_version_id: str | None = None
    prompt_file_mappings: Sequence[JsonValue] = Field(default_factory=list)


@@ -20,11 +20,13 @@ from core.helper.trace_id_helper import ParentTraceContext
 from core.ops.entities.trace_entity import TraceTaskName
 from core.ops.ops_trace_manager import TraceQueueManager, TraceTask
 from core.repositories.factory import WorkflowExecutionRepository, WorkflowNodeExecutionRepository
+from core.workflow.node_execution_process_data import preserve_workflow_agent_binding_id
 from core.workflow.system_variables import SystemVariableKey
 from core.workflow.variable_prefixes import SYSTEM_VARIABLE_NODE_ID
 from core.workflow.workflow_run_outputs import project_node_outputs_for_workflow_run
 from graphon.entities import WorkflowExecution, WorkflowNodeExecution
 from graphon.enums import (
+    BuiltinNodeTypes,
    WorkflowExecutionStatus,
    WorkflowNodeExecutionMetadataKey,
    WorkflowNodeExecutionStatus,
@@ -241,7 +243,10 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
        )

        self._node_execution_cache[event.id] = domain_execution
-        self._workflow_node_execution_repository.save(domain_execution)
+        if event.node_type == BuiltinNodeTypes.AGENT and event.node_version == "2":
+            self._workflow_node_execution_repository.save_synchronously(domain_execution)
+        else:
+            self._workflow_node_execution_repository.save(domain_execution)

        snapshot = _NodeRuntimeSnapshot(
            node_id=event.node_id,
@@ -361,7 +366,11 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
    def _append_retry_history(self, execution: WorkflowNodeExecution, event: NodeRunRetryEvent) -> None:
        """Append a validated full attempt before repository truncation or offload."""
        finished_at = naive_utc_now()
-        process_data = dict(execution.process_data or {})
+        process_data = preserve_workflow_agent_binding_id(
+            event.node_run_result.process_data,
+            execution.process_data,
+        )
+        process_data = dict(process_data or {})
        raw_history = process_data.get(RETRY_HISTORY_PROCESS_DATA_KEY)
        history = list(raw_history) if isinstance(raw_history, list) else []
        projected_outputs = project_node_outputs_for_workflow_run(
@@ -390,11 +399,12 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
        next_process_data: Mapping[str, Any] | None,
    ) -> Mapping[str, Any] | None:
        """Keep internal retry history while replacing node-specific Process Data."""
+        merged_process_data = preserve_workflow_agent_binding_id(existing_process_data, next_process_data)
        raw_history = (existing_process_data or {}).get(RETRY_HISTORY_PROCESS_DATA_KEY)
        if not isinstance(raw_history, list) or not raw_history:
-            return next_process_data
+            return merged_process_data

-        merged_process_data = dict(next_process_data or {})
+        merged_process_data = dict(merged_process_data or {})
        merged_process_data[RETRY_HISTORY_PROCESS_DATA_KEY] = raw_history
        return merged_process_data

@@ -440,6 +450,11 @@ class WorkflowPersistenceLayer(GraphEngineLayer):
                outputs=projected_outputs,
                metadata=node_result.metadata,
            )
+        else:
+            domain_execution.process_data = preserve_workflow_agent_binding_id(
+                node_result.process_data,
+                domain_execution.process_data,
+            )

        self._workflow_node_execution_repository.save(domain_execution)
        self._workflow_node_execution_repository.save_execution_data(domain_execution)
@@ -39,15 +39,16 @@ class Jinja2TemplateTransformer(TemplateTransformer):
    @override
    def get_runner_script(cls) -> str:
        runner_script = dedent(f"""
-            import jinja2
            import json
            from base64 import b64decode
+            from jinja2.sandbox import SandboxedEnvironment

            # declare main function
            def main(**inputs):
                # Decode base64-encoded template to handle special characters safely
                template_code = b64decode('{cls._template_b64_placeholder}').decode('utf-8')
-                template = jinja2.Template(template_code)
+                env = SandboxedEnvironment()
+                template = env.from_string(template_code)
                return template.render(**inputs)

            # decode and prepare input dict
@@ -67,12 +68,13 @@ class Jinja2TemplateTransformer(TemplateTransformer):
    @override
    def get_preload_script(cls) -> str:
        preload_script = dedent("""
-            import jinja2
+            from jinja2.sandbox import SandboxedEnvironment
            from base64 import b64decode

            def _jinja2_preload_():
-                # prepare jinja2 environment, load template and render before to avoid sandbox issue
-                template = jinja2.Template('{{s}}')
+                # prepare jinja2 sandboxed environment, load template and render
+                env = SandboxedEnvironment()
+                template = env.from_string('{{s}}')
                template.render(s='a')

            if __name__ == '__main__':
@@ -519,7 +519,7 @@ class IndexingRunner:
    def filter_string(text):
        text = re.sub(r"<\|", "<", text)
        text = re.sub(r"\|>", ">", text)
-        text = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F\xEF\xBF\xBE]", "", text)
+        text = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]", "", text)
        # Unicode  U+FFFE
        text = re.sub("\ufffe", "", text)
        return text
@@ -66,7 +66,7 @@ from services.enterprise.plugin_manager_service import (
    PreUninstallPluginRequest,
 )
 from services.errors.plugin import PluginInstallationForbiddenError
-from services.feature_service import FeatureService, PluginInstallationScope
+from services.feature_service import FeatureService, PluginInstallationPermissionModel, PluginInstallationScope

 logger = logging.getLogger(__name__)
 _provider_entities_adapter: TypeAdapter[list[ProviderEntity]] = TypeAdapter(list[ProviderEntity])
@@ -434,14 +434,18 @@ class PluginService:
                    exc_info=True,
                )

+    @classmethod
+    def _fetch_plugin_model_providers_uncached(
+        cls, tenant_id: str, client: PluginModelClient | None
+    ) -> tuple[ProviderEntity, ...]:
+        model_client = client or PluginModelClient()
+        return tuple(cls._to_provider_entity(provider) for provider in model_client.fetch_model_providers(tenant_id))
+
    @classmethod
    def _fetch_and_cache_plugin_model_providers(
        cls, tenant_id: str, client: PluginModelClient | None, *, refresh_generation: int | None
    ) -> tuple[ProviderEntity, ...]:
-        model_client = client or PluginModelClient()
-        providers = tuple(
-            cls._to_provider_entity(provider) for provider in model_client.fetch_model_providers(tenant_id)
-        )
+        providers = cls._fetch_plugin_model_providers_uncached(tenant_id, client)
        generation = cls._load_plugin_model_providers_generation(tenant_id)
        if generation is not None and generation == refresh_generation:
            cls._store_cached_plugin_model_providers(tenant_id, generation, providers)
@@ -471,6 +475,9 @@ class PluginService:
        are intentionally owned by this service so tenant isolation and cache
        expiry are handled in one place.
        """
+        if not dify_config.PLUGIN_MODEL_PROVIDERS_CACHE_ENABLED:
+            return cls._fetch_plugin_model_providers_uncached(tenant_id, client)
+
        deadline = time.monotonic() + cls.PLUGIN_MODEL_PROVIDERS_LOCK_WAIT_TIMEOUT

        while True:
@@ -597,22 +604,30 @@ class PluginService:
            return result

    @staticmethod
-    def _check_marketplace_only_permission():
+    def _check_marketplace_only_permission() -> None:
        """
        Check if the marketplace only permission is enabled
        """
-        features = FeatureService.get_system_features()
-        if features.plugin_installation_permission.restrict_to_marketplace_only:
+        permission = PluginService._get_plugin_installation_permission()
+        if permission.restrict_to_marketplace_only:
            raise PluginInstallationForbiddenError("Plugin installation is restricted to marketplace only")

    @staticmethod
-    def _check_plugin_installation_scope(plugin_verification: PluginVerification | None):
+    def _get_plugin_installation_permission() -> PluginInstallationPermissionModel:
+        """Resolve the validated policy and reject deny-all before any installation side effect."""
+        permission = FeatureService.get_plugin_installation_permission()
+        if permission.plugin_installation_scope == PluginInstallationScope.NONE:
+            raise PluginInstallationForbiddenError("Installing plugins is not allowed")
+        return permission
+
+    @staticmethod
+    def _check_plugin_installation_scope(plugin_verification: PluginVerification | None) -> None:
        """
        Check the plugin installation scope
        """
-        features = FeatureService.get_system_features()
+        permission = PluginService._get_plugin_installation_permission()

-        match features.plugin_installation_permission.plugin_installation_scope:
+        match permission.plugin_installation_scope:
            case PluginInstallationScope.OFFICIAL_ONLY:
                if (
                    plugin_verification is None
@@ -627,10 +642,10 @@ class PluginService:
                    raise PluginInstallationForbiddenError(
                        "Plugin installation is restricted to official and specific partners"
                    )
-            case PluginInstallationScope.NONE:
-                raise PluginInstallationForbiddenError("Installing plugins is not allowed")
            case PluginInstallationScope.ALL:
                pass
+            case _:
+                raise PluginInstallationForbiddenError("Plugin installation policy is invalid")

    @staticmethod
    def get_debugging_key(tenant_id: str) -> str:
@@ -900,7 +915,7 @@ class PluginService:
        # check if plugin pkg is already downloaded
        manager = PluginInstaller()

-        features = FeatureService.get_system_features()
+        permission = PluginService._get_plugin_installation_permission()

        try:
            manager.fetch_plugin_manifest(tenant_id, new_plugin_unique_identifier)
@@ -912,7 +927,7 @@ class PluginService:
            response = manager.upload_pkg(
                tenant_id,
                pkg,
-                verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
+                verify_signature=permission.restrict_to_marketplace_only,
            )

            # check if the plugin is available to install
@@ -967,11 +982,11 @@ class PluginService:
        """
        PluginService._check_marketplace_only_permission()
        manager = PluginInstaller()
-        features = FeatureService.get_system_features()
+        permission = PluginService._get_plugin_installation_permission()
        response = manager.upload_pkg(
            tenant_id,
            pkg,
-            verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
+            verify_signature=permission.restrict_to_marketplace_only,
        )
        PluginService._check_plugin_installation_scope(response.verification)

@@ -989,13 +1004,13 @@ class PluginService:
        pkg = download_with_size_limit(
            f"https://github.com/{repo}/releases/download/{version}/{package}", dify_config.PLUGIN_MAX_PACKAGE_SIZE
        )
-        features = FeatureService.get_system_features()
+        permission = PluginService._get_plugin_installation_permission()

        manager = PluginInstaller()
        response = manager.upload_pkg(
            tenant_id,
            pkg,
-            verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
+            verify_signature=permission.restrict_to_marketplace_only,
        )
        PluginService._check_plugin_installation_scope(response.verification)

@@ -1069,7 +1084,7 @@ class PluginService:
        if not dify_config.MARKETPLACE_ENABLED:
            raise ValueError("marketplace is not enabled")

-        features = FeatureService.get_system_features()
+        permission = PluginService._get_plugin_installation_permission()

        manager = PluginInstaller()
        try:
@@ -1079,7 +1094,7 @@ class PluginService:
            response = manager.upload_pkg(
                tenant_id,
                pkg,
-                verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
+                verify_signature=permission.restrict_to_marketplace_only,
            )
            # check if the plugin is available to install
            PluginService._check_plugin_installation_scope(response.verification)
@@ -1101,7 +1116,7 @@ class PluginService:
        # collect actual plugin_unique_identifiers
        actual_plugin_unique_identifiers = []
        metas = []
-        features = FeatureService.get_system_features()
+        permission = PluginService._get_plugin_installation_permission()

        # check if already downloaded
        for plugin_unique_identifier in plugin_unique_identifiers:
@@ -1119,7 +1134,7 @@ class PluginService:
                response = manager.upload_pkg(
                    tenant_id,
                    pkg,
-                    verify_signature=features.plugin_installation_permission.restrict_to_marketplace_only,
+                    verify_signature=permission.restrict_to_marketplace_only,
                )
                # check if the plugin is available to install
                PluginService._check_plugin_installation_scope(response.verification)
@@ -9,7 +9,7 @@ class CleanProcessor:
        # remove invalid symbol
        text = re.sub(r"<\|", "<", text)
        text = re.sub(r"\|>", ">", text)
-        text = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F\xEF\xBF\xBE]", "", text)
+        text = re.sub(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]", "", text)
        # Unicode  U+FFFE
        text = re.sub("\ufffe", "", text)

@@ -2023,6 +2023,8 @@ class DatasetRetrieval:
            redis_client.zremrangebyscore(key, 0, current_time - 60000)
            request_count = redis_client.zcard(key)
            if request_count > knowledge_rate_limit.limit:
+                # The rate-limit exception is raised after this block, so commit the audit row
+                # explicitly instead of relying on the Session context, which only closes it.
                with session_factory.create_session() as session:
                    rate_limit_log = RateLimitLog(
                        tenant_id=tenant_id,
@@ -2030,6 +2032,7 @@ class DatasetRetrieval:
                        operation="knowledge",
                    )
                    session.add(rate_limit_log)
+                    session.commit()
                raise exc.RateLimitExceededError(
                    "you have reached the knowledge base request rate limit of your subscription."
                )
@@ -91,8 +91,8 @@ class FixedRecursiveCharacterTextSplitter(EnhanceRecursiveCharacterTextSplitter)
                splits = re.split(r" +", text)
            else:
                splits = text.split(separator)
-                if self._keep_separator:
-                    splits = [s + separator for s in splits[:-1]] + splits[-1:]
+            if self._keep_separator:
+                splits = [s + separator for s in splits[:-1]] + splits[-1:]
        else:
            splits = list(text)
        if separator == "\n":
@@ -16,6 +16,9 @@ from core.repositories.factory import (
    OrderConfig,
    WorkflowNodeExecutionRepository,
 )
+from core.repositories.sqlalchemy_workflow_node_execution_repository import (
+    SQLAlchemyWorkflowNodeExecutionRepository,
+)
 from graphon.entities import WorkflowNodeExecution
 from models import Account, CreatorUserRole, EndUser
 from models.workflow import WorkflowNodeExecutionTriggeredFrom
@@ -49,6 +52,7 @@ class CeleryWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
    _creator_user_role: CreatorUserRole
    _execution_cache: dict[str, WorkflowNodeExecution]
    _workflow_execution_mapping: dict[str, list[str]]
+    _sql_repository: SQLAlchemyWorkflowNodeExecutionRepository

    def __init__(
        self,
@@ -98,6 +102,13 @@ class CeleryWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):

        # Cache for mapping workflow_execution_ids to execution IDs for efficient retrieval
        self._workflow_execution_mapping = {}
+        self._sql_repository = SQLAlchemyWorkflowNodeExecutionRepository(
+            session_factory=session_factory,
+            tenant_id=tenant_id,
+            user=user,
+            app_id=app_id,
+            triggered_from=triggered_from,
+        )

        logger.info(
            "Initialized CeleryWorkflowNodeExecutionRepository for tenant %s, app %s, triggered_from %s",
@@ -149,6 +160,17 @@ class CeleryWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
            # For now, we'll re-raise the exception
            raise

+    @override
+    def save_synchronously(self, execution: WorkflowNodeExecution) -> None:
+        """Create the Agent v2 caller row before runtime participant allocation."""
+
+        self._sql_repository.save_synchronously(execution)
+        self._execution_cache[execution.id] = execution
+        if execution.workflow_execution_id:
+            execution_ids = self._workflow_execution_mapping.setdefault(execution.workflow_execution_id, [])
+            if execution.id not in execution_ids:
+                execution_ids.append(execution.id)
+
    @override
    def get_by_workflow_execution(
        self,
@@ -35,6 +35,8 @@ class WorkflowExecutionRepository(Protocol):
 class WorkflowNodeExecutionRepository(Protocol):
    def save(self, execution: WorkflowNodeExecution): ...

+    def save_synchronously(self, execution: WorkflowNodeExecution) -> None: ...
+
    def save_execution_data(self, execution: WorkflowNodeExecution): ...

    def get_by_workflow_execution(
@@ -18,6 +18,7 @@ from tenacity import before_sleep_log, retry, retry_if_exception, stop_after_att

 from configs import dify_config
 from core.repositories.factory import OrderConfig, WorkflowNodeExecutionRepository
+from core.workflow.node_execution_process_data import preserve_workflow_agent_binding_id
 from extensions.ext_storage import storage
 from graphon.entities import WorkflowNodeExecution
 from graphon.enums import WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
@@ -372,6 +373,12 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
            logger.exception("Failed to save workflow node execution after all retries")
            raise

+    @override
+    def save_synchronously(self, execution: WorkflowNodeExecution) -> None:
+        """Persist a caller row before an Agent v2 participant is materialized."""
+
+        self.save(execution)
+
    def _persist_to_database(self, db_model: WorkflowNodeExecutionModel):
        """
        Persist the database model to the database.
@@ -386,6 +393,13 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
            existing = session.get(WorkflowNodeExecutionModel, db_model.id)

            if existing:
+                merged_process_data = preserve_workflow_agent_binding_id(
+                    existing.process_data_dict,
+                    db_model.process_data_dict,
+                )
+                db_model.process_data = (
+                    _deterministic_json_dump(merged_process_data) if merged_process_data is not None else None
+                )
                # Update existing record by copying all non-private attributes
                for key, value in db_model.__dict__.items():
                    if not key.startswith("_"):
@@ -442,18 +456,25 @@ class SQLAlchemyWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository)
            else:
                db_model.outputs = self._json_encode(domain_model.outputs)

-        if domain_model.process_data is not None:
+        process_data = preserve_workflow_agent_binding_id(db_model.process_data_dict, domain_model.process_data)
+        if process_data is not None:
            result = self._truncate_and_upload(
-                domain_model.process_data,
+                process_data,
                domain_model.id,
                ExecutionOffLoadType.PROCESS_DATA,
            )
            if result is not None:
-                db_model.process_data = self._json_encode(result.truncated_value)
-                domain_model.set_truncated_process_data(result.truncated_value)
+                truncated_process_data = preserve_workflow_agent_binding_id(
+                    process_data,
+                    result.truncated_value,
+                )
+                if truncated_process_data is None:
+                    raise ValueError("truncated process data is unavailable")
+                db_model.process_data = self._json_encode(truncated_process_data)
+                domain_model.set_truncated_process_data(truncated_process_data)
                offload_data = _replace_or_append_offload(offload_data, result.offload)
            else:
-                db_model.process_data = self._json_encode(domain_model.process_data)
+                db_model.process_data = self._json_encode(process_data)

        db_model.offload_data = offload_data
        with self._session_factory() as session, session.begin():
@@ -107,6 +107,8 @@ class MCPTool(Tool):
        if self.entity.output_schema and result.structuredContent:
            for k, v in result.structuredContent.items():
                yield self.create_variable_message(k, v)
+        elif result.structuredContent:
+            yield self.create_json_message(result.structuredContent)

    def _process_text_content(self, content: TextContent) -> Generator[ToolInvokeMessage, None, None]:
        """Process text content and yield appropriate messages."""
@@ -0,0 +1,27 @@
+from collections.abc import Mapping
+from typing import Any
+
+WORKFLOW_AGENT_BINDING_ID_KEY = "workflow_agent_binding_id"
+
+
+def preserve_workflow_agent_binding_id(
+    identity_source: Mapping[str, Any] | None,
+    process_data: Mapping[str, Any] | None,
+) -> dict[str, Any] | None:
+    source_id = (identity_source or {}).get(WORKFLOW_AGENT_BINDING_ID_KEY)
+    target_id = (process_data or {}).get(WORKFLOW_AGENT_BINDING_ID_KEY)
+    for value in (source_id, target_id):
+        if value is not None and not isinstance(value, str):
+            raise ValueError("workflow_agent_binding_id must be a string")
+    if source_id is not None and target_id is not None and source_id != target_id:
+        raise ValueError("workflow_agent_binding_id does not match")
+
+    if process_data is None and source_id is None:
+        return None
+    merged = dict(process_data or {})
+    if source_id is not None:
+        merged[WORKFLOW_AGENT_BINDING_ID_KEY] = source_id
+    return merged
+
+
+__all__ = ["WORKFLOW_AGENT_BINDING_ID_KEY", "preserve_workflow_agent_binding_id"]
@@ -487,7 +487,7 @@ class DifyNodeFactory(NodeFactory):
            from core.workflow.nodes.agent_v2.output_failure_orchestrator import OutputFailureOrchestrator
            from core.workflow.nodes.agent_v2.output_file_rebacker import reback_tool_file_output
            from core.workflow.nodes.agent_v2.output_type_checker import PerOutputTypeChecker
-            from core.workflow.nodes.agent_v2.session_store import WorkflowAgentRuntimeSessionStore
+            from core.workflow.nodes.agent_v2.session_store import WorkflowAgentWorkspaceStore

            return {
                "binding_resolver": WorkflowAgentBindingResolver(),
@@ -497,6 +497,7 @@ class DifyNodeFactory(NodeFactory):
                ),
                "agent_backend_client": create_agent_backend_run_client(
                    base_url=dify_config.AGENT_BACKEND_BASE_URL,
+                    api_token=dify_config.AGENT_BACKEND_API_TOKEN,
                    use_fake=dify_config.AGENT_BACKEND_USE_FAKE,
                    fake_scenario=dify_config.AGENT_BACKEND_FAKE_SCENARIO,
                    stream_read_timeout_seconds=dify_config.AGENT_BACKEND_STREAM_READ_TIMEOUT_SECONDS,
@@ -511,7 +512,7 @@ class DifyNodeFactory(NodeFactory):
                # tenant validator resolves ToolFile (canonical) + UploadFile refs.
                "type_checker": PerOutputTypeChecker(file_validator=AgentOutputFileTenantValidator()),
                "failure_orchestrator": OutputFailureOrchestrator(),
-                "session_store": WorkflowAgentRuntimeSessionStore(),
+                "session_store": WorkflowAgentWorkspaceStore(),
            }
        return {
            "strategy_resolver": self._agent_strategy_resolver,
@@ -198,8 +198,23 @@ class AgentRuntimeSupport:
                    if model_schema:
                        model_schema = self._remove_unsupported_model_features_for_old_version(model_schema)
                        value["entity"] = model_schema.model_dump(mode="json")
+                        # The model selector value from the workflow frontend only
+                        # carries provider/model/mode — it does NOT include
+                        # completion_params.  AgentStrategy plugins (cot_agent,
+                        # function_calling) read completion_params to build the
+                        # LLMModelConfig that is backwards-invoked, and some model
+                        # providers raise KeyError('required') when
+                        # completion_params is empty because their parameter_rules
+                        # declare required fields with no default.  Populate
+                        # completion_params with the defaults declared in the model
+                        # schema so the plugin daemon always receives a valid set
+                        # of model parameters.
+                        if "completion_params" not in value:
+                            value["completion_params"] = self._extract_default_completion_params(model_schema)
                    else:
                        value["entity"] = None
+                        if "completion_params" not in value:
+                            value["completion_params"] = {}
            result[parameter_name] = value

        return result
@@ -275,6 +290,24 @@ class AgentRuntimeSupport:
                    model_schema.features.remove(feature)
        return model_schema

+    @staticmethod
+    def _extract_default_completion_params(model_schema: AIModelEntity) -> dict[str, Any]:
+        """Build a completion_params dict from the model schema's parameter_rules.
+
+        The workflow Agent node's model-selector parameter only stores
+        provider/model/mode — it never carries completion_params.  When the
+        value is forwarded to the plugin daemon, AgentModelConfig defaults
+        completion_params to ``{}``, which causes some model providers to fail
+        because their parameter_rules declare required fields.  This helper
+        collects the ``default`` value of every parameter_rule that has one so
+        the plugin daemon receives a valid, non-empty set of model parameters.
+        """
+        completion_params: dict[str, Any] = {}
+        for rule in model_schema.parameter_rules:
+            if rule.default is not None:
+                completion_params[rule.name] = rule.default
+        return completion_params
+
    @staticmethod
    def _filter_mcp_type_tool(
        strategy: ResolvedAgentStrategy,
@@ -18,13 +18,10 @@ from clients.agent_backend import (
    AgentBackendRunEventAdapter,
    AgentBackendRunFailedInternalEvent,
    AgentBackendRunSucceededInternalEvent,
-    AgentBackendSessionCleanupPayload,
    AgentBackendStreamError,
    AgentBackendStreamInternalEvent,
    AgentBackendTransportError,
    AgentBackendValidationError,
-    RuntimeLayerSpec,
-    extract_runtime_layer_specs,
 )
 from core.app.entities.app_invoke_entities import DIFY_RUN_CONTEXT_KEY, DifyRunContext
 from core.repositories.human_input_repository import HumanInputFormRepository, HumanInputFormRepositoryImpl
@@ -33,11 +30,12 @@ from core.workflow.nodes.human_input.session_binding import default_session_bind
 from core.workflow.system_variables import SystemVariableKey, get_system_text
 from graphon.entities.pause_reason import HitlRequired, SchedulingPause
 from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
-from graphon.node_events import NodeEventBase, NodeRunResult, PauseRequestedEvent, StreamCompletedEvent
+from graphon.graph_events import NodeRunPauseRequestedEvent
+from graphon.node_events import NodeEventBase, NodeRunResult, StreamCompletedEvent
 from graphon.nodes.base.node import Node
 from models.agent_config_entities import AgentSoulConfig, WorkflowNodeJobConfig
 from services.agent.prompt_mentions import extract_workflow_node_output_selectors
-from tasks.agent_backend_session_cleanup_task import cleanup_workflow_agent_runtime_session
+from services.agent.workspace_service import AgentWorkspaceNotFoundError

 from .ask_human_hitl import AskHumanFormBuildError, build_ask_human_pause_reason
 from .ask_human_resume import build_deferred_tool_results, resolve_ask_human_form
@@ -56,7 +54,7 @@ from .runtime_request_builder import (
    WorkflowAgentRuntimeRequestBuilder,
    WorkflowAgentRuntimeRequestBuildError,
 )
-from .session_store import WorkflowAgentRuntimeSessionStore, WorkflowAgentSessionScope
+from .session_store import WorkflowAgentSessionScope, WorkflowAgentWorkspaceStore

 if TYPE_CHECKING:
    from graphon.entities import GraphInitParams
@@ -68,7 +66,7 @@ logger = logging.getLogger(__name__)
 # Stage 4 §5+§7: the terminal events that `_consume_event_stream` may return.
 # Stream + started events are filtered out before we yield; transport errors
 # are surfaced as a separate StreamCompletedEvent in the second tuple slot.
-_TerminalAgentBackendEvent = (
+type _TerminalAgentBackendEvent = (
    AgentBackendRunSucceededInternalEvent
    | AgentBackendRunFailedInternalEvent
    | AgentBackendRunCancelledInternalEvent
@@ -93,7 +91,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
        output_adapter: WorkflowAgentOutputAdapter,
        type_checker: PerOutputTypeChecker,
        failure_orchestrator: OutputFailureOrchestrator,
-        session_store: WorkflowAgentRuntimeSessionStore | None = None,
+        session_store: WorkflowAgentWorkspaceStore,
    ) -> None:
        super().__init__(
            node_id=node_id,
@@ -130,7 +128,34 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
        return reason

    @override
-    def _run(self) -> Generator[NodeEventBase, None, None]:
+    def _run(self) -> Generator[NodeEventBase | NodeRunPauseRequestedEvent, None, None]:
+        inputs: dict[str, Any] = {}
+        process_data: dict[str, Any] = {}
+        metadata: dict[str, Any] = {
+            "agent_backend": {
+                "status": "not_started",
+            }
+        }
+        try:
+            yield from self._run_inner(inputs=inputs, process_data=process_data, metadata=metadata)
+        except Exception as error:
+            if not process_data:
+                raise
+            yield self._failure_event(
+                inputs=inputs,
+                process_data=process_data,
+                metadata=metadata,
+                error=str(error),
+                error_type="agent_workflow_node_runtime_error",
+            )
+
+    def _run_inner(
+        self,
+        *,
+        inputs: dict[str, Any],
+        process_data: dict[str, Any],
+        metadata: dict[str, Any],
+    ) -> Generator[NodeEventBase | NodeRunPauseRequestedEvent, None, None]:
        dify_ctx = DifyRunContext.model_validate(self.require_run_context_value(DIFY_RUN_CONTEXT_KEY))
        workflow_id = self.graph_init_params.workflow_id
        workflow_run_id = get_system_text(
@@ -143,21 +168,24 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
            self.graph_runtime_state.variable_pool,
            SystemVariableKey.CONVERSATION_ID,
        )
-        inputs: dict[str, Any] = {}
-        process_data: dict[str, Any] = {}
-        metadata: dict[str, Any] = {
-            "agent_backend": {
-                "status": "not_started",
-            }
-        }

        # ──── Setup: resolve binding once + extract declared outputs for stage 4 checks ────
        try:
+            existing_scope = self._session_store.load_existing_node_execution_scope(
+                tenant_id=dify_ctx.tenant_id,
+                app_id=dify_ctx.app_id,
+                workflow_id=workflow_id,
+                workflow_run_id=workflow_run_id,
+                node_id=self._node_id,
+                node_execution_id=self.execution_id,
+            )
            bundle = self._binding_resolver.resolve(
                tenant_id=dify_ctx.tenant_id,
                app_id=dify_ctx.app_id,
                workflow_id=workflow_id,
                node_id=self._node_id,
+                binding_id=existing_scope.workflow_agent_binding_id if existing_scope is not None else None,
+                snapshot_id=existing_scope.agent_config_snapshot_id if existing_scope is not None else None,
            )
        except WorkflowAgentBindingError as error:
            yield self._failure_event(
@@ -168,20 +196,31 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                error_type=error.error_code,
            )
            return
+        except AgentWorkspaceNotFoundError as error:
+            yield self._failure_event(
+                inputs=inputs,
+                process_data=process_data,
+                metadata=metadata,
+                error=str(error),
+                error_type="agent_workflow_node_runtime_error",
+            )
+            return

-        process_data = {
-            "agent_id": bundle.agent.id,
-            "agent_config_snapshot_id": bundle.snapshot.id,
-            "binding_id": bundle.binding.id,
-        }
-        session_scope = WorkflowAgentSessionScope(
+        process_data.update(
+            {
+                "agent_id": bundle.agent.id,
+                "agent_config_snapshot_id": bundle.snapshot.id,
+                "workflow_agent_binding_id": bundle.binding.id,
+            }
+        )
+        session_scope = existing_scope or WorkflowAgentSessionScope(
            tenant_id=dify_ctx.tenant_id,
            app_id=dify_ctx.app_id,
            workflow_id=workflow_id,
            workflow_run_id=workflow_run_id,
            node_id=self._node_id,
-            node_execution_id=self.id,
-            binding_id=bundle.binding.id,
+            node_execution_id=self.execution_id,
+            workflow_agent_binding_id=bundle.binding.id,
            agent_id=bundle.agent.id,
            agent_config_snapshot_id=bundle.snapshot.id,
        )
@@ -200,47 +239,53 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
        # the second Agent run as deferred_tool_results; if it is somehow still
        # waiting, re-emit the same pause defensively.
        deferred_tool_results = None
-        if self._session_store is not None:
-            stored_session = self._session_store.load_active_session(session_scope)
-            if stored_session is not None and stored_session.pending_form_id is not None:
-                resume_outcome = resolve_ask_human_form(
-                    form_id=stored_session.pending_form_id,
-                    tenant_id=dify_ctx.tenant_id,
-                    node_id=self._node_id,
+        stored_session = self._session_store.load_or_create_node_execution_session(
+            session_scope,
+            home_snapshot_id=bundle.snapshot.home_snapshot_id,
+        )
+        if stored_session.pending_form_id is not None:
+            resume_outcome = resolve_ask_human_form(
+                form_id=stored_session.pending_form_id,
+                tenant_id=dify_ctx.tenant_id,
+                node_id=self._node_id,
+            )
+            if resume_outcome is not None and resume_outcome.repause is not None:
+                yield self._pause_event(
+                    reason=resume_outcome.repause,
+                    inputs=inputs,
+                    process_data=process_data,
+                    metadata=metadata,
+                )
+                return
+            if (
+                resume_outcome is not None
+                and resume_outcome.deferred_result is not None
+                and stored_session.pending_tool_call_id is not None
+            ):
+                deferred_tool_results = build_deferred_tool_results(
+                    tool_call_id=stored_session.pending_tool_call_id,
+                    result=resume_outcome.deferred_result,
                )
-                if resume_outcome is not None and resume_outcome.repause is not None:
-                    yield PauseRequestedEvent(reason=self._to_graph_pause_reason(resume_outcome.repause))
-                    return
-                if (
-                    resume_outcome is not None
-                    and resume_outcome.deferred_result is not None
-                    and stored_session.pending_tool_call_id is not None
-                ):
-                    deferred_tool_results = build_deferred_tool_results(
-                        tool_call_id=stored_session.pending_tool_call_id,
-                        result=resume_outcome.deferred_result,
-                    )

        # ──── Retry loop (Stage 4 §7) ────
        attempt = 0
        while True:
            try:
-                session_snapshot = None
-                if self._session_store is not None:
-                    session_snapshot = self._session_store.load_active_snapshot(session_scope)
                runtime_request = self._runtime_request_builder.build(
                    WorkflowAgentRuntimeBuildContext(
                        dify_context=dify_ctx,
                        workflow_id=workflow_id,
                        workflow_run_id=workflow_run_id,
                        node_id=self._node_id,
-                        node_execution_id=self.id,
+                        node_execution_id=self.execution_id,
                        variable_pool=self.graph_runtime_state.variable_pool,
                        binding=bundle.binding,
                        agent=bundle.agent,
                        snapshot=bundle.snapshot,
+                        binding_id=stored_session.binding_id,
+                        backend_binding_ref=stored_session.backend_binding_ref,
                        attempt=attempt,
-                        session_snapshot=session_snapshot,
+                        session_snapshot=stored_session.session_snapshot,
                        deferred_tool_results=deferred_tool_results,
                    )
                )
@@ -266,8 +311,9 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
            # Capture inputs only from the first attempt so retry doesn't churn the
            # node's "inputs" payload that ends up in the workflow detail view.
            if attempt == 0:
-                inputs = {"agent_backend_request": runtime_request.redacted_request}
-            metadata = dict(runtime_request.metadata)
+                inputs["agent_backend_request"] = runtime_request.redacted_request
+            metadata.clear()
+            metadata.update(runtime_request.metadata)
            metadata["attempt"] = attempt

            try:
@@ -288,7 +334,12 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                "status": create_response.status,
            }

-            terminal_event, exhausted = self._consume_event_stream(create_response.run_id, metadata)
+            terminal_event, exhausted = self._consume_event_stream(
+                create_response.run_id,
+                inputs=inputs,
+                process_data=process_data,
+                metadata=metadata,
+            )
            if exhausted is not None:
                # Streaming error / unexpected end — surface immediately without
                # retrying because the failure is transport-level.
@@ -349,29 +400,23 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                    )
                self._save_session_snapshot(
                    session_scope=session_scope,
-                    backend_run_id=terminal_event.run_id,
+                    binding_id=stored_session.binding_id,
                    snapshot=terminal_event.session_snapshot,
-                    runtime_layer_specs=extract_runtime_layer_specs(runtime_request.request.composition),
                    metadata=metadata,
                    pending_form_id=pending_form_id,
                    pending_tool_call_id=pending_tool_call_id,
                )
-                yield PauseRequestedEvent(reason=self._to_graph_pause_reason(pause_reason))
-                return
-
-            # Non-success terminal (failed / cancelled) skips per-output
-            # post-processing — the backend itself already failed. We also retire
-            # the local ACTIVE session row so a workflow loop back into the same
-            # Agent node cannot resume from a stale snapshot. The failed agent
-            # backend layers (suspended per ``on_exit``) are left for agent
-            # backend's own GC; this row will no longer be picked up by the
-            # workflow-terminal cleanup layer.
-            if not isinstance(terminal_event, AgentBackendRunSucceededInternalEvent):
-                self._mark_session_cleaned_on_failure(
-                    session_scope=session_scope,
-                    backend_run_id=terminal_event.run_id,
+                yield self._pause_event(
+                    reason=pause_reason,
+                    inputs=inputs,
+                    process_data=process_data,
                    metadata=metadata,
                )
+                return
+
+            # A failed attempt does not retire the product-owned Binding. The
+            # Workflow Run terminal lifecycle event owns that transition.
+            if not isinstance(terminal_event, AgentBackendRunSucceededInternalEvent):
                yield StreamCompletedEvent(
                    node_run_result=self._output_adapter.build_failure_result(
                        event=terminal_event,
@@ -384,9 +429,8 @@ class DifyAgentNode(Node[DifyAgentNodeData]):

            self._save_session_snapshot(
                session_scope=session_scope,
-                backend_run_id=terminal_event.run_id,
+                binding_id=stored_session.binding_id,
                snapshot=terminal_event.session_snapshot,
-                runtime_layer_specs=extract_runtime_layer_specs(runtime_request.request.composition),
                metadata=metadata,
            )

@@ -458,6 +502,9 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
    def _consume_event_stream(
        self,
        run_id: str,
+        *,
+        inputs: dict[str, Any],
+        process_data: dict[str, Any],
        metadata: dict[str, Any],
    ) -> tuple[
        _TerminalAgentBackendEvent | None,
@@ -507,8 +554,8 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                        return internal_event, None
                    self._cancel_backend_run(run_id, reason="unexpected_event")
                    return None, self._failure_event(
-                        inputs={},
-                        process_data={},
+                        inputs=inputs,
+                        process_data=process_data,
                        metadata=metadata,
                        error=f"Unexpected internal event type {internal_event.type!r}",
                        error_type="agent_backend_stream_error",
@@ -516,8 +563,8 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
        except AgentBackendError as error:
            self._cancel_backend_run(run_id, reason=self._stream_stop_reason())
            return None, self._failure_event(
-                inputs={},
-                process_data={},
+                inputs=inputs,
+                process_data=process_data,
                metadata=metadata,
                error=str(error),
                error_type=self._agent_backend_error_type(error),
@@ -525,8 +572,8 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
        except Exception as error:
            self._cancel_backend_run(run_id, reason=self._stream_stop_reason())
            return None, self._failure_event(
-                inputs={},
-                process_data={},
+                inputs=inputs,
+                process_data=process_data,
                metadata=metadata,
                error=str(error),
                error_type="agent_backend_stream_error",
@@ -596,21 +643,17 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
        self,
        *,
        session_scope: WorkflowAgentSessionScope,
-        backend_run_id: str,
+        binding_id: str,
        snapshot: CompositorSessionSnapshot | None,
-        runtime_layer_specs: list[RuntimeLayerSpec],
        metadata: dict[str, Any],
        pending_form_id: str | None = None,
        pending_tool_call_id: str | None = None,
    ) -> None:
-        if self._session_store is None:
-            return
        try:
            self._session_store.save_active_snapshot(
                scope=session_scope,
-                backend_run_id=backend_run_id,
+                binding_id=binding_id,
                snapshot=snapshot,
-                runtime_layer_specs=runtime_layer_specs,
                pending_form_id=pending_form_id,
                pending_tool_call_id=pending_tool_call_id,
            )
@@ -619,88 +662,18 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
            metadata["agent_backend"] = agent_backend
        except Exception:
            logger.warning(
-                "Failed to persist workflow Agent runtime session snapshot: "
-                "tenant_id=%s workflow_run_id=%s node_id=%s binding_id=%s agent_id=%s backend_run_id=%s",
+                "Failed to persist workflow Agent Binding session snapshot: "
+                "tenant_id=%s workflow_run_id=%s node_id=%s binding_id=%s agent_id=%s",
                session_scope.tenant_id,
                session_scope.workflow_run_id,
                session_scope.node_id,
-                session_scope.binding_id,
+                session_scope.workflow_agent_binding_id,
                session_scope.agent_id,
-                backend_run_id,
                exc_info=True,
            )
            agent_backend = dict(metadata.get("agent_backend") or {})
            agent_backend["session_snapshot_persisted"] = False
-            agent_backend["session_snapshot_persist_error"] = "workflow_agent_runtime_session_store_error"
-            metadata["agent_backend"] = agent_backend
-
-    def _mark_session_cleaned_on_failure(
-        self,
-        *,
-        session_scope: WorkflowAgentSessionScope,
-        backend_run_id: str,
-        metadata: dict[str, Any],
-    ) -> None:
-        if self._session_store is None:
-            return
-        stored_session = self._session_store.load_active_session(session_scope)
-        try:
-            if stored_session is not None and stored_session.runtime_layer_specs:
-                payload = AgentBackendSessionCleanupPayload(
-                    session_snapshot=stored_session.session_snapshot,
-                    runtime_layer_specs=stored_session.runtime_layer_specs,
-                    idempotency_key=(
-                        f"{session_scope.tenant_id}:{session_scope.workflow_run_id}:{session_scope.node_id}:"
-                        f"{session_scope.binding_id}:workflow-agent-failure-cleanup:"
-                        f"{stored_session.backend_run_id or 'no-stored-run'}:{backend_run_id}"
-                    ),
-                    metadata={
-                        "tenant_id": session_scope.tenant_id,
-                        "app_id": session_scope.app_id,
-                        "workflow_id": session_scope.workflow_id,
-                        "workflow_run_id": session_scope.workflow_run_id,
-                        "node_id": session_scope.node_id,
-                        "node_execution_id": session_scope.node_execution_id,
-                        "binding_id": session_scope.binding_id,
-                        "agent_id": session_scope.agent_id,
-                        "agent_config_snapshot_id": session_scope.agent_config_snapshot_id,
-                        "previous_agent_backend_run_id": stored_session.backend_run_id,
-                        "failed_agent_backend_run_id": backend_run_id,
-                    },
-                )
-                cleanup_workflow_agent_runtime_session.delay(payload.model_dump(mode="json"))
-        except Exception:
-            logger.warning(
-                "Failed to enqueue workflow Agent backend cleanup on agent run failure: "
-                "tenant_id=%s workflow_run_id=%s node_id=%s binding_id=%s agent_id=%s backend_run_id=%s",
-                session_scope.tenant_id,
-                session_scope.workflow_run_id,
-                session_scope.node_id,
-                session_scope.binding_id,
-                session_scope.agent_id,
-                backend_run_id,
-                exc_info=True,
-            )
-        try:
-            self._session_store.mark_cleaned(scope=session_scope, backend_run_id=backend_run_id)
-            agent_backend = dict(metadata.get("agent_backend") or {})
-            agent_backend["session_snapshot_cleaned_on_failure"] = True
-            metadata["agent_backend"] = agent_backend
-        except Exception:
-            logger.warning(
-                "Failed to mark workflow Agent runtime session cleaned on agent run failure: "
-                "tenant_id=%s workflow_run_id=%s node_id=%s binding_id=%s agent_id=%s backend_run_id=%s",
-                session_scope.tenant_id,
-                session_scope.workflow_run_id,
-                session_scope.node_id,
-                session_scope.binding_id,
-                session_scope.agent_id,
-                backend_run_id,
-                exc_info=True,
-            )
-            agent_backend = dict(metadata.get("agent_backend") or {})
-            agent_backend["session_snapshot_cleaned_on_failure"] = False
-            agent_backend["session_snapshot_cleanup_error"] = "workflow_agent_runtime_session_store_error"
+            agent_backend["session_snapshot_persist_error"] = "workflow_agent_workspace_store_error"
            metadata["agent_backend"] = agent_backend

    @staticmethod
@@ -742,6 +715,27 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
            )
        )

+    def _pause_event(
+        self,
+        *,
+        reason: HumanInputRequired | SchedulingPause,
+        inputs: dict[str, Any],
+        process_data: dict[str, Any],
+        metadata: dict[str, Any],
+    ) -> NodeRunPauseRequestedEvent:
+        return NodeRunPauseRequestedEvent(
+            id=self.execution_id,
+            node_id=self._node_id,
+            node_type=self.node_type,
+            node_run_result=NodeRunResult(
+                status=WorkflowNodeExecutionStatus.PAUSED,
+                inputs=inputs,
+                process_data=process_data,
+                metadata={WorkflowNodeExecutionMetadataKey.AGENT_LOG: metadata},
+            ),
+            reason=self._to_graph_pause_reason(reason),
+        )
+
    @staticmethod
    def _agent_backend_error_type(error: AgentBackendError) -> str:
        if isinstance(error, AgentBackendValidationError):
@@ -41,18 +41,27 @@ class WorkflowAgentBindingResolver:
        app_id: str,
        workflow_id: str,
        node_id: str,
+        binding_id: str | None = None,
+        snapshot_id: str | None = None,
    ) -> WorkflowAgentBindingBundle:
-        with session_factory.create_session() as session:
-            binding = session.scalar(
-                select(WorkflowAgentNodeBinding)
-                .where(
-                    WorkflowAgentNodeBinding.tenant_id == tenant_id,
-                    WorkflowAgentNodeBinding.app_id == app_id,
-                    WorkflowAgentNodeBinding.workflow_id == workflow_id,
-                    WorkflowAgentNodeBinding.node_id == node_id,
-                )
-                .limit(1)
+        """Resolve the current binding, optionally at a generation pinned by an existing execution."""
+
+        if (binding_id is None) != (snapshot_id is None):
+            raise WorkflowAgentBindingError(
+                "agent_binding_generation_invalid",
+                "Workflow Agent binding and config snapshot must be pinned together.",
            )
+
+        with session_factory.create_session() as session:
+            binding_stmt = select(WorkflowAgentNodeBinding).where(
+                WorkflowAgentNodeBinding.tenant_id == tenant_id,
+                WorkflowAgentNodeBinding.app_id == app_id,
+                WorkflowAgentNodeBinding.workflow_id == workflow_id,
+                WorkflowAgentNodeBinding.node_id == node_id,
+            )
+            if binding_id is not None:
+                binding_stmt = binding_stmt.where(WorkflowAgentNodeBinding.id == binding_id)
+            binding = session.scalar(binding_stmt.limit(1))
            if binding is None:
                raise WorkflowAgentBindingError(
                    "agent_binding_not_found",
@@ -77,12 +86,16 @@ class WorkflowAgentBindingResolver:
                    f"Agent {binding.agent_id} is not available or has not been published.",
                )

-            snapshot_id = (
-                agent.active_config_snapshot_id
-                if binding.binding_type == WorkflowAgentBindingType.ROSTER_AGENT
-                else binding.current_snapshot_id
+            effective_snapshot_id = (
+                (
+                    agent.active_config_snapshot_id
+                    if binding.binding_type == WorkflowAgentBindingType.ROSTER_AGENT
+                    else binding.current_snapshot_id
+                )
+                if snapshot_id is None
+                else snapshot_id
            )
-            if snapshot_id is None:
+            if effective_snapshot_id is None:
                raise WorkflowAgentBindingError(
                    "agent_config_snapshot_not_found",
                    "Workflow Agent binding has no current config snapshot.",
@@ -93,14 +106,14 @@ class WorkflowAgentBindingResolver:
                .where(
                    AgentConfigSnapshot.tenant_id == tenant_id,
                    AgentConfigSnapshot.agent_id == agent.id,
-                    AgentConfigSnapshot.id == snapshot_id,
+                    AgentConfigSnapshot.id == effective_snapshot_id,
                )
                .limit(1)
            )
            if snapshot is None:
                raise WorkflowAgentBindingError(
                    "agent_config_snapshot_not_found",
-                    f"Agent config snapshot {snapshot_id} not found.",
+                    f"Agent config snapshot {effective_snapshot_id} not found.",
                )

            session.expunge(binding)
@@ -33,7 +33,6 @@ from dify_agent.layers.shell import (
    DifyShellCliToolConfig,
    DifyShellEnvVarConfig,
    DifyShellLayerConfig,
-    DifyShellSandboxConfig,
    DifyShellSecretRefConfig,
 )
 from dify_agent.protocol import CreateRunRequest, DeferredToolResultsPayload
@@ -136,6 +135,8 @@ class WorkflowAgentRuntimeBuildContext:
    binding: WorkflowAgentNodeBinding
    agent: Agent
    snapshot: AgentConfigSnapshot
+    binding_id: str
+    backend_binding_ref: str
    # Stage 4 §7 / D-4: 0 for the first run, then incremented per retry. Drives the
    # idempotency key so the backend treats each retry as a fresh request.
    attempt: int = 0
@@ -251,6 +252,7 @@ class WorkflowAgentRuntimeRequestBuilder:
                    agent_mode=self._agent_backend_agent_mode(context.dify_context.invoke_from),
                    invoke_from=cast(DifyExecutionContextInvokeFrom, context.dify_context.invoke_from.value),
                ),
+                backend_binding_ref=context.backend_binding_ref,
                agent_soul_prompt=soul_prompt or None,
                workflow_node_job_prompt=workflow_job_prompt,
                user_prompt=user_prompt,
@@ -739,7 +741,6 @@ class WorkflowAgentRuntimeRequestBuilder:

 def build_shell_layer_config(agent_soul: AgentSoulConfig) -> DifyShellLayerConfig:
    """Map Agent Soul shell-adjacent fields into the Agent backend shell config."""
-    sandbox_config = _plain_mapping(agent_soul.sandbox.config)
    return DifyShellLayerConfig(
        cli_tools=[
            tool
@@ -750,12 +751,6 @@ def build_shell_layer_config(agent_soul: AgentSoulConfig) -> DifyShellLayerConfi
        secret_refs=[
            secret for secret in (_shell_secret_ref(item) for item in agent_soul.env.secret_refs) if secret is not None
        ],
-        sandbox=DifyShellSandboxConfig(
-            provider=agent_soul.sandbox.provider,
-            config=sandbox_config,
-        )
-        if agent_soul.sandbox.provider or sandbox_config
-        else None,
    )


@@ -1,126 +0,0 @@
-"""Workflow terminal layer that retires Agent backend sessions asynchronously."""
-
-from __future__ import annotations
-
-import logging
-from typing import override
-
-from clients.agent_backend import AgentBackendSessionCleanupPayload
-from core.workflow.system_variables import SystemVariableKey, get_system_text
-from graphon.graph_engine.layers import GraphEngineLayer
-from graphon.graph_events import (
-    GraphEngineEvent,
-    GraphRunAbortedEvent,
-    GraphRunFailedEvent,
-    GraphRunPartialSucceededEvent,
-    GraphRunSucceededEvent,
-)
-from tasks.agent_backend_session_cleanup_task import cleanup_workflow_agent_runtime_session
-
-from .session_store import StoredWorkflowAgentSession, WorkflowAgentRuntimeSessionStore
-
-logger = logging.getLogger(__name__)
-
-
-class WorkflowAgentSessionCleanupLayer(GraphEngineLayer):
-    """Retire workflow-owned Agent runtime sessions when the workflow ends.
-
-    Workflow termination is a product-lifecycle boundary: once the run reaches a
-    terminal graph event, the local session row must no longer be resumable. The
-    actual Agent backend cleanup is therefore dispatched asynchronously with the
-    persisted snapshot/specs payload, while the local row is marked CLEANED
-    immediately afterwards regardless of enqueue outcome.
-    """
-
-    _TERMINAL_EVENTS = (
-        GraphRunSucceededEvent,
-        GraphRunPartialSucceededEvent,
-        GraphRunFailedEvent,
-        GraphRunAbortedEvent,
-    )
-
-    def __init__(self, *, session_store: WorkflowAgentRuntimeSessionStore) -> None:
-        super().__init__()
-        self._session_store = session_store
-
-    @override
-    def on_graph_start(self) -> None:
-        return
-
-    @override
-    def on_event(self, event: GraphEngineEvent) -> None:
-        if not isinstance(event, self._TERMINAL_EVENTS):
-            return
-        workflow_run_id = get_system_text(
-            self.graph_runtime_state.variable_pool,
-            SystemVariableKey.WORKFLOW_EXECUTION_ID,
-        )
-        if not workflow_run_id:
-            logger.warning("Skipping workflow Agent session cleanup: workflow_run_id is missing.")
-            return
-
-        for stored_session in self._session_store.list_active_sessions(workflow_run_id=workflow_run_id):
-            self._cleanup_session(stored_session)
-
-    @override
-    def on_graph_end(self, error: Exception | None) -> None:
-        return
-
-    def _cleanup_session(self, stored_session: StoredWorkflowAgentSession) -> None:
-        scope = stored_session.scope
-        try:
-            if stored_session.runtime_layer_specs:
-                payload = AgentBackendSessionCleanupPayload(
-                    session_snapshot=stored_session.session_snapshot,
-                    runtime_layer_specs=stored_session.runtime_layer_specs,
-                    idempotency_key=f"{scope.workflow_run_id}:{scope.node_id}:{scope.binding_id}:agent-session-cleanup",
-                    metadata={
-                        "tenant_id": scope.tenant_id,
-                        "app_id": scope.app_id,
-                        "workflow_id": scope.workflow_id,
-                        "workflow_run_id": scope.workflow_run_id,
-                        "node_id": scope.node_id,
-                        "node_execution_id": scope.node_execution_id,
-                        "binding_id": scope.binding_id,
-                        "agent_id": scope.agent_id,
-                        "agent_config_snapshot_id": scope.agent_config_snapshot_id,
-                        "previous_agent_backend_run_id": stored_session.backend_run_id,
-                    },
-                )
-                cleanup_workflow_agent_runtime_session.delay(payload.model_dump(mode="json"))
-            else:
-                logger.warning(
-                    "Skipping workflow Agent backend cleanup enqueue: no runtime_layer_specs persisted. "
-                    "workflow_run_id=%s node_id=%s agent_id=%s",
-                    scope.workflow_run_id,
-                    scope.node_id,
-                    scope.agent_id,
-                )
-        except Exception:
-            logger.warning(
-                "Failed to enqueue workflow Agent backend cleanup: "
-                "workflow_run_id=%s node_id=%s agent_id=%s previous_run_id=%s",
-                scope.workflow_run_id,
-                scope.node_id,
-                scope.agent_id,
-                stored_session.backend_run_id,
-                exc_info=True,
-            )
-        finally:
-            try:
-                self._session_store.mark_cleaned(scope=scope, backend_run_id=stored_session.backend_run_id)
-            except Exception:
-                logger.warning(
-                    "Failed to retire workflow Agent runtime session after cleanup enqueue: "
-                    "workflow_run_id=%s node_id=%s agent_id=%s previous_run_id=%s",
-                    scope.workflow_run_id,
-                    scope.node_id,
-                    scope.agent_id,
-                    stored_session.backend_run_id,
-                    exc_info=True,
-                )
-
-
-def build_workflow_agent_session_cleanup_layer() -> WorkflowAgentSessionCleanupLayer:
-    """Wire the cleanup layer with the standard workflow-owned session store."""
-    return WorkflowAgentSessionCleanupLayer(session_store=WorkflowAgentRuntimeSessionStore())
@@ -1,31 +1,32 @@
+"""Workflow Agent participant persistence keyed by node execution."""
+
 from __future__ import annotations

-from dataclasses import dataclass, field
+import json
+import time
+from dataclasses import dataclass

 from agenton.compositor import CompositorSessionSnapshot
-from dify_agent.protocol import RuntimeLayerSpec
-from pydantic import TypeAdapter
 from sqlalchemy import select
+from sqlalchemy.orm import Session

 from core.db.session_factory import session_factory
-from libs.datetime_utils import naive_utc_now
 from models.agent import (
-    AgentRuntimeSessionOwnerType,
-    WorkflowAgentRuntimeSession,
-    WorkflowAgentRuntimeSessionStatus,
+    AgentConfigVersionKind,
+    AgentWorkingResourceStatus,
+    AgentWorkspace,
+    AgentWorkspaceBinding,
+    AgentWorkspaceOwnerType,
+)
+from models.workflow import WorkflowNodeExecutionModel
+from services.agent.workspace_service import (
+    AgentWorkspaceNotFoundError,
+    AgentWorkspaceService,
+    WorkspaceOwnerScope,
 )

-_SPECS_ADAPTER: TypeAdapter[list[RuntimeLayerSpec]] = TypeAdapter(list[RuntimeLayerSpec])
-
-
-def _serialize_specs(specs: list[RuntimeLayerSpec]) -> str:
-    return _SPECS_ADAPTER.dump_json(specs).decode()
-
-
-def _deserialize_specs(value: str | None) -> list[RuntimeLayerSpec]:
-    if not value:
-        return []
-    return _SPECS_ADAPTER.validate_json(value)
+_CALLER_VISIBILITY_ATTEMPTS = 60
+_CALLER_VISIBILITY_INTERVAL_SECONDS = 0.05


@dataclass(frozen=True, slots=True)
@@ -36,171 +37,251 @@ class WorkflowAgentSessionScope:
    workflow_run_id: str | None
    node_id: str
    node_execution_id: str
-    binding_id: str
+    workflow_agent_binding_id: str
    agent_id: str
    agent_config_snapshot_id: str

+    @property
+    def workspace_owner(self) -> WorkspaceOwnerScope:
+        return WorkspaceOwnerScope(
+            tenant_id=self.tenant_id,
+            app_id=self.app_id,
+            owner_type=AgentWorkspaceOwnerType.WORKFLOW_RUN,
+            owner_id=self.workflow_run_id or self.node_execution_id,
+            owner_scope_key=f"{self.node_id}:{self.workflow_agent_binding_id}",
+        )
+

@dataclass(frozen=True, slots=True)
 class StoredWorkflowAgentSession:
    scope: WorkflowAgentSessionScope
-    session_snapshot: CompositorSessionSnapshot
-    backend_run_id: str | None
-    runtime_layer_specs: list[RuntimeLayerSpec] = field(default_factory=list)
-    # ENG-637: set while the session is paused on a dify.ask_human deferred call.
+    binding_id: str
+    workspace_id: str
+    backend_binding_ref: str
+    session_snapshot: CompositorSessionSnapshot | None
    pending_form_id: str | None = None
    pending_tool_call_id: str | None = None


-class WorkflowAgentRuntimeSessionStore:
-    """Stores Agent backend session snapshots for workflow Agent node re-entry."""
+class WorkflowAgentWorkspaceStore:
+    """Load or create the participant named by a node execution caller row."""

-    def load_active_snapshot(self, scope: WorkflowAgentSessionScope) -> CompositorSessionSnapshot | None:
-        stored = self.load_active_session(scope)
-        return stored.session_snapshot if stored is not None else None
-
-    def load_active_session(self, scope: WorkflowAgentSessionScope) -> StoredWorkflowAgentSession | None:
-        """Load the active session row including any pending ask_human correlation."""
-        if scope.workflow_run_id is None:
-            return None
+    def load_existing_node_execution_scope(
+        self,
+        *,
+        tenant_id: str,
+        app_id: str,
+        workflow_id: str,
+        workflow_run_id: str | None,
+        node_id: str,
+        node_execution_id: str,
+    ) -> WorkflowAgentSessionScope | None:
+        """Return the generation pinned by an existing node execution participant."""

        with session_factory.create_session() as session:
-            row = session.scalar(
-                select(WorkflowAgentRuntimeSession).where(
-                    WorkflowAgentRuntimeSession.tenant_id == scope.tenant_id,
-                    WorkflowAgentRuntimeSession.workflow_run_id == scope.workflow_run_id,
-                    WorkflowAgentRuntimeSession.node_id == scope.node_id,
-                    WorkflowAgentRuntimeSession.binding_id == scope.binding_id,
-                    WorkflowAgentRuntimeSession.agent_id == scope.agent_id,
-                    WorkflowAgentRuntimeSession.status == WorkflowAgentRuntimeSessionStatus.ACTIVE,
-                )
+            execution = self._load_execution_by_identity(
+                session=session,
+                tenant_id=tenant_id,
+                app_id=app_id,
+                workflow_id=workflow_id,
+                workflow_run_id=workflow_run_id,
+                node_id=node_id,
+                node_execution_id=node_execution_id,
            )
-            if row is None:
+            binding_id = execution.agent_workspace_binding_id
+            if binding_id is None:
                return None
-            return StoredWorkflowAgentSession(
-                scope=scope,
-                session_snapshot=CompositorSessionSnapshot.model_validate_json(row.session_snapshot),
-                backend_run_id=row.backend_run_id,
-                runtime_layer_specs=_deserialize_specs(row.composition_layer_specs),
-                pending_form_id=row.pending_form_id,
-                pending_tool_call_id=row.pending_tool_call_id,
+            process_data = execution.process_data_dict
+            if not isinstance(process_data, dict):
+                raise AgentWorkspaceNotFoundError("Workflow node execution caller identity is invalid")
+            workflow_agent_binding_id = process_data.get("workflow_agent_binding_id")
+            if not isinstance(workflow_agent_binding_id, str):
+                raise AgentWorkspaceNotFoundError("Workflow node execution caller identity is missing")
+            owner_scope = WorkspaceOwnerScope(
+                tenant_id=tenant_id,
+                app_id=app_id,
+                owner_type=AgentWorkspaceOwnerType.WORKFLOW_RUN,
+                owner_id=workflow_run_id or node_execution_id,
+                owner_scope_key=f"{node_id}:{workflow_agent_binding_id}",
+            )
+            binding = AgentWorkspaceService.get_active_binding(
+                session=session,
+                tenant_id=tenant_id,
+                binding_id=binding_id,
+                expected_owner_scope=owner_scope,
+            )
+            if binding is None or binding.agent_config_version_kind != AgentConfigVersionKind.SNAPSHOT:
+                raise AgentWorkspaceNotFoundError("Workflow node participant Binding is unavailable")
+            return WorkflowAgentSessionScope(
+                tenant_id=tenant_id,
+                app_id=app_id,
+                workflow_id=workflow_id,
+                workflow_run_id=workflow_run_id,
+                node_id=node_id,
+                node_execution_id=node_execution_id,
+                workflow_agent_binding_id=workflow_agent_binding_id,
+                agent_id=binding.agent_id,
+                agent_config_snapshot_id=binding.agent_config_version_id,
            )

-    def list_active_sessions(self, *, workflow_run_id: str) -> list[StoredWorkflowAgentSession]:
+    def load_or_create_node_execution_session(
+        self, scope: WorkflowAgentSessionScope, *, home_snapshot_id: str
+    ) -> StoredWorkflowAgentSession:
        with session_factory.create_session() as session:
-            rows = session.scalars(
-                select(WorkflowAgentRuntimeSession).where(
-                    WorkflowAgentRuntimeSession.workflow_run_id == workflow_run_id,
-                    WorkflowAgentRuntimeSession.status == WorkflowAgentRuntimeSessionStatus.ACTIVE,
+            execution = self._load_execution(session=session, scope=scope)
+            process_data = execution.process_data_dict
+            if process_data is None:
+                process_data = {}
+            if not isinstance(process_data, dict):
+                raise AgentWorkspaceNotFoundError("Workflow node execution caller identity is invalid")
+            stored_workflow_binding_id = process_data.get("workflow_agent_binding_id")
+            if stored_workflow_binding_id is not None and stored_workflow_binding_id != scope.workflow_agent_binding_id:
+                raise AgentWorkspaceNotFoundError("Workflow node execution caller identity does not match")
+
+            binding_id = execution.agent_workspace_binding_id
+            if binding_id is None:
+                binding = AgentWorkspaceService.create_binding(
+                    session=session,
+                    scope=scope.workspace_owner,
+                    agent_id=scope.agent_id,
+                    base_home_snapshot_id=home_snapshot_id,
+                    agent_config_version_id=scope.agent_config_snapshot_id,
+                    agent_config_version_kind=AgentConfigVersionKind.SNAPSHOT,
                )
-            ).all()
-            return [
-                StoredWorkflowAgentSession(
-                    scope=WorkflowAgentSessionScope(
-                        tenant_id=row.tenant_id,
-                        app_id=row.app_id,
-                        # These columns are nullable on the unified runtime-session
-                        # table (workflow_run ⊕ conversation owner), but are always
-                        # populated for a workflow-owned row; coerce for the typed scope.
-                        workflow_id=row.workflow_id or "",
-                        workflow_run_id=row.workflow_run_id,
-                        node_id=row.node_id or "",
-                        node_execution_id=row.node_execution_id or "",
-                        binding_id=row.binding_id or "",
-                        agent_id=row.agent_id,
-                        agent_config_snapshot_id=row.agent_config_snapshot_id or "",
-                    ),
-                    session_snapshot=CompositorSessionSnapshot.model_validate_json(row.session_snapshot),
-                    backend_run_id=row.backend_run_id,
-                    runtime_layer_specs=_deserialize_specs(row.composition_layer_specs),
+                execution.agent_workspace_binding_id = binding.id
+                execution.process_data = json.dumps(
+                    {
+                        **process_data,
+                        "workflow_agent_binding_id": scope.workflow_agent_binding_id,
+                    },
+                    ensure_ascii=False,
                )
-                for row in rows
-            ]
+                session.commit()
+            else:
+                if stored_workflow_binding_id is None:
+                    raise AgentWorkspaceNotFoundError("Workflow node execution caller identity is missing")
+                resolved_binding = AgentWorkspaceService.get_active_binding(
+                    session=session,
+                    tenant_id=scope.tenant_id,
+                    binding_id=binding_id,
+                    expected_owner_scope=scope.workspace_owner,
+                )
+                if resolved_binding is None or resolved_binding.agent_id != scope.agent_id:
+                    raise AgentWorkspaceNotFoundError("Workflow node participant Binding is unavailable")
+                binding = resolved_binding
+                AgentWorkspaceService.validate_binding_generation(
+                    binding,
+                    base_home_snapshot_id=home_snapshot_id,
+                    agent_config_version_id=scope.agent_config_snapshot_id,
+                    agent_config_version_kind=AgentConfigVersionKind.SNAPSHOT,
+                )
+            return self._stored(scope, binding)

    def save_active_snapshot(
        self,
        *,
        scope: WorkflowAgentSessionScope,
-        backend_run_id: str,
+        binding_id: str,
        snapshot: CompositorSessionSnapshot | None,
-        runtime_layer_specs: list[RuntimeLayerSpec],
        pending_form_id: str | None = None,
        pending_tool_call_id: str | None = None,
    ) -> None:
-        if scope.workflow_run_id is None or snapshot is None:
+        if snapshot is None:
            return
+        AgentWorkspaceService.save_binding_session_snapshot(
+            tenant_id=scope.tenant_id,
+            binding_id=binding_id,
+            session_snapshot=snapshot.model_dump_json(),
+            pending_form_id=pending_form_id,
+            pending_tool_call_id=pending_tool_call_id,
+        )

-        snapshot_json = snapshot.model_dump_json()
-        specs_json = _serialize_specs(runtime_layer_specs)
+    def retire_workflow_run(self, *, tenant_id: str, app_id: str, workflow_run_id: str) -> list[str]:
+        """Retire active Workspaces, commit, and return active or already-retired IDs for collection."""
+
+        retired: list[str] = []
        with session_factory.create_session() as session:
-            row = session.scalar(
-                select(WorkflowAgentRuntimeSession).where(
-                    WorkflowAgentRuntimeSession.tenant_id == scope.tenant_id,
-                    WorkflowAgentRuntimeSession.workflow_run_id == scope.workflow_run_id,
-                    WorkflowAgentRuntimeSession.node_id == scope.node_id,
-                    WorkflowAgentRuntimeSession.binding_id == scope.binding_id,
-                    WorkflowAgentRuntimeSession.agent_id == scope.agent_id,
+            workspaces = session.scalars(
+                select(AgentWorkspace).where(
+                    AgentWorkspace.tenant_id == tenant_id,
+                    AgentWorkspace.app_id == app_id,
+                    AgentWorkspace.owner_type == AgentWorkspaceOwnerType.WORKFLOW_RUN,
+                    AgentWorkspace.owner_id == workflow_run_id,
+                    AgentWorkspace.status.in_((AgentWorkingResourceStatus.ACTIVE, AgentWorkingResourceStatus.RETIRED)),
                )
-            )
-            if row is None:
-                row = WorkflowAgentRuntimeSession(
-                    tenant_id=scope.tenant_id,
-                    app_id=scope.app_id,
-                    owner_type=AgentRuntimeSessionOwnerType.WORKFLOW_RUN,
-                    workflow_id=scope.workflow_id,
-                    workflow_run_id=scope.workflow_run_id,
-                    node_id=scope.node_id,
-                    node_execution_id=scope.node_execution_id,
-                    binding_id=scope.binding_id,
-                    agent_id=scope.agent_id,
-                    agent_config_snapshot_id=scope.agent_config_snapshot_id,
-                    backend_run_id=backend_run_id,
-                    session_snapshot=snapshot_json,
-                    composition_layer_specs=specs_json,
-                    status=WorkflowAgentRuntimeSessionStatus.ACTIVE,
-                    pending_form_id=pending_form_id,
-                    pending_tool_call_id=pending_tool_call_id,
+            ).all()
+            for workspace in workspaces:
+                if workspace.status == AgentWorkingResourceStatus.RETIRED:
+                    retired.append(workspace.id)
+                    continue
+                workspace_id = AgentWorkspaceService.retire_workspace(
+                    session=session,
+                    tenant_id=tenant_id,
+                    workspace_id=workspace.id,
                )
-                session.add(row)
-            else:
-                row.node_execution_id = scope.node_execution_id
-                row.agent_config_snapshot_id = scope.agent_config_snapshot_id
-                row.backend_run_id = backend_run_id
-                row.session_snapshot = snapshot_json
-                row.composition_layer_specs = specs_json
-                row.status = WorkflowAgentRuntimeSessionStatus.ACTIVE
-                row.cleaned_at = None
-                # Set (or clear, when omitted) the ask_human pause correlation.
-                row.pending_form_id = pending_form_id
-                row.pending_tool_call_id = pending_tool_call_id
+                if workspace_id is not None:
+                    retired.append(workspace_id)
            session.commit()
+        return retired

-    def mark_cleaned(self, *, scope: WorkflowAgentSessionScope, backend_run_id: str | None = None) -> None:
-        if scope.workflow_run_id is None:
-            return
+    @staticmethod
+    def _load_execution(*, session: Session, scope: WorkflowAgentSessionScope) -> WorkflowNodeExecutionModel:
+        return WorkflowAgentWorkspaceStore._load_execution_by_identity(
+            session=session,
+            tenant_id=scope.tenant_id,
+            app_id=scope.app_id,
+            workflow_id=scope.workflow_id,
+            workflow_run_id=scope.workflow_run_id,
+            node_id=scope.node_id,
+            node_execution_id=scope.node_execution_id,
+        )

-        with session_factory.create_session() as session:
-            row = session.scalar(
-                select(WorkflowAgentRuntimeSession).where(
-                    WorkflowAgentRuntimeSession.tenant_id == scope.tenant_id,
-                    WorkflowAgentRuntimeSession.workflow_run_id == scope.workflow_run_id,
-                    WorkflowAgentRuntimeSession.node_id == scope.node_id,
-                    WorkflowAgentRuntimeSession.binding_id == scope.binding_id,
-                    WorkflowAgentRuntimeSession.agent_id == scope.agent_id,
-                    WorkflowAgentRuntimeSession.status == WorkflowAgentRuntimeSessionStatus.ACTIVE,
-                )
-            )
-            if row is None:
-                return
-            if backend_run_id is not None:
-                row.backend_run_id = backend_run_id
-            row.status = WorkflowAgentRuntimeSessionStatus.CLEANED
-            row.cleaned_at = naive_utc_now()
-            session.commit()
+    @staticmethod
+    def _load_execution_by_identity(
+        *,
+        session: Session,
+        tenant_id: str,
+        app_id: str,
+        workflow_id: str,
+        workflow_run_id: str | None,
+        node_id: str,
+        node_execution_id: str,
+    ) -> WorkflowNodeExecutionModel:
+        """Wait briefly for the already-emitted node-start event to persist its caller row."""
+
+        stmt = select(WorkflowNodeExecutionModel).where(
+            WorkflowNodeExecutionModel.id == node_execution_id,
+            WorkflowNodeExecutionModel.tenant_id == tenant_id,
+            WorkflowNodeExecutionModel.app_id == app_id,
+            WorkflowNodeExecutionModel.workflow_id == workflow_id,
+            WorkflowNodeExecutionModel.node_id == node_id,
+            WorkflowNodeExecutionModel.workflow_run_id == workflow_run_id,
+        )
+        for attempt in range(_CALLER_VISIBILITY_ATTEMPTS):
+            execution = session.scalar(stmt)
+            if execution is not None:
+                return execution
+            if attempt < _CALLER_VISIBILITY_ATTEMPTS - 1:
+                time.sleep(_CALLER_VISIBILITY_INTERVAL_SECONDS)
+
+        raise AgentWorkspaceNotFoundError("Workflow node execution caller is unavailable")
+
+    @staticmethod
+    def _stored(scope: WorkflowAgentSessionScope, binding: AgentWorkspaceBinding) -> StoredWorkflowAgentSession:
+        snapshot = (
+            CompositorSessionSnapshot.model_validate_json(binding.session_snapshot)
+            if binding.session_snapshot
+            else None
+        )
+        return StoredWorkflowAgentSession(
+            scope=scope,
+            binding_id=binding.id,
+            workspace_id=binding.workspace_id,
+            backend_binding_ref=binding.backend_binding_ref,
+            session_snapshot=snapshot,
+            pending_form_id=binding.pending_form_id,
+            pending_tool_call_id=binding.pending_tool_call_id,
+        )


-__all__ = [
-    "StoredWorkflowAgentSession",
-    "WorkflowAgentRuntimeSessionStore",
-    "WorkflowAgentSessionScope",
-]
+__all__ = ["StoredWorkflowAgentSession", "WorkflowAgentSessionScope", "WorkflowAgentWorkspaceStore"]
@@ -0,0 +1,89 @@
+"""Retire Workflow Agent Workspaces when the Workflow Run terminates."""
+
+from __future__ import annotations
+
+import logging
+from typing import override
+
+from core.app.entities.app_invoke_entities import DifyRunContext
+from core.workflow.nodes.agent_v2.session_store import WorkflowAgentWorkspaceStore
+from core.workflow.system_variables import SystemVariableKey, get_system_text
+from graphon.graph_engine.layers import GraphEngineLayer
+from graphon.graph_events import (
+    GraphEngineEvent,
+    GraphRunAbortedEvent,
+    GraphRunFailedEvent,
+    GraphRunPartialSucceededEvent,
+    GraphRunSucceededEvent,
+)
+from tasks.collect_agent_resources_task import enqueue_agent_resource_collection
+
+logger = logging.getLogger(__name__)
+
+
+class WorkflowAgentWorkspaceRetirementLayer(GraphEngineLayer):
+    """Synchronously retire run Workspaces, then enqueue physical collection."""
+
+    _TERMINAL_EVENTS = (
+        GraphRunSucceededEvent,
+        GraphRunPartialSucceededEvent,
+        GraphRunFailedEvent,
+        GraphRunAbortedEvent,
+    )
+
+    def __init__(
+        self,
+        *,
+        dify_run_context: DifyRunContext,
+    ) -> None:
+        super().__init__()
+        self._dify_run_context = dify_run_context
+
+    @override
+    def on_graph_start(self) -> None:
+        return
+
+    @override
+    def on_event(self, event: GraphEngineEvent) -> None:
+        if not isinstance(event, self._TERMINAL_EVENTS):
+            return
+        workflow_run_id = get_system_text(
+            self.graph_runtime_state.variable_pool,
+            SystemVariableKey.WORKFLOW_EXECUTION_ID,
+        )
+        if not workflow_run_id:
+            logger.warning("Skipping Workflow Agent Workspace retirement: workflow_run_id is missing")
+            return
+        try:
+            workspace_ids = WorkflowAgentWorkspaceStore().retire_workflow_run(
+                tenant_id=self._dify_run_context.tenant_id,
+                app_id=self._dify_run_context.app_id,
+                workflow_run_id=workflow_run_id,
+            )
+        except Exception:
+            logger.exception(
+                "Failed to retire Workflow Agent Workspaces",
+                extra={
+                    "tenant_id": self._dify_run_context.tenant_id,
+                    "app_id": self._dify_run_context.app_id,
+                    "workflow_run_id": workflow_run_id,
+                },
+            )
+            return
+        enqueue_agent_resource_collection(
+            tenant_id=self._dify_run_context.tenant_id,
+            workspace_ids=workspace_ids,
+        )
+
+    @override
+    def on_graph_end(self, error: Exception | None) -> None:
+        return
+
+
+def build_workflow_agent_workspace_retirement_layer(
+    *, dify_run_context: DifyRunContext
+) -> WorkflowAgentWorkspaceRetirementLayer:
+    return WorkflowAgentWorkspaceRetirementLayer(dify_run_context=dify_run_context)
+
+
+__all__ = ["WorkflowAgentWorkspaceRetirementLayer", "build_workflow_agent_workspace_retirement_layer"]
@@ -5,6 +5,7 @@ from typing import Any
 import pytz  # type: ignore[import-untyped]
 from celery import Celery, Task
 from celery.schedules import crontab
+from celery.signals import beat_init
 from typing_extensions import TypedDict

 from configs import dify_config
@@ -36,6 +37,19 @@ class CeleryBeatScheduleEntry(TypedDict):
    schedule: crontab | timedelta


+def _enqueue_initial_community_telemetry_heartbeat(sender: Any, **_: Any) -> None:
+    task_name = "community_telemetry.send_heartbeat"
+    if "community_telemetry_heartbeat" not in sender.app.conf.beat_schedule:
+        return
+
+    task = sender.app.tasks.get(task_name)
+    if task is not None:
+        task.apply_async()
+
+
+beat_init.connect(_enqueue_initial_community_telemetry_heartbeat, weak=False)
+
+
 def get_celery_ssl_options() -> CelerySSLOptionsDict | None:
    """Get SSL configuration for Celery broker/backend connections."""
    # Only apply SSL if we're using Redis as broker/backend
@@ -152,6 +166,7 @@ def init_app(app: DifyApp) -> Celery:

    imports = [
        "tasks.async_workflow_tasks",  # trigger workers
+        "tasks.collect_agent_resources_task",  # retired Agent resource collection
        "tasks.trigger_processing_tasks",  # async trigger processing
        "tasks.generate_summary_index_task",  # summary index generation
        "tasks.regenerate_summary_index_task",  # summary index regeneration
@@ -260,6 +275,19 @@ def init_app(app: DifyApp) -> Celery:
            "schedule": timedelta(minutes=dify_config.API_TOKEN_LAST_USED_UPDATE_INTERVAL),
        }

+    if (
+        dify_config.EDITION == "SELF_HOSTED"
+        and not dify_config.ENTERPRISE_ENABLED
+        and not dify_config.DISABLE_TELEMETRY
+        and not dify_config.DO_NOT_TRACK
+        and not dify_config.CI
+    ):
+        imports.append("tasks.community_telemetry_task")
+        beat_schedule["community_telemetry_heartbeat"] = {
+            "task": "community_telemetry.send_heartbeat",
+            "schedule": timedelta(minutes=dify_config.TELEMETRY_HEARTBEAT_INTERVAL_MINUTES),
+        }
+
    if dify_config.ENTERPRISE_ENABLED and dify_config.ENTERPRISE_TELEMETRY_ENABLED:
        imports.append("tasks.enterprise_telemetry_task")
    celery_app.conf.update(beat_schedule=beat_schedule, imports=imports)
@@ -277,6 +277,12 @@ class LogstoreWorkflowNodeExecutionRepository(WorkflowNodeExecutionRepository):
                logger.exception("Failed to dual-write node execution to SQL database: id=%s", execution.id)
                # Don't raise - LogStore write succeeded, SQL is just a backup

+    @override
+    def save_synchronously(self, execution: WorkflowNodeExecution) -> None:
+        """Create the SQL caller row required by Agent v2 participant ownership."""
+
+        self.sql_repository.save_synchronously(execution)
+
    @override
    def save_execution_data(self, execution: WorkflowNodeExecution) -> None:
        """
@@ -383,6 +383,7 @@ class AgentAppComposerResponse(ResponseModel):
    variant: Literal[ComposerVariant.AGENT_APP]
    agent: AgentComposerAgentResponse
    active_config_snapshot: AgentConfigSnapshotSummaryResponse | None = None
+    active_config_is_published: bool
    draft: AgentConfigDraftSummaryResponse | None = None
    agent_soul: AgentSoulConfig
    save_options: list[ComposerSaveStrategy]
@@ -34,7 +34,7 @@ class _SessionResponseSource[SourceT]:
        self._session = session

    def __getattr__(self, name: str) -> object:
-        return getattr(self._source, name)  # noqa: no-new-getattr response adapter delegates model fields
+        return getattr(self._source, name)  # guard-ignore: no-new-getattr -- delegates model fields


 class _FeedbackResponseSource(_SessionResponseSource[MessageFeedback]):
@@ -227,7 +227,7 @@ class DatasetDetailResponseSource:
        return self.dataset.get_total_available_documents(session=self.session)

    def __getattr__(self, name: str) -> Any:
-        return getattr(self.dataset, name)  # noqa: no-new-getattr response adapter delegates model fields
+        return getattr(self.dataset, name)  # guard-ignore: no-new-getattr -- delegates model fields


 def dataset_detail_response_source(dataset: Any, *, session: Session) -> DatasetDetailResponseSource:
@@ -90,7 +90,7 @@ class DocumentWithSession:
        return self.document.get_doc_metadata_details(session=self.session)

    def __getattr__(self, name: str) -> Any:
-        return getattr(self.document, name)  # noqa: no-new-getattr response adapter delegates model fields
+        return getattr(self.document, name)  # guard-ignore: no-new-getattr -- delegates model fields


 def document_response(document: Document, *, session: Session) -> DocumentResponse:
@@ -289,7 +289,11 @@ UUIDStr = Annotated[str, AfterValidator(_strict_uuid)]

 def alphanumeric(value: str):
    # check if the value is alphanumeric and underlined
-    if re.match(r"^[a-zA-Z0-9_]+$", value):
+    # Use re.fullmatch instead of re.match to reject trailing newlines.
+    # In Python, '$' matches at end-of-string OR just before a trailing newline,
+    # so re.match accepts "tool_name\n". re.fullmatch requires the entire
+    # string to match. Regression for #39666 (sibling of #39234 / #39548).
+    if re.fullmatch(r"^[a-zA-Z0-9_]+$", value):
        return value

    raise ValueError(f"{value} is not a valid alphanumeric value")
@@ -0,0 +1,64 @@
+"""add agent home snapshot ledger
+
+Revision ID: 2f39536b3feb
+Revises: 6f5a9c2d8e1b
+Create Date: 2026-07-21 22:51:07.268658
+
+"""
+from alembic import op
+import models as models
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '2f39536b3feb'
+down_revision = '6f5a9c2d8e1b'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('agent_home_snapshots',
+    sa.Column('id', models.types.StringUUID(), nullable=False),
+    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+    sa.Column('agent_id', models.types.StringUUID(), nullable=False),
+    sa.Column('snapshot_ref', sa.String(length=255), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='agent_home_snapshot_pkey')
+    )
+    with op.batch_alter_table('agent_home_snapshots', schema=None) as batch_op:
+        batch_op.create_index('agent_home_snapshot_tenant_agent_idx', ['tenant_id', 'agent_id'], unique=False)
+
+    with op.batch_alter_table('agent_config_drafts', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('home_snapshot_id', models.types.StringUUID(), nullable=False))
+
+    with op.batch_alter_table('agent_config_snapshots', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('home_snapshot_id', models.types.StringUUID(), nullable=False))
+
+    with op.batch_alter_table('agent_runtime_sessions', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('home_snapshot_id', models.types.StringUUID(), nullable=False))
+        batch_op.drop_index(batch_op.f('agent_runtime_session_conversation_scope_unique'), postgresql_where='(conversation_id IS NOT NULL)')
+        batch_op.create_index('agent_runtime_session_conversation_scope_unique', ['tenant_id', 'conversation_id', 'agent_id', 'agent_config_snapshot_id', 'home_snapshot_id'], unique=True, postgresql_where=sa.text('conversation_id IS NOT NULL'))
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('agent_runtime_sessions', schema=None) as batch_op:
+        batch_op.drop_index('agent_runtime_session_conversation_scope_unique', postgresql_where=sa.text('conversation_id IS NOT NULL'))
+        batch_op.create_index(batch_op.f('agent_runtime_session_conversation_scope_unique'), ['tenant_id', 'conversation_id', 'agent_id', 'agent_config_snapshot_id'], unique=True, postgresql_where='(conversation_id IS NOT NULL)')
+        batch_op.drop_column('home_snapshot_id')
+
+    with op.batch_alter_table('agent_config_snapshots', schema=None) as batch_op:
+        batch_op.drop_column('home_snapshot_id')
+
+    with op.batch_alter_table('agent_config_drafts', schema=None) as batch_op:
+        batch_op.drop_column('home_snapshot_id')
+
+    with op.batch_alter_table('agent_home_snapshots', schema=None) as batch_op:
+        batch_op.drop_index('agent_home_snapshot_tenant_agent_idx')
+
+    op.drop_table('agent_home_snapshots')
+    # ### end Alembic commands ###
@@ -0,0 +1,153 @@
+"""replace agent runtime sessions with workspaces and bindings
+
+Revision ID: f6e4c5686857
+Revises: 2f39536b3feb
+Create Date: 2026-07-23 02:03:05.641638
+
+"""
+from alembic import op
+import models as models
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = 'f6e4c5686857'
+down_revision = '2f39536b3feb'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('agent_workspace_bindings',
+    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+    sa.Column('app_id', models.types.StringUUID(), nullable=False),
+    sa.Column('workspace_id', models.types.StringUUID(), nullable=False),
+    sa.Column('agent_id', models.types.StringUUID(), nullable=False),
+    sa.Column('base_home_snapshot_id', models.types.StringUUID(), nullable=False),
+    sa.Column('agent_config_version_id', models.types.StringUUID(), nullable=False),
+    sa.Column('agent_config_version_kind', sa.String(length=32), nullable=False),
+    sa.Column('backend_binding_ref', sa.String(length=255), nullable=False),
+    sa.Column('session_snapshot', models.types.LongText(), nullable=True),
+    sa.Column('status', sa.String(length=32), server_default='active', nullable=False),
+    sa.Column('retired_at', sa.DateTime(), nullable=True),
+    sa.Column('pending_form_id', models.types.StringUUID(), nullable=True),
+    sa.Column('pending_tool_call_id', sa.String(length=255), nullable=True),
+    sa.Column('id', models.types.StringUUID(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='agent_workspace_binding_pkey')
+    )
+    with op.batch_alter_table('agent_workspace_bindings', schema=None) as batch_op:
+        batch_op.create_index('agent_workspace_binding_agent_status_idx', ['tenant_id', 'agent_id', 'status'], unique=False)
+        batch_op.create_index('agent_workspace_binding_status_retired_idx', ['status', 'retired_at'], unique=False)
+        batch_op.create_index('agent_workspace_binding_workspace_status_idx', ['tenant_id', 'workspace_id', 'status'], unique=False)
+
+    op.create_table('agent_workspaces',
+    sa.Column('tenant_id', models.types.StringUUID(), nullable=False),
+    sa.Column('app_id', models.types.StringUUID(), nullable=False),
+    sa.Column('owner_type', sa.String(length=32), nullable=False),
+    sa.Column('owner_id', models.types.StringUUID(), nullable=False),
+    sa.Column('owner_scope_key', sa.String(length=255), nullable=False),
+    sa.Column('backend_workspace_ref', sa.String(length=255), nullable=False),
+    sa.Column('status', sa.String(length=32), server_default='active', nullable=False),
+    sa.Column('active_guard', sa.SmallInteger(), server_default='1', nullable=True),
+    sa.Column('retired_at', sa.DateTime(), nullable=True),
+    sa.Column('id', models.types.StringUUID(), nullable=False),
+    sa.Column('created_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=False),
+    sa.PrimaryKeyConstraint('id', name='agent_workspace_pkey')
+    )
+    with op.batch_alter_table('agent_workspaces', schema=None) as batch_op:
+        batch_op.create_index('agent_workspace_owner_active_unique', ['tenant_id', 'owner_type', 'owner_id', 'owner_scope_key', 'active_guard'], unique=True)
+        batch_op.create_index('agent_workspace_status_retired_idx', ['status', 'retired_at'], unique=False)
+        batch_op.create_index('agent_workspace_tenant_app_status_idx', ['tenant_id', 'app_id', 'status'], unique=False)
+        batch_op.create_index('agent_workspace_tenant_status_idx', ['tenant_id', 'status'], unique=False)
+
+    with op.batch_alter_table('agent_runtime_sessions', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('agent_runtime_session_backend_run_idx'))
+        batch_op.drop_index(batch_op.f('agent_runtime_session_conversation_lookup_idx'))
+        batch_op.drop_index(batch_op.f('agent_runtime_session_conversation_scope_unique'), postgresql_where='(conversation_id IS NOT NULL)')
+        batch_op.drop_index(batch_op.f('agent_runtime_session_workflow_lookup_idx'))
+        batch_op.drop_index(batch_op.f('agent_runtime_session_workflow_scope_unique'), postgresql_where='(workflow_run_id IS NOT NULL)')
+
+    op.drop_table('agent_runtime_sessions')
+    with op.batch_alter_table('agent_home_snapshots', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('status', sa.String(length=32), server_default='active', nullable=False))
+        batch_op.add_column(sa.Column('retired_at', sa.DateTime(), nullable=True))
+        batch_op.create_index('agent_home_snapshot_status_retired_idx', ['status', 'retired_at'], unique=False)
+
+    with op.batch_alter_table('conversations', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('agent_workspace_binding_id', models.types.StringUUID(), nullable=True))
+
+    with op.batch_alter_table('agent_config_drafts', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('agent_workspace_binding_id', models.types.StringUUID(), nullable=True))
+
+    with op.batch_alter_table('workflow_node_executions', schema=None) as batch_op:
+        batch_op.add_column(sa.Column('agent_workspace_binding_id', models.types.StringUUID(), nullable=True))
+
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table('workflow_node_executions', schema=None) as batch_op:
+        batch_op.drop_column('agent_workspace_binding_id')
+
+    with op.batch_alter_table('agent_config_drafts', schema=None) as batch_op:
+        batch_op.drop_column('agent_workspace_binding_id')
+
+    with op.batch_alter_table('conversations', schema=None) as batch_op:
+        batch_op.drop_column('agent_workspace_binding_id')
+
+    with op.batch_alter_table('agent_home_snapshots', schema=None) as batch_op:
+        batch_op.drop_index('agent_home_snapshot_status_retired_idx')
+        batch_op.drop_column('retired_at')
+        batch_op.drop_column('status')
+
+    op.create_table('agent_runtime_sessions',
+    sa.Column('id', sa.UUID(), server_default=sa.text('uuidv7()'), autoincrement=False, nullable=False),
+    sa.Column('tenant_id', sa.UUID(), autoincrement=False, nullable=False),
+    sa.Column('app_id', sa.UUID(), autoincrement=False, nullable=False),
+    sa.Column('owner_type', sa.VARCHAR(length=32), autoincrement=False, nullable=False),
+    sa.Column('agent_id', sa.UUID(), autoincrement=False, nullable=False),
+    sa.Column('backend_run_id', sa.VARCHAR(length=255), autoincrement=False, nullable=True),
+    sa.Column('session_snapshot', sa.TEXT(), autoincrement=False, nullable=False),
+    sa.Column('workflow_id', sa.UUID(), autoincrement=False, nullable=True),
+    sa.Column('workflow_run_id', sa.UUID(), autoincrement=False, nullable=True),
+    sa.Column('node_id', sa.VARCHAR(length=255), autoincrement=False, nullable=True),
+    sa.Column('node_execution_id', sa.VARCHAR(length=255), autoincrement=False, nullable=True),
+    sa.Column('binding_id', sa.UUID(), autoincrement=False, nullable=True),
+    sa.Column('agent_config_snapshot_id', sa.UUID(), autoincrement=False, nullable=True),
+    sa.Column('composition_layer_specs', sa.TEXT(), autoincrement=False, nullable=False),
+    sa.Column('conversation_id', sa.UUID(), autoincrement=False, nullable=True),
+    sa.Column('status', sa.VARCHAR(length=32), server_default=sa.text("'active'::character varying"), autoincrement=False, nullable=False),
+    sa.Column('cleaned_at', postgresql.TIMESTAMP(), autoincrement=False, nullable=True),
+    sa.Column('created_at', postgresql.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP'), autoincrement=False, nullable=False),
+    sa.Column('updated_at', postgresql.TIMESTAMP(), server_default=sa.text('CURRENT_TIMESTAMP'), autoincrement=False, nullable=False),
+    sa.Column('pending_form_id', sa.UUID(), autoincrement=False, nullable=True),
+    sa.Column('pending_tool_call_id', sa.VARCHAR(length=255), autoincrement=False, nullable=True),
+    sa.Column('home_snapshot_id', sa.UUID(), autoincrement=False, nullable=False),
+    sa.PrimaryKeyConstraint('id', name=op.f('agent_runtime_session_pkey'))
+    )
+    with op.batch_alter_table('agent_runtime_sessions', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('agent_runtime_session_workflow_scope_unique'), ['tenant_id', 'workflow_run_id', 'node_id', 'binding_id', 'agent_id'], unique=True, postgresql_where='(workflow_run_id IS NOT NULL)')
+        batch_op.create_index(batch_op.f('agent_runtime_session_workflow_lookup_idx'), ['tenant_id', 'workflow_run_id', 'node_id', 'status'], unique=False)
+        batch_op.create_index(batch_op.f('agent_runtime_session_conversation_scope_unique'), ['tenant_id', 'conversation_id', 'agent_id', 'agent_config_snapshot_id', 'home_snapshot_id'], unique=True, postgresql_where='(conversation_id IS NOT NULL)')
+        batch_op.create_index(batch_op.f('agent_runtime_session_conversation_lookup_idx'), ['tenant_id', 'conversation_id', 'status'], unique=False)
+        batch_op.create_index(batch_op.f('agent_runtime_session_backend_run_idx'), ['backend_run_id'], unique=False)
+
+    with op.batch_alter_table('agent_workspaces', schema=None) as batch_op:
+        batch_op.drop_index('agent_workspace_tenant_status_idx')
+        batch_op.drop_index('agent_workspace_tenant_app_status_idx')
+        batch_op.drop_index('agent_workspace_status_retired_idx')
+        batch_op.drop_index('agent_workspace_owner_active_unique')
+
+    op.drop_table('agent_workspaces')
+    with op.batch_alter_table('agent_workspace_bindings', schema=None) as batch_op:
+        batch_op.drop_index('agent_workspace_binding_workspace_status_idx')
+        batch_op.drop_index('agent_workspace_binding_status_retired_idx')
+        batch_op.drop_index('agent_workspace_binding_agent_status_idx')
+
+    op.drop_table('agent_workspace_bindings')
+    # ### end Alembic commands ###
@@ -0,0 +1,30 @@
+"""add telemetry fields to dify_setups
+
+Revision ID: 6f5a9c2d8e1b
+Revises: d2825e7b9c10
+Create Date: 2026-07-23 12:00:00.000000
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "6f5a9c2d8e1b"
+down_revision = "d2825e7b9c10"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    with op.batch_alter_table("dify_setups", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("instance_id", sa.String(length=255), nullable=True))
+        batch_op.add_column(sa.Column("install_reported_at", sa.DateTime(), nullable=True))
+        batch_op.add_column(sa.Column("last_heartbeat_at", sa.DateTime(), nullable=True))
+
+
+def downgrade():
+    with op.batch_alter_table("dify_setups", schema=None) as batch_op:
+        batch_op.drop_column("last_heartbeat_at")
+        batch_op.drop_column("install_reported_at")
+        batch_op.drop_column("instance_id")
@@ -15,21 +15,22 @@ from .agent import (
    AgentConfigRevision,
    AgentConfigRevisionOperation,
    AgentConfigSnapshot,
+    AgentConfigVersionKind,
    AgentDebugConversation,
    AgentDriveFile,
    AgentDriveFileKind,
+    AgentHomeSnapshot,
    AgentIconType,
    AgentKind,
-    AgentRuntimeSession,
-    AgentRuntimeSessionOwnerType,
-    AgentRuntimeSessionStatus,
    AgentScope,
    AgentSource,
    AgentStatus,
+    AgentWorkingResourceStatus,
+    AgentWorkspace,
+    AgentWorkspaceBinding,
+    AgentWorkspaceOwnerType,
    WorkflowAgentBindingType,
    WorkflowAgentNodeBinding,
-    WorkflowAgentRuntimeSession,
-    WorkflowAgentRuntimeSessionStatus,
 )
 from .api_based_extension import APIBasedExtension, APIBasedExtensionPoint
 from .comment import (
@@ -164,17 +165,20 @@ __all__ = [
    "AgentConfigRevision",
    "AgentConfigRevisionOperation",
    "AgentConfigSnapshot",
+    "AgentConfigVersionKind",
    "AgentDebugConversation",
    "AgentDriveFile",
    "AgentDriveFileKind",
+    "AgentHomeSnapshot",
    "AgentIconType",
    "AgentKind",
-    "AgentRuntimeSession",
-    "AgentRuntimeSessionOwnerType",
-    "AgentRuntimeSessionStatus",
    "AgentScope",
    "AgentSource",
    "AgentStatus",
+    "AgentWorkingResourceStatus",
+    "AgentWorkspace",
+    "AgentWorkspaceBinding",
+    "AgentWorkspaceOwnerType",
    "ApiRequest",
    "ApiToken",
    "ApiToolProvider",
@@ -271,8 +275,6 @@ __all__ = [
    "Workflow",
    "WorkflowAgentBindingType",
    "WorkflowAgentNodeBinding",
-    "WorkflowAgentRuntimeSession",
-    "WorkflowAgentRuntimeSessionStatus",
    "WorkflowAppLog",
    "WorkflowAppLogCreatedFrom",
    "WorkflowArchiveLog",
@@ -116,35 +116,29 @@ class WorkflowAgentBindingType(StrEnum):
    INLINE_AGENT = "inline_agent"


-class AgentRuntimeSessionStatus(StrEnum):
-    """Lifecycle state of an Agent backend session snapshot.
+class AgentWorkingResourceStatus(StrEnum):
+    """Product lifecycle state for a persistent working-environment resource."""

-    Owner-agnostic: applies both to workflow Agent Node runs (owner =
-    workflow_run) and to Agent App conversations (owner = conversation).
-    """
-
-    # Snapshot can be reused by a later Agent run in the same session.
    ACTIVE = "active"
-    # Snapshot has been retired and must not be submitted to Agent backend again.
-    CLEANED = "cleaned"
+    RETIRED = "retired"


-class AgentRuntimeSessionOwnerType(StrEnum):
-    """Which product surface owns an Agent runtime session row."""
+class AgentWorkspaceOwnerType(StrEnum):
+    """Product scope that owns a Workspace."""

-    # Owned by one workflow Agent Node execution scope.
    WORKFLOW_RUN = "workflow_run"
-    # Owned by one Agent App conversation (multi-turn chat).
    CONVERSATION = "conversation"
+    BUILD_DRAFT = "build_draft"


-# Back-compat alias: the workflow lifecycle code (shipped in PR #36724) imports
-# the old name. Kept so unifying the table does not churn that path.
-WorkflowAgentRuntimeSessionStatus = AgentRuntimeSessionStatus
+class AgentConfigVersionKind(StrEnum):
+    SNAPSHOT = "snapshot"
+    DRAFT = "draft"
+    BUILD_DRAFT = "build_draft"


 class Agent(DefaultFieldsMixin, Base):
-    """Workspace-scoped Agent identity used by Agent Roster and workflow-only agents."""
+    """Agent Soul and source lineage; ``AgentWorkspaceBinding.id`` identifies each materialized participant."""

    __tablename__ = "agents"
    __table_args__ = (
@@ -221,14 +215,42 @@ class Agent(DefaultFieldsMixin, Base):
    archived_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)


-class AgentDebugConversation(DefaultFieldsMixin, Base):
-    """Per-account, per-draft console debug conversation for an Agent App.
+class AgentHomeSnapshot(Base):
+    """Append-only mapping from one Agent-owned Home identity to its backend ref.

-    Agent App preview state must be isolated by editor account. The Agent row is
-    shared by everyone in the workspace, so this table owns the user-specific
-    conversation pointers used by console debug chat. ``draft`` is the Preview
-    conversation and ``debug_build`` is the Build conversation; they must never
-    share persisted messages or runtime sessions.
+    Product tables reference ``id``. ``snapshot_ref`` remains an opaque
+    deployment-specific handle and is only consumed at Dify Agent boundaries.
+    Snapshot bytes and ``snapshot_ref`` are immutable. Lifecycle metadata can
+    transition ACTIVE -> RETIRED; successful physical collection deletes row.
+    """
+
+    __tablename__ = "agent_home_snapshots"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="agent_home_snapshot_pkey"),
+        Index("agent_home_snapshot_tenant_agent_idx", "tenant_id", "agent_id"),
+        Index("agent_home_snapshot_status_retired_idx", "status", "retired_at"),
+    )
+
+    id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuidv7()))
+    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    agent_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    snapshot_ref: Mapped[str] = mapped_column(String(255), nullable=False)
+    status: Mapped[AgentWorkingResourceStatus] = mapped_column(
+        EnumText(AgentWorkingResourceStatus, length=32),
+        nullable=False,
+        default=AgentWorkingResourceStatus.ACTIVE,
+        server_default=AgentWorkingResourceStatus.ACTIVE.value,
+    )
+    retired_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, server_default=func.current_timestamp())
+
+
+class AgentDebugConversation(DefaultFieldsMixin, Base):
+    """Current console Conversation pointer for one account and draft surface.
+
+    This row owns no Binding or runtime. A Preview Conversation holds its
+    CONVERSATION Binding pointer, while a DEBUG_BUILD AgentConfigDraft holds its
+    BUILD_DRAFT Binding pointer.
    """

    __tablename__ = "agent_debug_conversations"
@@ -259,7 +281,11 @@ class AgentDebugConversation(DefaultFieldsMixin, Base):


 class AgentConfigDraft(DefaultFieldsMixin, Base):
-    """Editable Agent Soul draft separated from immutable published snapshots."""
+    """Editable Agent Soul draft separated from immutable published snapshots.
+
+    A DEBUG_BUILD draft owns its materialized participant through
+    ``agent_workspace_binding_id``. Normal drafts leave that pointer unset.
+    """

    __tablename__ = "agent_config_drafts"
    __table_args__ = (
@@ -281,6 +307,8 @@ class AgentConfigDraft(DefaultFieldsMixin, Base):
    account_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    draft_owner_key: Mapped[str] = mapped_column(String(255), nullable=False, default="")
    base_snapshot_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
+    home_snapshot_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    agent_workspace_binding_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    config_snapshot: Mapped[Any] = mapped_column(JSONModelColumn(AgentSoulConfig), nullable=False)
    created_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    updated_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
@@ -316,6 +344,7 @@ class AgentConfigSnapshot(DefaultFieldsMixin, Base):
    agent_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    version: Mapped[int] = mapped_column(sa.Integer, nullable=False)
    config_snapshot: Mapped[Any] = mapped_column(JSONModelColumn(AgentSoulConfig), nullable=False)
+    home_snapshot_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    summary: Mapped[str | None] = mapped_column(LongText, nullable=True)
    version_note: Mapped[str | None] = mapped_column(LongText, nullable=True)
    created_by: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
@@ -432,102 +461,83 @@ class WorkflowAgentNodeBinding(DefaultFieldsMixin, Base):
        return dict(self.node_job_config)


-class AgentRuntimeSession(DefaultFieldsMixin, Base):
-    """Persisted Agent backend session snapshot, owner-agnostic.
+class AgentWorkspace(DefaultFieldsMixin, Base):
+    """Mutable Workspace owned by one product scope, independent of Agents."""

-    One unified table serves both owners (decision Q2):
-    - workflow Agent Node runs: ``owner_type = workflow_run``; the
-      ``workflow_id / workflow_run_id / node_id / binding_id /
-      agent_config_snapshot_id / composition_layer_specs`` columns are set.
-    - Agent App conversations: ``owner_type = conversation``; the
-      ``conversation_id`` column is set and the workflow columns stay NULL.
-      Runtime state is scoped by ``agent_config_snapshot_id``. For published
-      web/API runs this points to an immutable AgentConfigSnapshot; for console
-      debugger/build runs it points to the editable AgentConfigDraft row.
-
-    The snapshot is runtime state returned by Agent backend, kept separate from
-    Agent Soul snapshots and workflow node-job config.
-    """
-
-    __tablename__ = "agent_runtime_sessions"
+    __tablename__ = "agent_workspaces"
    __table_args__ = (
-        sa.PrimaryKeyConstraint("id", name="agent_runtime_session_pkey"),
-        # Workflow owner uniqueness (partial: only rows with a workflow_run_id).
+        sa.PrimaryKeyConstraint("id", name="agent_workspace_pkey"),
        Index(
-            "agent_runtime_session_workflow_scope_unique",
+            "agent_workspace_owner_active_unique",
            "tenant_id",
-            "workflow_run_id",
-            "node_id",
-            "binding_id",
-            "agent_id",
+            "owner_type",
+            "owner_id",
+            "owner_scope_key",
+            "active_guard",
            unique=True,
-            postgresql_where=sa.text("workflow_run_id IS NOT NULL"),
        ),
-        # Conversation owner uniqueness (partial: only rows with a conversation_id).
-        Index(
-            "agent_runtime_session_conversation_scope_unique",
-            "tenant_id",
-            "conversation_id",
-            "agent_id",
-            "agent_config_snapshot_id",
-            unique=True,
-            postgresql_where=sa.text("conversation_id IS NOT NULL"),
-        ),
-        Index(
-            "agent_runtime_session_workflow_lookup_idx",
-            "tenant_id",
-            "workflow_run_id",
-            "node_id",
-            "status",
-        ),
-        Index(
-            "agent_runtime_session_conversation_lookup_idx",
-            "tenant_id",
-            "conversation_id",
-            "status",
-        ),
-        Index("agent_runtime_session_backend_run_idx", "backend_run_id"),
+        Index("agent_workspace_tenant_status_idx", "tenant_id", "status"),
+        Index("agent_workspace_tenant_app_status_idx", "tenant_id", "app_id", "status"),
+        Index("agent_workspace_status_retired_idx", "status", "retired_at"),
    )

    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
-    owner_type: Mapped[AgentRuntimeSessionOwnerType] = mapped_column(
-        EnumText(AgentRuntimeSessionOwnerType, length=32), nullable=False
+    owner_type: Mapped[AgentWorkspaceOwnerType] = mapped_column(
+        EnumText(AgentWorkspaceOwnerType, length=32), nullable=False
    )
-    agent_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
-    backend_run_id: Mapped[str | None] = mapped_column(String(255), nullable=True)
-    session_snapshot: Mapped[str] = mapped_column(LongText, nullable=False)
-    # Workflow-owner columns (NULL for conversation owner).
-    workflow_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
-    workflow_run_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
-    node_id: Mapped[str | None] = mapped_column(String(255), nullable=True)
-    node_execution_id: Mapped[str | None] = mapped_column(String(255), nullable=True)
-    binding_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
-    agent_config_snapshot_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
-    # JSON-encoded list of non-sensitive runtime layer specs ({name, type, deps,
-    # config}). The persisted schema keeps its original name because the sandbox
-    # refactor intentionally avoids a storage migration.
-    composition_layer_specs: Mapped[str] = mapped_column(LongText, nullable=False, server_default="[]")
-    # Conversation-owner column (NULL for workflow owner).
-    conversation_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
-    status: Mapped[AgentRuntimeSessionStatus] = mapped_column(
-        EnumText(AgentRuntimeSessionStatus, length=32),
+    owner_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    owner_scope_key: Mapped[str] = mapped_column(String(255), nullable=False)
+    backend_workspace_ref: Mapped[str] = mapped_column(String(255), nullable=False)
+    status: Mapped[AgentWorkingResourceStatus] = mapped_column(
+        EnumText(AgentWorkingResourceStatus, length=32),
        nullable=False,
-        default=AgentRuntimeSessionStatus.ACTIVE,
+        default=AgentWorkingResourceStatus.ACTIVE,
+        server_default=AgentWorkingResourceStatus.ACTIVE.value,
    )
-    cleaned_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
-    # ENG-637: when a run pauses for a dify.ask_human deferred call, these link
-    # the session to the awaiting HITL form and the deferred tool_call_id, so a
-    # resumed node can map the submitted form back into deferred_tool_results.
-    # Both NULL whenever the session is not paused on human input.
+    active_guard: Mapped[int | None] = mapped_column(sa.SmallInteger, nullable=True, default=1, server_default="1")
+    retired_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+
+
+class AgentWorkspaceBinding(DefaultFieldsMixin, Base):
+    """One materialized Agent participant and session attached to a Workspace.
+
+    All resource IDs are logical associations rather than database foreign
+    keys, so RETIRED rows can outlive their Workspace or base Home Snapshot.
+    ``agent_id`` identifies the source Agent Soul; this row's ``id`` identifies
+    the participant and its private Materialized Home.
+    """
+
+    __tablename__ = "agent_workspace_bindings"
+    __table_args__ = (
+        sa.PrimaryKeyConstraint("id", name="agent_workspace_binding_pkey"),
+        Index("agent_workspace_binding_workspace_status_idx", "tenant_id", "workspace_id", "status"),
+        Index("agent_workspace_binding_agent_status_idx", "tenant_id", "agent_id", "status"),
+        Index("agent_workspace_binding_status_retired_idx", "status", "retired_at"),
+    )
+
+    tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    workspace_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    agent_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    base_home_snapshot_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    agent_config_version_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
+    agent_config_version_kind: Mapped[AgentConfigVersionKind] = mapped_column(
+        EnumText(AgentConfigVersionKind, length=32), nullable=False
+    )
+    backend_binding_ref: Mapped[str] = mapped_column(String(255), nullable=False)
+    session_snapshot: Mapped[str | None] = mapped_column(LongText, nullable=True)
+    status: Mapped[AgentWorkingResourceStatus] = mapped_column(
+        EnumText(AgentWorkingResourceStatus, length=32),
+        nullable=False,
+        default=AgentWorkingResourceStatus.ACTIVE,
+        server_default=AgentWorkingResourceStatus.ACTIVE.value,
+    )
+    retired_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
    pending_form_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    pending_tool_call_id: Mapped[str | None] = mapped_column(String(255), nullable=True)


-# Back-compat alias for the shipped workflow lifecycle code (PR #36724).
-WorkflowAgentRuntimeSession = AgentRuntimeSession
-
-
 class AgentDriveFileKind(StrEnum):
    """Kind of existing file record an agent-drive KV entry points at."""

@@ -362,6 +362,9 @@ class DifySetup(TypeBase):
    __table_args__ = (sa.PrimaryKeyConstraint("version", name="dify_setup_pkey"),)

    version: Mapped[str] = mapped_column(String(255), nullable=False)
+    instance_id: Mapped[str | None] = mapped_column(String(255), nullable=True, default=None)
+    install_reported_at: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True, default=None)
+    last_heartbeat_at: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True, default=None)
    setup_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
@@ -1114,14 +1117,14 @@ class ExporleBanner(TypeBase):
    status: Mapped[BannerStatus] = mapped_column(
        EnumText(BannerStatus, length=255),
        nullable=False,
-        server_default=sa.text("'enabled'::character varying"),
+        server_default=sa.text("'enabled'"),
        default=BannerStatus.ENABLED,
    )
    created_at: Mapped[datetime] = mapped_column(
        sa.DateTime, nullable=False, server_default=func.current_timestamp(), init=False
    )
    language: Mapped[str] = mapped_column(
-        String(255), nullable=False, server_default=sa.text("'en-US'::character varying"), default="en-US"
+        String(255), nullable=False, server_default=sa.text("'en-US'"), default="en-US"
    )


@@ -1157,6 +1160,13 @@ class OAuthProviderApp(TypeBase):


 class Conversation(Base):
+    """Conversation state, including the exact Agent participant when applicable.
+
+    ``agent_workspace_binding_id`` is a logical pointer rather than a foreign
+    key because retired Binding ledger rows may be collected before the
+    conversation history is deleted.
+    """
+
    __tablename__ = "conversations"
    __table_args__ = (
        sa.PrimaryKeyConstraint("id", name="conversation_pkey"),
@@ -1178,6 +1188,7 @@ class Conversation(Base):
    id: Mapped[str] = mapped_column(StringUUID, default=lambda: str(uuid4()))
    app_id = mapped_column(StringUUID, nullable=False)
    app_model_config_id = mapped_column(StringUUID, nullable=True)
+    agent_workspace_binding_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    model_provider = mapped_column(String(255), nullable=True)
    override_model_configs = mapped_column(LongText)
    model_id = mapped_column(String(255), nullable=True)
@@ -1031,6 +1031,7 @@ class WorkflowNodeExecutionModel(Base):  # This model is expected to have `offlo
    node_id: Mapped[str] = mapped_column(String(255))
    node_type: Mapped[str] = mapped_column(String(255))
    title: Mapped[str] = mapped_column(String(255))
+    agent_workspace_binding_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
    inputs: Mapped[str | None] = mapped_column(LongText)
    process_data: Mapped[str | None] = mapped_column(LongText)
    outputs: Mapped[str | None] = mapped_column(LongText)
@@ -964,12 +964,6 @@ Stop a running Agent App chat message generation
 | ---- | ---------- | ----------- | -------- | ------ |
 | agent_id | path |  | Yes | string (uuid) |

-#### Request Body
-
-| Required | Schema |
-| -------- | ------ |
-|  No | **application/json**: [AgentDebugConversationRefreshPayload](#agentdebugconversationrefreshpayload)<br> |
-
 #### Responses

 | Code | Description | Schema |
@@ -1261,7 +1255,8 @@ Get basic information for an Agent App conversation sandbox
 | Name | Located in | Description | Required | Schema |
 | ---- | ---------- | ----------- | -------- | ------ |
 | agent_id | path | Agent ID | Yes | string (uuid) |
-| conversation_id | query | Agent App conversation ID | Yes | string |
+| caller_id | query | Agent App caller ID | Yes | string |
+| caller_type | query |  | Yes | string, <br>**Available values:** "build_draft", "conversation" |

 #### Responses

@@ -1277,7 +1272,8 @@ List a directory in an Agent App conversation sandbox
 | Name | Located in | Description | Required | Schema |
 | ---- | ---------- | ----------- | -------- | ------ |
 | agent_id | path | Agent ID | Yes | string (uuid) |
-| conversation_id | query | Agent App conversation ID | Yes | string |
+| caller_id | query | Agent App caller ID | Yes | string |
+| caller_type | query |  | Yes | string, <br>**Available values:** "build_draft", "conversation" |
 | path | query | Directory path relative to the sandbox workspace | No | string, <br>**Default:** . |

 #### Responses
@@ -1294,7 +1290,8 @@ Read a text/binary preview file in an Agent App conversation sandbox
 | Name | Located in | Description | Required | Schema |
 | ---- | ---------- | ----------- | -------- | ------ |
 | agent_id | path | Agent ID | Yes | string (uuid) |
-| conversation_id | query | Agent App conversation ID | Yes | string |
+| caller_id | query | Agent App caller ID | Yes | string |
+| caller_type | query |  | Yes | string, <br>**Available values:** "build_draft", "conversation" |
 | path | query | File path relative to the sandbox workspace | Yes | string |

 #### Responses
@@ -1672,6 +1669,23 @@ Create a new application
 | 200 | Import confirmed | **application/json**: [Import](#import)<br> |
 | 400 | Import failed | **application/json**: [Import](#import)<br> |

+### [GET] /apps/recent
+**Return the lightweight app cards needed by the Explore home page**
+
+Get recently modified apps for the home Continue Work section
+
+#### Parameters
+
+| Name | Located in | Description | Required | Schema |
+| ---- | ---------- | ----------- | -------- | ------ |
+| limit | query | Number of recently modified apps to return (1-8) | No | integer, <br>**Default:** 8 |
+
+#### Responses
+
+| Code | Description | Schema |
+| ---- | ----------- | ------ |
+| 200 | Success | **application/json**: [RecentAppListResponse](#recentapplistresponse)<br> |
+
 ### [GET] /apps/starred
 Get applications starred by the current account

@@ -3790,7 +3804,7 @@ List a directory in a workflow Agent node sandbox
 | app_id | path | Application ID | Yes | string (uuid) |
 | node_id | path | Workflow Agent node ID | Yes | string |
 | workflow_run_id | path | Workflow run ID | Yes | string (uuid) |
-| node_execution_id | query | Optional workflow node execution ID. When omitted, the latest active session for the node is used. | No | string |
+| node_execution_id | query | Workflow node execution ID | Yes | string |
 | path | query | Directory path relative to the sandbox workspace | No | string, <br>**Default:** . |

 #### Responses
@@ -3809,7 +3823,7 @@ Read a text/binary preview file in a workflow Agent node sandbox
 | app_id | path | Application ID | Yes | string (uuid) |
 | node_id | path | Workflow Agent node ID | Yes | string |
 | workflow_run_id | path | Workflow run ID | Yes | string (uuid) |
-| node_execution_id | query | Optional workflow node execution ID. When omitted, the latest active session for the node is used. | No | string |
+| node_execution_id | query | Workflow node execution ID | Yes | string |
 | path | query | File path relative to the sandbox workspace | Yes | string |

 #### Responses
@@ -13243,6 +13257,7 @@ Model class for AI model.

 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
+| active_config_is_published | boolean |  | Yes |
 | active_config_snapshot | [AgentConfigSnapshotSummaryResponse](#agentconfigsnapshotsummaryresponse) |  | No |
 | agent | [AgentComposerAgentResponse](#agentcomposeragentresponse) |  | Yes |
 | agent_soul | [AgentSoulConfig](#agentsoulconfig) |  | Yes |
@@ -13282,7 +13297,6 @@ Model class for AI model.
 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
 | access_mode | string |  | No |
-| active_config_is_published | boolean |  | No |
 | api_base_url | string |  | No |
 | app_id | string |  | No |
 | backing_app_id | string |  | No |
@@ -13926,12 +13940,6 @@ Stable Agent Soul reference to one normalized skill archive.
 | date | string |  | Yes |
 | message_count | integer |  | Yes |

-#### AgentDebugConversationRefreshPayload
-
-| Name | Type | Description | Required |
-| ---- | ---- | ----------- | -------- |
-| draft_type | [AgentConfigDraftType](#agentconfigdrafttype) | Agent draft surface whose conversation should be refreshed | No |
-
 #### AgentDebugConversationRefreshResponse

 | Name | Type | Description | Required |
@@ -14649,7 +14657,8 @@ section may be empty, which is how callers express "no knowledge layer".

 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| conversation_id | string | Agent App conversation ID | Yes |
+| caller_id | string | Agent App caller ID | Yes |
+| caller_type | string, <br>**Available values:** "build_draft", "conversation" | *Enum:* `"build_draft"`, `"conversation"` | Yes |
 | path | string | File path relative to the sandbox workspace | Yes |

 #### AgentScope
@@ -21018,6 +21027,28 @@ Whitelist scopes accepted by RBAC app and dataset access config APIs.
 | result | string |  | Yes |
 | updated_at | integer |  | Yes |

+#### RecentAppListResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| data | [ [RecentAppResponse](#recentappresponse) ] |  | Yes |
+
+#### RecentAppResponse
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| author_name | string |  | No |
+| icon | string |  | No |
+| icon_background | string |  | No |
+| icon_type | [IconType](#icontype) |  | No |
+| icon_url | string |  | Yes |
+| id | string |  | Yes |
+| maintainer | string |  | No |
+| mode | string, <br>**Available values:** "advanced-chat", "agent-chat", "chat", "completion", "workflow" | *Enum:* `"advanced-chat"`, `"agent-chat"`, `"chat"`, `"completion"`, `"workflow"` | Yes |
+| name | string |  | Yes |
+| permission_keys | [ string ] |  | No |
+| updated_at | integer |  | Yes |
+
 #### RecommendedAppDetailNullableResponse

 | Name | Type | Description | Required |
@@ -21308,7 +21339,6 @@ Whitelist scopes accepted by RBAC app and dataset access config APIs.

 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| session_id | string |  | Yes |
 | workspace_cwd | string |  | Yes |

 #### SandboxListResponse
@@ -23128,7 +23158,7 @@ How a workflow node is bound to an Agent.

 | Name | Type | Description | Required |
 | ---- | ---- | ----------- | -------- |
-| node_execution_id | string | Optional workflow node execution ID. When omitted, the latest active session for the node is used. | No |
+| node_execution_id | string | Workflow node execution ID | Yes |
 | path | string | File path relative to the sandbox workspace | Yes |

 #### WorkflowAppLogPaginationResponse
@@ -965,6 +965,12 @@ Returns Server-Sent Events stream.
 | ---- | ---- | ----------- | -------- |
 | tool_icons | object | Tool icon metadata keyed by tool name | No |

+#### AppMode
+
+| Name | Type | Description | Required |
+| ---- | ---- | ----------- | -------- |
+| AppMode | string |  |  |
+
 #### AppPermissionQuery

 | Name | Type | Description | Required |
@@ -1646,6 +1652,7 @@ in form definition, or a variable while the workflow is running.
 | custom_config | [WebAppCustomConfigResponse](#webappcustomconfigresponse) |  | No |
 | enable_site | boolean |  | Yes |
 | end_user_id | string |  | No |
+| mode | [AppMode](#appmode) |  | Yes |
 | model_config | [WebModelConfigResponse](#webmodelconfigresponse) |  | No |
 | plan | string |  | Yes |
 | site | [WebSiteResponse](#websiteresponse) |  | Yes |
@@ -1,3 +1,5 @@
+"""Unit tests for Aliyun trace utility transformations and database lookups."""
+
 import json
 from collections.abc import Mapping
 from typing import Any, cast
@@ -25,11 +27,13 @@ from dify_trace_aliyun.utils import (
    serialize_json_data,
 )
 from opentelemetry.trace import Link, StatusCode
+from sqlalchemy.orm import Session

 from core.rag.models.document import Document
 from graphon.entities import WorkflowNodeExecution
 from graphon.enums import WorkflowNodeExecutionStatus
 from models import EndUser
+from models.enums import EndUserType


 def test_get_user_id_from_message_data_no_end_user(monkeypatch: pytest.MonkeyPatch):
@@ -40,35 +44,40 @@ def test_get_user_id_from_message_data_no_end_user(monkeypatch: pytest.MonkeyPat
    assert get_user_id_from_message_data(message_data) == "account_id"


-def test_get_user_id_from_message_data_with_end_user(monkeypatch: pytest.MonkeyPatch):
+@pytest.mark.parametrize("sqlite3_session", [(EndUser,)], indirect=True)
+def test_get_user_id_from_message_data_with_end_user(monkeypatch: pytest.MonkeyPatch, sqlite3_session: Session) -> None:
    message_data = MagicMock()
    message_data.from_account_id = "account_id"
    message_data.from_end_user_id = "end_user_id"

-    end_user_data = MagicMock(spec=EndUser)
-    end_user_data.session_id = "session_id"
-
-    mock_session = MagicMock()
-    mock_session.get.return_value = end_user_data
+    end_user_data = EndUser(
+        id="end_user_id",
+        tenant_id="tenant_id",
+        app_id="app_id",
+        type=EndUserType.BROWSER,
+        session_id="session_id",
+    )
+    sqlite3_session.add(end_user_data)
+    sqlite3_session.commit()

    from dify_trace_aliyun.utils import db

-    monkeypatch.setattr(db, "session", mock_session)
+    monkeypatch.setattr(db, "session", sqlite3_session)

    assert get_user_id_from_message_data(message_data) == "session_id"


-def test_get_user_id_from_message_data_end_user_not_found(monkeypatch: pytest.MonkeyPatch):
+@pytest.mark.parametrize("sqlite3_session", [(EndUser,)], indirect=True)
+def test_get_user_id_from_message_data_end_user_not_found(
+    monkeypatch: pytest.MonkeyPatch, sqlite3_session: Session
+) -> None:
    message_data = MagicMock()
    message_data.from_account_id = "account_id"
    message_data.from_end_user_id = "end_user_id"

-    mock_session = MagicMock()
-    mock_session.get.return_value = None
-
    from dify_trace_aliyun.utils import db

-    monkeypatch.setattr(db, "session", mock_session)
+    monkeypatch.setattr(db, "session", sqlite3_session)

    assert get_user_id_from_message_data(message_data) == "account_id"

@@ -1,5 +1,8 @@
+"""Unit tests for LangSmith trace translation with SQLite-backed lookups."""
+
 import collections
 from datetime import datetime, timedelta
+from types import SimpleNamespace
 from typing import override
 from unittest.mock import MagicMock

@@ -11,6 +14,7 @@ from dify_trace_langsmith.entities.langsmith_trace_entity import (
    LangSmithRunUpdateModel,
 )
 from dify_trace_langsmith.langsmith_trace import LangSmithDataTrace
+from sqlalchemy.orm import Session

 from core.ops.entities.trace_entity import (
    DatasetRetrievalTraceInfo,
@@ -24,6 +28,7 @@ from core.ops.entities.trace_entity import (
 )
 from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey
 from models import EndUser
+from models.enums import EndUserType


 def _dt() -> datetime:
@@ -108,7 +113,8 @@ def test_trace_dispatch(trace_instance, monkeypatch: pytest.MonkeyPatch):
    mocks["generate_name_trace"].assert_called_once_with(info)


-def test_workflow_trace(trace_instance, monkeypatch: pytest.MonkeyPatch):
+@pytest.mark.parametrize("sqlite3_session", [()], indirect=True)
+def test_workflow_trace(trace_instance, monkeypatch: pytest.MonkeyPatch, sqlite3_session: Session) -> None:
    # Setup trace info
    workflow_data = MagicMock()
    workflow_data.created_at = _dt()
@@ -137,10 +143,10 @@ def test_workflow_trace(trace_instance, monkeypatch: pytest.MonkeyPatch):
        workflow_data=workflow_data,
    )

-    # Mock dependencies
-    mock_session = MagicMock()
-    monkeypatch.setattr("dify_trace_langsmith.langsmith_trace.sessionmaker", lambda bind: lambda: mock_session)
-    monkeypatch.setattr("dify_trace_langsmith.langsmith_trace.db", MagicMock(engine="engine"))
+    monkeypatch.setattr(
+        "dify_trace_langsmith.langsmith_trace.db",
+        SimpleNamespace(engine=sqlite3_session.get_bind(), session=sqlite3_session),
+    )

    # Mock node executions
    node_llm = MagicMock()
@@ -228,7 +234,10 @@ def test_workflow_trace(trace_instance, monkeypatch: pytest.MonkeyPatch):
    assert call_args[4].run_type == LangSmithRunType.retriever


-def test_workflow_trace_no_start_time(trace_instance, monkeypatch: pytest.MonkeyPatch):
+@pytest.mark.parametrize("sqlite3_session", [()], indirect=True)
+def test_workflow_trace_no_start_time(
+    trace_instance, monkeypatch: pytest.MonkeyPatch, sqlite3_session: Session
+) -> None:
    workflow_data = MagicMock()
    workflow_data.created_at = _dt()
    workflow_data.finished_at = _dt() + timedelta(seconds=1)
@@ -256,9 +265,10 @@ def test_workflow_trace_no_start_time(trace_instance, monkeypatch: pytest.Monkey
        workflow_data=workflow_data,
    )

-    mock_session = MagicMock()
-    monkeypatch.setattr("dify_trace_langsmith.langsmith_trace.sessionmaker", lambda bind: lambda: mock_session)
-    monkeypatch.setattr("dify_trace_langsmith.langsmith_trace.db", MagicMock(engine="engine"))
+    monkeypatch.setattr(
+        "dify_trace_langsmith.langsmith_trace.db",
+        SimpleNamespace(engine=sqlite3_session.get_bind(), session=sqlite3_session),
+    )
    repo = MagicMock()
    repo.get_by_workflow_execution.return_value = []
    mock_factory = MagicMock()
@@ -271,7 +281,10 @@ def test_workflow_trace_no_start_time(trace_instance, monkeypatch: pytest.Monkey
    assert trace_instance.add_run.called


-def test_workflow_trace_missing_app_id(trace_instance, monkeypatch: pytest.MonkeyPatch):
+@pytest.mark.parametrize("sqlite3_session", [()], indirect=True)
+def test_workflow_trace_missing_app_id(
+    trace_instance, monkeypatch: pytest.MonkeyPatch, sqlite3_session: Session
+) -> None:
    trace_info = MagicMock(spec=WorkflowTraceInfo)
    trace_info.trace_id = "trace-1"
    trace_info.message_id = None
@@ -287,15 +300,17 @@ def test_workflow_trace_missing_app_id(trace_instance, monkeypatch: pytest.Monke
    trace_info.workflow_run_outputs = {}
    trace_info.error = ""

-    mock_session = MagicMock()
-    monkeypatch.setattr("dify_trace_langsmith.langsmith_trace.sessionmaker", lambda bind: lambda: mock_session)
-    monkeypatch.setattr("dify_trace_langsmith.langsmith_trace.db", MagicMock(engine="engine"))
+    monkeypatch.setattr(
+        "dify_trace_langsmith.langsmith_trace.db",
+        SimpleNamespace(engine=sqlite3_session.get_bind(), session=sqlite3_session),
+    )

    with pytest.raises(ValueError, match="No app_id found in trace_info metadata"):
        trace_instance.workflow_trace(trace_info)


-def test_message_trace(trace_instance, monkeypatch: pytest.MonkeyPatch):
+@pytest.mark.parametrize("sqlite3_session", [(EndUser,)], indirect=True)
+def test_message_trace(trace_instance, monkeypatch: pytest.MonkeyPatch, sqlite3_session: Session) -> None:
    message_data = MagicMock()
    message_data.id = "msg-1"
    message_data.from_account_id = "acc-1"
@@ -321,10 +336,19 @@ def test_message_trace(trace_instance, monkeypatch: pytest.MonkeyPatch):
        message_file_data=MagicMock(url="file-url"),
    )

-    # Mock EndUser lookup
-    mock_end_user = MagicMock(spec=EndUser)
-    mock_end_user.session_id = "session-id-123"
-    monkeypatch.setattr("dify_trace_langsmith.langsmith_trace.db.session.get", lambda model, pk: mock_end_user)
+    end_user = EndUser(
+        id="end-user-1",
+        tenant_id="tenant-1",
+        app_id="app-1",
+        type=EndUserType.BROWSER,
+        session_id="session-id-123",
+    )
+    sqlite3_session.add(end_user)
+    sqlite3_session.commit()
+    monkeypatch.setattr(
+        "dify_trace_langsmith.langsmith_trace.db",
+        SimpleNamespace(engine=sqlite3_session.get_bind(), session=sqlite3_session),
+    )

    trace_instance.add_run = MagicMock()

@@ -521,9 +545,13 @@ def test_update_run_error(trace_instance):
        trace_instance.update_run(update_data)


+@pytest.mark.parametrize("sqlite3_session", [()], indirect=True)
 def test_workflow_trace_usage_extraction_error(
-    trace_instance, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
-):
+    trace_instance,
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+    sqlite3_session: Session,
+) -> None:
    workflow_data = MagicMock()
    workflow_data.created_at = _dt()
    workflow_data.finished_at = _dt() + timedelta(seconds=1)
@@ -576,8 +604,10 @@ def test_workflow_trace_usage_extraction_error(
    mock_factory = MagicMock()
    mock_factory.create_workflow_node_execution_repository.return_value = repo
    monkeypatch.setattr("dify_trace_langsmith.langsmith_trace.DifyCoreRepositoryFactory", mock_factory)
-    monkeypatch.setattr("dify_trace_langsmith.langsmith_trace.sessionmaker", lambda bind: lambda: MagicMock())
-    monkeypatch.setattr("dify_trace_langsmith.langsmith_trace.db", MagicMock(engine="engine"))
+    monkeypatch.setattr(
+        "dify_trace_langsmith.langsmith_trace.db",
+        SimpleNamespace(engine=sqlite3_session.get_bind(), session=sqlite3_session),
+    )
    monkeypatch.setattr(trace_instance, "get_service_account_with_tenant", lambda app_id: MagicMock())

    trace_instance.add_run = MagicMock()
@@ -644,9 +674,11 @@ def _make_workflow_trace_info(
    )


-def _patch_workflow_trace_deps(monkeypatch, trace_instance):
-    monkeypatch.setattr("dify_trace_langsmith.langsmith_trace.sessionmaker", lambda bind: lambda: MagicMock())
-    monkeypatch.setattr("dify_trace_langsmith.langsmith_trace.db", MagicMock(engine="engine"))
+def _patch_workflow_trace_deps(monkeypatch, trace_instance, sqlite3_session: Session) -> None:
+    monkeypatch.setattr(
+        "dify_trace_langsmith.langsmith_trace.db",
+        SimpleNamespace(engine=sqlite3_session.get_bind(), session=sqlite3_session),
+    )
    repo = MagicMock()
    repo.get_by_workflow_execution.return_value = []
    factory = MagicMock()
@@ -656,14 +688,17 @@ def _patch_workflow_trace_deps(monkeypatch, trace_instance):
    trace_instance.add_run = MagicMock()


-def test_workflow_trace_id_uses_message_id_not_external(trace_instance, monkeypatch: pytest.MonkeyPatch):
+@pytest.mark.parametrize("sqlite3_session", [()], indirect=True)
+def test_workflow_trace_id_uses_message_id_not_external(
+    trace_instance, monkeypatch: pytest.MonkeyPatch, sqlite3_session: Session
+) -> None:
    """Chatflow with external trace_id: LangSmith trace_id must be message_id, not external."""
    trace_info = _make_workflow_trace_info(
        message_id="msg-abc",
        workflow_run_id="run-xyz",
        trace_id="external-999",
    )
-    _patch_workflow_trace_deps(monkeypatch, trace_instance)
+    _patch_workflow_trace_deps(monkeypatch, trace_instance, sqlite3_session)

    trace_instance.workflow_trace(trace_info)

@@ -677,14 +712,17 @@ def test_workflow_trace_id_uses_message_id_not_external(trace_instance, monkeypa
    assert trace_info.metadata.get("external_trace_id") == "external-999"


-def test_workflow_trace_id_pure_workflow_uses_run_id(trace_instance, monkeypatch: pytest.MonkeyPatch):
+@pytest.mark.parametrize("sqlite3_session", [()], indirect=True)
+def test_workflow_trace_id_pure_workflow_uses_run_id(
+    trace_instance, monkeypatch: pytest.MonkeyPatch, sqlite3_session: Session
+) -> None:
    """Pure workflow (no message_id) with external trace_id: trace_id must be workflow_run_id."""
    trace_info = _make_workflow_trace_info(
        message_id=None,
        workflow_run_id="run-xyz",
        trace_id="external-999",
    )
-    _patch_workflow_trace_deps(monkeypatch, trace_instance)
+    _patch_workflow_trace_deps(monkeypatch, trace_instance, sqlite3_session)

    trace_instance.workflow_trace(trace_info)

@@ -316,10 +316,10 @@ class OracleVector(BaseVector):
                    entities.append(current_entity)
            else:
                try:
-                    nltk.data.find("tokenizers/punkt")
+                    nltk.data.find("tokenizers/punkt_tab")
                    nltk.data.find("corpora/stopwords")
                except LookupError:
-                    raise LookupError("Unable to find the required NLTK data package: punkt and stopwords")
+                    raise LookupError("Unable to find the required NLTK data package: punkt_tab and stopwords")
                e_str = re.sub(r"[^\w ]", "", query)
                all_tokens = nltk.word_tokenize(e_str)
                stop_words = stopwords.words("english")
@@ -1,12 +1,12 @@
 [project]
 name = "dify-api"
-version = "1.16.0"
+version = "1.16.1"
 requires-python = "~=3.12.0"

 dependencies = [
    # Legacy: mature and widely deployed
    "bleach>=6.4.0,<7.0.0",
-    "boto3>=1.43.46,<2.0.0",
+    "boto3>=1.43.56,<2.0.0",
    "celery>=5.6.3,<6.0.0",
    "croniter>=6.2.2,<7.0.0",
    "dify-agent",
@@ -193,10 +193,10 @@ dev = [
 ############################################################
 storage = [
    "azure-storage-blob>=12.30.0,<13.0.0",
-    "bce-python-sdk==0.9.72",
+    "bce-python-sdk==0.9.76",
    "cos-python-sdk-v5>=1.9.44,<2.0.0",
    "esdk-obs-python>=3.26.6,<4.0.0",
-    "google-cloud-storage>=3.12.1,<4.0.0",
+    "google-cloud-storage>=3.13.0,<4.0.0",
    "opendal==0.46.0",
    "oss2>=2.19.1,<3.0.0",
    "supabase>=2.31.0,<3.0.0",
@@ -206,7 +206,7 @@ storage = [
 ############################################################
 # [ Tools ] dependency group
 ############################################################
-tools = ["cloudscraper>=1.2.71,<2.0.0", "nltk>=3.9.1,<4.0.0"]
+tools = ["cloudscraper>=1.2.71,<2.0.0", "nltk>=3.10.0,<4.0.0"]

 ############################################################
 # [ VDB ] workspace plugins — hollow packages under providers/vdb/*
@@ -75,6 +75,7 @@ from services.errors.account import (
 from services.errors.workspace import WorkSpaceNotAllowedCreateError, WorkspacesLimitExceededError
 from services.feature_service import FeatureService
 from services.plugin.plugin_auto_upgrade_service import PluginAutoUpgradeService
+from services.telemetry_service import CommunityTelemetryService
 from tasks.delete_account_task import delete_account_task
 from tasks.mail_account_deletion_task import send_account_deletion_verification_code
 from tasks.mail_change_mail_task import (
@@ -1953,7 +1954,7 @@ class RegisterService:

            TenantService.create_owner_tenant_if_not_exist(account=account, is_setup=True, session=session)

-            dify_setup = DifySetup(version=dify_config.project.version)
+            dify_setup = DifySetup(version=dify_config.project.version, instance_id=str(uuid.uuid4()))
            session.add(dify_setup)
            session.commit()
        except Exception as e:
@@ -1966,6 +1967,11 @@ class RegisterService:
            logger.exception("Setup account failed, email: %s, name: %s", email, name)
            raise ValueError(f"Setup failed: {e}")

+        try:
+            CommunityTelemetryService.report_install(session=session)
+        except Exception:
+            logger.debug("Failed to report install telemetry", exc_info=True)
+
    @classmethod
    def register(
        cls,
@@ -9,7 +9,7 @@ from sqlalchemy.sql.elements import ColumnElement

 from core.agent.publish_visibility import agent_has_workflow_callable_active_snapshot
 from libs.helper import to_timestamp
-from models import Account
+from models import Account, Conversation
 from models.agent import (
    APP_BACKED_AGENT_SOURCES,
    Agent,
@@ -18,12 +18,15 @@ from models.agent import (
    AgentConfigRevision,
    AgentConfigRevisionOperation,
    AgentConfigSnapshot,
+    AgentConfigVersionKind,
+    AgentDebugConversation,
    AgentDriveFile,
    AgentIconType,
    AgentKind,
    AgentScope,
    AgentSource,
    AgentStatus,
+    AgentWorkspaceOwnerType,
    WorkflowAgentBindingType,
    WorkflowAgentNodeBinding,
 )
@@ -35,6 +38,7 @@ from models.workflow import Workflow
 from services.agent.agent_soul_state import agent_soul_has_model
 from services.agent.composer_validator import ComposerConfigValidator
 from services.agent.errors import (
+    AgentBuildSandboxNotFoundError,
    AgentModelNotConfiguredError,
    AgentNameConflictError,
    AgentNotFoundError,
@@ -42,11 +46,17 @@ from services.agent.errors import (
    AgentVersionNotFoundError,
    InvalidComposerConfigError,
 )
+from services.agent.home_snapshot_service import (
+    AgentHomeSnapshotService,
+    validate_home_snapshot_binding,
+)
 from services.agent.knowledge_datasets import (
    get_tenant_knowledge_dataset_rows,
    list_missing_tenant_knowledge_dataset_ids,
 )
+from services.agent.retirement_service import WorkflowAgentRetirementService
 from services.agent.roster_service import AgentRosterService
+from services.agent.workspace_service import AgentWorkspaceNotFoundError, AgentWorkspaceService, WorkspaceOwnerScope
 from services.app_service import AppService, CreateAppParams
 from services.entities.agent_entities import (
    AgentSoulConfig,
@@ -56,6 +66,7 @@ from services.entities.agent_entities import (
    ComposerVariant,
    WorkflowNodeJobConfig,
 )
+from tasks.collect_agent_resources_task import enqueue_agent_resource_collection

 # WorkflowAgentNodeBinding.workflow_version tag for the draft workflow row.
 # Mirrors Workflow.version when it is "draft" (see models/workflow.py).
@@ -198,6 +209,18 @@ class AgentComposerService:
        binding = cls._get_workflow_binding(
            session=session, tenant_id=tenant_id, workflow_id=workflow.id, node_id=node_id
        )
+        retirement_candidates = (
+            {binding.agent_id}
+            if binding is not None
+            and binding.binding_type == WorkflowAgentBindingType.INLINE_AGENT
+            and binding.agent_id
+            and payload.save_strategy
+            in {
+                ComposerSaveStrategy.SAVE_AS_NEW_AGENT,
+                ComposerSaveStrategy.SAVE_TO_ROSTER,
+            }
+            else set()
+        )

        match payload.save_strategy:
            case ComposerSaveStrategy.NODE_JOB_ONLY:
@@ -232,7 +255,11 @@ class AgentComposerService:
                )
            case ComposerSaveStrategy.SAVE_TO_ROSTER:
                binding = cls._save_to_roster(
-                    session=session, tenant_id=tenant_id, account_id=account_id, binding=binding, payload=payload
+                    session=session,
+                    tenant_id=tenant_id,
+                    account_id=account_id,
+                    binding=binding,
+                    payload=payload,
                )

        session.flush()
@@ -257,6 +284,17 @@ class AgentComposerService:
            payload=payload,
            agent_id=binding.agent_id,
        )
+        session.commit()
+        binding_ids, home_snapshot_ids = WorkflowAgentRetirementService.retire_unowned(
+            tenant_id=tenant_id,
+            agent_ids=retirement_candidates,
+            account_id=account_id,
+        )
+        enqueue_agent_resource_collection(
+            tenant_id=tenant_id,
+            binding_ids=binding_ids,
+            home_snapshot_ids=home_snapshot_ids,
+        )
        return state

    @classmethod
@@ -390,12 +428,10 @@ class AgentComposerService:

    @classmethod
    def _load_agent_composer_for_agent(cls, *, session: Session, tenant_id: str, agent: Agent) -> dict[str, Any]:
-        draft = cls._get_or_create_agent_draft(
+        draft = cls.get_or_create_normal_agent_draft(
            session=session,
            tenant_id=tenant_id,
            agent=agent,
-            draft_type=AgentConfigDraftType.DRAFT,
-            account_id=None,
            created_by=agent.updated_by or agent.created_by,
        )
        version = cls._get_version_if_present(
@@ -405,6 +441,7 @@ class AgentComposerService:
            "variant": ComposerVariant.AGENT_APP.value,
            "agent": cls._serialize_agent(agent),
            "active_config_snapshot": cls._serialize_version(version),
+            "active_config_is_published": bool(agent.active_config_snapshot_id and agent.active_config_is_published),
            "draft": cls._serialize_draft(draft),
            "agent_soul": draft.config_snapshot_dict,
            "save_options": [ComposerSaveStrategy.SAVE_TO_CURRENT_VERSION.value],
@@ -416,7 +453,34 @@ class AgentComposerService:

    @classmethod
    def save_agent_app_composer(
-        cls, *, session: Session, tenant_id: str, app_id: str, account_id: str, payload: ComposerSavePayload
+        cls,
+        *,
+        session: Session,
+        tenant_id: str,
+        app_id: str,
+        account_id: str,
+        payload: ComposerSavePayload,
+    ) -> dict[str, Any]:
+        try:
+            return cls._save_agent_app_composer_impl(
+                session=session,
+                tenant_id=tenant_id,
+                app_id=app_id,
+                account_id=account_id,
+                payload=payload,
+            )
+        except IntegrityError as exc:
+            raise AgentNameConflictError() from exc
+
+    @classmethod
+    def _save_agent_app_composer_impl(
+        cls,
+        *,
+        session: Session,
+        tenant_id: str,
+        app_id: str,
+        account_id: str,
+        payload: ComposerSavePayload,
    ) -> dict[str, Any]:
        if payload.variant != ComposerVariant.AGENT_APP:
            raise ValueError("Agent App composer endpoint only accepts agent_app variant")
@@ -445,11 +509,25 @@ class AgentComposerService:
                updated_by=account_id,
            )
            session.add(agent)
-            try:
-                session.flush()
-            except IntegrityError as exc:
-                session.rollback()
-                raise AgentNameConflictError() from exc
+            session.flush()
+            home_snapshot = AgentHomeSnapshotService.create_initial(
+                session=session,
+                tenant_id=tenant_id,
+                agent_id=agent.id,
+            )
+            initial_version = cls._create_config_version(
+                session=session,
+                tenant_id=tenant_id,
+                agent_id=agent.id,
+                account_id=account_id,
+                agent_soul=AgentSoulConfig(),
+                operation=AgentConfigRevisionOperation.CREATE_VERSION,
+                version_note=None,
+                home_snapshot_id=home_snapshot.id,
+            )
+            agent.active_config_snapshot_id = initial_version.id
+            agent.active_config_has_model = False
+            agent.active_config_is_published = False
        return cls._save_agent_composer_for_agent(
            session=session,
            tenant_id=tenant_id,
@@ -487,7 +565,7 @@ class AgentComposerService:
    ) -> dict[str, Any]:
        if payload.agent_soul is None:
            raise ValueError("agent_soul is required")
-        cls._save_agent_draft(
+        draft = cls._save_agent_draft(
            session=session,
            tenant_id=tenant_id,
            agent=agent,
@@ -502,6 +580,7 @@ class AgentComposerService:
            tenant_id=tenant_id,
            agent=agent,
            agent_soul=payload.agent_soul,
+            home_snapshot_id=draft.home_snapshot_id,
        )

        session.flush()
@@ -522,6 +601,7 @@ class AgentComposerService:
        tenant_id: str,
        agent: Agent,
        agent_soul: AgentSoulConfig,
+        home_snapshot_id: str,
    ) -> bool:
        if not agent.active_config_snapshot_id:
            return False
@@ -537,7 +617,9 @@ class AgentComposerService:
        if not agent_has_workflow_callable_active_snapshot(session=session, agent=agent):
            return False

-        return _agent_soul_config_json(agent_soul) == _agent_soul_config_json(active_version.config_snapshot_dict)
+        return home_snapshot_id == active_version.home_snapshot_id and _agent_soul_config_json(
+            agent_soul
+        ) == _agent_soul_config_json(active_version.config_snapshot_dict)

    @classmethod
    def publish_agent_app_draft(
@@ -566,6 +648,11 @@ class AgentComposerService:
        if not agent_soul_has_model(agent_soul):
            raise AgentModelNotConfiguredError()
        cls.validate_knowledge_datasets(session=session, tenant_id=tenant_id, agent_soul=agent_soul)
+        validate_home_snapshot_binding(
+            session=session,
+            agent=agent,
+            home_snapshot_id=draft.home_snapshot_id,
+        )
        version = cls._create_config_version(
            session=session,
            tenant_id=tenant_id,
@@ -575,6 +662,7 @@ class AgentComposerService:
            operation=AgentConfigRevisionOperation.PUBLISH_DRAFT,
            version_note=version_note,
            previous_snapshot_id=agent.active_config_snapshot_id,
+            home_snapshot_id=draft.home_snapshot_id,
        )
        agent.active_config_snapshot_id = version.id
        agent.active_config_has_model = agent_soul_has_model(agent_soul)
@@ -594,6 +682,29 @@ class AgentComposerService:
    def checkout_agent_app_build_draft(
        cls, *, session: Session, tenant_id: str, agent_id: str, account_id: str, force: bool = False
    ) -> dict[str, Any]:
+        try:
+            result, retired_binding_id = cls._checkout_agent_app_build_draft_in_transaction(
+                session=session,
+                tenant_id=tenant_id,
+                agent_id=agent_id,
+                account_id=account_id,
+                force=force,
+            )
+            session.commit()
+        except Exception:
+            session.rollback()
+            raise
+        if retired_binding_id is not None:
+            enqueue_agent_resource_collection(
+                tenant_id=tenant_id,
+                binding_ids=(retired_binding_id,),
+            )
+        return result
+
+    @classmethod
+    def _checkout_agent_app_build_draft_in_transaction(
+        cls, *, session: Session, tenant_id: str, agent_id: str, account_id: str, force: bool
+    ) -> tuple[dict[str, Any], str | None]:
        agent = cls._require_agent(session=session, tenant_id=tenant_id, agent_id=agent_id)
        normal_draft = cls._get_or_create_agent_draft(
            session=session,
@@ -611,7 +722,23 @@ class AgentComposerService:
            account_id=account_id,
        )
        if build_draft is not None and not force:
-            return cls._serialize_build_draft_state(build_draft)
+            return cls._serialize_build_draft_state(build_draft), None
+        retired_binding_id: str | None = None
+        if build_draft is not None and build_draft.agent_workspace_binding_id is not None:
+            cls._validate_active_build_draft_binding(
+                session=session,
+                tenant_id=tenant_id,
+                agent=agent,
+                build_draft=build_draft,
+            )
+            retired_binding_id = AgentWorkspaceService.retire_binding(
+                session=session,
+                tenant_id=tenant_id,
+                binding_id=build_draft.agent_workspace_binding_id,
+            )
+            if retired_binding_id is None:
+                raise AgentBuildSandboxNotFoundError()
+            build_draft.agent_workspace_binding_id = None
        if build_draft is None:
            build_draft = AgentConfigDraft(
                tenant_id=tenant_id,
@@ -623,10 +750,44 @@ class AgentComposerService:
            )
            session.add(build_draft)
        build_draft.base_snapshot_id = normal_draft.base_snapshot_id
+        build_draft.home_snapshot_id = normal_draft.home_snapshot_id
        build_draft.config_snapshot = AgentSoulConfig.model_validate(normal_draft.config_snapshot_dict)
        build_draft.updated_by = account_id
        session.flush()
-        return cls._serialize_build_draft_state(build_draft)
+        return cls._serialize_build_draft_state(build_draft), retired_binding_id
+
+    @classmethod
+    def _validate_active_build_draft_binding(
+        cls,
+        *,
+        session: Session,
+        tenant_id: str,
+        agent: Agent,
+        build_draft: AgentConfigDraft,
+    ) -> None:
+        binding_id = build_draft.agent_workspace_binding_id
+        runtime_app_id = AgentRosterService.runtime_backing_app_id(agent)
+        if binding_id is None or runtime_app_id is None:
+            raise AgentBuildSandboxNotFoundError()
+        binding = AgentWorkspaceService.get_active_binding(
+            session=session,
+            tenant_id=tenant_id,
+            binding_id=binding_id,
+            expected_owner_scope=WorkspaceOwnerScope(
+                tenant_id=tenant_id,
+                app_id=runtime_app_id,
+                owner_type=AgentWorkspaceOwnerType.BUILD_DRAFT,
+                owner_id=build_draft.id,
+            ),
+        )
+        if binding is None or binding.agent_id != agent.id:
+            raise AgentBuildSandboxNotFoundError()
+        AgentWorkspaceService.validate_binding_generation(
+            binding,
+            base_home_snapshot_id=build_draft.home_snapshot_id,
+            agent_config_version_id=build_draft.id,
+            agent_config_version_kind=AgentConfigVersionKind.BUILD_DRAFT,
+        )

    @classmethod
    def load_agent_app_build_draft(
@@ -668,6 +829,29 @@ class AgentComposerService:
    def apply_agent_app_build_draft(
        cls, *, session: Session, tenant_id: str, agent_id: str, account_id: str
    ) -> dict[str, Any]:
+        try:
+            result, retired_binding_ids = cls._apply_agent_app_build_draft_in_transaction(
+                session=session,
+                tenant_id=tenant_id,
+                agent_id=agent_id,
+                account_id=account_id,
+            )
+            session.commit()
+        except Exception:
+            session.rollback()
+            raise
+        enqueue_agent_resource_collection(tenant_id=tenant_id, binding_ids=retired_binding_ids)
+        return result
+
+    @classmethod
+    def _apply_agent_app_build_draft_in_transaction(
+        cls,
+        *,
+        session: Session,
+        tenant_id: str,
+        agent_id: str,
+        account_id: str,
+    ) -> tuple[dict[str, Any], list[str]]:
        agent = cls._require_agent(session=session, tenant_id=tenant_id, agent_id=agent_id)
        build_draft = cls._get_agent_draft(
            session=session,
@@ -679,6 +863,21 @@ class AgentComposerService:
        if build_draft is None:
            raise AgentVersionNotFoundError()
        applied_agent_soul = AgentSoulConfig.model_validate(build_draft.config_snapshot_dict)
+        ComposerConfigValidator.validate_publish_payload(
+            ComposerSavePayload(
+                variant=ComposerVariant.AGENT_APP,
+                agent_soul=applied_agent_soul,
+                save_strategy=ComposerSaveStrategy.SAVE_AS_NEW_VERSION,
+            )
+        )
+        cls.validate_knowledge_datasets(session=session, tenant_id=tenant_id, agent_soul=applied_agent_soul)
+        source_binding_id = build_draft.agent_workspace_binding_id
+        if source_binding_id is None:
+            raise AgentBuildSandboxNotFoundError()
+        home_snapshot = AgentHomeSnapshotService.create_for_build_apply(
+            session=session,
+            build_draft=build_draft,
+        )
        normal_draft = cls._save_agent_draft(
            session=session,
            tenant_id=tenant_id,
@@ -689,32 +888,145 @@ class AgentComposerService:
            account_id_for_audit=account_id,
            base_snapshot_id=build_draft.base_snapshot_id,
        )
+        retired_binding_ids = cls._retire_normal_preview_bindings(
+            session=session,
+            tenant_id=tenant_id,
+            agent=agent,
+            normal_draft=normal_draft,
+        )
+        normal_draft.home_snapshot_id = home_snapshot.id
        agent.active_config_is_published = cls._agent_soul_matches_active_config(
            session=session,
            tenant_id=tenant_id,
            agent=agent,
            agent_soul=applied_agent_soul,
+            home_snapshot_id=home_snapshot.id,
        )
        agent.updated_by = account_id
+        retired_binding_id = AgentWorkspaceService.retire_binding(
+            session=session,
+            tenant_id=tenant_id,
+            binding_id=source_binding_id,
+        )
+        if retired_binding_id is None:
+            raise AgentBuildSandboxNotFoundError()
+        retired_binding_ids.append(source_binding_id)
        session.delete(build_draft)
-        session.flush()
-        return {"result": "success", "draft": cls._serialize_draft(normal_draft)}
+        return {"result": "success", "draft": cls._serialize_draft(normal_draft)}, retired_binding_ids
+
+    @classmethod
+    def _retire_normal_preview_bindings(
+        cls,
+        *,
+        session: Session,
+        tenant_id: str,
+        agent: Agent,
+        normal_draft: AgentConfigDraft,
+    ) -> list[str]:
+        """Retire Preview participants before Build Apply replaces the shared Draft Home."""
+
+        mappings = session.scalars(
+            select(AgentDebugConversation).where(
+                AgentDebugConversation.tenant_id == tenant_id,
+                AgentDebugConversation.agent_id == agent.id,
+                AgentDebugConversation.draft_type == AgentConfigDraftType.DRAFT,
+            )
+        ).all()
+        retired_binding_ids: list[str] = []
+        for mapping in mappings:
+            conversation = session.scalar(
+                select(Conversation).where(
+                    Conversation.id == mapping.conversation_id,
+                    Conversation.app_id == mapping.app_id,
+                    Conversation.is_deleted.is_(False),
+                )
+            )
+            if conversation is None or conversation.agent_workspace_binding_id is None:
+                continue
+            binding_id = conversation.agent_workspace_binding_id
+            binding = AgentWorkspaceService.get_active_binding(
+                session=session,
+                tenant_id=tenant_id,
+                binding_id=binding_id,
+                expected_owner_scope=WorkspaceOwnerScope(
+                    tenant_id=tenant_id,
+                    app_id=mapping.app_id,
+                    owner_type=AgentWorkspaceOwnerType.CONVERSATION,
+                    owner_id=conversation.id,
+                ),
+            )
+            if binding is None or binding.agent_id != agent.id:
+                raise AgentWorkspaceNotFoundError("Agent Preview participant Binding is unavailable")
+            AgentWorkspaceService.validate_binding_generation(
+                binding,
+                base_home_snapshot_id=normal_draft.home_snapshot_id,
+                agent_config_version_id=normal_draft.id,
+                agent_config_version_kind=AgentConfigVersionKind.DRAFT,
+            )
+            retired_binding_id = AgentWorkspaceService.retire_binding(
+                session=session,
+                tenant_id=tenant_id,
+                binding_id=binding_id,
+            )
+            if retired_binding_id is None:
+                raise AgentWorkspaceNotFoundError("Agent Preview participant Binding is unavailable")
+            conversation.agent_workspace_binding_id = None
+            retired_binding_ids.append(binding_id)
+        return retired_binding_ids

    @classmethod
    def discard_agent_app_build_draft(
        cls, *, session: Session, tenant_id: str, agent_id: str, account_id: str
    ) -> dict[str, Any]:
+        try:
+            result, retired_binding_id = cls._discard_agent_app_build_draft_in_transaction(
+                session=session,
+                tenant_id=tenant_id,
+                agent_id=agent_id,
+                account_id=account_id,
+            )
+            session.commit()
+        except Exception:
+            session.rollback()
+            raise
+        if retired_binding_id is not None:
+            enqueue_agent_resource_collection(
+                tenant_id=tenant_id,
+                binding_ids=(retired_binding_id,),
+            )
+        return result
+
+    @classmethod
+    def _discard_agent_app_build_draft_in_transaction(
+        cls, *, session: Session, tenant_id: str, agent_id: str, account_id: str
+    ) -> tuple[dict[str, Any], str | None]:
+        agent = cls._require_agent(session=session, tenant_id=tenant_id, agent_id=agent_id)
        build_draft = cls._get_agent_draft(
            session=session,
            tenant_id=tenant_id,
-            agent_id=agent_id,
+            agent_id=agent.id,
            draft_type=AgentConfigDraftType.DEBUG_BUILD,
            account_id=account_id,
        )
-        if build_draft is not None:
-            session.delete(build_draft)
-            session.flush()
-        return {"result": "success"}
+        if build_draft is None:
+            return {"result": "success"}, None
+        retired_binding_id: str | None = None
+        if build_draft.agent_workspace_binding_id is not None:
+            cls._validate_active_build_draft_binding(
+                session=session,
+                tenant_id=tenant_id,
+                agent=agent,
+                build_draft=build_draft,
+            )
+            retired_binding_id = AgentWorkspaceService.retire_binding(
+                session=session,
+                tenant_id=tenant_id,
+                binding_id=build_draft.agent_workspace_binding_id,
+            )
+            if retired_binding_id is None:
+                raise AgentBuildSandboxNotFoundError()
+        session.delete(build_draft)
+        return {"result": "success"}, retired_binding_id

    @classmethod
    def collect_validation_findings(
@@ -1278,6 +1590,12 @@ class AgentComposerService:
        binding = cls._require_binding(binding)
        if not binding.agent_id or payload.agent_soul is None:
            raise ValueError("agent_id and agent_soul are required")
+        current_snapshot = cls._require_version(
+            session=session,
+            tenant_id=tenant_id,
+            agent_id=binding.agent_id,
+            version_id=binding.current_snapshot_id,
+        )
        version = cls._create_config_version(
            session=session,
            tenant_id=tenant_id,
@@ -1286,6 +1604,7 @@ class AgentComposerService:
            agent_soul=payload.agent_soul,
            operation=AgentConfigRevisionOperation.SAVE_NEW_VERSION,
            version_note=payload.version_note,
+            home_snapshot_id=current_snapshot.home_snapshot_id,
        )
        agent = cls._require_agent(session=session, tenant_id=tenant_id, agent_id=binding.agent_id)
        agent.active_config_snapshot_id = version.id
@@ -1448,6 +1767,11 @@ class AgentComposerService:
        )
        session.add(agent)
        session.flush()
+        home_snapshot = AgentHomeSnapshotService.create_initial(
+            session=session,
+            tenant_id=tenant_id,
+            agent_id=agent.id,
+        )
        version = cls._create_config_version(
            session=session,
            tenant_id=tenant_id,
@@ -1456,6 +1780,7 @@ class AgentComposerService:
            agent_soul=agent_soul,
            operation=AgentConfigRevisionOperation.CREATE_VERSION,
            version_note=None,
+            home_snapshot_id=home_snapshot.id,
        )
        agent.active_config_snapshot_id = version.id
        agent.active_config_has_model = agent_soul_has_model(agent_soul)
@@ -1589,7 +1914,6 @@ class AgentComposerService:
                session=session,
            )
        except IntegrityError as exc:
-            session.rollback()
            raise AgentNameConflictError() from exc

        agent = AgentRosterService(session).get_app_backing_agent(tenant_id=tenant_id, app_id=app.id)
@@ -1627,6 +1951,7 @@ class AgentComposerService:
        agent_soul: AgentSoulConfig,
        operation: AgentConfigRevisionOperation,
        version_note: str | None,
+        home_snapshot_id: str,
        previous_snapshot_id: str | None = None,
    ) -> AgentConfigSnapshot:
        next_version = (
@@ -1643,6 +1968,7 @@ class AgentComposerService:
            agent_id=agent_id,
            version=next_version,
            config_snapshot=agent_soul,
+            home_snapshot_id=home_snapshot_id,
            version_note=version_note,
            created_by=account_id,
        )
@@ -1682,6 +2008,7 @@ class AgentComposerService:
            operation=operation,
            version_note=version_note,
            previous_snapshot_id=current_snapshot.id,
+            home_snapshot_id=current_snapshot.home_snapshot_id,
        )

    @classmethod
@@ -1748,7 +2075,10 @@ class AgentComposerService:
        agent: Agent,
        created_by: str | None,
    ) -> AgentConfigDraft:
-        """Resolve the shared Preview draft, rebasing inline agents when needed."""
+        """Resolve the normal Draft, rebasing only stale WORKFLOW_ONLY DRAFT rows whose account_id is None.
+
+        Roster and DEBUG_BUILD Drafts are never rebased.
+        """
        return cls._get_or_create_agent_draft(
            session=session,
            tenant_id=tenant_id,
@@ -1766,6 +2096,8 @@ class AgentComposerService:
        snapshot: AgentConfigSnapshot,
        updated_by: str | None,
    ) -> bool:
+        """Sync a stale normal Draft's base_snapshot_id, home_snapshot_id, config_snapshot, and updated_by."""
+
        if (
            agent.scope != AgentScope.WORKFLOW_ONLY
            or draft.draft_type != AgentConfigDraftType.DRAFT
@@ -1776,6 +2108,7 @@ class AgentComposerService:
        ):
            return False
        draft.base_snapshot_id = snapshot.id
+        draft.home_snapshot_id = snapshot.home_snapshot_id
        draft.config_snapshot = AgentSoulConfig.model_validate(snapshot.config_snapshot_dict)
        draft.updated_by = updated_by
        return True
@@ -1812,7 +2145,9 @@ class AgentComposerService:
                    agent_id=agent.id,
                    version_id=agent.active_config_snapshot_id,
                )
-                if active_snapshot is not None and cls._rebase_workflow_only_normal_draft(
+                if active_snapshot is None:
+                    raise AgentVersionNotFoundError()
+                if cls._rebase_workflow_only_normal_draft(
                    agent=agent,
                    draft=draft,
                    snapshot=active_snapshot,
@@ -1826,18 +2161,17 @@ class AgentComposerService:
            agent_id=agent.id,
            version_id=agent.active_config_snapshot_id,
        )
-        agent_soul = (
-            AgentSoulConfig.model_validate(base_snapshot.config_snapshot_dict)
-            if base_snapshot is not None
-            else AgentSoulConfig()
-        )
+        if base_snapshot is None:
+            raise AgentVersionNotFoundError()
+        agent_soul = AgentSoulConfig.model_validate(base_snapshot.config_snapshot_dict)
        draft = AgentConfigDraft(
            tenant_id=tenant_id,
            agent_id=agent.id,
            draft_type=draft_type,
            account_id=account_id if draft_type == AgentConfigDraftType.DEBUG_BUILD else None,
            draft_owner_key=account_id if draft_type == AgentConfigDraftType.DEBUG_BUILD and account_id else "",
-            base_snapshot_id=base_snapshot.id if base_snapshot else None,
+            base_snapshot_id=base_snapshot.id,
+            home_snapshot_id=base_snapshot.home_snapshot_id,
            config_snapshot=agent_soul,
            created_by=created_by,
            updated_by=created_by,
@@ -2139,7 +2473,7 @@ class AgentComposerService:

        from services.agent.roster_service import AgentRosterService

-        return AgentRosterService(session).get_or_create_agent_app_debug_conversation_id(
+        return AgentRosterService(session).get_or_create_build_conversation(
            tenant_id=tenant_id,
            agent_id=agent.id,
            account_id=account_id,
--- a/Show More
+++ b/Show More