fix: sharepoint unquote (#8786)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

.github/actions/build-backend-image/action.yml

@@ -0,0 +1,73 @@
+name: "Build Backend Image"
+description: "Builds and pushes the backend Docker image with cache reuse"
+inputs:
+  runs-on-ecr-cache:
+    description: "ECR cache registry from runs-on/action"
+    required: true
+  ref-name:
+    description: "Git ref name used for cache suffix fallback"
+    required: true
+  pr-number:
+    description: "Optional PR number for cache suffix"
+    required: false
+    default: ""
+  github-sha:
+    description: "Commit SHA used for cache keys"
+    required: true
+  run-id:
+    description: "GitHub run ID used in output image tag"
+    required: true
+  docker-username:
+    description: "Docker Hub username"
+    required: true
+  docker-token:
+    description: "Docker Hub token"
+    required: true
+  docker-no-cache:
+    description: "Set to 'true' to disable docker build cache"
+    required: false
+    default: "false"
+runs:
+  using: "composite"
+  steps:
+    - name: Format branch name for cache
+      id: format-branch
+      shell: bash
+      env:
+        PR_NUMBER: ${{ inputs.pr-number }}
+        REF_NAME: ${{ inputs.ref-name }}
+      run: |
+        if [ -n "${PR_NUMBER}" ]; then
+          CACHE_SUFFIX="${PR_NUMBER}"
+        else
+          # shellcheck disable=SC2001
+          CACHE_SUFFIX=$(echo "${REF_NAME}" | sed 's/[^A-Za-z0-9._-]/-/g')
+        fi
+        echo "cache-suffix=${CACHE_SUFFIX}" >> "$GITHUB_OUTPUT"
+
+    - name: Set up Docker Buildx
+      uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # ratchet:docker/setup-buildx-action@v3
+
+    - name: Login to Docker Hub
+      uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # ratchet:docker/login-action@v3
+      with:
+        username: ${{ inputs.docker-username }}
+        password: ${{ inputs.docker-token }}
+
+    - name: Build and push Backend Docker image
+      uses: docker/build-push-action@263435318d21b8e681c14492fe198d362a7d2c83 # ratchet:docker/build-push-action@v6
+      with:
+        context: ./backend
+        file: ./backend/Dockerfile
+        push: true
+        tags: ${{ inputs.runs-on-ecr-cache }}:nightly-llm-it-backend-${{ inputs.run-id }}
+        cache-from: |
+          type=registry,ref=${{ inputs.runs-on-ecr-cache }}:backend-cache-${{ inputs.github-sha }}
+          type=registry,ref=${{ inputs.runs-on-ecr-cache }}:backend-cache-${{ steps.format-branch.outputs.cache-suffix }}
+          type=registry,ref=${{ inputs.runs-on-ecr-cache }}:backend-cache
+          type=registry,ref=onyxdotapp/onyx-backend:latest
+        cache-to: |
+          type=registry,ref=${{ inputs.runs-on-ecr-cache }}:backend-cache-${{ inputs.github-sha }},mode=max
+          type=registry,ref=${{ inputs.runs-on-ecr-cache }}:backend-cache-${{ steps.format-branch.outputs.cache-suffix }},mode=max
+          type=registry,ref=${{ inputs.runs-on-ecr-cache }}:backend-cache,mode=max
+        no-cache: ${{ inputs.docker-no-cache == 'true' }}

.github/actions/build-integration-image/action.yml

@@ -0,0 +1,75 @@
+name: "Build Integration Image"
+description: "Builds and pushes the integration test image with docker bake"
+inputs:
+  runs-on-ecr-cache:
+    description: "ECR cache registry from runs-on/action"
+    required: true
+  ref-name:
+    description: "Git ref name used for cache suffix fallback"
+    required: true
+  pr-number:
+    description: "Optional PR number for cache suffix"
+    required: false
+    default: ""
+  github-sha:
+    description: "Commit SHA used for cache keys"
+    required: true
+  run-id:
+    description: "GitHub run ID used in output image tag"
+    required: true
+  docker-username:
+    description: "Docker Hub username"
+    required: true
+  docker-token:
+    description: "Docker Hub token"
+    required: true
+runs:
+  using: "composite"
+  steps:
+    - name: Set up Docker Buildx
+      uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # ratchet:docker/setup-buildx-action@v3
+
+    - name: Login to Docker Hub
+      uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # ratchet:docker/login-action@v3
+      with:
+        username: ${{ inputs.docker-username }}
+        password: ${{ inputs.docker-token }}
+
+    - name: Format branch name for cache
+      id: format-branch
+      shell: bash
+      env:
+        PR_NUMBER: ${{ inputs.pr-number }}
+        REF_NAME: ${{ inputs.ref-name }}
+      run: |
+        if [ -n "${PR_NUMBER}" ]; then
+          CACHE_SUFFIX="${PR_NUMBER}"
+        else
+          # shellcheck disable=SC2001
+          CACHE_SUFFIX=$(echo "${REF_NAME}" | sed 's/[^A-Za-z0-9._-]/-/g')
+        fi
+        echo "cache-suffix=${CACHE_SUFFIX}" >> "$GITHUB_OUTPUT"
+
+    - name: Build and push integration test image with Docker Bake
+      shell: bash
+      env:
+        RUNS_ON_ECR_CACHE: ${{ inputs.runs-on-ecr-cache }}
+        TAG: nightly-llm-it-${{ inputs.run-id }}
+        CACHE_SUFFIX: ${{ steps.format-branch.outputs.cache-suffix }}
+        HEAD_SHA: ${{ inputs.github-sha }}
+      run: |
+        docker buildx bake --push \
+          --set backend.cache-from=type=registry,ref=${RUNS_ON_ECR_CACHE}:backend-cache-${HEAD_SHA} \
+          --set backend.cache-from=type=registry,ref=${RUNS_ON_ECR_CACHE}:backend-cache-${CACHE_SUFFIX} \
+          --set backend.cache-from=type=registry,ref=${RUNS_ON_ECR_CACHE}:backend-cache \
+          --set backend.cache-from=type=registry,ref=onyxdotapp/onyx-backend:latest \
+          --set backend.cache-to=type=registry,ref=${RUNS_ON_ECR_CACHE}:backend-cache-${HEAD_SHA},mode=max \
+          --set backend.cache-to=type=registry,ref=${RUNS_ON_ECR_CACHE}:backend-cache-${CACHE_SUFFIX},mode=max \
+          --set backend.cache-to=type=registry,ref=${RUNS_ON_ECR_CACHE}:backend-cache,mode=max \
+          --set integration.cache-from=type=registry,ref=${RUNS_ON_ECR_CACHE}:integration-cache-${HEAD_SHA} \
+          --set integration.cache-from=type=registry,ref=${RUNS_ON_ECR_CACHE}:integration-cache-${CACHE_SUFFIX} \
+          --set integration.cache-from=type=registry,ref=${RUNS_ON_ECR_CACHE}:integration-cache \
+          --set integration.cache-to=type=registry,ref=${RUNS_ON_ECR_CACHE}:integration-cache-${HEAD_SHA},mode=max \
+          --set integration.cache-to=type=registry,ref=${RUNS_ON_ECR_CACHE}:integration-cache-${CACHE_SUFFIX},mode=max \
+          --set integration.cache-to=type=registry,ref=${RUNS_ON_ECR_CACHE}:integration-cache,mode=max \
+          integration

.github/actions/build-model-server-image/action.yml

@@ -0,0 +1,68 @@
+name: "Build Model Server Image"
+description: "Builds and pushes the model server Docker image with cache reuse"
+inputs:
+  runs-on-ecr-cache:
+    description: "ECR cache registry from runs-on/action"
+    required: true
+  ref-name:
+    description: "Git ref name used for cache suffix fallback"
+    required: true
+  pr-number:
+    description: "Optional PR number for cache suffix"
+    required: false
+    default: ""
+  github-sha:
+    description: "Commit SHA used for cache keys"
+    required: true
+  run-id:
+    description: "GitHub run ID used in output image tag"
+    required: true
+  docker-username:
+    description: "Docker Hub username"
+    required: true
+  docker-token:
+    description: "Docker Hub token"
+    required: true
+runs:
+  using: "composite"
+  steps:
+    - name: Format branch name for cache
+      id: format-branch
+      shell: bash
+      env:
+        PR_NUMBER: ${{ inputs.pr-number }}
+        REF_NAME: ${{ inputs.ref-name }}
+      run: |
+        if [ -n "${PR_NUMBER}" ]; then
+          CACHE_SUFFIX="${PR_NUMBER}"
+        else
+          # shellcheck disable=SC2001
+          CACHE_SUFFIX=$(echo "${REF_NAME}" | sed 's/[^A-Za-z0-9._-]/-/g')
+        fi
+        echo "cache-suffix=${CACHE_SUFFIX}" >> "$GITHUB_OUTPUT"
+
+    - name: Set up Docker Buildx
+      uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # ratchet:docker/setup-buildx-action@v3
+
+    - name: Login to Docker Hub
+      uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # ratchet:docker/login-action@v3
+      with:
+        username: ${{ inputs.docker-username }}
+        password: ${{ inputs.docker-token }}
+
+    - name: Build and push Model Server Docker image
+      uses: docker/build-push-action@263435318d21b8e681c14492fe198d362a7d2c83 # ratchet:docker/build-push-action@v6
+      with:
+        context: ./backend
+        file: ./backend/Dockerfile.model_server
+        push: true
+        tags: ${{ inputs.runs-on-ecr-cache }}:nightly-llm-it-model-server-${{ inputs.run-id }}
+        cache-from: |
+          type=registry,ref=${{ inputs.runs-on-ecr-cache }}:model-server-cache-${{ inputs.github-sha }}
+          type=registry,ref=${{ inputs.runs-on-ecr-cache }}:model-server-cache-${{ steps.format-branch.outputs.cache-suffix }}
+          type=registry,ref=${{ inputs.runs-on-ecr-cache }}:model-server-cache
+          type=registry,ref=onyxdotapp/onyx-model-server:latest
+        cache-to: |
+          type=registry,ref=${{ inputs.runs-on-ecr-cache }}:model-server-cache-${{ inputs.github-sha }},mode=max
+          type=registry,ref=${{ inputs.runs-on-ecr-cache }}:model-server-cache-${{ steps.format-branch.outputs.cache-suffix }},mode=max
+          type=registry,ref=${{ inputs.runs-on-ecr-cache }}:model-server-cache,mode=max

.github/actions/run-nightly-provider-chat-test/action.yml

@@ -0,0 +1,120 @@
+name: "Run Nightly Provider Chat Test"
+description: "Starts required compose services and runs nightly provider integration test"
+inputs:
+  provider:
+    description: "Provider slug for NIGHTLY_LLM_PROVIDER"
+    required: true
+  models:
+    description: "Comma-separated model list for NIGHTLY_LLM_MODELS"
+    required: true
+  provider-api-key:
+    description: "API key for NIGHTLY_LLM_API_KEY"
+    required: true
+  strict:
+    description: "String true/false for NIGHTLY_LLM_STRICT"
+    required: true
+  api-base:
+    description: "Optional NIGHTLY_LLM_API_BASE"
+    required: false
+    default: ""
+  custom-config-json:
+    description: "Optional NIGHTLY_LLM_CUSTOM_CONFIG_JSON"
+    required: false
+    default: ""
+  runs-on-ecr-cache:
+    description: "ECR cache registry from runs-on/action"
+    required: true
+  run-id:
+    description: "GitHub run ID used in image tags"
+    required: true
+  docker-username:
+    description: "Docker Hub username"
+    required: true
+  docker-token:
+    description: "Docker Hub token"
+    required: true
+runs:
+  using: "composite"
+  steps:
+    - name: Login to Docker Hub
+      uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # ratchet:docker/login-action@v3
+      with:
+        username: ${{ inputs.docker-username }}
+        password: ${{ inputs.docker-token }}
+
+    - name: Create .env file for Docker Compose
+      shell: bash
+      env:
+        ECR_CACHE: ${{ inputs.runs-on-ecr-cache }}
+        RUN_ID: ${{ inputs.run-id }}
+      run: |
+        cat <<EOF2 > deployment/docker_compose/.env
+        COMPOSE_PROFILES=s3-filestore
+        ENABLE_PAID_ENTERPRISE_EDITION_FEATURES=true
+        LICENSE_ENFORCEMENT_ENABLED=false
+        AUTH_TYPE=basic
+        POSTGRES_POOL_PRE_PING=true
+        POSTGRES_USE_NULL_POOL=true
+        REQUIRE_EMAIL_VERIFICATION=false
+        DISABLE_TELEMETRY=true
+        INTEGRATION_TESTS_MODE=true
+        AUTO_LLM_UPDATE_INTERVAL_SECONDS=10
+        ONYX_BACKEND_IMAGE=${ECR_CACHE}:nightly-llm-it-backend-${RUN_ID}
+        ONYX_MODEL_SERVER_IMAGE=${ECR_CACHE}:nightly-llm-it-model-server-${RUN_ID}
+        EOF2
+
+    - name: Start Docker containers
+      shell: bash
+      run: |
+        cd deployment/docker_compose
+        docker compose -f docker-compose.yml -f docker-compose.dev.yml up -d --wait \
+          relational_db \
+          index \
+          cache \
+          minio \
+          api_server \
+          inference_model_server
+
+    - name: Run nightly provider integration test
+      uses: nick-fields/retry@ce71cc2ab81d554ebbe88c79ab5975992d79ba08 # ratchet:nick-fields/retry@v3
+      env:
+        MODELS: ${{ inputs.models }}
+        NIGHTLY_LLM_PROVIDER: ${{ inputs.provider }}
+        NIGHTLY_LLM_API_KEY: ${{ inputs.provider-api-key }}
+        NIGHTLY_LLM_API_BASE: ${{ inputs.api-base }}
+        NIGHTLY_LLM_CUSTOM_CONFIG_JSON: ${{ inputs.custom-config-json }}
+        NIGHTLY_LLM_STRICT: ${{ inputs.strict }}
+        RUNS_ON_ECR_CACHE: ${{ inputs.runs-on-ecr-cache }}
+        RUN_ID: ${{ inputs.run-id }}
+      with:
+        timeout_minutes: 20
+        max_attempts: 2
+        retry_wait_seconds: 10
+        command: |
+          if [ -z "${MODELS}" ]; then
+            echo "Input 'models' must be non-empty for provider '${NIGHTLY_LLM_PROVIDER}'."
+            exit 1
+          fi
+
+          docker run --rm --network onyx_default \
+            --name test-runner \
+            -e POSTGRES_HOST=relational_db \
+            -e POSTGRES_USER=postgres \
+            -e POSTGRES_PASSWORD=password \
+            -e POSTGRES_DB=postgres \
+            -e DB_READONLY_USER=db_readonly_user \
+            -e DB_READONLY_PASSWORD=password \
+            -e POSTGRES_POOL_PRE_PING=true \
+            -e POSTGRES_USE_NULL_POOL=true \
+            -e VESPA_HOST=index \
+            -e REDIS_HOST=cache \
+            -e API_SERVER_HOST=api_server \
+            -e TEST_WEB_HOSTNAME=test-runner \
+            -e NIGHTLY_LLM_PROVIDER="${NIGHTLY_LLM_PROVIDER}" \
+            -e NIGHTLY_LLM_MODELS="${MODELS}" \
+            -e NIGHTLY_LLM_API_KEY="${NIGHTLY_LLM_API_KEY}" \
+            -e NIGHTLY_LLM_API_BASE="${NIGHTLY_LLM_API_BASE}" \
+            -e NIGHTLY_LLM_CUSTOM_CONFIG_JSON="${NIGHTLY_LLM_CUSTOM_CONFIG_JSON}" \
+            -e NIGHTLY_LLM_STRICT="${NIGHTLY_LLM_STRICT}" \
+            ${RUNS_ON_ECR_CACHE}:nightly-llm-it-${RUN_ID} \
+            /app/tests/integration/tests/llm_workflows/test_nightly_provider_chat_workflow.py

.github/pull_request_template.md

@@ -8,5 +8,5 @@
 
 ## Additional Options
 
-- [ ] [Required] I have considered whether this PR needs to be cherry-picked to the latest beta branch.
+- [ ] [Optional] Please cherry-pick this PR to the latest release version.
 - [ ] [Optional] Override Linear Check

.github/workflows/helm-chart-releases.yml

@@ -33,7 +33,7 @@ jobs:
           helm repo add cloudnative-pg https://cloudnative-pg.github.io/charts
           helm repo add ot-container-kit https://ot-container-kit.github.io/helm-charts
           helm repo add minio https://charts.min.io/
-          helm repo add code-interpreter https://onyx-dot-app.github.io/code-interpreter/
+          helm repo add code-interpreter https://onyx-dot-app.github.io/python-sandbox/
           helm repo update
 
       - name: Build chart dependencies

.github/workflows/nightly-llm-provider-chat-openai.yml

@@ -0,0 +1,44 @@
+name: Nightly LLM Provider Chat Tests (OpenAI)
+concurrency:
+  group: Nightly-LLM-Provider-Chat-OpenAI-${{ github.workflow }}-${{ github.ref_name }}
+  cancel-in-progress: true
+
+on:
+  schedule:
+    # Runs daily at 10:30 UTC (2:30 AM PST / 3:30 AM PDT)
+    - cron: "30 10 * * *"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  openai-provider-chat-test:
+    uses: ./.github/workflows/reusable-nightly-llm-provider-chat.yml
+    with:
+      provider: openai
+      models: ${{ vars.NIGHTLY_LLM_OPENAI_MODELS }}
+      strict: true
+    secrets:
+      provider_api_key: ${{ secrets.OPENAI_API_KEY }}
+      DOCKER_USERNAME: ${{ secrets.DOCKER_USERNAME }}
+      DOCKER_TOKEN: ${{ secrets.DOCKER_TOKEN }}
+
+  notify-slack-on-failure:
+    needs: [openai-provider-chat-test]
+    if: failure() && github.event_name == 'schedule'
+    runs-on: ubuntu-slim
+    timeout-minutes: 5
+    steps:
+      - name: Checkout
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # ratchet:actions/checkout@v6
+        with:
+          persist-credentials: false
+
+      - name: Send Slack notification
+        uses: ./.github/actions/slack-notify
+        with:
+          webhook-url: ${{ secrets.SLACK_WEBHOOK }}
+          failed-jobs: openai-provider-chat-test
+          title: "🚨 Scheduled OpenAI Provider Chat Tests failed!"
+          ref-name: ${{ github.ref_name }}

.github/workflows/post-merge-beta-cherry-pick.yml

@@ -0,0 +1,161 @@
+name: Post-Merge Beta Cherry-Pick
+
+on:
+  push:
+    branches:
+      - main
+
+permissions:
+  contents: write
+  pull-requests: write
+
+jobs:
+  cherry-pick-to-latest-release:
+    outputs:
+      should_cherrypick: ${{ steps.gate.outputs.should_cherrypick }}
+      pr_number: ${{ steps.gate.outputs.pr_number }}
+      cherry_pick_reason: ${{ steps.run_cherry_pick.outputs.reason }}
+      cherry_pick_details: ${{ steps.run_cherry_pick.outputs.details }}
+    runs-on: ubuntu-latest
+    timeout-minutes: 45
+    steps:
+      - name: Resolve merged PR and checkbox state
+        id: gate
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          # For the commit that triggered this workflow (HEAD on main), fetch all
+          # associated PRs and keep only the PR that was actually merged into main
+          # with this exact merge commit SHA.
+          pr_numbers="$(gh api "repos/${GITHUB_REPOSITORY}/commits/${GITHUB_SHA}/pulls" | jq -r --arg sha "${GITHUB_SHA}" '.[] | select(.merged_at != null and .base.ref == "main" and .merge_commit_sha == $sha) | .number')"
+          match_count="$(printf '%s\n' "$pr_numbers" | sed '/^[[:space:]]*$/d' | wc -l | tr -d ' ')"
+          pr_number="$(printf '%s\n' "$pr_numbers" | sed '/^[[:space:]]*$/d' | head -n 1)"
+
+          if [ "${match_count}" -gt 1 ]; then
+            echo "::warning::Multiple merged PRs matched commit ${GITHUB_SHA}. Using PR #${pr_number}."
+          fi
+
+          if [ -z "$pr_number" ]; then
+            echo "No merged PR associated with commit ${GITHUB_SHA}; skipping."
+            echo "should_cherrypick=false" >> "$GITHUB_OUTPUT"
+            exit 0
+          fi
+
+          # Read the PR once so we can gate behavior and infer preferred actor.
+          pr_json="$(gh api "repos/${GITHUB_REPOSITORY}/pulls/${pr_number}")"
+          pr_body="$(printf '%s' "$pr_json" | jq -r '.body // ""')"
+          merged_by="$(printf '%s' "$pr_json" | jq -r '.merged_by.login // ""')"
+
+          echo "pr_number=$pr_number" >> "$GITHUB_OUTPUT"
+          echo "merged_by=$merged_by" >> "$GITHUB_OUTPUT"
+
+          if echo "$pr_body" | grep -qiE "\\[x\\][[:space:]]*(\\[[^]]+\\][[:space:]]*)?Please cherry-pick this PR to the latest release version"; then
+            echo "should_cherrypick=true" >> "$GITHUB_OUTPUT"
+            echo "Cherry-pick checkbox checked for PR #${pr_number}."
+            exit 0
+          fi
+
+          echo "should_cherrypick=false" >> "$GITHUB_OUTPUT"
+          echo "Cherry-pick checkbox not checked for PR #${pr_number}. Skipping."
+
+      - name: Checkout repository
+        if: steps.gate.outputs.should_cherrypick == 'true'
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # ratchet:actions/checkout@v6
+        with:
+          fetch-depth: 0
+          persist-credentials: true
+          ref: main
+
+      - name: Install the latest version of uv
+        if: steps.gate.outputs.should_cherrypick == 'true'
+        uses: astral-sh/setup-uv@61cb8a9741eeb8a550a1b8544337180c0fc8476b # ratchet:astral-sh/setup-uv@v7
+        with:
+          enable-cache: false
+          version: "0.9.9"
+
+      - name: Configure git identity
+        if: steps.gate.outputs.should_cherrypick == 'true'
+        run: |
+          git config user.name "github-actions[bot]"
+          git config user.email "github-actions[bot]@users.noreply.github.com"
+
+      - name: Create cherry-pick PR to latest release
+        id: run_cherry_pick
+        if: steps.gate.outputs.should_cherrypick == 'true'
+        continue-on-error: true
+        env:
+          GH_TOKEN: ${{ github.token }}
+          GITHUB_TOKEN: ${{ github.token }}
+          CHERRY_PICK_ASSIGNEE: ${{ steps.gate.outputs.merged_by }}
+        run: |
+          set -o pipefail
+          output_file="$(mktemp)"
+          uv run --no-sync --with onyx-devtools ods cherry-pick "${GITHUB_SHA}" --yes --no-verify 2>&1 | tee "$output_file"
+          exit_code="${PIPESTATUS[0]}"
+
+          if [ "${exit_code}" -eq 0 ]; then
+            echo "status=success" >> "$GITHUB_OUTPUT"
+            exit 0
+          fi
+
+          echo "status=failure" >> "$GITHUB_OUTPUT"
+
+          reason="command-failed"
+          if grep -qiE "merge conflict during cherry-pick|CONFLICT|could not apply|cherry-pick in progress with staged changes" "$output_file"; then
+            reason="merge-conflict"
+          fi
+          echo "reason=${reason}" >> "$GITHUB_OUTPUT"
+
+          {
+            echo "details<<EOF"
+            tail -n 40 "$output_file"
+            echo "EOF"
+          } >> "$GITHUB_OUTPUT"
+
+      - name: Mark workflow as failed if cherry-pick failed
+        if: steps.gate.outputs.should_cherrypick == 'true' && steps.run_cherry_pick.outputs.status == 'failure'
+        run: |
+          echo "::error::Automated cherry-pick failed (${{ steps.run_cherry_pick.outputs.reason }})."
+          exit 1
+
+  notify-slack-on-cherry-pick-failure:
+    needs:
+      - cherry-pick-to-latest-release
+    if: always() && needs.cherry-pick-to-latest-release.outputs.should_cherrypick == 'true' && needs.cherry-pick-to-latest-release.result != 'success'
+    runs-on: ubuntu-slim
+    timeout-minutes: 10
+    steps:
+      - name: Checkout
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # ratchet:actions/checkout@v6
+        with:
+          persist-credentials: false
+
+      - name: Build cherry-pick failure summary
+        id: failure-summary
+        env:
+          SOURCE_PR_NUMBER: ${{ needs.cherry-pick-to-latest-release.outputs.pr_number }}
+          CHERRY_PICK_REASON: ${{ needs.cherry-pick-to-latest-release.outputs.cherry_pick_reason }}
+          CHERRY_PICK_DETAILS: ${{ needs.cherry-pick-to-latest-release.outputs.cherry_pick_details }}
+        run: |
+          source_pr_url="https://github.com/${GITHUB_REPOSITORY}/pull/${SOURCE_PR_NUMBER}"
+
+          reason_text="cherry-pick command failed"
+          if [ "${CHERRY_PICK_REASON}" = "merge-conflict" ]; then
+            reason_text="merge conflict during cherry-pick"
+          fi
+
+          details_excerpt="$(printf '%s' "${CHERRY_PICK_DETAILS}" | tail -n 8 | tr '\n' ' ' | sed "s/[[:space:]]\\+/ /g" | sed "s/\"/'/g" | cut -c1-350)"
+          failed_jobs="• cherry-pick-to-latest-release\\n• source PR: ${source_pr_url}\\n• reason: ${reason_text}"
+          if [ -n "${details_excerpt}" ]; then
+            failed_jobs="${failed_jobs}\\n• excerpt: ${details_excerpt}"
+          fi
+
+          echo "jobs=${failed_jobs}" >> "$GITHUB_OUTPUT"
+
+      - name: Notify #cherry-pick-prs about cherry-pick failure
+        uses: ./.github/actions/slack-notify
+        with:
+          webhook-url: ${{ secrets.CHERRY_PICK_PRS_WEBHOOK }}
+          failed-jobs: ${{ steps.failure-summary.outputs.jobs }}
+          title: "🚨 Automated Cherry-Pick Failed"
+          ref-name: ${{ github.ref_name }}

.github/workflows/pr-beta-cherrypick-check.yml

@@ -1,28 +0,0 @@
-name: Require beta cherry-pick consideration
-concurrency:
-  group: Require-Beta-Cherrypick-Consideration-${{ github.workflow }}-${{ github.head_ref || github.event.workflow_run.head_branch || github.run_id }}
-  cancel-in-progress: true
-
-on:
-  pull_request:
-    types: [opened, edited, reopened, synchronize]
-
-permissions:
-  contents: read
-
-jobs:
-  beta-cherrypick-check:
-    runs-on: ubuntu-latest
-    timeout-minutes: 45
-    steps:
-      - name: Check PR body for beta cherry-pick consideration
-        env:
-          PR_BODY: ${{ github.event.pull_request.body }}
-        run: |
-          if echo "$PR_BODY" | grep -qiE "\\[x\\][[:space:]]*\\[Required\\][[:space:]]*I have considered whether this PR needs to be cherry[- ]picked to the latest beta branch"; then
-            echo "Cherry-pick consideration box is checked. Check passed."
-            exit 0
-          fi
-
-          echo "::error::Please check the 'I have considered whether this PR needs to be cherry-picked to the latest beta branch' box in the PR description."
-          exit 1

.github/workflows/pr-external-dependency-unit-tests.yml

@@ -45,9 +45,6 @@ env:
   # TODO: debug why this is failing and enable
   CODE_INTERPRETER_BASE_URL: http://localhost:8000
 
-  # OpenSearch
-  OPENSEARCH_ADMIN_PASSWORD: "StrongPassword123!"
-
 jobs:
   discover-test-dirs:
     # NOTE: Github-hosted runners have about 20s faster queue times and are preferred here.
@@ -118,9 +115,9 @@ jobs:
       - name: Create .env file for Docker Compose
         run: |
           cat <<EOF > deployment/docker_compose/.env
-          COMPOSE_PROFILES=s3-filestore
-          CODE_INTERPRETER_BETA_ENABLED=true
+          COMPOSE_PROFILES=s3-filestore,opensearch-enabled
           DISABLE_TELEMETRY=true
+          OPENSEARCH_FOR_ONYX_ENABLED=true
           EOF
 
       - name: Set up Standard Dependencies
@@ -129,7 +126,6 @@ jobs:
           docker compose \
             -f docker-compose.yml \
             -f docker-compose.dev.yml \
-            -f docker-compose.opensearch.yml \
             up -d \
             minio \
             relational_db \

.github/workflows/pr-helm-chart-testing.yml

@@ -91,7 +91,7 @@ jobs:
           helm repo add cloudnative-pg https://cloudnative-pg.github.io/charts
           helm repo add ot-container-kit https://ot-container-kit.github.io/helm-charts
           helm repo add minio https://charts.min.io/
-          helm repo add code-interpreter https://onyx-dot-app.github.io/code-interpreter/
+          helm repo add code-interpreter https://onyx-dot-app.github.io/python-sandbox/
           helm repo update
 
       - name: Install Redis operator

.github/workflows/pr-integration-tests.yml

@@ -20,6 +20,7 @@ env:
   # Test Environment Variables
   OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
   SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}
+  SLACK_BOT_TOKEN_TEST_SPACE: ${{ secrets.SLACK_BOT_TOKEN_TEST_SPACE }}
   CONFLUENCE_TEST_SPACE_URL: ${{ vars.CONFLUENCE_TEST_SPACE_URL }}
   CONFLUENCE_USER_NAME: ${{ vars.CONFLUENCE_USER_NAME }}
   CONFLUENCE_ACCESS_TOKEN: ${{ secrets.CONFLUENCE_ACCESS_TOKEN }}
@@ -423,6 +424,7 @@ jobs:
               -e OPENAI_API_KEY=${OPENAI_API_KEY} \
               -e EXA_API_KEY=${EXA_API_KEY} \
               -e SLACK_BOT_TOKEN=${SLACK_BOT_TOKEN} \
+              -e SLACK_BOT_TOKEN_TEST_SPACE=${SLACK_BOT_TOKEN_TEST_SPACE} \
               -e CONFLUENCE_TEST_SPACE_URL=${CONFLUENCE_TEST_SPACE_URL} \
               -e CONFLUENCE_USER_NAME=${CONFLUENCE_USER_NAME} \
               -e CONFLUENCE_ACCESS_TOKEN=${CONFLUENCE_ACCESS_TOKEN} \
@@ -443,6 +445,7 @@ jobs:
               -e TEST_WEB_HOSTNAME=test-runner \
               -e MOCK_CONNECTOR_SERVER_HOST=mock_connector_server \
               -e MOCK_CONNECTOR_SERVER_PORT=8001 \
+              -e ENABLE_PAID_ENTERPRISE_EDITION_FEATURES=${{ matrix.edition == 'ee' && 'true' || 'false' }} \
               ${{ env.RUNS_ON_ECR_CACHE }}:integration-test-${{ github.run_id }} \
               /app/tests/integration/${{ matrix.test-dir.path }}
 
@@ -701,6 +704,7 @@ jobs:
             -e OPENAI_API_KEY=${OPENAI_API_KEY} \
             -e EXA_API_KEY=${EXA_API_KEY} \
             -e SLACK_BOT_TOKEN=${SLACK_BOT_TOKEN} \
+            -e SLACK_BOT_TOKEN_TEST_SPACE=${SLACK_BOT_TOKEN_TEST_SPACE} \
             -e TEST_WEB_HOSTNAME=test-runner \
             -e AUTH_TYPE=cloud \
             -e MULTI_TENANT=true \

.github/workflows/reusable-nightly-llm-provider-chat.yml

@@ -0,0 +1,206 @@
+name: Reusable Nightly LLM Provider Chat Tests
+
+on:
+  workflow_call:
+    inputs:
+      provider:
+        description: "Provider slug passed to NIGHTLY_LLM_PROVIDER (e.g. openai, anthropic)"
+        required: true
+        type: string
+      models:
+        description: "Comma-separated model list passed to NIGHTLY_LLM_MODELS"
+        required: true
+        type: string
+      strict:
+        description: "Pass-through value for NIGHTLY_LLM_STRICT"
+        required: false
+        default: true
+        type: boolean
+      api_base:
+        description: "Optional NIGHTLY_LLM_API_BASE override"
+        required: false
+        default: ""
+        type: string
+      custom_config_json:
+        description: "Optional NIGHTLY_LLM_CUSTOM_CONFIG_JSON override"
+        required: false
+        default: ""
+        type: string
+    secrets:
+      provider_api_key:
+        description: "Provider API key passed to NIGHTLY_LLM_API_KEY"
+        required: true
+      DOCKER_USERNAME:
+        required: true
+      DOCKER_TOKEN:
+        required: true
+
+permissions:
+  contents: read
+
+env:
+  NIGHTLY_LLM_PROVIDER: ${{ inputs.provider }}
+  NIGHTLY_LLM_MODELS: ${{ inputs.models }}
+  NIGHTLY_LLM_API_KEY: ${{ secrets.provider_api_key }}
+  NIGHTLY_LLM_API_BASE: ${{ inputs.api_base }}
+  NIGHTLY_LLM_CUSTOM_CONFIG_JSON: ${{ inputs.custom_config_json }}
+  NIGHTLY_LLM_STRICT: ${{ inputs.strict && 'true' || 'false' }}
+
+jobs:
+  validate-inputs:
+    # NOTE: Keep this cheap and fail before image builds if required inputs are missing.
+    runs-on: ubuntu-slim
+    timeout-minutes: 5
+    steps:
+      - name: Validate required nightly provider inputs
+        run: |
+          if [ -z "${NIGHTLY_LLM_MODELS}" ]; then
+            echo "Input 'models' must be non-empty for provider '${NIGHTLY_LLM_PROVIDER}'."
+            exit 1
+          fi
+
+  build-backend-image:
+    needs: [validate-inputs]
+    runs-on:
+      [
+        runs-on,
+        runner=1cpu-linux-arm64,
+        "run-id=${{ github.run_id }}-build-backend-image",
+        "extras=ecr-cache",
+      ]
+    timeout-minutes: 45
+    steps:
+      - uses: runs-on/action@cd2b598b0515d39d78c38a02d529db87d2196d1e # ratchet:runs-on/action@v2
+
+      - name: Checkout code
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # ratchet:actions/checkout@v6
+        with:
+          persist-credentials: false
+
+      - name: Build backend image
+        uses: ./.github/actions/build-backend-image
+        with:
+          runs-on-ecr-cache: ${{ env.RUNS_ON_ECR_CACHE }}
+          ref-name: ${{ github.ref_name }}
+          pr-number: ${{ github.event.pull_request.number }}
+          github-sha: ${{ github.sha }}
+          run-id: ${{ github.run_id }}
+          docker-username: ${{ secrets.DOCKER_USERNAME }}
+          docker-token: ${{ secrets.DOCKER_TOKEN }}
+          docker-no-cache: ${{ vars.DOCKER_NO_CACHE == 'true' && 'true' || 'false' }}
+
+  build-model-server-image:
+    needs: [validate-inputs]
+    runs-on:
+      [
+        runs-on,
+        runner=1cpu-linux-arm64,
+        "run-id=${{ github.run_id }}-build-model-server-image",
+        "extras=ecr-cache",
+      ]
+    timeout-minutes: 45
+    steps:
+      - uses: runs-on/action@cd2b598b0515d39d78c38a02d529db87d2196d1e # ratchet:runs-on/action@v2
+
+      - name: Checkout code
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # ratchet:actions/checkout@v6
+        with:
+          persist-credentials: false
+
+      - name: Build model server image
+        uses: ./.github/actions/build-model-server-image
+        with:
+          runs-on-ecr-cache: ${{ env.RUNS_ON_ECR_CACHE }}
+          ref-name: ${{ github.ref_name }}
+          pr-number: ${{ github.event.pull_request.number }}
+          github-sha: ${{ github.sha }}
+          run-id: ${{ github.run_id }}
+          docker-username: ${{ secrets.DOCKER_USERNAME }}
+          docker-token: ${{ secrets.DOCKER_TOKEN }}
+
+  build-integration-image:
+    needs: [validate-inputs]
+    runs-on:
+      [
+        runs-on,
+        runner=2cpu-linux-arm64,
+        "run-id=${{ github.run_id }}-build-integration-image",
+        "extras=ecr-cache",
+      ]
+    timeout-minutes: 45
+    steps:
+      - uses: runs-on/action@cd2b598b0515d39d78c38a02d529db87d2196d1e # ratchet:runs-on/action@v2
+
+      - name: Checkout code
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # ratchet:actions/checkout@v6
+        with:
+          persist-credentials: false
+
+      - name: Build integration image
+        uses: ./.github/actions/build-integration-image
+        with:
+          runs-on-ecr-cache: ${{ env.RUNS_ON_ECR_CACHE }}
+          ref-name: ${{ github.ref_name }}
+          pr-number: ${{ github.event.pull_request.number }}
+          github-sha: ${{ github.sha }}
+          run-id: ${{ github.run_id }}
+          docker-username: ${{ secrets.DOCKER_USERNAME }}
+          docker-token: ${{ secrets.DOCKER_TOKEN }}
+
+  provider-chat-test:
+    needs:
+      [build-backend-image, build-model-server-image, build-integration-image]
+    runs-on:
+      - runs-on
+      - runner=4cpu-linux-arm64
+      - "run-id=${{ github.run_id }}-nightly-${{ inputs.provider }}-provider-chat-test"
+      - extras=ecr-cache
+    timeout-minutes: 45
+    steps:
+      - uses: runs-on/action@cd2b598b0515d39d78c38a02d529db87d2196d1e # ratchet:runs-on/action@v2
+
+      - name: Checkout code
+        uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # ratchet:actions/checkout@v6
+        with:
+          persist-credentials: false
+
+      - name: Run nightly provider chat test
+        uses: ./.github/actions/run-nightly-provider-chat-test
+        with:
+          provider: ${{ env.NIGHTLY_LLM_PROVIDER }}
+          models: ${{ env.NIGHTLY_LLM_MODELS }}
+          provider-api-key: ${{ secrets.provider_api_key }}
+          strict: ${{ env.NIGHTLY_LLM_STRICT }}
+          api-base: ${{ env.NIGHTLY_LLM_API_BASE }}
+          custom-config-json: ${{ env.NIGHTLY_LLM_CUSTOM_CONFIG_JSON }}
+          runs-on-ecr-cache: ${{ env.RUNS_ON_ECR_CACHE }}
+          run-id: ${{ github.run_id }}
+          docker-username: ${{ secrets.DOCKER_USERNAME }}
+          docker-token: ${{ secrets.DOCKER_TOKEN }}
+
+      - name: Dump API server logs
+        if: always()
+        run: |
+          cd deployment/docker_compose
+          docker compose logs --no-color api_server > $GITHUB_WORKSPACE/api_server.log || true
+
+      - name: Dump all-container logs
+        if: always()
+        run: |
+          cd deployment/docker_compose
+          docker compose logs --no-color > $GITHUB_WORKSPACE/docker-compose.log || true
+
+      - name: Upload logs
+        if: always()
+        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f
+        with:
+          name: docker-all-logs-nightly-${{ inputs.provider }}-llm-provider
+          path: |
+            ${{ github.workspace }}/api_server.log
+            ${{ github.workspace }}/docker-compose.log
+
+      - name: Stop Docker containers
+        if: always()
+        run: |
+          cd deployment/docker_compose
+          docker compose down -v

AGENTS.md

@@ -548,7 +548,7 @@ class in the utils over directly calling the APIs with a library like `requests`
 calling the utilities directly (e.g. do NOT create admin users with
 `admin_user = UserManager.create(name="admin_user")`, instead use the `admin_user` fixture).
 
-A great example of this type of test is `backend/tests/integration/dev_apis/test_simple_chat_api.py`.
+A great example of this type of test is `backend/tests/integration/tests/streaming_endpoints/test_chat_stream.py`.
 
 To run them:
 
@@ -616,3 +616,9 @@ This is a minimal list - feel free to include more. Do NOT write code as part of
 Keep it high level. You can reference certain files or functions though.
 
 Before writing your plan, make sure to do research. Explore the relevant sections in the codebase.
+
+## Best Practices
+
+In addition to the other content in this file, best practices for contributing
+to the codebase can be found at `contributing_guides/best_practices.md`.
+Understand its contents and follow them.

backend/alembic/run_multitenant_migrations.py

@@ -21,15 +21,14 @@ import sys
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor, as_completed
-from typing import List, NamedTuple
+from typing import NamedTuple
 
 from alembic.config import Config
 from alembic.script import ScriptDirectory
-from sqlalchemy import text
 
-from onyx.db.engine.sql_engine import is_valid_schema_name
 from onyx.db.engine.sql_engine import SqlEngine
 from onyx.db.engine.tenant_utils import get_all_tenant_ids
+from onyx.db.engine.tenant_utils import get_schemas_needing_migration
 from shared_configs.configs import TENANT_ID_PREFIX
 
 
@@ -105,56 +104,6 @@ def get_head_revision() -> str | None:
     return script.get_current_head()
 
 
-def get_schemas_needing_migration(
-    tenant_schemas: List[str], head_rev: str
-) -> List[str]:
-    """Return only schemas whose current alembic version is not at head."""
-    if not tenant_schemas:
-        return []
-
-    engine = SqlEngine.get_engine()
-
-    with engine.connect() as conn:
-        # Find which schemas actually have an alembic_version table
-        rows = conn.execute(
-            text(
-                "SELECT table_schema FROM information_schema.tables "
-                "WHERE table_name = 'alembic_version' "
-                "AND table_schema = ANY(:schemas)"
-            ),
-            {"schemas": tenant_schemas},
-        )
-        schemas_with_table = set(row[0] for row in rows)
-
-        # Schemas without the table definitely need migration
-        needs_migration = [s for s in tenant_schemas if s not in schemas_with_table]
-
-        if not schemas_with_table:
-            return needs_migration
-
-        # Validate schema names before interpolating into SQL
-        for schema in schemas_with_table:
-            if not is_valid_schema_name(schema):
-                raise ValueError(f"Invalid schema name: {schema}")
-
-        # Single query to get every schema's current revision at once.
-        # Use integer tags instead of interpolating schema names into
-        # string literals to avoid quoting issues.
-        schema_list = list(schemas_with_table)
-        union_parts = [
-            f'SELECT {i} AS idx, version_num FROM "{schema}".alembic_version'
-            for i, schema in enumerate(schema_list)
-        ]
-        rows = conn.execute(text(" UNION ALL ".join(union_parts)))
-        version_by_schema = {schema_list[row[0]]: row[1] for row in rows}
-
-        needs_migration.extend(
-            s for s in schemas_with_table if version_by_schema.get(s) != head_rev
-        )
-
-    return needs_migration
-
-
 def run_migrations_parallel(
     schemas: list[str],
     max_workers: int,

backend/alembic/versions/07b98176f1de_code_interpreter_seed.py

@@ -0,0 +1,29 @@
+"""code interpreter seed
+
+Revision ID: 07b98176f1de
+Revises: 7cb492013621
+Create Date: 2026-02-23 15:55:07.606784
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "07b98176f1de"
+down_revision = "7cb492013621"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # Seed the single instance of code_interpreter_server
+    # NOTE: There should only exist at most and at minimum 1 code_interpreter_server row
+    op.execute(
+        sa.text("INSERT INTO code_interpreter_server (server_enabled) VALUES (true)")
+    )
+
+
+def downgrade() -> None:
+    op.execute(sa.text("DELETE FROM code_interpreter_server"))

backend/alembic/versions/0bb4558f35df_add_scim_username_to_scim_user_mapping.py

@@ -0,0 +1,28 @@
+"""add scim_username to scim_user_mapping
+
+Revision ID: 0bb4558f35df
+Revises: 631fd2504136
+Create Date: 2026-02-20 10:45:30.340188
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "0bb4558f35df"
+down_revision = "631fd2504136"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "scim_user_mapping",
+        sa.Column("scim_username", sa.String(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("scim_user_mapping", "scim_username")

backend/alembic/versions/7616121f6e97_add_enterprise_fields_to_scim_user_mapping.py

@@ -0,0 +1,48 @@
+"""add enterprise and name fields to scim_user_mapping
+
+Revision ID: 7616121f6e97
+Revises: 07b98176f1de
+Create Date: 2026-02-23 12:00:00.000000
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "7616121f6e97"
+down_revision = "07b98176f1de"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "scim_user_mapping",
+        sa.Column("department", sa.String(), nullable=True),
+    )
+    op.add_column(
+        "scim_user_mapping",
+        sa.Column("manager", sa.String(), nullable=True),
+    )
+    op.add_column(
+        "scim_user_mapping",
+        sa.Column("given_name", sa.String(), nullable=True),
+    )
+    op.add_column(
+        "scim_user_mapping",
+        sa.Column("family_name", sa.String(), nullable=True),
+    )
+    op.add_column(
+        "scim_user_mapping",
+        sa.Column("scim_emails_json", sa.Text(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("scim_user_mapping", "scim_emails_json")
+    op.drop_column("scim_user_mapping", "family_name")
+    op.drop_column("scim_user_mapping", "given_name")
+    op.drop_column("scim_user_mapping", "manager")
+    op.drop_column("scim_user_mapping", "department")

backend/alembic/versions/7cb492013621_code_interpreter_server_model.py

@@ -0,0 +1,31 @@
+"""code interpreter server model
+
+Revision ID: 7cb492013621
+Revises: 0bb4558f35df
+Create Date: 2026-02-22 18:54:54.007265
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "7cb492013621"
+down_revision = "0bb4558f35df"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "code_interpreter_server",
+        sa.Column("id", sa.Integer, primary_key=True),
+        sa.Column(
+            "server_enabled", sa.Boolean, nullable=False, server_default=sa.true()
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("code_interpreter_server")

backend/alembic/versions/8ffcc2bcfc11_add_needs_persona_sync_to_user_file.py

@@ -0,0 +1,33 @@
+"""add needs_persona_sync to user_file
+
+Revision ID: 8ffcc2bcfc11
+Revises: 7616121f6e97
+Create Date: 2026-02-23 10:48:48.343826
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "8ffcc2bcfc11"
+down_revision = "7616121f6e97"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "user_file",
+        sa.Column(
+            "needs_persona_sync",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.text("false"),
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("user_file", "needs_persona_sync")

backend/ee/onyx/db/scim.py

@@ -34,6 +34,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from ee.onyx.server.scim.filtering import ScimFilter
 from ee.onyx.server.scim.filtering import ScimFilterOperator
+from ee.onyx.server.scim.models import ScimMappingFields
 from onyx.db.dal import DAL
 from onyx.db.models import ScimGroupMapping
 from onyx.db.models import ScimToken
@@ -127,9 +128,21 @@ class ScimDAL(DAL):
         self,
         external_id: str,
         user_id: UUID,
+        scim_username: str | None = None,
+        fields: ScimMappingFields | None = None,
     ) -> ScimUserMapping:
         """Create a mapping between a SCIM externalId and an Onyx user."""
-        mapping = ScimUserMapping(external_id=external_id, user_id=user_id)
+        f = fields or ScimMappingFields()
+        mapping = ScimUserMapping(
+            external_id=external_id,
+            user_id=user_id,
+            scim_username=scim_username,
+            department=f.department,
+            manager=f.manager,
+            given_name=f.given_name,
+            family_name=f.family_name,
+            scim_emails_json=f.scim_emails_json,
+        )
         self._session.add(mapping)
         self._session.flush()
         return mapping
@@ -248,11 +261,11 @@ class ScimDAL(DAL):
         scim_filter: ScimFilter | None,
         start_index: int = 1,
         count: int = 100,
-    ) -> tuple[list[tuple[User, str | None]], int]:
+    ) -> tuple[list[tuple[User, ScimUserMapping | None]], int]:
         """Query users with optional SCIM filter and pagination.
 
         Returns:
-            A tuple of (list of (user, external_id) pairs, total_count).
+            A tuple of (list of (user, mapping) pairs, total_count).
 
         Raises:
             ValueError: If the filter uses an unsupported attribute.
@@ -292,33 +305,117 @@ class ScimDAL(DAL):
         users = list(
             self._session.scalars(
                 query.order_by(User.id).offset(offset).limit(count)  # type: ignore[arg-type]
-            ).all()
+            )
+            .unique()
+            .all()
         )
 
-        # Batch-fetch external IDs to avoid N+1 queries
-        ext_id_map = self._get_user_external_ids([u.id for u in users])
-        return [(u, ext_id_map.get(u.id)) for u in users], total
+        # Batch-fetch SCIM mappings to avoid N+1 queries
+        mapping_map = self._get_user_mappings_batch([u.id for u in users])
+        return [(u, mapping_map.get(u.id)) for u in users], total
 
-    def sync_user_external_id(self, user_id: UUID, new_external_id: str | None) -> None:
-        """Create, update, or delete the external ID mapping for a user."""
+    def sync_user_external_id(
+        self,
+        user_id: UUID,
+        new_external_id: str | None,
+        scim_username: str | None = None,
+        fields: ScimMappingFields | None = None,
+    ) -> None:
+        """Create, update, or delete the external ID mapping for a user.
+
+        When *fields* is provided, all mapping fields are written
+        unconditionally — including ``None`` values — so that a caller can
+        clear a previously-set field (e.g. removing a department).
+        """
         mapping = self.get_user_mapping_by_user_id(user_id)
         if new_external_id:
             if mapping:
                 if mapping.external_id != new_external_id:
                     mapping.external_id = new_external_id
+                if scim_username is not None:
+                    mapping.scim_username = scim_username
+                if fields is not None:
+                    mapping.department = fields.department
+                    mapping.manager = fields.manager
+                    mapping.given_name = fields.given_name
+                    mapping.family_name = fields.family_name
+                    mapping.scim_emails_json = fields.scim_emails_json
             else:
-                self.create_user_mapping(external_id=new_external_id, user_id=user_id)
+                self.create_user_mapping(
+                    external_id=new_external_id,
+                    user_id=user_id,
+                    scim_username=scim_username,
+                    fields=fields,
+                )
         elif mapping:
             self.delete_user_mapping(mapping.id)
 
-    def _get_user_external_ids(self, user_ids: list[UUID]) -> dict[UUID, str]:
-        """Batch-fetch external IDs for a list of user IDs."""
+    def _get_user_mappings_batch(
+        self, user_ids: list[UUID]
+    ) -> dict[UUID, ScimUserMapping]:
+        """Batch-fetch SCIM user mappings keyed by user ID."""
         if not user_ids:
             return {}
         mappings = self._session.scalars(
             select(ScimUserMapping).where(ScimUserMapping.user_id.in_(user_ids))
         ).all()
-        return {m.user_id: m.external_id for m in mappings}
+        return {m.user_id: m for m in mappings}
+
+    def get_user_groups(self, user_id: UUID) -> list[tuple[int, str]]:
+        """Get groups a user belongs to as ``(group_id, group_name)`` pairs.
+
+        Excludes groups marked for deletion.
+        """
+        rels = self._session.scalars(
+            select(User__UserGroup).where(User__UserGroup.user_id == user_id)
+        ).all()
+
+        group_ids = [r.user_group_id for r in rels]
+        if not group_ids:
+            return []
+
+        groups = self._session.scalars(
+            select(UserGroup).where(
+                UserGroup.id.in_(group_ids),
+                UserGroup.is_up_for_deletion.is_(False),
+            )
+        ).all()
+        return [(g.id, g.name) for g in groups]
+
+    def get_users_groups_batch(
+        self, user_ids: list[UUID]
+    ) -> dict[UUID, list[tuple[int, str]]]:
+        """Batch-fetch group memberships for multiple users.
+
+        Returns a mapping of ``user_id → [(group_id, group_name), ...]``.
+        Avoids N+1 queries when building user list responses.
+        """
+        if not user_ids:
+            return {}
+
+        rels = self._session.scalars(
+            select(User__UserGroup).where(User__UserGroup.user_id.in_(user_ids))
+        ).all()
+
+        group_ids = list({r.user_group_id for r in rels})
+        if not group_ids:
+            return {}
+
+        groups = self._session.scalars(
+            select(UserGroup).where(
+                UserGroup.id.in_(group_ids),
+                UserGroup.is_up_for_deletion.is_(False),
+            )
+        ).all()
+        groups_by_id = {g.id: g.name for g in groups}
+
+        result: dict[UUID, list[tuple[int, str]]] = {}
+        for r in rels:
+            if r.user_id and r.user_group_id in groups_by_id:
+                result.setdefault(r.user_id, []).append(
+                    (r.user_group_id, groups_by_id[r.user_group_id])
+                )
+        return result
 
     # ------------------------------------------------------------------
     # Group mapping operations
@@ -483,9 +580,13 @@ class ScimDAL(DAL):
         if not user_ids:
             return []
 
-        users = self._session.scalars(
-            select(User).where(User.id.in_(user_ids))  # type: ignore[attr-defined]
-        ).all()
+        users = (
+            self._session.scalars(
+                select(User).where(User.id.in_(user_ids))  # type: ignore[attr-defined]
+            )
+            .unique()
+            .all()
+        )
         users_by_id = {u.id: u for u in users}
 
         return [
@@ -504,9 +605,13 @@ class ScimDAL(DAL):
         """
         if not uuids:
             return []
-        existing_users = self._session.scalars(
-            select(User).where(User.id.in_(uuids))  # type: ignore[attr-defined]
-        ).all()
+        existing_users = (
+            self._session.scalars(
+                select(User).where(User.id.in_(uuids))  # type: ignore[attr-defined]
+            )
+            .unique()
+            .all()
+        )
         existing_ids = {u.id for u in existing_users}
         return [uid for uid in uuids if uid not in existing_ids]
 

backend/ee/onyx/db/user_group.py

@@ -9,6 +9,7 @@ from sqlalchemy import Select
 from sqlalchemy import select
 from sqlalchemy import update
 from sqlalchemy.dialects.postgresql import insert
+from sqlalchemy.orm import selectinload
 from sqlalchemy.orm import Session
 
 from ee.onyx.server.user_group.models import SetCuratorRequest
@@ -18,11 +19,15 @@ from onyx.db.connector_credential_pair import get_connector_credential_pair_from
 from onyx.db.enums import AccessType
 from onyx.db.enums import ConnectorCredentialPairStatus
 from onyx.db.models import ConnectorCredentialPair
+from onyx.db.models import Credential
 from onyx.db.models import Credential__UserGroup
 from onyx.db.models import Document
 from onyx.db.models import DocumentByConnectorCredentialPair
+from onyx.db.models import DocumentSet
 from onyx.db.models import DocumentSet__UserGroup
+from onyx.db.models import FederatedConnector__DocumentSet
 from onyx.db.models import LLMProvider__UserGroup
+from onyx.db.models import Persona
 from onyx.db.models import Persona__UserGroup
 from onyx.db.models import TokenRateLimit__UserGroup
 from onyx.db.models import User
@@ -195,8 +200,60 @@ def fetch_user_group(db_session: Session, user_group_id: int) -> UserGroup | Non
     return db_session.scalar(stmt)
 
 
+def _add_user_group_snapshot_eager_loads(
+    stmt: Select,
+) -> Select:
+    """Add eager loading options needed by UserGroup.from_model snapshot creation."""
+    return stmt.options(
+        selectinload(UserGroup.users),
+        selectinload(UserGroup.user_group_relationships),
+        selectinload(UserGroup.cc_pair_relationships)
+        .selectinload(UserGroup__ConnectorCredentialPair.cc_pair)
+        .options(
+            selectinload(ConnectorCredentialPair.connector),
+            selectinload(ConnectorCredentialPair.credential).selectinload(
+                Credential.user
+            ),
+        ),
+        selectinload(UserGroup.document_sets).options(
+            selectinload(DocumentSet.connector_credential_pairs).selectinload(
+                ConnectorCredentialPair.connector
+            ),
+            selectinload(DocumentSet.users),
+            selectinload(DocumentSet.groups),
+            selectinload(DocumentSet.federated_connectors).selectinload(
+                FederatedConnector__DocumentSet.federated_connector
+            ),
+        ),
+        selectinload(UserGroup.personas).options(
+            selectinload(Persona.tools),
+            selectinload(Persona.hierarchy_nodes),
+            selectinload(Persona.attached_documents).selectinload(
+                Document.parent_hierarchy_node
+            ),
+            selectinload(Persona.labels),
+            selectinload(Persona.document_sets).options(
+                selectinload(DocumentSet.connector_credential_pairs).selectinload(
+                    ConnectorCredentialPair.connector
+                ),
+                selectinload(DocumentSet.users),
+                selectinload(DocumentSet.groups),
+                selectinload(DocumentSet.federated_connectors).selectinload(
+                    FederatedConnector__DocumentSet.federated_connector
+                ),
+            ),
+            selectinload(Persona.user),
+            selectinload(Persona.user_files),
+            selectinload(Persona.users),
+            selectinload(Persona.groups),
+        ),
+    )
+
+
 def fetch_user_groups(
-    db_session: Session, only_up_to_date: bool = True
+    db_session: Session,
+    only_up_to_date: bool = True,
+    eager_load_for_snapshot: bool = False,
 ) -> Sequence[UserGroup]:
     """
     Fetches user groups from the database.
@@ -209,6 +266,8 @@ def fetch_user_groups(
         db_session (Session): The SQLAlchemy session used to query the database.
         only_up_to_date (bool, optional): Flag to determine whether to filter the results
             to include only up to date user groups. Defaults to `True`.
+        eager_load_for_snapshot: If True, adds eager loading for all relationships
+            needed by UserGroup.from_model snapshot creation.
 
     Returns:
         Sequence[UserGroup]: A sequence of `UserGroup` objects matching the query criteria.
@@ -216,11 +275,16 @@ def fetch_user_groups(
     stmt = select(UserGroup)
     if only_up_to_date:
         stmt = stmt.where(UserGroup.is_up_to_date == True)  # noqa: E712
-    return db_session.scalars(stmt).all()
+    if eager_load_for_snapshot:
+        stmt = _add_user_group_snapshot_eager_loads(stmt)
+    return db_session.scalars(stmt).unique().all()
 
 
 def fetch_user_groups_for_user(
-    db_session: Session, user_id: UUID, only_curator_groups: bool = False
+    db_session: Session,
+    user_id: UUID,
+    only_curator_groups: bool = False,
+    eager_load_for_snapshot: bool = False,
 ) -> Sequence[UserGroup]:
     stmt = (
         select(UserGroup)
@@ -230,7 +294,9 @@ def fetch_user_groups_for_user(
     )
     if only_curator_groups:
         stmt = stmt.where(User__UserGroup.is_curator == True)  # noqa: E712
-    return db_session.scalars(stmt).all()
+    if eager_load_for_snapshot:
+        stmt = _add_user_group_snapshot_eager_loads(stmt)
+    return db_session.scalars(stmt).unique().all()
 
 
 def construct_document_id_select_by_usergroup(

backend/ee/onyx/external_permissions/sharepoint/group_sync.py

@@ -1,9 +1,13 @@
 from collections.abc import Generator
 
+from office365.sharepoint.client_context import ClientContext  # type: ignore[import-untyped]
+
 from ee.onyx.db.external_perm import ExternalUserGroup
 from ee.onyx.external_permissions.sharepoint.permission_utils import (
     get_sharepoint_external_groups,
 )
+from onyx.configs.app_configs import SHAREPOINT_EXHAUSTIVE_AD_ENUMERATION
+from onyx.connectors.sharepoint.connector import acquire_token_for_rest
 from onyx.connectors.sharepoint.connector import SharepointConnector
 from onyx.db.models import ConnectorCredentialPair
 from onyx.utils.logger import setup_logger
@@ -43,14 +47,27 @@ def sharepoint_group_sync(
 
     logger.info(f"Processing {len(site_descriptors)} sites for group sync")
 
-    # Process each site
+    enumerate_all = connector_config.get(
+        "exhaustive_ad_enumeration", SHAREPOINT_EXHAUSTIVE_AD_ENUMERATION
+    )
+
+    msal_app = connector.msal_app
+    sp_tenant_domain = connector.sp_tenant_domain
+    sp_domain_suffix = connector.sharepoint_domain_suffix
     for site_descriptor in site_descriptors:
         logger.debug(f"Processing site: {site_descriptor.url}")
 
-        ctx = connector._create_rest_client_context(site_descriptor.url)
+        ctx = ClientContext(site_descriptor.url).with_access_token(
+            lambda: acquire_token_for_rest(msal_app, sp_tenant_domain, sp_domain_suffix)
+        )
 
-        # Get external groups for this site
-        external_groups = get_sharepoint_external_groups(ctx, connector.graph_client)
+        external_groups = get_sharepoint_external_groups(
+            ctx,
+            connector.graph_client,
+            graph_api_base=connector.graph_api_base,
+            get_access_token=connector._get_graph_access_token,
+            enumerate_all_ad_groups=enumerate_all,
+        )
 
         # Yield each group
         for group in external_groups:

backend/ee/onyx/external_permissions/sharepoint/permission_utils.py

@@ -1,9 +1,12 @@
 import re
+import time
 from collections import deque
+from collections.abc import Callable
+from collections.abc import Generator
 from typing import Any
-from urllib.parse import unquote
 from urllib.parse import urlparse
 
+import requests as _requests
 from office365.graph_client import GraphClient  # type: ignore[import-untyped]
 from office365.onedrive.driveitems.driveItem import DriveItem  # type: ignore[import-untyped]
 from office365.runtime.client_request import ClientRequestException  # type: ignore
@@ -14,7 +17,10 @@ from pydantic import BaseModel
 from ee.onyx.db.external_perm import ExternalUserGroup
 from onyx.access.models import ExternalAccess
 from onyx.access.utils import build_ext_group_name_for_onyx
+from onyx.configs.app_configs import REQUEST_TIMEOUT_SECONDS
 from onyx.configs.constants import DocumentSource
+from onyx.connectors.sharepoint.connector import GRAPH_API_MAX_RETRIES
+from onyx.connectors.sharepoint.connector import GRAPH_API_RETRYABLE_STATUSES
 from onyx.connectors.sharepoint.connector import SHARED_DOCUMENTS_MAP_REVERSE
 from onyx.connectors.sharepoint.connector import sleep_and_retry
 from onyx.utils.logger import setup_logger
@@ -33,6 +39,70 @@ LIMITED_ACCESS_ROLE_TYPES = [1, 9]
 LIMITED_ACCESS_ROLE_NAMES = ["Limited Access", "Web-Only Limited Access"]
 
 
+AD_GROUP_ENUMERATION_THRESHOLD = 100_000
+
+
+def _graph_api_get(
+    url: str,
+    get_access_token: Callable[[], str],
+    params: dict[str, str] | None = None,
+) -> dict[str, Any]:
+    """Authenticated Graph API GET with retry on transient errors."""
+    for attempt in range(GRAPH_API_MAX_RETRIES + 1):
+        access_token = get_access_token()
+        headers = {"Authorization": f"Bearer {access_token}"}
+        try:
+            resp = _requests.get(
+                url, headers=headers, params=params, timeout=REQUEST_TIMEOUT_SECONDS
+            )
+            if (
+                resp.status_code in GRAPH_API_RETRYABLE_STATUSES
+                and attempt < GRAPH_API_MAX_RETRIES
+            ):
+                wait = min(int(resp.headers.get("Retry-After", str(2**attempt))), 60)
+                logger.warning(
+                    f"Graph API {resp.status_code} on attempt {attempt + 1}, "
+                    f"retrying in {wait}s: {url}"
+                )
+                time.sleep(wait)
+                continue
+            resp.raise_for_status()
+            return resp.json()
+        except (_requests.ConnectionError, _requests.Timeout, _requests.HTTPError):
+            if attempt < GRAPH_API_MAX_RETRIES:
+                wait = min(2**attempt, 60)
+                logger.warning(
+                    f"Graph API connection error on attempt {attempt + 1}, "
+                    f"retrying in {wait}s: {url}"
+                )
+                time.sleep(wait)
+                continue
+            raise
+    raise RuntimeError(
+        f"Graph API request failed after {GRAPH_API_MAX_RETRIES + 1} attempts: {url}"
+    )
+
+
+def _iter_graph_collection(
+    initial_url: str,
+    get_access_token: Callable[[], str],
+    params: dict[str, str] | None = None,
+) -> Generator[dict[str, Any], None, None]:
+    """Paginate through a Graph API collection, yielding items one at a time."""
+    url: str | None = initial_url
+    while url:
+        data = _graph_api_get(url, get_access_token, params)
+        params = None
+        yield from data.get("value", [])
+        url = data.get("@odata.nextLink")
+
+
+def _normalize_email(email: str) -> str:
+    if MICROSOFT_DOMAIN in email:
+        return email.replace(MICROSOFT_DOMAIN, "")
+    return email
+
+
 class SharepointGroup(BaseModel):
     model_config = {"frozen": True}
 
@@ -527,8 +597,12 @@ def get_external_access_from_sharepoint(
         )
     elif site_page:
         site_url = site_page.get("webUrl")
-        # Prefer server-relative URL to avoid OData filters that break on apostrophes
-        server_relative_url = unquote(urlparse(site_url).path)
+        # Keep percent-encoding intact so the path matches the encoding
+        # used by the Office365 library's SPResPath.create_relative(),
+        # which compares against urlparse(context.base_url).path.
+        # Decoding (e.g. %27 → ') causes a mismatch that duplicates
+        # the site prefix in the constructed URL.
+        server_relative_url = urlparse(site_url).path
         file_obj = client_context.web.get_file_by_server_relative_url(
             server_relative_url
         )
@@ -572,8 +646,65 @@ def get_external_access_from_sharepoint(
     )
 
 
+def _enumerate_ad_groups_paginated(
+    get_access_token: Callable[[], str],
+    already_resolved: set[str],
+    graph_api_base: str,
+) -> Generator[ExternalUserGroup, None, None]:
+    """Paginate through all Azure AD groups and yield ExternalUserGroup for each.
+
+    Skips groups whose suffixed name is already in *already_resolved*.
+    Stops early if the number of groups exceeds AD_GROUP_ENUMERATION_THRESHOLD.
+    """
+    groups_url = f"{graph_api_base}/groups"
+    groups_params: dict[str, str] = {"$select": "id,displayName", "$top": "999"}
+    total_groups = 0
+
+    for group_json in _iter_graph_collection(
+        groups_url, get_access_token, groups_params
+    ):
+        group_id: str = group_json.get("id", "")
+        display_name: str = group_json.get("displayName", "")
+        if not group_id or not display_name:
+            continue
+
+        total_groups += 1
+        if total_groups > AD_GROUP_ENUMERATION_THRESHOLD:
+            logger.warning(
+                f"Azure AD group enumeration exceeded {AD_GROUP_ENUMERATION_THRESHOLD} "
+                "groups — stopping to avoid excessive memory/API usage. "
+                "Remaining groups will be resolved from role assignments only."
+            )
+            return
+
+        name = f"{display_name}_{group_id}"
+        if name in already_resolved:
+            continue
+
+        member_emails: list[str] = []
+        members_url = f"{graph_api_base}/groups/{group_id}/members"
+        members_params: dict[str, str] = {
+            "$select": "userPrincipalName,mail",
+            "$top": "999",
+        }
+        for member_json in _iter_graph_collection(
+            members_url, get_access_token, members_params
+        ):
+            email = member_json.get("userPrincipalName") or member_json.get("mail")
+            if email:
+                member_emails.append(_normalize_email(email))
+
+        yield ExternalUserGroup(id=name, user_emails=member_emails)
+
+    logger.info(f"Enumerated {total_groups} Azure AD groups via paginated Graph API")
+
+
 def get_sharepoint_external_groups(
-    client_context: ClientContext, graph_client: GraphClient
+    client_context: ClientContext,
+    graph_client: GraphClient,
+    graph_api_base: str,
+    get_access_token: Callable[[], str] | None = None,
+    enumerate_all_ad_groups: bool = False,
 ) -> list[ExternalUserGroup]:
 
     groups: set[SharepointGroup] = set()
@@ -629,57 +760,22 @@ def get_sharepoint_external_groups(
         client_context, graph_client, groups, is_group_sync=True
     )
 
-    # get all Azure AD groups because if any group is assigned to the drive item, we don't want to miss them
-    # We can't assign sharepoint groups to drive items or drives, so we don't need to get all sharepoint groups
-    azure_ad_groups = sleep_and_retry(
-        graph_client.groups.get_all(page_loaded=lambda _: None),
-        "get_sharepoint_external_groups:get_azure_ad_groups",
-    )
-    logger.info(f"Azure AD Groups: {len(azure_ad_groups)}")
-    identified_groups: set[str] = set(groups_and_members.groups_to_emails.keys())
-    ad_groups_to_emails: dict[str, set[str]] = {}
-    for group in azure_ad_groups:
-        # If the group is already identified, we don't need to get the members
-        if group.display_name in identified_groups:
-            continue
-        # AD groups allows same display name for multiple groups, so we need to add the GUID to the name
-        name = group.display_name
-        name = _get_group_name_with_suffix(group.id, name, graph_client)
+    external_user_groups: list[ExternalUserGroup] = [
+        ExternalUserGroup(id=group_name, user_emails=list(emails))
+        for group_name, emails in groups_and_members.groups_to_emails.items()
+    ]
 
-        members = sleep_and_retry(
-            group.members.get_all(page_loaded=lambda _: None),
-            "get_sharepoint_external_groups:get_azure_ad_groups:get_members",
+    if not enumerate_all_ad_groups or get_access_token is None:
+        logger.info(
+            "Skipping exhaustive Azure AD group enumeration. "
+            "Only groups found in site role assignments are included."
         )
-        for member in members:
-            member_data = member.to_json()
-            user_principal_name = member_data.get("userPrincipalName")
-            mail = member_data.get("mail")
-            if not ad_groups_to_emails.get(name):
-                ad_groups_to_emails[name] = set()
-            if user_principal_name:
-                if MICROSOFT_DOMAIN in user_principal_name:
-                    user_principal_name = user_principal_name.replace(
-                        MICROSOFT_DOMAIN, ""
-                    )
-                ad_groups_to_emails[name].add(user_principal_name)
-            elif mail:
-                if MICROSOFT_DOMAIN in mail:
-                    mail = mail.replace(MICROSOFT_DOMAIN, "")
-                ad_groups_to_emails[name].add(mail)
+        return external_user_groups
 
-    external_user_groups: list[ExternalUserGroup] = []
-    for group_name, emails in groups_and_members.groups_to_emails.items():
-        external_user_group = ExternalUserGroup(
-            id=group_name,
-            user_emails=list(emails),
-        )
-        external_user_groups.append(external_user_group)
-
-    for group_name, emails in ad_groups_to_emails.items():
-        external_user_group = ExternalUserGroup(
-            id=group_name,
-            user_emails=list(emails),
-        )
-        external_user_groups.append(external_user_group)
+    already_resolved = set(groups_and_members.groups_to_emails.keys())
+    for group in _enumerate_ad_groups_paginated(
+        get_access_token, already_resolved, graph_api_base
+    ):
+        external_user_groups.append(group)
 
     return external_user_groups

backend/ee/onyx/server/query_and_chat/models.py

@@ -34,7 +34,7 @@ class SendSearchQueryRequest(BaseModel):
     filters: BaseFilters | None = None
     num_docs_fed_to_llm_selection: int | None = None
     run_query_expansion: bool = False
-    num_hits: int = 50
+    num_hits: int = 30
 
     include_content: bool = False
     stream: bool = False

backend/ee/onyx/server/scim/api.py

@@ -26,21 +26,23 @@ from sqlalchemy.orm import Session
 from ee.onyx.db.scim import ScimDAL
 from ee.onyx.server.scim.auth import verify_scim_token
 from ee.onyx.server.scim.filtering import parse_scim_filter
-from ee.onyx.server.scim.models import ScimEmail
+from ee.onyx.server.scim.models import SCIM_LIST_RESPONSE_SCHEMA
 from ee.onyx.server.scim.models import ScimError
 from ee.onyx.server.scim.models import ScimGroupMember
 from ee.onyx.server.scim.models import ScimGroupResource
 from ee.onyx.server.scim.models import ScimListResponse
-from ee.onyx.server.scim.models import ScimMeta
+from ee.onyx.server.scim.models import ScimMappingFields
 from ee.onyx.server.scim.models import ScimName
 from ee.onyx.server.scim.models import ScimPatchRequest
-from ee.onyx.server.scim.models import ScimResourceType
-from ee.onyx.server.scim.models import ScimSchemaDefinition
 from ee.onyx.server.scim.models import ScimServiceProviderConfig
 from ee.onyx.server.scim.models import ScimUserResource
 from ee.onyx.server.scim.patch import apply_group_patch
 from ee.onyx.server.scim.patch import apply_user_patch
 from ee.onyx.server.scim.patch import ScimPatchError
+from ee.onyx.server.scim.providers.base import get_default_provider
+from ee.onyx.server.scim.providers.base import ScimProvider
+from ee.onyx.server.scim.providers.base import serialize_emails
+from ee.onyx.server.scim.schema_definitions import ENTERPRISE_USER_SCHEMA_DEF
 from ee.onyx.server.scim.schema_definitions import GROUP_RESOURCE_TYPE
 from ee.onyx.server.scim.schema_definitions import GROUP_SCHEMA_DEF
 from ee.onyx.server.scim.schema_definitions import SERVICE_PROVIDER_CONFIG
@@ -48,21 +50,45 @@ from ee.onyx.server.scim.schema_definitions import USER_RESOURCE_TYPE
 from ee.onyx.server.scim.schema_definitions import USER_SCHEMA_DEF
 from onyx.db.engine.sql_engine import get_session
 from onyx.db.models import ScimToken
+from onyx.db.models import ScimUserMapping
 from onyx.db.models import User
 from onyx.db.models import UserGroup
 from onyx.db.models import UserRole
+from onyx.utils.logger import setup_logger
 from onyx.utils.variable_functionality import fetch_ee_implementation_or_noop
 
+logger = setup_logger()
+
+
+class ScimJSONResponse(JSONResponse):
+    """JSONResponse with Content-Type: application/scim+json (RFC 7644 §3.1)."""
+
+    media_type = "application/scim+json"
+
 
 # NOTE: All URL paths in this router (/ServiceProviderConfig, /ResourceTypes,
 # /Schemas, /Users, /Groups) are mandated by the SCIM spec (RFC 7643/7644).
 # IdPs like Okta and Azure AD hardcode these exact paths, so they cannot be
 # changed to kebab-case.
+
+
 scim_router = APIRouter(prefix="/scim/v2", tags=["SCIM"])
 
 _pw_helper = PasswordHelper()
 
 
+def _get_provider(
+    _token: ScimToken = Depends(verify_scim_token),
+) -> ScimProvider:
+    """Resolve the SCIM provider for the current request.
+
+    Currently returns OktaProvider for all requests. When multi-provider
+    support is added (ENG-3652), this will resolve based on token metadata
+    or tenant configuration — no endpoint changes required.
+    """
+    return get_default_provider()
+
+
 # ---------------------------------------------------------------------------
 # Service Discovery Endpoints (unauthenticated)
 # ---------------------------------------------------------------------------
@@ -75,15 +101,39 @@ def get_service_provider_config() -> ScimServiceProviderConfig:
 
 
 @scim_router.get("/ResourceTypes")
-def get_resource_types() -> list[ScimResourceType]:
-    """List available SCIM resource types (RFC 7643 §6)."""
-    return [USER_RESOURCE_TYPE, GROUP_RESOURCE_TYPE]
+def get_resource_types() -> ScimJSONResponse:
+    """List available SCIM resource types (RFC 7643 §6).
+
+    Wrapped in a ListResponse envelope (RFC 7644 §3.4.2) because IdPs
+    like Entra ID expect a JSON object, not a bare array.
+    """
+    resources = [USER_RESOURCE_TYPE, GROUP_RESOURCE_TYPE]
+    return ScimJSONResponse(
+        content={
+            "schemas": [SCIM_LIST_RESPONSE_SCHEMA],
+            "totalResults": len(resources),
+            "Resources": [
+                r.model_dump(exclude_none=True, by_alias=True) for r in resources
+            ],
+        }
+    )
 
 
 @scim_router.get("/Schemas")
-def get_schemas() -> list[ScimSchemaDefinition]:
-    """Return SCIM schema definitions (RFC 7643 §7)."""
-    return [USER_SCHEMA_DEF, GROUP_SCHEMA_DEF]
+def get_schemas() -> ScimJSONResponse:
+    """Return SCIM schema definitions (RFC 7643 §7).
+
+    Wrapped in a ListResponse envelope (RFC 7644 §3.4.2) because IdPs
+    like Entra ID expect a JSON object, not a bare array.
+    """
+    schemas = [USER_SCHEMA_DEF, GROUP_SCHEMA_DEF, ENTERPRISE_USER_SCHEMA_DEF]
+    return ScimJSONResponse(
+        content={
+            "schemas": [SCIM_LIST_RESPONSE_SCHEMA],
+            "totalResults": len(schemas),
+            "Resources": [s.model_dump(exclude_none=True) for s in schemas],
+        }
+    )
 
 
 # ---------------------------------------------------------------------------
@@ -91,35 +141,43 @@ def get_schemas() -> list[ScimSchemaDefinition]:
 # ---------------------------------------------------------------------------
 
 
-def _scim_error_response(status: int, detail: str) -> JSONResponse:
+def _scim_error_response(status: int, detail: str) -> ScimJSONResponse:
     """Build a SCIM-compliant error response (RFC 7644 §3.12)."""
+    logger.warning("SCIM error response: status=%s detail=%s", status, detail)
     body = ScimError(status=str(status), detail=detail)
-    return JSONResponse(
+    return ScimJSONResponse(
         status_code=status,
         content=body.model_dump(exclude_none=True),
     )
 
 
-def _user_to_scim(user: User, external_id: str | None = None) -> ScimUserResource:
-    """Convert an Onyx User to a SCIM User resource representation."""
-    name = None
-    if user.personal_name:
-        parts = user.personal_name.split(" ", 1)
-        name = ScimName(
-            givenName=parts[0],
-            familyName=parts[1] if len(parts) > 1 else None,
-            formatted=user.personal_name,
-        )
+def _parse_excluded_attributes(raw: str | None) -> set[str]:
+    """Parse the ``excludedAttributes`` query parameter (RFC 7644 §3.4.2.5).
 
-    return ScimUserResource(
-        id=str(user.id),
-        externalId=external_id,
-        userName=user.email,
-        name=name,
-        emails=[ScimEmail(value=user.email, type="work", primary=True)],
-        active=user.is_active,
-        meta=ScimMeta(resourceType="User"),
-    )
+    Returns a set of lowercased attribute names to omit from responses.
+    """
+    if not raw:
+        return set()
+    return {attr.strip().lower() for attr in raw.split(",") if attr.strip()}
+
+
+def _apply_exclusions(
+    resource: ScimUserResource | ScimGroupResource,
+    excluded: set[str],
+) -> dict:
+    """Serialize a SCIM resource, omitting attributes the IdP excluded.
+
+    RFC 7644 §3.4.2.5 lets the IdP pass ``?excludedAttributes=groups,emails``
+    to reduce response payload size. We strip those fields after serialization
+    so the rest of the pipeline doesn't need to know about them.
+    """
+    data = resource.model_dump(exclude_none=True, by_alias=True)
+    for attr in excluded:
+        # Match case-insensitively against the camelCase field names
+        keys_to_remove = [k for k in data if k.lower() == attr]
+        for k in keys_to_remove:
+            del data[k]
+    return data
 
 
 def _check_seat_availability(dal: ScimDAL) -> str | None:
@@ -135,7 +193,7 @@ def _check_seat_availability(dal: ScimDAL) -> str | None:
     return None
 
 
-def _fetch_user_or_404(user_id: str, dal: ScimDAL) -> User | JSONResponse:
+def _fetch_user_or_404(user_id: str, dal: ScimDAL) -> User | ScimJSONResponse:
     """Parse *user_id* as UUID, look up the user, or return a 404 error."""
     try:
         uid = UUID(user_id)
@@ -155,8 +213,94 @@ def _scim_name_to_str(name: ScimName | None) -> str | None:
     """
     if not name:
         return None
-    return name.formatted or " ".join(
-        part for part in [name.givenName, name.familyName] if part
+    # If the client explicitly provides ``formatted``, prefer it — the client
+    # knows what display string it wants. Otherwise build from components.
+    if name.formatted:
+        return name.formatted
+    parts = " ".join(part for part in [name.givenName, name.familyName] if part)
+    return parts or None
+
+
+def _scim_resource_response(
+    resource: ScimUserResource | ScimGroupResource | ScimListResponse,
+    status_code: int = 200,
+) -> ScimJSONResponse:
+    """Serialize a SCIM resource as ``application/scim+json``."""
+    content = resource.model_dump(exclude_none=True, by_alias=True)
+    return ScimJSONResponse(
+        status_code=status_code,
+        content=content,
+    )
+
+
+def _build_list_response(
+    resources: list[ScimUserResource | ScimGroupResource],
+    total: int,
+    start_index: int,
+    count: int,
+    excluded: set[str] | None = None,
+) -> ScimListResponse | ScimJSONResponse:
+    """Build a SCIM list response, optionally applying attribute exclusions.
+
+    RFC 7644 §3.4.2.5 — IdPs may request certain attributes be omitted via
+    the ``excludedAttributes`` query parameter.
+    """
+    if excluded:
+        envelope = ScimListResponse(
+            totalResults=total,
+            startIndex=start_index,
+            itemsPerPage=count,
+        )
+        data = envelope.model_dump(exclude_none=True)
+        data["Resources"] = [_apply_exclusions(r, excluded) for r in resources]
+        return ScimJSONResponse(content=data)
+
+    return _scim_resource_response(
+        ScimListResponse(
+            totalResults=total,
+            startIndex=start_index,
+            itemsPerPage=count,
+            Resources=resources,
+        )
+    )
+
+
+def _extract_enterprise_fields(
+    resource: ScimUserResource,
+) -> tuple[str | None, str | None]:
+    """Extract department and manager from enterprise extension."""
+    ext = resource.enterprise_extension
+    if not ext:
+        return None, None
+    department = ext.department
+    manager = ext.manager.value if ext.manager else None
+    return department, manager
+
+
+def _mapping_to_fields(
+    mapping: ScimUserMapping | None,
+) -> ScimMappingFields | None:
+    """Extract round-trip fields from a SCIM user mapping."""
+    if not mapping:
+        return None
+    return ScimMappingFields(
+        department=mapping.department,
+        manager=mapping.manager,
+        given_name=mapping.given_name,
+        family_name=mapping.family_name,
+        scim_emails_json=mapping.scim_emails_json,
+    )
+
+
+def _fields_from_resource(resource: ScimUserResource) -> ScimMappingFields:
+    """Build mapping fields from an incoming SCIM user resource."""
+    department, manager = _extract_enterprise_fields(resource)
+    return ScimMappingFields(
+        department=department,
+        manager=manager,
+        given_name=resource.name.givenName if resource.name else None,
+        family_name=resource.name.familyName if resource.name else None,
+        scim_emails_json=serialize_emails(resource.emails),
     )
 
 
@@ -168,11 +312,13 @@ def _scim_name_to_str(name: ScimName | None) -> str | None:
 @scim_router.get("/Users", response_model=None)
 def list_users(
     filter: str | None = Query(None),
+    excludedAttributes: str | None = None,
     startIndex: int = Query(1, ge=1),
     count: int = Query(100, ge=0, le=500),
     _token: ScimToken = Depends(verify_scim_token),
+    provider: ScimProvider = Depends(_get_provider),
     db_session: Session = Depends(get_session),
-) -> ScimListResponse | JSONResponse:
+) -> ScimListResponse | ScimJSONResponse:
     """List users with optional SCIM filter and pagination."""
     dal = ScimDAL(db_session)
     dal.update_token_last_used(_token.id)
@@ -183,52 +329,78 @@ def list_users(
         return _scim_error_response(400, str(e))
 
     try:
-        users_with_ext_ids, total = dal.list_users(scim_filter, startIndex, count)
+        users_with_mappings, total = dal.list_users(scim_filter, startIndex, count)
     except ValueError as e:
         return _scim_error_response(400, str(e))
 
+    user_groups_map = dal.get_users_groups_batch([u.id for u, _ in users_with_mappings])
     resources: list[ScimUserResource | ScimGroupResource] = [
-        _user_to_scim(user, ext_id) for user, ext_id in users_with_ext_ids
+        provider.build_user_resource(
+            user,
+            mapping.external_id if mapping else None,
+            groups=user_groups_map.get(user.id, []),
+            scim_username=mapping.scim_username if mapping else None,
+            fields=_mapping_to_fields(mapping),
+        )
+        for user, mapping in users_with_mappings
     ]
 
-    return ScimListResponse(
-        totalResults=total,
-        startIndex=startIndex,
-        itemsPerPage=count,
-        Resources=resources,
+    return _build_list_response(
+        resources,
+        total,
+        startIndex,
+        count,
+        excluded=_parse_excluded_attributes(excludedAttributes),
     )
 
 
 @scim_router.get("/Users/{user_id}", response_model=None)
 def get_user(
     user_id: str,
+    excludedAttributes: str | None = None,
     _token: ScimToken = Depends(verify_scim_token),
+    provider: ScimProvider = Depends(_get_provider),
     db_session: Session = Depends(get_session),
-) -> ScimUserResource | JSONResponse:
+) -> ScimUserResource | ScimJSONResponse:
     """Get a single user by ID."""
     dal = ScimDAL(db_session)
     dal.update_token_last_used(_token.id)
 
     result = _fetch_user_or_404(user_id, dal)
-    if isinstance(result, JSONResponse):
+    if isinstance(result, ScimJSONResponse):
         return result
     user = result
 
     mapping = dal.get_user_mapping_by_user_id(user.id)
-    return _user_to_scim(user, mapping.external_id if mapping else None)
+
+    resource = provider.build_user_resource(
+        user,
+        mapping.external_id if mapping else None,
+        groups=dal.get_user_groups(user.id),
+        scim_username=mapping.scim_username if mapping else None,
+        fields=_mapping_to_fields(mapping),
+    )
+
+    # RFC 7644 §3.4.2.5 — IdP may request certain attributes be omitted
+    excluded = _parse_excluded_attributes(excludedAttributes)
+    if excluded:
+        return ScimJSONResponse(content=_apply_exclusions(resource, excluded))
+
+    return _scim_resource_response(resource)
 
 
 @scim_router.post("/Users", status_code=201, response_model=None)
 def create_user(
     user_resource: ScimUserResource,
     _token: ScimToken = Depends(verify_scim_token),
+    provider: ScimProvider = Depends(_get_provider),
     db_session: Session = Depends(get_session),
-) -> ScimUserResource | JSONResponse:
+) -> ScimUserResource | ScimJSONResponse:
     """Create a new user from a SCIM provisioning request."""
     dal = ScimDAL(db_session)
     dal.update_token_last_used(_token.id)
 
-    email = user_resource.userName.strip().lower()
+    email = user_resource.userName.strip()
 
     # externalId is how the IdP correlates this user on subsequent requests.
     # Without it, the IdP can't find the user and will try to re-create,
@@ -264,11 +436,26 @@ def create_user(
 
     # Create SCIM mapping (externalId is validated above, always present)
     external_id = user_resource.externalId
-    dal.create_user_mapping(external_id=external_id, user_id=user.id)
+    scim_username = user_resource.userName.strip()
+    fields = _fields_from_resource(user_resource)
+    dal.create_user_mapping(
+        external_id=external_id,
+        user_id=user.id,
+        scim_username=scim_username,
+        fields=fields,
+    )
 
     dal.commit()
 
-    return _user_to_scim(user, external_id)
+    return _scim_resource_response(
+        provider.build_user_resource(
+            user,
+            external_id,
+            scim_username=scim_username,
+            fields=fields,
+        ),
+        status_code=201,
+    )
 
 
 @scim_router.put("/Users/{user_id}", response_model=None)
@@ -276,14 +463,15 @@ def replace_user(
     user_id: str,
     user_resource: ScimUserResource,
     _token: ScimToken = Depends(verify_scim_token),
+    provider: ScimProvider = Depends(_get_provider),
     db_session: Session = Depends(get_session),
-) -> ScimUserResource | JSONResponse:
+) -> ScimUserResource | ScimJSONResponse:
     """Replace a user entirely (RFC 7644 §3.5.1)."""
     dal = ScimDAL(db_session)
     dal.update_token_last_used(_token.id)
 
     result = _fetch_user_or_404(user_id, dal)
-    if isinstance(result, JSONResponse):
+    if isinstance(result, ScimJSONResponse):
         return result
     user = result
 
@@ -293,19 +481,36 @@ def replace_user(
         if seat_error:
             return _scim_error_response(403, seat_error)
 
+    personal_name = _scim_name_to_str(user_resource.name)
+
     dal.update_user(
         user,
-        email=user_resource.userName.strip().lower(),
+        email=user_resource.userName.strip(),
         is_active=user_resource.active,
-        personal_name=_scim_name_to_str(user_resource.name),
+        personal_name=personal_name,
     )
 
     new_external_id = user_resource.externalId
-    dal.sync_user_external_id(user.id, new_external_id)
+    scim_username = user_resource.userName.strip()
+    fields = _fields_from_resource(user_resource)
+    dal.sync_user_external_id(
+        user.id,
+        new_external_id,
+        scim_username=scim_username,
+        fields=fields,
+    )
 
     dal.commit()
 
-    return _user_to_scim(user, new_external_id)
+    return _scim_resource_response(
+        provider.build_user_resource(
+            user,
+            new_external_id,
+            groups=dal.get_user_groups(user.id),
+            scim_username=scim_username,
+            fields=fields,
+        )
+    )
 
 
 @scim_router.patch("/Users/{user_id}", response_model=None)
@@ -313,8 +518,9 @@ def patch_user(
     user_id: str,
     patch_request: ScimPatchRequest,
     _token: ScimToken = Depends(verify_scim_token),
+    provider: ScimProvider = Depends(_get_provider),
     db_session: Session = Depends(get_session),
-) -> ScimUserResource | JSONResponse:
+) -> ScimUserResource | ScimJSONResponse:
     """Partially update a user (RFC 7644 §3.5.2).
 
     This is the primary endpoint for user deprovisioning — Okta sends
@@ -324,17 +530,27 @@ def patch_user(
     dal.update_token_last_used(_token.id)
 
     result = _fetch_user_or_404(user_id, dal)
-    if isinstance(result, JSONResponse):
+    if isinstance(result, ScimJSONResponse):
         return result
     user = result
 
     mapping = dal.get_user_mapping_by_user_id(user.id)
     external_id = mapping.external_id if mapping else None
+    current_scim_username = mapping.scim_username if mapping else None
+    current_fields = _mapping_to_fields(mapping)
 
-    current = _user_to_scim(user, external_id)
+    current = provider.build_user_resource(
+        user,
+        external_id,
+        groups=dal.get_user_groups(user.id),
+        scim_username=current_scim_username,
+        fields=current_fields,
+    )
 
     try:
-        patched = apply_user_patch(patch_request.Operations, current)
+        patched, ent_data = apply_user_patch(
+            patch_request.Operations, current, provider.ignored_patch_paths
+        )
     except ScimPatchError as e:
         return _scim_error_response(e.status, e.detail)
 
@@ -345,22 +561,60 @@ def patch_user(
             if seat_error:
                 return _scim_error_response(403, seat_error)
 
+    # Track the scim_username — if userName was patched, update it
+    new_scim_username = patched.userName.strip() if patched.userName else None
+
+    # If displayName was explicitly patched (different from the original), use
+    # it as personal_name directly.  Otherwise, derive from name components.
+    personal_name: str | None
+    if patched.displayName and patched.displayName != current.displayName:
+        personal_name = patched.displayName
+    else:
+        personal_name = _scim_name_to_str(patched.name)
+
     dal.update_user(
         user,
         email=(
-            patched.userName.strip().lower()
-            if patched.userName.lower() != user.email
+            patched.userName.strip()
+            if patched.userName.strip().lower() != user.email.lower()
             else None
         ),
         is_active=patched.active if patched.active != user.is_active else None,
-        personal_name=_scim_name_to_str(patched.name),
+        personal_name=personal_name,
     )
 
-    dal.sync_user_external_id(user.id, patched.externalId)
+    # Build updated fields by merging PATCH enterprise data with current values
+    cf = current_fields or ScimMappingFields()
+    fields = ScimMappingFields(
+        department=ent_data.get("department", cf.department),
+        manager=ent_data.get("manager", cf.manager),
+        given_name=patched.name.givenName if patched.name else cf.given_name,
+        family_name=patched.name.familyName if patched.name else cf.family_name,
+        scim_emails_json=(
+            serialize_emails(patched.emails)
+            if patched.emails is not None
+            else cf.scim_emails_json
+        ),
+    )
+
+    dal.sync_user_external_id(
+        user.id,
+        patched.externalId,
+        scim_username=new_scim_username,
+        fields=fields,
+    )
 
     dal.commit()
 
-    return _user_to_scim(user, patched.externalId)
+    return _scim_resource_response(
+        provider.build_user_resource(
+            user,
+            patched.externalId,
+            groups=dal.get_user_groups(user.id),
+            scim_username=new_scim_username,
+            fields=fields,
+        )
+    )
 
 
 @scim_router.delete("/Users/{user_id}", status_code=204, response_model=None)
@@ -368,25 +622,29 @@ def delete_user(
     user_id: str,
     _token: ScimToken = Depends(verify_scim_token),
     db_session: Session = Depends(get_session),
-) -> Response | JSONResponse:
+) -> Response | ScimJSONResponse:
     """Delete a user (RFC 7644 §3.6).
 
     Deactivates the user and removes the SCIM mapping. Note that Okta
     typically uses PATCH active=false instead of DELETE.
+    A second DELETE returns 404 per RFC 7644 §3.6.
     """
     dal = ScimDAL(db_session)
     dal.update_token_last_used(_token.id)
 
     result = _fetch_user_or_404(user_id, dal)
-    if isinstance(result, JSONResponse):
+    if isinstance(result, ScimJSONResponse):
         return result
     user = result
 
-    dal.deactivate_user(user)
-
+    # If no SCIM mapping exists, the user was already deleted from
+    # SCIM's perspective — return 404 per RFC 7644 §3.6.
     mapping = dal.get_user_mapping_by_user_id(user.id)
-    if mapping:
-        dal.delete_user_mapping(mapping.id)
+    if not mapping:
+        return _scim_error_response(404, f"User {user_id} not found")
+
+    dal.deactivate_user(user)
+    dal.delete_user_mapping(mapping.id)
 
     dal.commit()
 
@@ -398,25 +656,7 @@ def delete_user(
 # ---------------------------------------------------------------------------
 
 
-def _group_to_scim(
-    group: UserGroup,
-    members: list[tuple[UUID, str | None]],
-    external_id: str | None = None,
-) -> ScimGroupResource:
-    """Convert an Onyx UserGroup to a SCIM Group resource."""
-    scim_members = [
-        ScimGroupMember(value=str(uid), display=email) for uid, email in members
-    ]
-    return ScimGroupResource(
-        id=str(group.id),
-        externalId=external_id,
-        displayName=group.name,
-        members=scim_members,
-        meta=ScimMeta(resourceType="Group"),
-    )
-
-
-def _fetch_group_or_404(group_id: str, dal: ScimDAL) -> UserGroup | JSONResponse:
+def _fetch_group_or_404(group_id: str, dal: ScimDAL) -> UserGroup | ScimJSONResponse:
     """Parse *group_id* as int, look up the group, or return a 404 error."""
     try:
         gid = int(group_id)
@@ -471,11 +711,13 @@ def _validate_and_parse_members(
 @scim_router.get("/Groups", response_model=None)
 def list_groups(
     filter: str | None = Query(None),
+    excludedAttributes: str | None = None,
     startIndex: int = Query(1, ge=1),
     count: int = Query(100, ge=0, le=500),
     _token: ScimToken = Depends(verify_scim_token),
+    provider: ScimProvider = Depends(_get_provider),
     db_session: Session = Depends(get_session),
-) -> ScimListResponse | JSONResponse:
+) -> ScimListResponse | ScimJSONResponse:
     """List groups with optional SCIM filter and pagination."""
     dal = ScimDAL(db_session)
     dal.update_token_last_used(_token.id)
@@ -491,45 +733,58 @@ def list_groups(
         return _scim_error_response(400, str(e))
 
     resources: list[ScimUserResource | ScimGroupResource] = [
-        _group_to_scim(group, dal.get_group_members(group.id), ext_id)
+        provider.build_group_resource(group, dal.get_group_members(group.id), ext_id)
         for group, ext_id in groups_with_ext_ids
     ]
 
-    return ScimListResponse(
-        totalResults=total,
-        startIndex=startIndex,
-        itemsPerPage=count,
-        Resources=resources,
+    return _build_list_response(
+        resources,
+        total,
+        startIndex,
+        count,
+        excluded=_parse_excluded_attributes(excludedAttributes),
     )
 
 
 @scim_router.get("/Groups/{group_id}", response_model=None)
 def get_group(
     group_id: str,
+    excludedAttributes: str | None = None,
     _token: ScimToken = Depends(verify_scim_token),
+    provider: ScimProvider = Depends(_get_provider),
     db_session: Session = Depends(get_session),
-) -> ScimGroupResource | JSONResponse:
+) -> ScimGroupResource | ScimJSONResponse:
     """Get a single group by ID."""
     dal = ScimDAL(db_session)
     dal.update_token_last_used(_token.id)
 
     result = _fetch_group_or_404(group_id, dal)
-    if isinstance(result, JSONResponse):
+    if isinstance(result, ScimJSONResponse):
         return result
     group = result
 
     mapping = dal.get_group_mapping_by_group_id(group.id)
     members = dal.get_group_members(group.id)
 
-    return _group_to_scim(group, members, mapping.external_id if mapping else None)
+    resource = provider.build_group_resource(
+        group, members, mapping.external_id if mapping else None
+    )
+
+    # RFC 7644 §3.4.2.5 — IdP may request certain attributes be omitted
+    excluded = _parse_excluded_attributes(excludedAttributes)
+    if excluded:
+        return ScimJSONResponse(content=_apply_exclusions(resource, excluded))
+
+    return _scim_resource_response(resource)
 
 
 @scim_router.post("/Groups", status_code=201, response_model=None)
 def create_group(
     group_resource: ScimGroupResource,
     _token: ScimToken = Depends(verify_scim_token),
+    provider: ScimProvider = Depends(_get_provider),
     db_session: Session = Depends(get_session),
-) -> ScimGroupResource | JSONResponse:
+) -> ScimGroupResource | ScimJSONResponse:
     """Create a new group from a SCIM provisioning request."""
     dal = ScimDAL(db_session)
     dal.update_token_last_used(_token.id)
@@ -565,7 +820,10 @@ def create_group(
     dal.commit()
 
     members = dal.get_group_members(db_group.id)
-    return _group_to_scim(db_group, members, external_id)
+    return _scim_resource_response(
+        provider.build_group_resource(db_group, members, external_id),
+        status_code=201,
+    )
 
 
 @scim_router.put("/Groups/{group_id}", response_model=None)
@@ -573,14 +831,15 @@ def replace_group(
     group_id: str,
     group_resource: ScimGroupResource,
     _token: ScimToken = Depends(verify_scim_token),
+    provider: ScimProvider = Depends(_get_provider),
     db_session: Session = Depends(get_session),
-) -> ScimGroupResource | JSONResponse:
+) -> ScimGroupResource | ScimJSONResponse:
     """Replace a group entirely (RFC 7644 §3.5.1)."""
     dal = ScimDAL(db_session)
     dal.update_token_last_used(_token.id)
 
     result = _fetch_group_or_404(group_id, dal)
-    if isinstance(result, JSONResponse):
+    if isinstance(result, ScimJSONResponse):
         return result
     group = result
 
@@ -595,7 +854,9 @@ def replace_group(
     dal.commit()
 
     members = dal.get_group_members(group.id)
-    return _group_to_scim(group, members, group_resource.externalId)
+    return _scim_resource_response(
+        provider.build_group_resource(group, members, group_resource.externalId)
+    )
 
 
 @scim_router.patch("/Groups/{group_id}", response_model=None)
@@ -603,8 +864,9 @@ def patch_group(
     group_id: str,
     patch_request: ScimPatchRequest,
     _token: ScimToken = Depends(verify_scim_token),
+    provider: ScimProvider = Depends(_get_provider),
     db_session: Session = Depends(get_session),
-) -> ScimGroupResource | JSONResponse:
+) -> ScimGroupResource | ScimJSONResponse:
     """Partially update a group (RFC 7644 §3.5.2).
 
     Handles member add/remove operations from Okta and Azure AD.
@@ -613,7 +875,7 @@ def patch_group(
     dal.update_token_last_used(_token.id)
 
     result = _fetch_group_or_404(group_id, dal)
-    if isinstance(result, JSONResponse):
+    if isinstance(result, ScimJSONResponse):
         return result
     group = result
 
@@ -621,11 +883,11 @@ def patch_group(
     external_id = mapping.external_id if mapping else None
 
     current_members = dal.get_group_members(group.id)
-    current = _group_to_scim(group, current_members, external_id)
+    current = provider.build_group_resource(group, current_members, external_id)
 
     try:
         patched, added_ids, removed_ids = apply_group_patch(
-            patch_request.Operations, current
+            patch_request.Operations, current, provider.ignored_patch_paths
         )
     except ScimPatchError as e:
         return _scim_error_response(e.status, e.detail)
@@ -652,7 +914,9 @@ def patch_group(
     dal.commit()
 
     members = dal.get_group_members(group.id)
-    return _group_to_scim(group, members, patched.externalId)
+    return _scim_resource_response(
+        provider.build_group_resource(group, members, patched.externalId)
+    )
 
 
 @scim_router.delete("/Groups/{group_id}", status_code=204, response_model=None)
@@ -660,13 +924,13 @@ def delete_group(
     group_id: str,
     _token: ScimToken = Depends(verify_scim_token),
     db_session: Session = Depends(get_session),
-) -> Response | JSONResponse:
+) -> Response | ScimJSONResponse:
     """Delete a group (RFC 7644 §3.6)."""
     dal = ScimDAL(db_session)
     dal.update_token_last_used(_token.id)
 
     result = _fetch_group_or_404(group_id, dal)
-    if isinstance(result, JSONResponse):
+    if isinstance(result, ScimJSONResponse):
         return result
     group = result
 

backend/ee/onyx/server/scim/models.py

@@ -7,12 +7,14 @@ SCIM protocol schemas follow the wire format defined in:
 Admin API schemas are internal to Onyx and used for SCIM token management.
 """
 
+from dataclasses import dataclass
 from datetime import datetime
 from enum import Enum
 
 from pydantic import BaseModel
 from pydantic import ConfigDict
 from pydantic import Field
+from pydantic import field_validator
 
 
 # ---------------------------------------------------------------------------
@@ -31,6 +33,9 @@ SCIM_SERVICE_PROVIDER_CONFIG_SCHEMA = (
 )
 SCIM_RESOURCE_TYPE_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:ResourceType"
 SCIM_SCHEMA_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:Schema"
+SCIM_ENTERPRISE_USER_SCHEMA = (
+    "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User"
+)
 
 
 # ---------------------------------------------------------------------------
@@ -63,6 +68,43 @@ class ScimMeta(BaseModel):
     location: str | None = None
 
 
+class ScimUserGroupRef(BaseModel):
+    """Group reference within a User resource (RFC 7643 §4.1.2, read-only)."""
+
+    value: str
+    display: str | None = None
+
+
+class ScimManagerRef(BaseModel):
+    """Manager sub-attribute for the enterprise extension (RFC 7643 §4.3)."""
+
+    value: str | None = None
+
+
+class ScimEnterpriseExtension(BaseModel):
+    """Enterprise User extension attributes (RFC 7643 §4.3)."""
+
+    department: str | None = None
+    manager: ScimManagerRef | None = None
+
+
+@dataclass
+class ScimMappingFields:
+    """Stored SCIM mapping fields that need to round-trip through the IdP.
+
+    Entra ID sends structured name components, email metadata, and enterprise
+    extension attributes that must be returned verbatim in subsequent GET
+    responses. These fields are persisted on ScimUserMapping and threaded
+    through the DAL, provider, and endpoint layers.
+    """
+
+    department: str | None = None
+    manager: str | None = None
+    given_name: str | None = None
+    family_name: str | None = None
+    scim_emails_json: str | None = None
+
+
 class ScimUserResource(BaseModel):
     """SCIM User resource representation (RFC 7643 §4.1).
 
@@ -71,14 +113,22 @@ class ScimUserResource(BaseModel):
     to match the SCIM wire format (not Python convention).
     """
 
+    model_config = ConfigDict(populate_by_name=True)
+
     schemas: list[str] = Field(default_factory=lambda: [SCIM_USER_SCHEMA])
     id: str | None = None  # Onyx's internal user ID, set on responses
     externalId: str | None = None  # IdP's identifier for this user
     userName: str  # Typically the user's email address
     name: ScimName | None = None
+    displayName: str | None = None
     emails: list[ScimEmail] = Field(default_factory=list)
     active: bool = True
+    groups: list[ScimUserGroupRef] = Field(default_factory=list)
     meta: ScimMeta | None = None
+    enterprise_extension: ScimEnterpriseExtension | None = Field(
+        default=None,
+        alias="urn:ietf:params:scim:schemas:extension:enterprise:2.0:User",
+    )
 
 
 class ScimGroupMember(BaseModel):
@@ -121,12 +171,53 @@ class ScimPatchOperationType(str, Enum):
     REMOVE = "remove"
 
 
+class ScimPatchResourceValue(BaseModel):
+    """Partial resource dict for path-less PATCH replace operations.
+
+    When an IdP sends a PATCH without a ``path``, the ``value`` is a dict
+    of resource attributes to set.  IdPs may include read-only fields
+    (``id``, ``schemas``, ``meta``) alongside actual changes — these are
+    stripped by the provider's ``ignored_patch_paths`` before processing.
+
+    ``extra="allow"`` lets unknown attributes pass through so the patch
+    handler can decide what to do with them (ignore or reject).
+    """
+
+    model_config = ConfigDict(extra="allow")
+
+    active: bool | None = None
+    userName: str | None = None
+    displayName: str | None = None
+    externalId: str | None = None
+    name: ScimName | None = None
+    members: list[ScimGroupMember] | None = None
+    id: str | None = None
+    schemas: list[str] | None = None
+    meta: ScimMeta | None = None
+
+
+ScimPatchValue = str | bool | list[ScimGroupMember] | ScimPatchResourceValue | None
+
+
 class ScimPatchOperation(BaseModel):
     """Single PATCH operation (RFC 7644 §3.5.2)."""
 
     op: ScimPatchOperationType
     path: str | None = None
-    value: str | list[dict[str, str]] | dict[str, str | bool] | bool | None = None
+    value: ScimPatchValue = None
+
+    @field_validator("op", mode="before")
+    @classmethod
+    def normalize_operation(cls, v: object) -> object:
+        """Normalize op to lowercase for case-insensitive matching.
+
+        Some IdPs (e.g. Entra ID) send capitalized ops like ``"Replace"``
+        instead of ``"replace"``. This is safe for all providers since the
+        enum values are lowercase. If a future provider requires other
+        pre-processing quirks, move patch deserialization into the provider
+        subclass instead of adding more special cases here.
+        """
+        return v.lower() if isinstance(v, str) else v
 
 
 class ScimPatchRequest(BaseModel):

backend/ee/onyx/server/scim/patch.py

@@ -14,13 +14,70 @@ responsible for persisting changes.
 
 from __future__ import annotations
 
+import logging
 import re
+from dataclasses import dataclass
+from dataclasses import field
+from typing import Any
 
+from ee.onyx.server.scim.models import SCIM_ENTERPRISE_USER_SCHEMA
+from ee.onyx.server.scim.models import ScimGroupMember
 from ee.onyx.server.scim.models import ScimGroupResource
 from ee.onyx.server.scim.models import ScimPatchOperation
 from ee.onyx.server.scim.models import ScimPatchOperationType
+from ee.onyx.server.scim.models import ScimPatchResourceValue
+from ee.onyx.server.scim.models import ScimPatchValue
 from ee.onyx.server.scim.models import ScimUserResource
 
+logger = logging.getLogger(__name__)
+
+# Lowercased enterprise extension URN for case-insensitive matching
+_ENTERPRISE_URN_LOWER = SCIM_ENTERPRISE_USER_SCHEMA.lower()
+
+# Pattern for email filter paths, e.g.:
+#   emails[primary eq true].value  (Okta)
+#   emails[type eq "work"].value   (Azure AD / Entra ID)
+_EMAIL_FILTER_RE = re.compile(
+    r"^emails\[.+\]\.value$",
+    re.IGNORECASE,
+)
+
+# Pattern for member removal path: members[value eq "user-id"]
+_MEMBER_FILTER_RE = re.compile(
+    r'^members\[value\s+eq\s+"([^"]+)"\]$',
+    re.IGNORECASE,
+)
+
+# ---------------------------------------------------------------------------
+# Dispatch tables for user PATCH paths
+#
+# Maps lowercased SCIM path → (camelCase key, target dict name).
+# "data" writes to the top-level resource dict, "name" writes to the
+# name sub-object dict. This replaces the elif chains for simple fields.
+# ---------------------------------------------------------------------------
+
+_USER_REPLACE_PATHS: dict[str, tuple[str, str]] = {
+    "active": ("active", "data"),
+    "username": ("userName", "data"),
+    "externalid": ("externalId", "data"),
+    "name.givenname": ("givenName", "name"),
+    "name.familyname": ("familyName", "name"),
+    "name.formatted": ("formatted", "name"),
+}
+
+_USER_REMOVE_PATHS: dict[str, tuple[str, str]] = {
+    "externalid": ("externalId", "data"),
+    "name.givenname": ("givenName", "name"),
+    "name.familyname": ("familyName", "name"),
+    "name.formatted": ("formatted", "name"),
+    "displayname": ("displayName", "data"),
+}
+
+_GROUP_REPLACE_PATHS: dict[str, tuple[str, str]] = {
+    "displayname": ("displayName", "data"),
+    "externalid": ("externalId", "data"),
+}
+
 
 class ScimPatchError(Exception):
     """Raised when a PATCH operation cannot be applied."""
@@ -31,94 +88,223 @@ class ScimPatchError(Exception):
         super().__init__(detail)
 
 
-# Pattern for member removal path: members[value eq "user-id"]
-_MEMBER_FILTER_RE = re.compile(
-    r'^members\[value\s+eq\s+"([^"]+)"\]$',
-    re.IGNORECASE,
-)
+@dataclass
+class _UserPatchCtx:
+    """Bundles the mutable state for user PATCH operations."""
+
+    data: dict[str, Any]
+    name_data: dict[str, Any]
+    ent_data: dict[str, str | None] = field(default_factory=dict)
+
+
+# ---------------------------------------------------------------------------
+# User PATCH
+# ---------------------------------------------------------------------------
 
 
 def apply_user_patch(
     operations: list[ScimPatchOperation],
     current: ScimUserResource,
-) -> ScimUserResource:
+    ignored_paths: frozenset[str] = frozenset(),
+) -> tuple[ScimUserResource, dict[str, str | None]]:
     """Apply SCIM PATCH operations to a user resource.
 
-    Returns a new ``ScimUserResource`` with the modifications applied.
-    The original object is not mutated.
+    Args:
+        operations: The PATCH operations to apply.
+        current: The current user resource state.
+        ignored_paths: SCIM attribute paths to silently skip (from provider).
+
+    Returns:
+        A tuple of (modified user resource, enterprise extension data dict).
+        The enterprise dict has keys ``"department"`` and ``"manager"``
+        with values set only when a PATCH operation touched them.
 
     Raises:
         ScimPatchError: If an operation targets an unsupported path.
     """
     data = current.model_dump()
-    name_data = data.get("name") or {}
+    ctx = _UserPatchCtx(data=data, name_data=data.get("name") or {})
 
     for op in operations:
-        if op.op == ScimPatchOperationType.REPLACE:
-            _apply_user_replace(op, data, name_data)
-        elif op.op == ScimPatchOperationType.ADD:
-            _apply_user_replace(op, data, name_data)
+        if op.op in (ScimPatchOperationType.REPLACE, ScimPatchOperationType.ADD):
+            _apply_user_replace(op, ctx, ignored_paths)
+        elif op.op == ScimPatchOperationType.REMOVE:
+            _apply_user_remove(op, ctx, ignored_paths)
         else:
             raise ScimPatchError(
                 f"Unsupported operation '{op.op.value}' on User resource"
             )
 
-    data["name"] = name_data
-    return ScimUserResource.model_validate(data)
+    ctx.data["name"] = ctx.name_data
+    return ScimUserResource.model_validate(ctx.data), ctx.ent_data
 
 
 def _apply_user_replace(
     op: ScimPatchOperation,
-    data: dict,
-    name_data: dict,
+    ctx: _UserPatchCtx,
+    ignored_paths: frozenset[str],
 ) -> None:
     """Apply a replace/add operation to user data."""
     path = (op.path or "").lower()
 
     if not path:
-        # No path — value is a dict of top-level attributes to set
-        if isinstance(op.value, dict):
-            for key, val in op.value.items():
-                _set_user_field(key.lower(), val, data, name_data)
+        # No path — value is a resource dict of top-level attributes to set.
+        if isinstance(op.value, ScimPatchResourceValue):
+            for key, val in op.value.model_dump(exclude_unset=True).items():
+                _set_user_field(key.lower(), val, ctx, ignored_paths, strict=False)
         else:
             raise ScimPatchError("Replace without path requires a dict value")
         return
 
-    _set_user_field(path, op.value, data, name_data)
+    _set_user_field(path, op.value, ctx, ignored_paths)
+
+
+def _apply_user_remove(
+    op: ScimPatchOperation,
+    ctx: _UserPatchCtx,
+    ignored_paths: frozenset[str],
+) -> None:
+    """Apply a remove operation to user data — clears the target field."""
+    path = (op.path or "").lower()
+    if not path:
+        raise ScimPatchError("Remove operation requires a path")
+
+    if path in ignored_paths:
+        return
+
+    entry = _USER_REMOVE_PATHS.get(path)
+    if entry:
+        key, target = entry
+        target_dict = ctx.data if target == "data" else ctx.name_data
+        target_dict[key] = None
+        return
+
+    raise ScimPatchError(f"Unsupported remove path '{path}' for User PATCH")
 
 
 def _set_user_field(
     path: str,
-    value: str | bool | dict | list | None,
-    data: dict,
-    name_data: dict,
+    value: ScimPatchValue,
+    ctx: _UserPatchCtx,
+    ignored_paths: frozenset[str],
+    *,
+    strict: bool = True,
 ) -> None:
-    """Set a single field on user data by SCIM path."""
-    if path == "active":
-        data["active"] = value
-    elif path == "username":
-        data["userName"] = value
-    elif path == "externalid":
-        data["externalId"] = value
-    elif path == "name.givenname":
-        name_data["givenName"] = value
-    elif path == "name.familyname":
-        name_data["familyName"] = value
-    elif path == "name.formatted":
-        name_data["formatted"] = value
-    elif path == "displayname":
-        # Some IdPs send displayName on users; map to formatted name
-        name_data["formatted"] = value
+    """Set a single field on user data by SCIM path.
+
+    Args:
+        strict: When ``False`` (path-less replace), unknown attributes are
+            silently skipped.  When ``True`` (explicit path), they raise.
+    """
+    if path in ignored_paths:
+        return
+
+    # Simple field writes handled by the dispatch table
+    entry = _USER_REPLACE_PATHS.get(path)
+    if entry:
+        key, target = entry
+        target_dict = ctx.data if target == "data" else ctx.name_data
+        target_dict[key] = value
+        return
+
+    # displayName sets both the top-level field and the name.formatted sub-field
+    if path == "displayname":
+        ctx.data["displayName"] = value
+        ctx.name_data["formatted"] = value
+    elif path == "name":
+        if isinstance(value, dict):
+            for k, v in value.items():
+                ctx.name_data[k] = v
+    elif path == "emails":
+        if isinstance(value, list):
+            ctx.data["emails"] = value
+    elif _EMAIL_FILTER_RE.match(path):
+        _update_primary_email(ctx.data, value)
+    elif path.startswith(_ENTERPRISE_URN_LOWER):
+        _set_enterprise_field(path, value, ctx.ent_data)
+    elif not strict:
+        return
     else:
         raise ScimPatchError(f"Unsupported path '{path}' for User PATCH")
 
 
+def _update_primary_email(data: dict[str, Any], value: ScimPatchValue) -> None:
+    """Update the primary email entry via an email filter path."""
+    emails: list[dict] = data.get("emails") or []
+    for email_entry in emails:
+        if email_entry.get("primary"):
+            email_entry["value"] = value
+            break
+    else:
+        emails.append({"value": value, "type": "work", "primary": True})
+    data["emails"] = emails
+
+
+def _to_dict(value: ScimPatchValue) -> dict | None:
+    """Coerce a SCIM patch value to a plain dict if possible.
+
+    Pydantic may parse raw dicts as ``ScimPatchResourceValue`` (which uses
+    ``extra="allow"``), so we also dump those back to a dict.
+    """
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, ScimPatchResourceValue):
+        return value.model_dump(exclude_unset=True)
+    return None
+
+
+def _set_enterprise_field(
+    path: str,
+    value: ScimPatchValue,
+    ent_data: dict[str, str | None],
+) -> None:
+    """Handle enterprise extension URN paths or value dicts."""
+    # Full URN as key with dict value (path-less PATCH)
+    # e.g. key="urn:...:user", value={"department": "Eng", "manager": {...}}
+    if path == _ENTERPRISE_URN_LOWER:
+        d = _to_dict(value)
+        if d is not None:
+            if "department" in d:
+                ent_data["department"] = d["department"]
+            if "manager" in d:
+                mgr = d["manager"]
+                if isinstance(mgr, dict):
+                    ent_data["manager"] = mgr.get("value")
+        return
+
+    # Dotted URN path, e.g. "urn:...:user:department"
+    suffix = path[len(_ENTERPRISE_URN_LOWER) :].lstrip(":").lower()
+    if suffix == "department":
+        ent_data["department"] = str(value) if value is not None else None
+    elif suffix == "manager":
+        d = _to_dict(value)
+        if d is not None:
+            ent_data["manager"] = d.get("value")
+        elif isinstance(value, str):
+            ent_data["manager"] = value
+    else:
+        # Unknown enterprise attributes are silently ignored rather than
+        # rejected — IdPs may send attributes we don't model yet.
+        logger.warning("Ignoring unknown enterprise extension attribute '%s'", suffix)
+
+
+# ---------------------------------------------------------------------------
+# Group PATCH
+# ---------------------------------------------------------------------------
+
+
 def apply_group_patch(
     operations: list[ScimPatchOperation],
     current: ScimGroupResource,
+    ignored_paths: frozenset[str] = frozenset(),
 ) -> tuple[ScimGroupResource, list[str], list[str]]:
     """Apply SCIM PATCH operations to a group resource.
 
+    Args:
+        operations: The PATCH operations to apply.
+        current: The current group resource state.
+        ignored_paths: SCIM attribute paths to silently skip (from provider).
+
     Returns:
         A tuple of (modified group, added member IDs, removed member IDs).
         The caller uses the member ID lists to update the database.
@@ -133,7 +319,9 @@ def apply_group_patch(
 
     for op in operations:
         if op.op == ScimPatchOperationType.REPLACE:
-            _apply_group_replace(op, data, current_members, added_ids, removed_ids)
+            _apply_group_replace(
+                op, data, current_members, added_ids, removed_ids, ignored_paths
+            )
         elif op.op == ScimPatchOperationType.ADD:
             _apply_group_add(op, current_members, added_ids)
         elif op.op == ScimPatchOperationType.REMOVE:
@@ -154,38 +342,48 @@ def _apply_group_replace(
     current_members: list[dict],
     added_ids: list[str],
     removed_ids: list[str],
+    ignored_paths: frozenset[str],
 ) -> None:
     """Apply a replace operation to group data."""
     path = (op.path or "").lower()
 
     if not path:
-        if isinstance(op.value, dict):
-            for key, val in op.value.items():
+        if isinstance(op.value, ScimPatchResourceValue):
+            dumped = op.value.model_dump(exclude_unset=True)
+            for key, val in dumped.items():
                 if key.lower() == "members":
                     _replace_members(val, current_members, added_ids, removed_ids)
                 else:
-                    _set_group_field(key.lower(), val, data)
+                    _set_group_field(key.lower(), val, data, ignored_paths)
         else:
             raise ScimPatchError("Replace without path requires a dict value")
         return
 
     if path == "members":
-        _replace_members(op.value, current_members, added_ids, removed_ids)
+        _replace_members(
+            _members_to_dicts(op.value), current_members, added_ids, removed_ids
+        )
         return
 
-    _set_group_field(path, op.value, data)
+    _set_group_field(path, op.value, data, ignored_paths)
+
+
+def _members_to_dicts(
+    value: str | bool | list[ScimGroupMember] | ScimPatchResourceValue | None,
+) -> list[dict]:
+    """Convert a member list value to a list of dicts for internal processing."""
+    if not isinstance(value, list):
+        raise ScimPatchError("Replace members requires a list value")
+    return [m.model_dump(exclude_none=True) for m in value]
 
 
 def _replace_members(
-    value: str | list | dict | bool | None,
+    value: list[dict],
     current_members: list[dict],
     added_ids: list[str],
     removed_ids: list[str],
 ) -> None:
     """Replace the entire group member list."""
-    if not isinstance(value, list):
-        raise ScimPatchError("Replace members requires a list value")
-
     old_ids = {m["value"] for m in current_members}
     new_ids = {m.get("value", "") for m in value}
 
@@ -197,16 +395,21 @@ def _replace_members(
 
 def _set_group_field(
     path: str,
-    value: str | bool | dict | list | None,
+    value: ScimPatchValue,
     data: dict,
+    ignored_paths: frozenset[str],
 ) -> None:
     """Set a single field on group data by SCIM path."""
-    if path == "displayname":
-        data["displayName"] = value
-    elif path == "externalid":
-        data["externalId"] = value
-    else:
-        raise ScimPatchError(f"Unsupported path '{path}' for Group PATCH")
+    if path in ignored_paths:
+        return
+
+    entry = _GROUP_REPLACE_PATHS.get(path)
+    if entry:
+        key, _ = entry
+        data[key] = value
+        return
+
+    raise ScimPatchError(f"Unsupported path '{path}' for Group PATCH")
 
 
 def _apply_group_add(
@@ -223,8 +426,10 @@ def _apply_group_add(
     if not isinstance(op.value, list):
         raise ScimPatchError("Add members requires a list value")
 
+    member_dicts = [m.model_dump(exclude_none=True) for m in op.value]
+
     existing_ids = {m["value"] for m in members}
-    for member_data in op.value:
+    for member_data in member_dicts:
         member_id = member_data.get("value", "")
         if member_id and member_id not in existing_ids:
             members.append(member_data)

backend/ee/onyx/server/scim/providers/base.py

@@ -0,0 +1,210 @@
+"""Base SCIM provider abstraction."""
+
+from __future__ import annotations
+
+import json
+import logging
+from abc import ABC
+from abc import abstractmethod
+from uuid import UUID
+
+from pydantic import ValidationError
+
+from ee.onyx.server.scim.models import SCIM_ENTERPRISE_USER_SCHEMA
+from ee.onyx.server.scim.models import SCIM_USER_SCHEMA
+from ee.onyx.server.scim.models import ScimEmail
+from ee.onyx.server.scim.models import ScimEnterpriseExtension
+from ee.onyx.server.scim.models import ScimGroupMember
+from ee.onyx.server.scim.models import ScimGroupResource
+from ee.onyx.server.scim.models import ScimManagerRef
+from ee.onyx.server.scim.models import ScimMappingFields
+from ee.onyx.server.scim.models import ScimMeta
+from ee.onyx.server.scim.models import ScimName
+from ee.onyx.server.scim.models import ScimUserGroupRef
+from ee.onyx.server.scim.models import ScimUserResource
+from onyx.db.models import User
+from onyx.db.models import UserGroup
+
+
+logger = logging.getLogger(__name__)
+
+COMMON_IGNORED_PATCH_PATHS: frozenset[str] = frozenset(
+    {
+        "id",
+        "schemas",
+        "meta",
+    }
+)
+
+
+class ScimProvider(ABC):
+    """Base class for provider-specific SCIM behavior.
+
+    Subclass this to handle IdP-specific quirks. The base class provides
+    RFC 7643-compliant response builders that populate all standard fields.
+    """
+
+    @property
+    @abstractmethod
+    def name(self) -> str:
+        """Short identifier for this provider (e.g. ``"okta"``)."""
+        ...
+
+    @property
+    @abstractmethod
+    def ignored_patch_paths(self) -> frozenset[str]:
+        """SCIM attribute paths to silently skip in PATCH value-object dicts.
+
+        IdPs may include read-only or meta fields alongside actual changes
+        (e.g. Okta sends ``{"id": "...", "active": false}``). Paths listed
+        here are silently dropped instead of raising an error.
+        """
+        ...
+
+    @property
+    def user_schemas(self) -> list[str]:
+        """Schema URIs to include in User resource responses.
+
+        Override in subclasses to advertise additional schemas (e.g. the
+        enterprise extension for Entra ID).
+        """
+        return [SCIM_USER_SCHEMA]
+
+    def build_user_resource(
+        self,
+        user: User,
+        external_id: str | None = None,
+        groups: list[tuple[int, str]] | None = None,
+        scim_username: str | None = None,
+        fields: ScimMappingFields | None = None,
+    ) -> ScimUserResource:
+        """Build a SCIM User response from an Onyx User.
+
+        Args:
+            user: The Onyx user model.
+            external_id: The IdP's external identifier for this user.
+            groups: List of ``(group_id, group_name)`` tuples for the
+                ``groups`` read-only attribute. Pass ``None`` or ``[]``
+                for newly-created users.
+            scim_username: The original-case userName from the IdP. Falls
+                back to ``user.email`` (lowercase) when not available.
+            fields: Stored mapping fields that the IdP expects round-tripped.
+        """
+        f = fields or ScimMappingFields()
+        group_refs = [
+            ScimUserGroupRef(value=str(gid), display=gname)
+            for gid, gname in (groups or [])
+        ]
+
+        username = scim_username or user.email
+
+        # Build enterprise extension when at least one value is present.
+        # Dynamically add the enterprise URN to schemas per RFC 7643 §3.0.
+        enterprise_ext: ScimEnterpriseExtension | None = None
+        schemas = list(self.user_schemas)
+        if f.department is not None or f.manager is not None:
+            manager_ref = (
+                ScimManagerRef(value=f.manager) if f.manager is not None else None
+            )
+            enterprise_ext = ScimEnterpriseExtension(
+                department=f.department,
+                manager=manager_ref,
+            )
+            if SCIM_ENTERPRISE_USER_SCHEMA not in schemas:
+                schemas.append(SCIM_ENTERPRISE_USER_SCHEMA)
+
+        name = self.build_scim_name(user, f)
+        emails = _deserialize_emails(f.scim_emails_json, username)
+
+        resource = ScimUserResource(
+            schemas=schemas,
+            id=str(user.id),
+            externalId=external_id,
+            userName=username,
+            name=name,
+            displayName=user.personal_name,
+            emails=emails,
+            active=user.is_active,
+            groups=group_refs,
+            meta=ScimMeta(resourceType="User"),
+        )
+        resource.enterprise_extension = enterprise_ext
+        return resource
+
+    def build_group_resource(
+        self,
+        group: UserGroup,
+        members: list[tuple[UUID, str | None]],
+        external_id: str | None = None,
+    ) -> ScimGroupResource:
+        """Build a SCIM Group response from an Onyx UserGroup."""
+        scim_members = [
+            ScimGroupMember(value=str(uid), display=email) for uid, email in members
+        ]
+        return ScimGroupResource(
+            id=str(group.id),
+            externalId=external_id,
+            displayName=group.name,
+            members=scim_members,
+            meta=ScimMeta(resourceType="Group"),
+        )
+
+    def build_scim_name(
+        self,
+        user: User,
+        fields: ScimMappingFields,
+    ) -> ScimName | None:
+        """Build SCIM name components for the response.
+
+        Round-trips stored ``given_name``/``family_name`` when available (so
+        the IdP gets back what it sent). Falls back to splitting
+        ``personal_name`` for users provisioned before we stored components.
+        Providers may override for custom behavior.
+        """
+        if fields.given_name is not None or fields.family_name is not None:
+            return ScimName(
+                givenName=fields.given_name,
+                familyName=fields.family_name,
+                formatted=user.personal_name,
+            )
+        if not user.personal_name:
+            return None
+        parts = user.personal_name.split(" ", 1)
+        return ScimName(
+            givenName=parts[0],
+            familyName=parts[1] if len(parts) > 1 else None,
+            formatted=user.personal_name,
+        )
+
+
+def _deserialize_emails(stored_json: str | None, username: str) -> list[ScimEmail]:
+    """Deserialize stored email entries or build a default work email."""
+    if stored_json:
+        try:
+            entries = json.loads(stored_json)
+            if isinstance(entries, list) and entries:
+                return [ScimEmail(**e) for e in entries]
+        except (json.JSONDecodeError, TypeError, ValidationError):
+            logger.warning(
+                "Corrupt scim_emails_json, falling back to default: %s", stored_json
+            )
+    return [ScimEmail(value=username, type="work", primary=True)]
+
+
+def serialize_emails(emails: list[ScimEmail]) -> str | None:
+    """Serialize SCIM email entries to JSON for storage."""
+    if not emails:
+        return None
+    return json.dumps([e.model_dump(exclude_none=True) for e in emails])
+
+
+def get_default_provider() -> ScimProvider:
+    """Return the default SCIM provider.
+
+    Currently returns ``OktaProvider`` since Okta is the primary supported
+    IdP. When provider detection is added (via token metadata or tenant
+    config), this can be replaced with dynamic resolution.
+    """
+    from ee.onyx.server.scim.providers.okta import OktaProvider
+
+    return OktaProvider()

backend/ee/onyx/server/scim/providers/entra.py

@@ -0,0 +1,36 @@
+"""Entra ID (Azure AD) SCIM provider."""
+
+from __future__ import annotations
+
+from ee.onyx.server.scim.models import SCIM_ENTERPRISE_USER_SCHEMA
+from ee.onyx.server.scim.models import SCIM_USER_SCHEMA
+from ee.onyx.server.scim.providers.base import COMMON_IGNORED_PATCH_PATHS
+from ee.onyx.server.scim.providers.base import ScimProvider
+
+_ENTRA_IGNORED_PATCH_PATHS = COMMON_IGNORED_PATCH_PATHS
+
+
+class EntraProvider(ScimProvider):
+    """Entra ID (Azure AD) SCIM provider.
+
+    Entra behavioral notes:
+      - Sends capitalized PATCH ops (``"Add"``, ``"Replace"``, ``"Remove"``)
+        — handled by ``ScimPatchOperation.normalize_op`` validator.
+      - Sends the enterprise extension URN as a key in path-less PATCH value
+        dicts — handled by ``_set_enterprise_field`` in ``patch.py`` to
+        store department/manager values.
+      - Expects the enterprise extension schema in ``schemas`` arrays and
+        ``/Schemas`` + ``/ResourceTypes`` discovery endpoints.
+    """
+
+    @property
+    def name(self) -> str:
+        return "entra"
+
+    @property
+    def ignored_patch_paths(self) -> frozenset[str]:
+        return _ENTRA_IGNORED_PATCH_PATHS
+
+    @property
+    def user_schemas(self) -> list[str]:
+        return [SCIM_USER_SCHEMA, SCIM_ENTERPRISE_USER_SCHEMA]

backend/ee/onyx/server/scim/providers/okta.py

@@ -0,0 +1,26 @@
+"""Okta SCIM provider."""
+
+from __future__ import annotations
+
+from ee.onyx.server.scim.providers.base import COMMON_IGNORED_PATCH_PATHS
+from ee.onyx.server.scim.providers.base import ScimProvider
+
+
+class OktaProvider(ScimProvider):
+    """Okta SCIM provider.
+
+    Okta behavioral notes:
+      - Uses ``PATCH {"active": false}`` for deprovisioning (not DELETE)
+      - Sends path-less PATCH with value dicts containing extra fields
+        (``id``, ``schemas``)
+      - Expects ``displayName`` and ``groups`` in user responses
+      - Only uses ``eq`` operator for ``userName`` filter
+    """
+
+    @property
+    def name(self) -> str:
+        return "okta"
+
+    @property
+    def ignored_patch_paths(self) -> frozenset[str]:
+        return COMMON_IGNORED_PATCH_PATHS

backend/ee/onyx/server/scim/schema_definitions.py

@@ -4,6 +4,7 @@ Pre-built at import time — these never change at runtime. Separated from
 api.py to keep the endpoint module focused on request handling.
 """
 
+from ee.onyx.server.scim.models import SCIM_ENTERPRISE_USER_SCHEMA
 from ee.onyx.server.scim.models import SCIM_GROUP_SCHEMA
 from ee.onyx.server.scim.models import SCIM_USER_SCHEMA
 from ee.onyx.server.scim.models import ScimResourceType
@@ -20,6 +21,9 @@ USER_RESOURCE_TYPE = ScimResourceType.model_validate(
         "endpoint": "/scim/v2/Users",
         "description": "SCIM User resource",
         "schema": SCIM_USER_SCHEMA,
+        "schemaExtensions": [
+            {"schema": SCIM_ENTERPRISE_USER_SCHEMA, "required": False}
+        ],
     }
 )
 
@@ -104,6 +108,31 @@ USER_SCHEMA_DEF = ScimSchemaDefinition(
     ],
 )
 
+ENTERPRISE_USER_SCHEMA_DEF = ScimSchemaDefinition(
+    id=SCIM_ENTERPRISE_USER_SCHEMA,
+    name="EnterpriseUser",
+    description="Enterprise User extension (RFC 7643 §4.3)",
+    attributes=[
+        ScimSchemaAttribute(
+            name="department",
+            type="string",
+            description="Department.",
+        ),
+        ScimSchemaAttribute(
+            name="manager",
+            type="complex",
+            description="The user's manager.",
+            subAttributes=[
+                ScimSchemaAttribute(
+                    name="value",
+                    type="string",
+                    description="Manager user ID.",
+                ),
+            ],
+        ),
+    ],
+)
+
 GROUP_SCHEMA_DEF = ScimSchemaDefinition(
     id=SCIM_GROUP_SCHEMA,
     name="Group",

backend/ee/onyx/server/user_group/api.py

@@ -37,12 +37,15 @@ def list_user_groups(
     db_session: Session = Depends(get_session),
 ) -> list[UserGroup]:
     if user.role == UserRole.ADMIN:
-        user_groups = fetch_user_groups(db_session, only_up_to_date=False)
+        user_groups = fetch_user_groups(
+            db_session, only_up_to_date=False, eager_load_for_snapshot=True
+        )
     else:
         user_groups = fetch_user_groups_for_user(
             db_session=db_session,
             user_id=user.id,
             only_curator_groups=user.role == UserRole.CURATOR,
+            eager_load_for_snapshot=True,
         )
     return [UserGroup.from_model(user_group) for user_group in user_groups]
 

backend/ee/onyx/server/user_group/models.py

@@ -53,7 +53,8 @@ class UserGroup(BaseModel):
                     id=cc_pair_relationship.cc_pair.id,
                     name=cc_pair_relationship.cc_pair.name,
                     connector=ConnectorSnapshot.from_connector_db_model(
-                        cc_pair_relationship.cc_pair.connector
+                        cc_pair_relationship.cc_pair.connector,
+                        credential_ids=[cc_pair_relationship.cc_pair.credential_id],
                     ),
                     credential=CredentialSnapshot.from_credential_db_model(
                         cc_pair_relationship.cc_pair.credential

backend/onyx/auth/users.py

@@ -277,13 +277,32 @@ def verify_email_domain(email: str) -> None:
             detail="Email is not valid",
         )
 
-    domain = email.split("@")[-1].lower()
+    local_part, domain = email.split("@")
+    domain = domain.lower()
+
+    if AUTH_TYPE == AuthType.CLOUD:
+        # Normalize googlemail.com to gmail.com (they deliver to the same inbox)
+        if domain == "googlemail.com":
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail={"reason": "Please use @gmail.com instead of @googlemail.com."},
+            )
+
+        if "+" in local_part and domain != "onyx.app":
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail={
+                    "reason": "Email addresses with '+' are not allowed. Please use your base email address."
+                },
+            )
 
     # Check if email uses a disposable/temporary domain
     if is_disposable_email(email):
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
-            detail="Disposable email addresses are not allowed. Please use a permanent email address.",
+            detail={
+                "reason": "Disposable email addresses are not allowed. Please use a permanent email address."
+            },
         )
 
     # Check domain whitelist if configured
@@ -1671,7 +1690,10 @@ def get_oauth_router(
         if redirect_url is not None:
             authorize_redirect_url = redirect_url
         else:
-            authorize_redirect_url = str(request.url_for(callback_route_name))
+            # Use WEB_DOMAIN instead of request.url_for() to prevent host
+            # header poisoning — request.url_for() trusts the Host header.
+            callback_path = request.app.url_path_for(callback_route_name)
+            authorize_redirect_url = f"{WEB_DOMAIN}{callback_path}"
 
         next_url = request.query_params.get("next", "/")
 

backend/onyx/background/celery/tasks/opensearch_migration/transformer.py

@@ -22,6 +22,7 @@ from onyx.document_index.vespa_constants import HIDDEN
 from onyx.document_index.vespa_constants import IMAGE_FILE_NAME
 from onyx.document_index.vespa_constants import METADATA_LIST
 from onyx.document_index.vespa_constants import METADATA_SUFFIX
+from onyx.document_index.vespa_constants import PERSONAS
 from onyx.document_index.vespa_constants import PRIMARY_OWNERS
 from onyx.document_index.vespa_constants import SECONDARY_OWNERS
 from onyx.document_index.vespa_constants import SEMANTIC_IDENTIFIER
@@ -58,6 +59,7 @@ FIELDS_NEEDED_FOR_TRANSFORMATION: list[str] = [
     METADATA_SUFFIX,
     DOCUMENT_SETS,
     USER_PROJECT,
+    PERSONAS,
     PRIMARY_OWNERS,
     SECONDARY_OWNERS,
     ACCESS_CONTROL_LIST,
@@ -276,6 +278,7 @@ def transform_vespa_chunks_to_opensearch_chunks(
                 )
             )
             user_projects: list[int] | None = vespa_chunk.get(USER_PROJECT)
+            personas: list[int] | None = vespa_chunk.get(PERSONAS)
             primary_owners: list[str] | None = vespa_chunk.get(PRIMARY_OWNERS)
             secondary_owners: list[str] | None = vespa_chunk.get(SECONDARY_OWNERS)
 
@@ -325,6 +328,7 @@ def transform_vespa_chunks_to_opensearch_chunks(
                 metadata_suffix=metadata_suffix,
                 document_sets=document_sets,
                 user_projects=user_projects,
+                personas=personas,
                 primary_owners=primary_owners,
                 secondary_owners=secondary_owners,
                 tenant_id=tenant_state,

backend/onyx/background/celery/tasks/user_file_processing/tasks.py

@@ -5,14 +5,18 @@ from uuid import UUID
 
 import httpx
 import sqlalchemy as sa
+from celery import Celery
 from celery import shared_task
 from celery import Task
+from redis import Redis
 from redis.lock import Lock as RedisLock
 from retry import retry
 from sqlalchemy import select
+from sqlalchemy.orm import selectinload
 from sqlalchemy.orm import Session
 
 from onyx.background.celery.apps.app_base import task_logger
+from onyx.background.celery.celery_redis import celery_get_queue_length
 from onyx.background.celery.celery_utils import httpx_init_vespa_pool
 from onyx.background.celery.tasks.shared.RetryDocumentIndex import RetryDocumentIndex
 from onyx.configs.app_configs import DISABLE_VECTOR_DB
@@ -21,12 +25,16 @@ from onyx.configs.app_configs import VESPA_CLOUD_CERT_PATH
 from onyx.configs.app_configs import VESPA_CLOUD_KEY_PATH
 from onyx.configs.constants import CELERY_GENERIC_BEAT_LOCK_TIMEOUT
 from onyx.configs.constants import CELERY_USER_FILE_PROCESSING_LOCK_TIMEOUT
+from onyx.configs.constants import CELERY_USER_FILE_PROCESSING_TASK_EXPIRES
 from onyx.configs.constants import CELERY_USER_FILE_PROJECT_SYNC_LOCK_TIMEOUT
+from onyx.configs.constants import CELERY_USER_FILE_PROJECT_SYNC_TASK_EXPIRES
 from onyx.configs.constants import DocumentSource
 from onyx.configs.constants import OnyxCeleryPriority
 from onyx.configs.constants import OnyxCeleryQueues
 from onyx.configs.constants import OnyxCeleryTask
 from onyx.configs.constants import OnyxRedisLocks
+from onyx.configs.constants import USER_FILE_PROCESSING_MAX_QUEUE_DEPTH
+from onyx.configs.constants import USER_FILE_PROJECT_SYNC_MAX_QUEUE_DEPTH
 from onyx.connectors.file.connector import LocalFileConnector
 from onyx.connectors.models import Document
 from onyx.connectors.models import HierarchyNode
@@ -57,14 +65,73 @@ def _user_file_lock_key(user_file_id: str | UUID) -> str:
     return f"{OnyxRedisLocks.USER_FILE_PROCESSING_LOCK_PREFIX}:{user_file_id}"
 
 
+def _user_file_queued_key(user_file_id: str | UUID) -> str:
+    """Key that exists while a process_single_user_file task is sitting in the queue.
+
+    The beat generator sets this with a TTL equal to CELERY_USER_FILE_PROCESSING_TASK_EXPIRES
+    before enqueuing and the worker deletes it as its first action.  This prevents
+    the beat from adding duplicate tasks for files that already have a live task
+    in flight.
+    """
+    return f"{OnyxRedisLocks.USER_FILE_QUEUED_PREFIX}:{user_file_id}"
+
+
 def _user_file_project_sync_lock_key(user_file_id: str | UUID) -> str:
     return f"{OnyxRedisLocks.USER_FILE_PROJECT_SYNC_LOCK_PREFIX}:{user_file_id}"
 
 
+def _user_file_project_sync_queued_key(user_file_id: str | UUID) -> str:
+    return f"{OnyxRedisLocks.USER_FILE_PROJECT_SYNC_QUEUED_PREFIX}:{user_file_id}"
+
+
 def _user_file_delete_lock_key(user_file_id: str | UUID) -> str:
     return f"{OnyxRedisLocks.USER_FILE_DELETE_LOCK_PREFIX}:{user_file_id}"
 
 
+def get_user_file_project_sync_queue_depth(celery_app: Celery) -> int:
+    redis_celery: Redis = celery_app.broker_connection().channel().client  # type: ignore
+    return celery_get_queue_length(
+        OnyxCeleryQueues.USER_FILE_PROJECT_SYNC, redis_celery
+    )
+
+
+def enqueue_user_file_project_sync_task(
+    *,
+    celery_app: Celery,
+    redis_client: Redis,
+    user_file_id: str | UUID,
+    tenant_id: str,
+    priority: OnyxCeleryPriority = OnyxCeleryPriority.HIGH,
+) -> bool:
+    """Enqueue a project-sync task if no matching queued task already exists."""
+    queued_key = _user_file_project_sync_queued_key(user_file_id)
+
+    # NX+EX gives us atomic dedupe and a self-healing TTL.
+    queued_guard_set = redis_client.set(
+        queued_key,
+        1,
+        nx=True,
+        ex=CELERY_USER_FILE_PROJECT_SYNC_TASK_EXPIRES,
+    )
+    if not queued_guard_set:
+        return False
+
+    try:
+        celery_app.send_task(
+            OnyxCeleryTask.PROCESS_SINGLE_USER_FILE_PROJECT_SYNC,
+            kwargs={"user_file_id": str(user_file_id), "tenant_id": tenant_id},
+            queue=OnyxCeleryQueues.USER_FILE_PROJECT_SYNC,
+            priority=priority,
+            expires=CELERY_USER_FILE_PROJECT_SYNC_TASK_EXPIRES,
+        )
+    except Exception:
+        # Roll back the queued guard if task publish fails.
+        redis_client.delete(queued_key)
+        raise
+
+    return True
+
+
 @retry(tries=3, delay=1, backoff=2, jitter=(0.0, 1.0))
 def _visit_chunks(
     *,
@@ -120,7 +187,24 @@ def _get_document_chunk_count(
 def check_user_file_processing(self: Task, *, tenant_id: str) -> None:
     """Scan for user files with PROCESSING status and enqueue per-file tasks.
 
-    Uses direct Redis locks to avoid overlapping runs.
+    Three mechanisms prevent queue runaway:
+
+    1. **Queue depth backpressure** – if the broker queue already has more than
+       USER_FILE_PROCESSING_MAX_QUEUE_DEPTH items we skip this beat cycle
+       entirely.  Workers are clearly behind; adding more tasks would only make
+       the backlog worse.
+
+    2. **Per-file queued guard** – before enqueuing a task we set a short-lived
+       Redis key (TTL = CELERY_USER_FILE_PROCESSING_TASK_EXPIRES).  If that key
+       already exists the file already has a live task in the queue, so we skip
+       it.  The worker deletes the key the moment it picks up the task so the
+       next beat cycle can re-enqueue if the file is still PROCESSING.
+
+    3. **Task expiry** – every enqueued task carries an `expires` value equal to
+       CELERY_USER_FILE_PROCESSING_TASK_EXPIRES.  If a task is still sitting in
+       the queue after that deadline, Celery discards it without touching the DB.
+       This is a belt-and-suspenders defence: even if the guard key is lost (e.g.
+       Redis restart), stale tasks evict themselves rather than piling up forever.
     """
     task_logger.info("check_user_file_processing - Starting")
 
@@ -135,7 +219,21 @@ def check_user_file_processing(self: Task, *, tenant_id: str) -> None:
         return None
 
     enqueued = 0
+    skipped_guard = 0
     try:
+        # --- Protection 1: queue depth backpressure ---
+        r_celery = self.app.broker_connection().channel().client  # type: ignore
+        queue_len = celery_get_queue_length(
+            OnyxCeleryQueues.USER_FILE_PROCESSING, r_celery
+        )
+        if queue_len > USER_FILE_PROCESSING_MAX_QUEUE_DEPTH:
+            task_logger.warning(
+                f"check_user_file_processing - Queue depth {queue_len} exceeds "
+                f"{USER_FILE_PROCESSING_MAX_QUEUE_DEPTH}, skipping enqueue for "
+                f"tenant={tenant_id}"
+            )
+            return None
+
         with get_session_with_current_tenant() as db_session:
             user_file_ids = (
                 db_session.execute(
@@ -148,12 +246,35 @@ def check_user_file_processing(self: Task, *, tenant_id: str) -> None:
             )
 
             for user_file_id in user_file_ids:
-                self.app.send_task(
-                    OnyxCeleryTask.PROCESS_SINGLE_USER_FILE,
-                    kwargs={"user_file_id": str(user_file_id), "tenant_id": tenant_id},
-                    queue=OnyxCeleryQueues.USER_FILE_PROCESSING,
-                    priority=OnyxCeleryPriority.HIGH,
+                # --- Protection 2: per-file queued guard ---
+                queued_key = _user_file_queued_key(user_file_id)
+                guard_set = redis_client.set(
+                    queued_key,
+                    1,
+                    ex=CELERY_USER_FILE_PROCESSING_TASK_EXPIRES,
+                    nx=True,
                 )
+                if not guard_set:
+                    skipped_guard += 1
+                    continue
+
+                # --- Protection 3: task expiry ---
+                # If task submission fails, clear the guard immediately so the
+                # next beat cycle can retry enqueuing this file.
+                try:
+                    self.app.send_task(
+                        OnyxCeleryTask.PROCESS_SINGLE_USER_FILE,
+                        kwargs={
+                            "user_file_id": str(user_file_id),
+                            "tenant_id": tenant_id,
+                        },
+                        queue=OnyxCeleryQueues.USER_FILE_PROCESSING,
+                        priority=OnyxCeleryPriority.HIGH,
+                        expires=CELERY_USER_FILE_PROCESSING_TASK_EXPIRES,
+                    )
+                except Exception:
+                    redis_client.delete(queued_key)
+                    raise
                 enqueued += 1
 
     finally:
@@ -161,7 +282,8 @@ def check_user_file_processing(self: Task, *, tenant_id: str) -> None:
             lock.release()
 
     task_logger.info(
-        f"check_user_file_processing - Enqueued {enqueued} tasks for tenant={tenant_id}"
+        f"check_user_file_processing - Enqueued {enqueued} skipped_guard={skipped_guard} "
+        f"tasks for tenant={tenant_id}"
     )
     return None
 
@@ -304,6 +426,12 @@ def process_single_user_file(
     start = time.monotonic()
 
     redis_client = get_redis_client(tenant_id=tenant_id)
+
+    # Clear the "queued" guard set by the beat generator so that the next beat
+    # cycle can re-enqueue this file if it is still in PROCESSING state after
+    # this task completes or fails.
+    redis_client.delete(_user_file_queued_key(user_file_id))
+
     file_lock: RedisLock = redis_client.lock(
         _user_file_lock_key(user_file_id),
         timeout=CELERY_USER_FILE_PROCESSING_LOCK_TIMEOUT,
@@ -557,8 +685,8 @@ def process_single_user_file_delete(
     ignore_result=True,
 )
 def check_for_user_file_project_sync(self: Task, *, tenant_id: str) -> None:
-    """Scan for user files with PROJECT_SYNC status and enqueue per-file tasks."""
-    task_logger.info("check_for_user_file_project_sync - Starting")
+    """Scan for user files needing project sync and enqueue per-file tasks."""
+    task_logger.info("Starting")
 
     redis_client = get_redis_client(tenant_id=tenant_id)
     lock: RedisLock = redis_client.lock(
@@ -570,13 +698,25 @@ def check_for_user_file_project_sync(self: Task, *, tenant_id: str) -> None:
         return None
 
     enqueued = 0
+    skipped_guard = 0
     try:
+        queue_depth = get_user_file_project_sync_queue_depth(self.app)
+        if queue_depth > USER_FILE_PROJECT_SYNC_MAX_QUEUE_DEPTH:
+            task_logger.warning(
+                f"Queue depth {queue_depth} exceeds "
+                f"{USER_FILE_PROJECT_SYNC_MAX_QUEUE_DEPTH}, skipping enqueue for tenant={tenant_id}"
+            )
+            return None
+
         with get_session_with_current_tenant() as db_session:
             user_file_ids = (
                 db_session.execute(
                     select(UserFile.id).where(
                         sa.and_(
-                            UserFile.needs_project_sync.is_(True),
+                            sa.or_(
+                                UserFile.needs_project_sync.is_(True),
+                                UserFile.needs_persona_sync.is_(True),
+                            ),
                             UserFile.status == UserFileStatus.COMPLETED,
                         )
                     )
@@ -586,19 +726,23 @@ def check_for_user_file_project_sync(self: Task, *, tenant_id: str) -> None:
             )
 
             for user_file_id in user_file_ids:
-                self.app.send_task(
-                    OnyxCeleryTask.PROCESS_SINGLE_USER_FILE_PROJECT_SYNC,
-                    kwargs={"user_file_id": str(user_file_id), "tenant_id": tenant_id},
-                    queue=OnyxCeleryQueues.USER_FILE_PROJECT_SYNC,
+                if not enqueue_user_file_project_sync_task(
+                    celery_app=self.app,
+                    redis_client=redis_client,
+                    user_file_id=user_file_id,
+                    tenant_id=tenant_id,
                     priority=OnyxCeleryPriority.HIGH,
-                )
+                ):
+                    skipped_guard += 1
+                    continue
                 enqueued += 1
     finally:
         if lock.owned():
             lock.release()
 
     task_logger.info(
-        f"check_for_user_file_project_sync - Enqueued {enqueued} tasks for tenant={tenant_id}"
+        f"Enqueued {enqueued} "
+        f"Skipped guard {skipped_guard} tasks for tenant={tenant_id}"
     )
     return None
 
@@ -617,6 +761,8 @@ def process_single_user_file_project_sync(
     )
 
     redis_client = get_redis_client(tenant_id=tenant_id)
+    redis_client.delete(_user_file_project_sync_queued_key(user_file_id))
+
     file_lock: RedisLock = redis_client.lock(
         _user_file_project_sync_lock_key(user_file_id),
         timeout=CELERY_USER_FILE_PROJECT_SYNC_LOCK_TIMEOUT,
@@ -630,7 +776,11 @@ def process_single_user_file_project_sync(
 
     try:
         with get_session_with_current_tenant() as db_session:
-            user_file = db_session.get(UserFile, _as_uuid(user_file_id))
+            user_file = db_session.execute(
+                select(UserFile)
+                .where(UserFile.id == _as_uuid(user_file_id))
+                .options(selectinload(UserFile.assistants))
+            ).scalar_one_or_none()
             if not user_file:
                 task_logger.info(
                     f"process_single_user_file_project_sync - User file not found id={user_file_id}"
@@ -658,13 +808,17 @@ def process_single_user_file_project_sync(
                 ]
 
                 project_ids = [project.id for project in user_file.projects]
+                persona_ids = [p.id for p in user_file.assistants if not p.deleted]
                 for retry_document_index in retry_document_indices:
                     retry_document_index.update_single(
                         doc_id=str(user_file.id),
                         tenant_id=tenant_id,
                         chunk_count=user_file.chunk_count,
                         fields=None,
-                        user_fields=VespaDocumentUserFields(user_projects=project_ids),
+                        user_fields=VespaDocumentUserFields(
+                            user_projects=project_ids,
+                            personas=persona_ids,
+                        ),
                     )
 
             task_logger.info(
@@ -672,6 +826,7 @@ def process_single_user_file_project_sync(
             )
 
             user_file.needs_project_sync = False
+            user_file.needs_persona_sync = False
             user_file.last_project_sync_at = datetime.datetime.now(
                 datetime.timezone.utc
             )

backend/onyx/background/indexing/run_docfetching.py

@@ -58,6 +58,8 @@ from onyx.file_store.document_batch_storage import DocumentBatchStorage
 from onyx.file_store.document_batch_storage import get_document_batch_storage
 from onyx.indexing.indexing_heartbeat import IndexingHeartbeatInterface
 from onyx.indexing.indexing_pipeline import index_doc_batch_prepare
+from onyx.indexing.postgres_sanitization import sanitize_document_for_postgres
+from onyx.indexing.postgres_sanitization import sanitize_hierarchy_nodes_for_postgres
 from onyx.redis.redis_hierarchy import cache_hierarchy_nodes_batch
 from onyx.redis.redis_hierarchy import ensure_source_node_exists
 from onyx.redis.redis_hierarchy import get_node_id_from_raw_id
@@ -156,36 +158,7 @@ def strip_null_characters(doc_batch: list[Document]) -> list[Document]:
             logger.warning(
                 f"doc {doc.id} too large, Document size: {sys.getsizeof(doc)}"
             )
-        cleaned_doc = doc.model_copy()
-
-        # Postgres cannot handle NUL characters in text fields
-        if "\x00" in cleaned_doc.id:
-            logger.warning(f"NUL characters found in document ID: {cleaned_doc.id}")
-            cleaned_doc.id = cleaned_doc.id.replace("\x00", "")
-
-        if cleaned_doc.title and "\x00" in cleaned_doc.title:
-            logger.warning(
-                f"NUL characters found in document title: {cleaned_doc.title}"
-            )
-            cleaned_doc.title = cleaned_doc.title.replace("\x00", "")
-
-        if "\x00" in cleaned_doc.semantic_identifier:
-            logger.warning(
-                f"NUL characters found in document semantic identifier: {cleaned_doc.semantic_identifier}"
-            )
-            cleaned_doc.semantic_identifier = cleaned_doc.semantic_identifier.replace(
-                "\x00", ""
-            )
-
-        for section in cleaned_doc.sections:
-            if section.link is not None:
-                section.link = section.link.replace("\x00", "")
-
-            # since text can be longer, just replace to avoid double scan
-            if isinstance(section, TextSection) and section.text is not None:
-                section.text = section.text.replace("\x00", "")
-
-        cleaned_batch.append(cleaned_doc)
+        cleaned_batch.append(sanitize_document_for_postgres(doc))
 
     return cleaned_batch
 
@@ -602,10 +575,13 @@ def connector_document_extraction(
 
                 # Process hierarchy nodes batch - upsert to Postgres and cache in Redis
                 if hierarchy_node_batch:
+                    hierarchy_node_batch_cleaned = (
+                        sanitize_hierarchy_nodes_for_postgres(hierarchy_node_batch)
+                    )
                     with get_session_with_current_tenant() as db_session:
                         upserted_nodes = upsert_hierarchy_nodes_batch(
                             db_session=db_session,
-                            nodes=hierarchy_node_batch,
+                            nodes=hierarchy_node_batch_cleaned,
                             source=db_connector.source,
                             commit=True,
                             is_connector_public=is_connector_public,
@@ -624,7 +600,7 @@ def connector_document_extraction(
                         )
 
                     logger.debug(
-                        f"Persisted and cached {len(hierarchy_node_batch)} hierarchy nodes "
+                        f"Persisted and cached {len(hierarchy_node_batch_cleaned)} hierarchy nodes "
                         f"for attempt={index_attempt_id}"
                     )
 

backend/onyx/chat/chat_utils.py

@@ -1,3 +1,4 @@
+import json
 import re
 from collections.abc import Callable
 from typing import cast
@@ -45,6 +46,7 @@ from onyx.utils.timing import log_function_time
 
 
 logger = setup_logger()
+IMAGE_GENERATION_TOOL_NAME = "generate_image"
 
 
 def create_chat_session_from_request(
@@ -422,6 +424,40 @@ def convert_chat_history_basic(
     return list(reversed(trimmed_reversed))
 
 
+def _build_tool_call_response_history_message(
+    tool_name: str,
+    generated_images: list[dict] | None,
+    tool_call_response: str | None,
+) -> str:
+    if tool_name != IMAGE_GENERATION_TOOL_NAME:
+        return TOOL_CALL_RESPONSE_CROSS_MESSAGE
+
+    if generated_images:
+        llm_image_context: list[dict[str, str]] = []
+        for image in generated_images:
+            file_id = image.get("file_id")
+            revised_prompt = image.get("revised_prompt")
+            if not isinstance(file_id, str):
+                continue
+
+            llm_image_context.append(
+                {
+                    "file_id": file_id,
+                    "revised_prompt": (
+                        revised_prompt if isinstance(revised_prompt, str) else ""
+                    ),
+                }
+            )
+
+        if llm_image_context:
+            return json.dumps(llm_image_context)
+
+    if tool_call_response:
+        return tool_call_response
+
+    return TOOL_CALL_RESPONSE_CROSS_MESSAGE
+
+
 def convert_chat_history(
     chat_history: list[ChatMessage],
     files: list[ChatLoadedFile],
@@ -582,10 +618,24 @@ def convert_chat_history(
 
                     # Add TOOL_CALL_RESPONSE messages for each tool call in this turn
                     for tool_call in turn_tool_calls:
+                        tool_name = tool_id_to_name_map.get(
+                            tool_call.tool_id, "unknown"
+                        )
+                        tool_response_message = (
+                            _build_tool_call_response_history_message(
+                                tool_name=tool_name,
+                                generated_images=tool_call.generated_images,
+                                tool_call_response=tool_call.tool_call_response,
+                            )
+                        )
                         simple_messages.append(
                             ChatMessageSimple(
-                                message=TOOL_CALL_RESPONSE_CROSS_MESSAGE,
-                                token_count=20,  # Tiny overestimate
+                                message=tool_response_message,
+                                token_count=(
+                                    token_counter(tool_response_message)
+                                    if tool_name == IMAGE_GENERATION_TOOL_NAME
+                                    else 20
+                                ),
                                 message_type=MessageType.TOOL_CALL_RESPONSE,
                                 tool_call_id=tool_call.tool_call_id,
                                 image_files=None,

backend/onyx/chat/llm_loop.py

@@ -30,6 +30,7 @@ from onyx.configs.constants import DocumentSource
 from onyx.configs.constants import MessageType
 from onyx.context.search.models import SearchDoc
 from onyx.context.search.models import SearchDocsResponse
+from onyx.db.engine.sql_engine import get_session_with_current_tenant
 from onyx.db.memory import add_memory
 from onyx.db.memory import update_memory_at_index
 from onyx.db.memory import UserMemoryContext
@@ -656,7 +657,12 @@ def run_llm_loop(
         fallback_extraction_attempted: bool = False
         citation_mapping: dict[int, str] = {}  # Maps citation_num -> document_id/URL
 
-        default_base_system_prompt: str = get_default_base_system_prompt(db_session)
+        # Fetch this in a short-lived session so the long-running stream loop does
+        # not pin a connection just to keep read state alive.
+        with get_session_with_current_tenant() as prompt_db_session:
+            default_base_system_prompt: str = get_default_base_system_prompt(
+                prompt_db_session
+            )
         system_prompt = None
         custom_agent_prompt_msg = None
 

backend/onyx/chat/process_message.py

@@ -856,6 +856,11 @@ def handle_stream_message_objects(
                 reserved_tokens=reserved_token_count,
             )
 
+        # Release any read transaction before entering the long-running LLM stream.
+        # Without this, the request-scoped session can keep a connection checked out
+        # for the full stream duration.
+        db_session.commit()
+
         # The stream generator can resume on a different worker thread after early yields.
         # Set this right before launching the LLM loop so run_in_background copies the right context.
         if new_msg_req.mock_llm_response is not None:

backend/onyx/chat/prompt_utils.py

@@ -190,7 +190,7 @@ def _build_user_information_section(
     if not sections:
         return ""
 
-    return USER_INFORMATION_HEADER + "".join(sections)
+    return USER_INFORMATION_HEADER + "\n".join(sections)
 
 
 def build_system_prompt(
@@ -228,23 +228,21 @@ def build_system_prompt(
         system_prompt += REQUIRE_CITATION_GUIDANCE
 
     if include_all_guidance:
-        system_prompt += (
-            TOOL_SECTION_HEADER
-            + TOOL_DESCRIPTION_SEARCH_GUIDANCE
-            + INTERNAL_SEARCH_GUIDANCE
-            + WEB_SEARCH_GUIDANCE.format(
+        tool_sections = [
+            TOOL_DESCRIPTION_SEARCH_GUIDANCE,
+            INTERNAL_SEARCH_GUIDANCE,
+            WEB_SEARCH_GUIDANCE.format(
                 site_colon_disabled=WEB_SEARCH_SITE_DISABLED_GUIDANCE
-            )
-            + OPEN_URLS_GUIDANCE
-            + PYTHON_TOOL_GUIDANCE
-            + GENERATE_IMAGE_GUIDANCE
-            + MEMORY_GUIDANCE
-        )
+            ),
+            OPEN_URLS_GUIDANCE,
+            PYTHON_TOOL_GUIDANCE,
+            GENERATE_IMAGE_GUIDANCE,
+            MEMORY_GUIDANCE,
+        ]
+        system_prompt += TOOL_SECTION_HEADER + "\n".join(tool_sections)
         return system_prompt
 
     if tools:
-        system_prompt += TOOL_SECTION_HEADER
-
         has_web_search = any(isinstance(tool, WebSearchTool) for tool in tools)
         has_internal_search = any(isinstance(tool, SearchTool) for tool in tools)
         has_open_urls = any(isinstance(tool, OpenURLTool) for tool in tools)
@@ -254,12 +252,14 @@ def build_system_prompt(
         )
         has_memory = any(isinstance(tool, MemoryTool) for tool in tools)
 
+        tool_guidance_sections: list[str] = []
+
         if has_web_search or has_internal_search or include_all_guidance:
-            system_prompt += TOOL_DESCRIPTION_SEARCH_GUIDANCE
+            tool_guidance_sections.append(TOOL_DESCRIPTION_SEARCH_GUIDANCE)
 
         # These are not included at the Tool level because the ordering may matter.
         if has_internal_search or include_all_guidance:
-            system_prompt += INTERNAL_SEARCH_GUIDANCE
+            tool_guidance_sections.append(INTERNAL_SEARCH_GUIDANCE)
 
         if has_web_search or include_all_guidance:
             site_disabled_guidance = ""
@@ -269,20 +269,23 @@ def build_system_prompt(
                 )
                 if web_search_tool and not web_search_tool.supports_site_filter:
                     site_disabled_guidance = WEB_SEARCH_SITE_DISABLED_GUIDANCE
-            system_prompt += WEB_SEARCH_GUIDANCE.format(
-                site_colon_disabled=site_disabled_guidance
+            tool_guidance_sections.append(
+                WEB_SEARCH_GUIDANCE.format(site_colon_disabled=site_disabled_guidance)
             )
 
         if has_open_urls or include_all_guidance:
-            system_prompt += OPEN_URLS_GUIDANCE
+            tool_guidance_sections.append(OPEN_URLS_GUIDANCE)
 
         if has_python or include_all_guidance:
-            system_prompt += PYTHON_TOOL_GUIDANCE
+            tool_guidance_sections.append(PYTHON_TOOL_GUIDANCE)
 
         if has_generate_image or include_all_guidance:
-            system_prompt += GENERATE_IMAGE_GUIDANCE
+            tool_guidance_sections.append(GENERATE_IMAGE_GUIDANCE)
 
         if has_memory or include_all_guidance:
-            system_prompt += MEMORY_GUIDANCE
+            tool_guidance_sections.append(MEMORY_GUIDANCE)
+
+        if tool_guidance_sections:
+            system_prompt += TOOL_SECTION_HEADER + "\n".join(tool_guidance_sections)
 
     return system_prompt

backend/onyx/configs/app_configs.py

@@ -210,10 +210,10 @@ AUTH_COOKIE_EXPIRE_TIME_SECONDS = int(
 REQUIRE_EMAIL_VERIFICATION = (
     os.environ.get("REQUIRE_EMAIL_VERIFICATION", "").lower() == "true"
 )
-SMTP_SERVER = os.environ.get("SMTP_SERVER") or "smtp.gmail.com"
+SMTP_SERVER = os.environ.get("SMTP_SERVER") or ""
 SMTP_PORT = int(os.environ.get("SMTP_PORT") or "587")
-SMTP_USER = os.environ.get("SMTP_USER", "your-email@gmail.com")
-SMTP_PASS = os.environ.get("SMTP_PASS", "your-gmail-password")
+SMTP_USER = os.environ.get("SMTP_USER") or ""
+SMTP_PASS = os.environ.get("SMTP_PASS") or ""
 EMAIL_FROM = os.environ.get("EMAIL_FROM") or SMTP_USER
 
 SENDGRID_API_KEY = os.environ.get("SENDGRID_API_KEY") or ""
@@ -251,7 +251,9 @@ DEFAULT_OPENSEARCH_QUERY_TIMEOUT_S = int(
     os.environ.get("DEFAULT_OPENSEARCH_QUERY_TIMEOUT_S") or 50
 )
 OPENSEARCH_ADMIN_USERNAME = os.environ.get("OPENSEARCH_ADMIN_USERNAME", "admin")
-OPENSEARCH_ADMIN_PASSWORD = os.environ.get("OPENSEARCH_ADMIN_PASSWORD", "")
+OPENSEARCH_ADMIN_PASSWORD = os.environ.get(
+    "OPENSEARCH_ADMIN_PASSWORD", "StrongPassword123!"
+)
 USING_AWS_MANAGED_OPENSEARCH = (
     os.environ.get("USING_AWS_MANAGED_OPENSEARCH", "").lower() == "true"
 )
@@ -282,6 +284,9 @@ OPENSEARCH_TEXT_ANALYZER = os.environ.get("OPENSEARCH_TEXT_ANALYZER") or "englis
 ENABLE_OPENSEARCH_INDEXING_FOR_ONYX = (
     os.environ.get("ENABLE_OPENSEARCH_INDEXING_FOR_ONYX", "").lower() == "true"
 )
+# NOTE: This effectively does nothing anymore, admins can now toggle whether
+# retrieval is through OpenSearch. This value is only used as a final fallback
+# in case that doesn't work for whatever reason.
 # Given that the "base" config above is true, this enables whether we want to
 # retrieve from OpenSearch or Vespa. We want to be able to quickly toggle this
 # in the event we see issues with OpenSearch retrieval in our dev environments.
@@ -637,6 +642,14 @@ SHAREPOINT_CONNECTOR_SIZE_THRESHOLD = int(
     os.environ.get("SHAREPOINT_CONNECTOR_SIZE_THRESHOLD", 20 * 1024 * 1024)
 )
 
+# When True, group sync enumerates every Azure AD group in the tenant (expensive).
+# When False (default), only groups found in site role assignments are synced.
+# Can be overridden per-connector via the "exhaustive_ad_enumeration" key in
+# connector_specific_config.
+SHAREPOINT_EXHAUSTIVE_AD_ENUMERATION = (
+    os.environ.get("SHAREPOINT_EXHAUSTIVE_AD_ENUMERATION", "").lower() == "true"
+)
+
 BLOB_STORAGE_SIZE_THRESHOLD = int(
     os.environ.get("BLOB_STORAGE_SIZE_THRESHOLD", 20 * 1024 * 1024)
 )

backend/onyx/configs/constants.py

@@ -157,6 +157,25 @@ CELERY_EXTERNAL_GROUP_SYNC_LOCK_TIMEOUT = 300  # 5 min
 
 CELERY_USER_FILE_PROCESSING_LOCK_TIMEOUT = 30 * 60  # 30 minutes (in seconds)
 
+# How long a queued user-file task is valid before workers discard it.
+# Should be longer than the beat interval (20 s) but short enough to prevent
+# indefinite queue growth.  Workers drop tasks older than this without touching
+# the DB, so a shorter value = faster drain of stale duplicates.
+CELERY_USER_FILE_PROCESSING_TASK_EXPIRES = 60  # 1 minute (in seconds)
+
+# Maximum number of tasks allowed in the user-file-processing queue before the
+# beat generator stops adding more.  Prevents unbounded queue growth when workers
+# fall behind.
+USER_FILE_PROCESSING_MAX_QUEUE_DEPTH = 500
+# How long a queued user-file-project-sync task remains valid.
+# Should be short enough to discard stale queue entries under load while still
+# allowing workers enough time to pick up new tasks.
+CELERY_USER_FILE_PROJECT_SYNC_TASK_EXPIRES = 60  # 1 minute (in seconds)
+
+# Max queue depth before user-file-project-sync producers stop enqueuing.
+# This applies backpressure when workers are falling behind.
+USER_FILE_PROJECT_SYNC_MAX_QUEUE_DEPTH = 500
+
 CELERY_USER_FILE_PROJECT_SYNC_LOCK_TIMEOUT = 5 * 60  # 5 minutes (in seconds)
 
 CELERY_SANDBOX_FILE_SYNC_LOCK_TIMEOUT = 5 * 60  # 5 minutes (in seconds)
@@ -443,8 +462,12 @@ class OnyxRedisLocks:
     # User file processing
     USER_FILE_PROCESSING_BEAT_LOCK = "da_lock:check_user_file_processing_beat"
     USER_FILE_PROCESSING_LOCK_PREFIX = "da_lock:user_file_processing"
+    # Short-lived key set when a task is enqueued; cleared when the worker picks it up.
+    # Prevents the beat from re-enqueuing the same file while a task is already queued.
+    USER_FILE_QUEUED_PREFIX = "da_lock:user_file_queued"
     USER_FILE_PROJECT_SYNC_BEAT_LOCK = "da_lock:check_user_file_project_sync_beat"
     USER_FILE_PROJECT_SYNC_LOCK_PREFIX = "da_lock:user_file_project_sync"
+    USER_FILE_PROJECT_SYNC_QUEUED_PREFIX = "da_lock:user_file_project_sync_queued"
     USER_FILE_DELETE_BEAT_LOCK = "da_lock:check_user_file_delete_beat"
     USER_FILE_DELETE_LOCK_PREFIX = "da_lock:user_file_delete"
 

backend/onyx/connectors/google_utils/google_utils.py

@@ -16,6 +16,22 @@ from onyx.utils.retry_wrapper import retry_builder
 
 logger = setup_logger()
 
+_RATE_LIMIT_REASONS = {"userRateLimitExceeded", "rateLimitExceeded"}
+
+
+def _is_rate_limit_error(error: HttpError) -> bool:
+    """Google sometimes returns rate-limit errors as 403 with reason
+    'userRateLimitExceeded' instead of 429. This helper detects both."""
+    if error.resp.status == 429:
+        return True
+    if error.resp.status != 403:
+        return False
+    error_details = getattr(error, "error_details", None) or []
+    for detail in error_details:
+        if isinstance(detail, dict) and detail.get("reason") in _RATE_LIMIT_REASONS:
+            return True
+    return "userRateLimitExceeded" in str(error) or "rateLimitExceeded" in str(error)
+
 
 # Google Drive APIs are quite flakey and may 500 for an
 # extended period of time. This is now addressed by checkpointing.
@@ -57,7 +73,7 @@ def _execute_with_retry(request: Any) -> Any:
         except HttpError as error:
             attempt += 1
 
-            if error.resp.status == 429:
+            if _is_rate_limit_error(error):
                 # Attempt to get 'Retry-After' from headers
                 retry_after = error.resp.get("Retry-After")
                 if retry_after:
@@ -140,16 +156,16 @@ def _execute_single_retrieval(
                 )
             logger.error(f"Error executing request: {e}")
             raise e
+        elif _is_rate_limit_error(e):
+            results = _execute_with_retry(
+                lambda: retrieval_function(**request_kwargs).execute()
+            )
         elif e.resp.status == 404 or e.resp.status == 403:
             if continue_on_404_or_403:
                 logger.debug(f"Error executing request: {e}")
                 results = {}
             else:
                 raise e
-        elif e.resp.status == 429:
-            results = _execute_with_retry(
-                lambda: retrieval_function(**request_kwargs).execute()
-            )
         else:
             logger.exception("Error executing request:")
             raise e

backend/onyx/connectors/microsoft_graph_env.py

@@ -0,0 +1,96 @@
+"""Inverse mapping from user-facing Microsoft host URLs to the SDK's AzureEnvironment.
+
+The office365 library's GraphClient requires an ``AzureEnvironment`` string
+(e.g. ``"Global"``, ``"GCC High"``) to route requests to the correct national
+cloud.  Our connectors instead expose free-text ``authority_host`` and
+``graph_api_host`` fields so the frontend doesn't need to know about SDK
+internals.
+
+This module bridges the gap: given the two host URLs the user configured, it
+resolves the matching ``AzureEnvironment`` value (and the implied SharePoint
+domain suffix) so callers can pass ``environment=…`` to ``GraphClient``.
+"""
+
+from office365.graph_client import AzureEnvironment  # type: ignore[import-untyped]
+from pydantic import BaseModel
+
+from onyx.connectors.exceptions import ConnectorValidationError
+
+
+class MicrosoftGraphEnvironment(BaseModel):
+    """One row of the inverse mapping."""
+
+    environment: str
+    graph_host: str
+    authority_host: str
+    sharepoint_domain_suffix: str
+
+
+_ENVIRONMENTS: list[MicrosoftGraphEnvironment] = [
+    MicrosoftGraphEnvironment(
+        environment=AzureEnvironment.Global,
+        graph_host="https://graph.microsoft.com",
+        authority_host="https://login.microsoftonline.com",
+        sharepoint_domain_suffix="sharepoint.com",
+    ),
+    MicrosoftGraphEnvironment(
+        environment=AzureEnvironment.USGovernmentHigh,
+        graph_host="https://graph.microsoft.us",
+        authority_host="https://login.microsoftonline.us",
+        sharepoint_domain_suffix="sharepoint.us",
+    ),
+    MicrosoftGraphEnvironment(
+        environment=AzureEnvironment.USGovernmentDoD,
+        graph_host="https://dod-graph.microsoft.us",
+        authority_host="https://login.microsoftonline.us",
+        sharepoint_domain_suffix="sharepoint.us",
+    ),
+    MicrosoftGraphEnvironment(
+        environment=AzureEnvironment.China,
+        graph_host="https://microsoftgraph.chinacloudapi.cn",
+        authority_host="https://login.chinacloudapi.cn",
+        sharepoint_domain_suffix="sharepoint.cn",
+    ),
+    MicrosoftGraphEnvironment(
+        environment=AzureEnvironment.Germany,
+        graph_host="https://graph.microsoft.de",
+        authority_host="https://login.microsoftonline.de",
+        sharepoint_domain_suffix="sharepoint.de",
+    ),
+]
+
+_GRAPH_HOST_INDEX: dict[str, MicrosoftGraphEnvironment] = {
+    env.graph_host: env for env in _ENVIRONMENTS
+}
+
+
+def resolve_microsoft_environment(
+    graph_api_host: str,
+    authority_host: str,
+) -> MicrosoftGraphEnvironment:
+    """Return the ``MicrosoftGraphEnvironment`` that matches the supplied hosts.
+
+    Raises ``ConnectorValidationError`` when the combination is unknown or
+    internally inconsistent (e.g. a GCC-High graph host paired with a
+    commercial authority host).
+    """
+    graph_api_host = graph_api_host.rstrip("/")
+    authority_host = authority_host.rstrip("/")
+
+    env = _GRAPH_HOST_INDEX.get(graph_api_host)
+    if env is None:
+        known = ", ".join(sorted(_GRAPH_HOST_INDEX))
+        raise ConnectorValidationError(
+            f"Unsupported Microsoft Graph API host '{graph_api_host}'. "
+            f"Recognised hosts: {known}"
+        )
+
+    if env.authority_host != authority_host:
+        raise ConnectorValidationError(
+            f"Authority host '{authority_host}' is inconsistent with "
+            f"graph API host '{graph_api_host}'. "
+            f"Expected authority host '{env.authority_host}' "
+            f"for the {env.environment} environment."
+        )
+
+    return env

backend/onyx/connectors/models.py

@@ -6,6 +6,7 @@ from typing import cast
 
 from pydantic import BaseModel
 from pydantic import Field
+from pydantic import field_validator
 from pydantic import model_validator
 
 from onyx.access.models import ExternalAccess
@@ -167,6 +168,14 @@ class DocumentBase(BaseModel):
     # list of strings.
     metadata: dict[str, str | list[str]]
 
+    @field_validator("metadata", mode="before")
+    @classmethod
+    def _coerce_metadata_values(cls, v: dict[str, Any]) -> dict[str, str | list[str]]:
+        return {
+            key: [str(item) for item in val] if isinstance(val, list) else str(val)
+            for key, val in v.items()
+        }
+
     # UTC time
     doc_updated_at: datetime | None = None
     chunk_count: int | None = None

backend/onyx/connectors/sharepoint/connector.py

@@ -47,6 +47,7 @@ from onyx.connectors.interfaces import GenerateSlimDocumentOutput
 from onyx.connectors.interfaces import IndexingHeartbeatInterface
 from onyx.connectors.interfaces import SecondsSinceUnixEpoch
 from onyx.connectors.interfaces import SlimConnectorWithPermSync
+from onyx.connectors.microsoft_graph_env import resolve_microsoft_environment
 from onyx.connectors.models import BasicExpertInfo
 from onyx.connectors.models import ConnectorCheckpoint
 from onyx.connectors.models import ConnectorFailure
@@ -83,7 +84,11 @@ SHARED_DOCUMENTS_MAP_REVERSE = {v: k for k, v in SHARED_DOCUMENTS_MAP.items()}
 
 ASPX_EXTENSION = ".aspx"
 
-GRAPH_API_BASE = "https://graph.microsoft.com/v1.0"
+DEFAULT_AUTHORITY_HOST = "https://login.microsoftonline.com"
+DEFAULT_GRAPH_API_HOST = "https://graph.microsoft.com"
+DEFAULT_SHAREPOINT_DOMAIN_SUFFIX = "sharepoint.com"
+
+GRAPH_API_BASE = f"{DEFAULT_GRAPH_API_HOST}/v1.0"
 GRAPH_API_MAX_RETRIES = 5
 GRAPH_API_RETRYABLE_STATUSES = frozenset({429, 500, 502, 503, 504})
 
@@ -142,7 +147,9 @@ class DriveItemData(BaseModel):
             self.id,
             ResourcePath("items", ResourcePath(self.drive_id, ResourcePath("drives"))),
         )
-        return DriveItem(graph_client, path)
+        item = DriveItem(graph_client, path)
+        item.set_property("id", self.id)
+        return item
 
 
 # The office365 library's ClientContext caches the access token from its
@@ -176,6 +183,25 @@ class CertificateData(BaseModel):
     thumbprint: str
 
 
+def _site_page_in_time_window(
+    page: dict[str, Any],
+    start: datetime | None,
+    end: datetime | None,
+) -> bool:
+    """Return True if the page's lastModifiedDateTime falls within [start, end]."""
+    if start is None and end is None:
+        return True
+    raw = page.get("lastModifiedDateTime")
+    if not raw:
+        return True
+    if not isinstance(raw, str):
+        raise ValueError(f"lastModifiedDateTime is not a string: {raw}")
+    last_modified = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+    return (start is None or last_modified >= start) and (
+        end is None or last_modified <= end
+    )
+
+
 def sleep_and_retry(
     query_obj: ClientQuery, method_name: str, max_retries: int = 3
 ) -> Any:
@@ -221,6 +247,12 @@ class SharepointConnectorCheckpoint(ConnectorCheckpoint):
     current_drive_name: str | None = None
     # Drive's web_url from the API - used as raw_node_id for DRIVE hierarchy nodes
     current_drive_web_url: str | None = None
+    # Resolved drive ID — avoids re-resolving on checkpoint resume
+    current_drive_id: str | None = None
+    # Next delta API page URL for per-page checkpointing within a drive.
+    # When set, Phase 3b fetches one page at a time so progress is persisted
+    # between pages.  None means BFS path or no active delta traversal.
+    current_drive_delta_next_link: str | None = None
 
     process_site_pages: bool = False
 
@@ -266,10 +298,12 @@ def load_certificate_from_pfx(pfx_data: bytes, password: str) -> CertificateData
 
 
 def acquire_token_for_rest(
-    msal_app: msal.ConfidentialClientApplication, sp_tenant_domain: str
+    msal_app: msal.ConfidentialClientApplication,
+    sp_tenant_domain: str,
+    sharepoint_domain_suffix: str,
 ) -> TokenResponse:
     token = msal_app.acquire_token_for_client(
-        scopes=[f"https://{sp_tenant_domain}.sharepoint.com/.default"]
+        scopes=[f"https://{sp_tenant_domain}.{sharepoint_domain_suffix}/.default"]
     )
     return TokenResponse.from_json(token)
 
@@ -384,12 +418,13 @@ def _download_via_graph_api(
     drive_id: str,
     item_id: str,
     bytes_allowed: int,
+    graph_api_base: str,
 ) -> bytes:
     """Download a drive item via the Graph API /content endpoint with a byte cap.
 
     Raises SizeCapExceeded if the cap is exceeded.
     """
-    url = f"{GRAPH_API_BASE}/drives/{drive_id}/items/{item_id}/content"
+    url = f"{graph_api_base}/drives/{drive_id}/items/{item_id}/content"
     headers = {"Authorization": f"Bearer {access_token}"}
     with requests.get(
         url, headers=headers, stream=True, timeout=REQUEST_TIMEOUT_SECONDS
@@ -410,6 +445,7 @@ def _convert_driveitem_to_document_with_permissions(
     drive_name: str,
     ctx: ClientContext | None,
     graph_client: GraphClient,
+    graph_api_base: str,
     include_permissions: bool = False,
     parent_hierarchy_raw_node_id: str | None = None,
     access_token: str | None = None,
@@ -466,6 +502,7 @@ def _convert_driveitem_to_document_with_permissions(
                 driveitem.drive_id,
                 driveitem.id,
                 SHAREPOINT_CONNECTOR_SIZE_THRESHOLD,
+                graph_api_base=graph_api_base,
             )
         except SizeCapExceeded:
             logger.warning(
@@ -785,6 +822,9 @@ class SharepointConnector(
         sites: list[str] = [],
         include_site_pages: bool = True,
         include_site_documents: bool = True,
+        authority_host: str = DEFAULT_AUTHORITY_HOST,
+        graph_api_host: str = DEFAULT_GRAPH_API_HOST,
+        sharepoint_domain_suffix: str = DEFAULT_SHAREPOINT_DOMAIN_SUFFIX,
     ) -> None:
         self.batch_size = batch_size
         self.sites = list(sites)
@@ -801,6 +841,20 @@ class SharepointConnector(
         self._cached_rest_ctx_url: str | None = None
         self._cached_rest_ctx_created_at: float = 0.0
 
+        resolved_env = resolve_microsoft_environment(graph_api_host, authority_host)
+        self._azure_environment = resolved_env.environment
+        self.authority_host = resolved_env.authority_host
+        self.graph_api_host = resolved_env.graph_host
+        self.graph_api_base = f"{self.graph_api_host}/v1.0"
+        self.sharepoint_domain_suffix = resolved_env.sharepoint_domain_suffix
+        if sharepoint_domain_suffix != resolved_env.sharepoint_domain_suffix:
+            logger.warning(
+                f"Configured sharepoint_domain_suffix '{sharepoint_domain_suffix}' "
+                f"differs from the expected suffix '{resolved_env.sharepoint_domain_suffix}' "
+                f"for the {resolved_env.environment} environment. "
+                f"Using '{resolved_env.sharepoint_domain_suffix}'."
+            )
+
     def validate_connector_settings(self) -> None:
         # Validate that at least one content type is enabled
         if not self.include_site_documents and not self.include_site_pages:
@@ -856,8 +910,9 @@ class SharepointConnector(
 
         msal_app = self.msal_app
         sp_tenant_domain = self.sp_tenant_domain
+        sp_domain_suffix = self.sharepoint_domain_suffix
         self._cached_rest_ctx = ClientContext(site_url).with_access_token(
-            lambda: acquire_token_for_rest(msal_app, sp_tenant_domain)
+            lambda: acquire_token_for_rest(msal_app, sp_tenant_domain, sp_domain_suffix)
         )
         self._cached_rest_ctx_url = site_url
         self._cached_rest_ctx_created_at = time.monotonic()
@@ -1117,76 +1172,36 @@ class SharepointConnector(
         site_descriptor: SiteDescriptor,
         start: datetime | None = None,
         end: datetime | None = None,
-    ) -> list[dict[str, Any]]:
-        """Fetch SharePoint site pages (.aspx files) using the SharePoint Pages API."""
+    ) -> Generator[dict[str, Any], None, None]:
+        """Yield SharePoint site pages (.aspx files) one at a time.
 
-        # Get the site to extract the site ID
+        Pages are fetched via the Graph Pages API and yielded lazily as each
+        API page arrives, so memory stays bounded regardless of total page count.
+        Time-window filtering is applied per-item before yielding.
+        """
         site = self.graph_client.sites.get_by_url(site_descriptor.url)
-        site.execute_query()  # Execute the query to actually fetch the data
+        site.execute_query()
         site_id = site.id
 
-        # Get the token acquisition function from the GraphClient
-        token_data = self._acquire_token()
-        access_token = token_data.get("access_token")
-        if not access_token:
-            raise RuntimeError("Failed to acquire access token")
-
-        # Construct the SharePoint Pages API endpoint
-        # Using API directly, since the Graph Client doesn't support the Pages API
-        pages_endpoint = f"https://graph.microsoft.com/v1.0/sites/{site_id}/pages/microsoft.graph.sitePage"
-
-        headers = {
-            "Authorization": f"Bearer {access_token}",
-            "Content-Type": "application/json",
-        }
-
-        # Add expand parameter to get canvas layout content
-        params = {"$expand": "canvasLayout"}
-
-        response = requests.get(
-            pages_endpoint,
-            headers=headers,
-            params=params,
-            timeout=REQUEST_TIMEOUT_SECONDS,
+        page_url: str | None = (
+            f"{self.graph_api_base}/sites/{site_id}" f"/pages/microsoft.graph.sitePage"
         )
-        response.raise_for_status()
-        pages_data = response.json()
-        all_pages = pages_data.get("value", [])
+        params: dict[str, str] | None = {"$expand": "canvasLayout"}
+        total_yielded = 0
 
-        # Handle pagination if there are more pages
-        # TODO: This accumulates all pages in memory and can be heavy on large tenants.
-        #       We should process each page incrementally to avoid unbounded growth.
-        while "@odata.nextLink" in pages_data:
-            next_url = pages_data["@odata.nextLink"]
-            response = requests.get(
-                next_url, headers=headers, timeout=REQUEST_TIMEOUT_SECONDS
-            )
-            response.raise_for_status()
-            pages_data = response.json()
-            all_pages.extend(pages_data.get("value", []))
+        while page_url:
+            data = self._graph_api_get_json(page_url, params)
+            params = None  # nextLink already embeds query params
 
-        logger.debug(f"Found {len(all_pages)} site pages in {site_descriptor.url}")
+            for page in data.get("value", []):
+                if not _site_page_in_time_window(page, start, end):
+                    continue
+                total_yielded += 1
+                yield page
 
-        # Filter pages based on time window if specified
-        if start is not None or end is not None:
-            filtered_pages: list[dict[str, Any]] = []
-            for page in all_pages:
-                page_modified = page.get("lastModifiedDateTime")
-                if page_modified:
-                    if isinstance(page_modified, str):
-                        page_modified = datetime.fromisoformat(
-                            page_modified.replace("Z", "+00:00")
-                        )
+            page_url = data.get("@odata.nextLink")
 
-                    if start is not None and page_modified < start:
-                        continue
-                    if end is not None and page_modified > end:
-                        continue
-
-                filtered_pages.append(page)
-            all_pages = filtered_pages
-
-        return all_pages
+        logger.debug(f"Yielded {total_yielded} site pages for {site_descriptor.url}")
 
     def _acquire_token(self) -> dict[str, Any]:
         """
@@ -1196,7 +1211,7 @@ class SharepointConnector(
             raise RuntimeError("MSAL app is not initialized")
 
         token = self.msal_app.acquire_token_for_client(
-            scopes=["https://graph.microsoft.com/.default"]
+            scopes=[f"{self.graph_api_host}/.default"]
         )
         return token
 
@@ -1269,9 +1284,10 @@ class SharepointConnector(
         Performs BFS folder traversal manually, fetching one page of children
         at a time so that memory usage stays bounded regardless of drive size.
         """
-        base = f"{GRAPH_API_BASE}/drives/{drive_id}"
+        base = f"{self.graph_api_base}/drives/{drive_id}"
         if folder_path:
-            start_url = f"{base}/root:/{folder_path}:/children"
+            encoded_path = quote(folder_path, safe="/")
+            start_url = f"{base}/root:/{encoded_path}:/children"
         else:
             start_url = f"{base}/root/children"
 
@@ -1329,7 +1345,7 @@ class SharepointConnector(
         """
         use_timestamp_token = start is not None and start > _EPOCH
 
-        initial_url = f"{GRAPH_API_BASE}/drives/{drive_id}/root/delta"
+        initial_url = f"{self.graph_api_base}/drives/{drive_id}/root/delta"
         if use_timestamp_token:
             assert start is not None  # mypy
             token = quote(start.isoformat(timespec="seconds"))
@@ -1375,7 +1391,7 @@ class SharepointConnector(
                         drive_id,
                     )
                     yield from self._iter_delta_pages(
-                        initial_url=f"{GRAPH_API_BASE}/drives/{drive_id}/root/delta",
+                        initial_url=f"{self.graph_api_base}/drives/{drive_id}/root/delta",
                         drive_id=drive_id,
                         start=start,
                         end=end,
@@ -1406,6 +1422,87 @@ class SharepointConnector(
             if not page_url:
                 break
 
+    def _build_delta_start_url(
+        self,
+        drive_id: str,
+        start: datetime | None = None,
+        page_size: int = 200,
+    ) -> str:
+        """Build the initial delta API URL with query parameters embedded.
+
+        Embeds ``$top`` (and optionally a timestamp ``token``) directly in the
+        URL so that the returned string is fully self-contained and can be
+        stored in a checkpoint without needing a separate params dict.
+        """
+        base_url = f"{self.graph_api_base}/drives/{drive_id}/root/delta"
+        params = [f"$top={page_size}"]
+        if start is not None and start > _EPOCH:
+            token = quote(start.isoformat(timespec="seconds"))
+            params.append(f"token={token}")
+        return f"{base_url}?{'&'.join(params)}"
+
+    def _fetch_one_delta_page(
+        self,
+        page_url: str,
+        drive_id: str,
+        start: datetime | None = None,
+        end: datetime | None = None,
+        page_size: int = 200,
+    ) -> tuple[list[DriveItemData], str | None]:
+        """Fetch a single page of delta API results.
+
+        Returns ``(items, next_page_url)``.  *next_page_url* is ``None`` when
+        the delta enumeration is complete (deltaLink with no nextLink).
+
+        On 410 Gone (expired token) returns ``([], full_resync_url)`` so
+        the caller can store the resync URL in the checkpoint and retry on
+        the next cycle.
+        """
+        try:
+            data = self._graph_api_get_json(page_url)
+        except requests.HTTPError as e:
+            if e.response is not None and e.response.status_code == 410:
+                logger.warning(
+                    "Delta token expired (410 Gone) for drive '%s'. "
+                    "Will restart with full delta enumeration.",
+                    drive_id,
+                )
+                full_url = (
+                    f"{self.graph_api_base}/drives/{drive_id}/root/delta"
+                    f"?$top={page_size}"
+                )
+                return [], full_url
+            raise
+
+        items: list[DriveItemData] = []
+        for item in data.get("value", []):
+            if "folder" in item or "deleted" in item:
+                continue
+            if start is not None or end is not None:
+                raw_ts = item.get("lastModifiedDateTime")
+                if raw_ts:
+                    mod_dt = datetime.fromisoformat(raw_ts.replace("Z", "+00:00"))
+                    if start is not None and mod_dt < start:
+                        continue
+                    if end is not None and mod_dt > end:
+                        continue
+            items.append(DriveItemData.from_graph_json(item))
+
+        next_url = data.get("@odata.nextLink")
+        if next_url:
+            return items, next_url
+        return items, None
+
+    @staticmethod
+    def _clear_drive_checkpoint_state(
+        checkpoint: "SharepointConnectorCheckpoint",
+    ) -> None:
+        """Reset all drive-level fields in the checkpoint."""
+        checkpoint.current_drive_name = None
+        checkpoint.current_drive_id = None
+        checkpoint.current_drive_web_url = None
+        checkpoint.current_drive_delta_next_link = None
+
     def _fetch_slim_documents_from_sharepoint(self) -> GenerateSlimDocumentOutput:
         site_descriptors = self.site_descriptors or self.fetch_sites()
 
@@ -1492,7 +1589,7 @@ class SharepointConnector(
         sp_private_key = credentials.get("sp_private_key")
         sp_certificate_password = credentials.get("sp_certificate_password")
 
-        authority_url = f"https://login.microsoftonline.com/{sp_directory_id}"
+        authority_url = f"{self.authority_host}/{sp_directory_id}"
 
         if auth_method == SharepointAuthMethod.CERTIFICATE.value:
             logger.info("Using certificate authentication")
@@ -1508,6 +1605,7 @@ class SharepointConnector(
             if certificate_data is None:
                 raise RuntimeError("Failed to load certificate")
 
+            logger.info(f"Creating MSAL app with authority url {authority_url}")
             self.msal_app = msal.ConfidentialClientApplication(
                 authority=authority_url,
                 client_id=sp_client_id,
@@ -1533,13 +1631,15 @@ class SharepointConnector(
                 raise ConnectorValidationError("MSAL app is not initialized")
 
             token = self.msal_app.acquire_token_for_client(
-                scopes=["https://graph.microsoft.com/.default"]
+                scopes=[f"{self.graph_api_host}/.default"]
             )
             if token is None:
                 raise ConnectorValidationError("Failed to acquire token for graph")
             return token
 
-        self._graph_client = GraphClient(_acquire_token_for_graph)
+        self._graph_client = GraphClient(
+            _acquire_token_for_graph, environment=self._azure_environment
+        )
         if auth_method == SharepointAuthMethod.CERTIFICATE.value:
             org = self.graph_client.organization.get().execute_query()
             if not org or len(org) == 0:
@@ -1847,14 +1947,13 @@ class SharepointConnector(
             # Return checkpoint to allow persistence after drive initialization
             return checkpoint
 
-        # Phase 3: Process documents from current drive
+        # Phase 3a: Initialize the next drive for processing
         if (
             checkpoint.current_site_descriptor
             and checkpoint.cached_drive_names
             and len(checkpoint.cached_drive_names) > 0
             and checkpoint.current_drive_name is None
         ):
-
             checkpoint.current_drive_name = checkpoint.cached_drive_names.popleft()
 
             start_dt = datetime.fromtimestamp(start, tz=timezone.utc)
@@ -1862,7 +1961,8 @@ class SharepointConnector(
             site_descriptor = checkpoint.current_site_descriptor
 
             logger.info(
-                f"Processing drive '{checkpoint.current_drive_name}' in site: {site_descriptor.url}"
+                f"Processing drive '{checkpoint.current_drive_name}' "
+                f"in site: {site_descriptor.url}"
             )
             logger.debug(f"Time range: {start_dt} to {end_dt}")
 
@@ -1871,35 +1971,35 @@ class SharepointConnector(
                 logger.warning("Current drive name is None, skipping")
                 return checkpoint
 
-            driveitems: Iterable[DriveItemData] = iter(())
-            drive_web_url: str | None = None
             try:
                 logger.info(
                     f"Fetching drive items for drive name: {current_drive_name}"
                 )
                 result = self._resolve_drive(site_descriptor, current_drive_name)
-                if result is not None:
-                    drive_id, drive_web_url = result
-                    driveitems = self._get_drive_items_for_drive_id(
-                        site_descriptor, drive_id, start_dt, end_dt
-                    )
-                    checkpoint.current_drive_web_url = drive_web_url
+                if result is None:
+                    logger.warning(f"Drive '{current_drive_name}' not found, skipping")
+                    self._clear_drive_checkpoint_state(checkpoint)
+                    return checkpoint
+
+                drive_id, drive_web_url = result
+                checkpoint.current_drive_id = drive_id
+                checkpoint.current_drive_web_url = drive_web_url
             except Exception as e:
                 logger.error(
-                    f"Failed to retrieve items from drive '{current_drive_name}' in site: {site_descriptor.url}: {e}"
+                    f"Failed to retrieve items from drive '{current_drive_name}' "
+                    f"in site: {site_descriptor.url}: {e}"
                 )
                 yield _create_entity_failure(
                     f"{site_descriptor.url}|{current_drive_name}",
-                    f"Failed to access drive '{current_drive_name}' in site '{site_descriptor.url}': {str(e)}",
+                    f"Failed to access drive '{current_drive_name}' "
+                    f"in site '{site_descriptor.url}': {str(e)}",
                     (start_dt, end_dt),
                     e,
                 )
-                checkpoint.current_drive_name = None
-                checkpoint.current_drive_web_url = None
+                self._clear_drive_checkpoint_state(checkpoint)
                 return checkpoint
 
-            # Normalize drive name (e.g., "Documents" -> "Shared Documents")
-            current_drive_name = SHARED_DOCUMENTS_MAP.get(
+            display_drive_name = SHARED_DOCUMENTS_MAP.get(
                 current_drive_name, current_drive_name
             )
 
@@ -1907,10 +2007,74 @@ class SharepointConnector(
                 yield from self._yield_drive_hierarchy_node(
                     site_descriptor.url,
                     drive_web_url,
-                    current_drive_name,
+                    display_drive_name,
                     checkpoint,
                 )
 
+            # For non-folder-scoped drives, use delta API with per-page
+            # checkpointing.  Build the initial URL and fall through to 3b.
+            if not site_descriptor.folder_path:
+                checkpoint.current_drive_delta_next_link = self._build_delta_start_url(
+                    drive_id, start_dt
+                )
+            # else: BFS path — delta_next_link stays None;
+            # Phase 3b will use _iter_drive_items_paged.
+
+        # Phase 3b: Process items from the current drive
+        if (
+            checkpoint.current_site_descriptor
+            and checkpoint.current_drive_name is not None
+            and checkpoint.current_drive_id is not None
+        ):
+            site_descriptor = checkpoint.current_site_descriptor
+            start_dt = datetime.fromtimestamp(start, tz=timezone.utc)
+            end_dt = datetime.fromtimestamp(end, tz=timezone.utc)
+            current_drive_name = SHARED_DOCUMENTS_MAP.get(
+                checkpoint.current_drive_name, checkpoint.current_drive_name
+            )
+            drive_web_url = checkpoint.current_drive_web_url
+
+            # --- determine item source ---
+            driveitems: Iterable[DriveItemData]
+            has_more_delta_pages = False
+
+            if checkpoint.current_drive_delta_next_link:
+                # Delta path: fetch one page at a time for checkpointing
+                try:
+                    page_items, next_url = self._fetch_one_delta_page(
+                        page_url=checkpoint.current_drive_delta_next_link,
+                        drive_id=checkpoint.current_drive_id,
+                        start=start_dt,
+                        end=end_dt,
+                    )
+                except Exception as e:
+                    logger.error(
+                        f"Failed to fetch delta page for drive "
+                        f"'{current_drive_name}': {e}"
+                    )
+                    yield _create_entity_failure(
+                        f"{site_descriptor.url}|{current_drive_name}",
+                        f"Failed to fetch delta page for drive "
+                        f"'{current_drive_name}': {str(e)}",
+                        (start_dt, end_dt),
+                        e,
+                    )
+                    self._clear_drive_checkpoint_state(checkpoint)
+                    return checkpoint
+
+                driveitems = page_items
+                has_more_delta_pages = next_url is not None
+                if next_url:
+                    checkpoint.current_drive_delta_next_link = next_url
+            else:
+                # BFS path (folder-scoped): process all items at once
+                driveitems = self._iter_drive_items_paged(
+                    drive_id=checkpoint.current_drive_id,
+                    folder_path=site_descriptor.folder_path,
+                    start=start_dt,
+                    end=end_dt,
+                )
+
             item_count = 0
             for driveitem in driveitems:
                 item_count += 1
@@ -1952,8 +2116,6 @@ class SharepointConnector(
                     if include_permissions:
                         ctx = self._create_rest_client_context(site_descriptor.url)
 
-                    # Re-acquire token in case it expired during a long traversal
-                    # MSAL has a cache that returns the same token while still valid.
                     access_token = self._get_graph_access_token()
                     doc_or_failure = _convert_driveitem_to_document_with_permissions(
                         driveitem,
@@ -1962,6 +2124,7 @@ class SharepointConnector(
                         self.graph_client,
                         include_permissions=include_permissions,
                         parent_hierarchy_raw_node_id=parent_hierarchy_url,
+                        graph_api_base=self.graph_api_base,
                         access_token=access_token,
                     )
 
@@ -1988,8 +2151,11 @@ class SharepointConnector(
                     )
 
             logger.info(f"Processed {item_count} items in drive '{current_drive_name}'")
-            checkpoint.current_drive_name = None
-            checkpoint.current_drive_web_url = None
+
+            if has_more_delta_pages:
+                return checkpoint
+
+            self._clear_drive_checkpoint_state(checkpoint)
 
         # Phase 4: Progression logic - determine next step
         # If we have more drives in current site, continue with current site

backend/onyx/connectors/slab/connector.py

@@ -11,6 +11,7 @@ from dateutil import parser
 
 from onyx.configs.app_configs import INDEX_BATCH_SIZE
 from onyx.configs.constants import DocumentSource
+from onyx.connectors.exceptions import ConnectorValidationError
 from onyx.connectors.interfaces import GenerateDocumentsOutput
 from onyx.connectors.interfaces import GenerateSlimDocumentOutput
 from onyx.connectors.interfaces import LoadConnector
@@ -258,3 +259,21 @@ class SlabConnector(LoadConnector, PollConnector, SlimConnectorWithPermSync):
                 slim_doc_batch = []
         if slim_doc_batch:
             yield slim_doc_batch
+
+    def validate_connector_settings(self) -> None:
+        """
+        Very basic validation, we could do more here
+        """
+        if not self.base_url.startswith("https://") and not self.base_url.startswith(
+            "http://"
+        ):
+            raise ConnectorValidationError(
+                "Base URL must start with https:// or http://"
+            )
+
+        try:
+            get_all_post_ids(self.slab_bot_token)
+        except ConnectorMissingCredentialError:
+            raise
+        except Exception as e:
+            raise ConnectorValidationError(f"Failed to fetch posts from Slab: {e}")

backend/onyx/connectors/teams/connector.py

@@ -23,6 +23,7 @@ from onyx.connectors.interfaces import CheckpointOutput
 from onyx.connectors.interfaces import GenerateSlimDocumentOutput
 from onyx.connectors.interfaces import SecondsSinceUnixEpoch
 from onyx.connectors.interfaces import SlimConnectorWithPermSync
+from onyx.connectors.microsoft_graph_env import resolve_microsoft_environment
 from onyx.connectors.models import ConnectorCheckpoint
 from onyx.connectors.models import ConnectorFailure
 from onyx.connectors.models import ConnectorMissingCredentialError
@@ -50,12 +51,15 @@ class TeamsCheckpoint(ConnectorCheckpoint):
     todo_team_ids: list[str] | None = None
 
 
+DEFAULT_AUTHORITY_HOST = "https://login.microsoftonline.com"
+DEFAULT_GRAPH_API_HOST = "https://graph.microsoft.com"
+
+
 class TeamsConnector(
     CheckpointedConnectorWithPermSync[TeamsCheckpoint],
     SlimConnectorWithPermSync,
 ):
     MAX_WORKERS = 10
-    AUTHORITY_URL_PREFIX = "https://login.microsoftonline.com/"
 
     def __init__(
         self,
@@ -63,12 +67,19 @@ class TeamsConnector(
         # are not necessarily guaranteed to be unique
         teams: list[str] = [],
         max_workers: int = MAX_WORKERS,
+        authority_host: str = DEFAULT_AUTHORITY_HOST,
+        graph_api_host: str = DEFAULT_GRAPH_API_HOST,
     ) -> None:
         self.graph_client: GraphClient | None = None
         self.msal_app: msal.ConfidentialClientApplication | None = None
         self.max_workers = max_workers
         self.requested_team_list: list[str] = teams
 
+        resolved_env = resolve_microsoft_environment(graph_api_host, authority_host)
+        self._azure_environment = resolved_env.environment
+        self.authority_host = resolved_env.authority_host
+        self.graph_api_host = resolved_env.graph_host
+
     # impls for BaseConnector
 
     def load_credentials(self, credentials: dict[str, Any]) -> dict[str, Any] | None:
@@ -76,7 +87,7 @@ class TeamsConnector(
         teams_client_secret = credentials["teams_client_secret"]
         teams_directory_id = credentials["teams_directory_id"]
 
-        authority_url = f"{TeamsConnector.AUTHORITY_URL_PREFIX}{teams_directory_id}"
+        authority_url = f"{self.authority_host}/{teams_directory_id}"
         self.msal_app = msal.ConfidentialClientApplication(
             authority=authority_url,
             client_id=teams_client_id,
@@ -91,7 +102,7 @@ class TeamsConnector(
                 raise RuntimeError("MSAL app is not initialized")
 
             token = self.msal_app.acquire_token_for_client(
-                scopes=["https://graph.microsoft.com/.default"]
+                scopes=[f"{self.graph_api_host}/.default"]
             )
 
             if not isinstance(token, dict):
@@ -99,7 +110,9 @@ class TeamsConnector(
 
             return token
 
-        self.graph_client = GraphClient(_acquire_token_func)
+        self.graph_client = GraphClient(
+            _acquire_token_func, environment=self._azure_environment
+        )
         return None
 
     def validate_connector_settings(self) -> None:

backend/onyx/context/search/federated/slack_search.py

@@ -32,6 +32,7 @@ from onyx.context.search.federated.slack_search_utils import should_include_mess
 from onyx.context.search.models import ChunkIndexRequest
 from onyx.context.search.models import InferenceChunk
 from onyx.db.document import DocumentSource
+from onyx.db.models import SearchSettings
 from onyx.db.search_settings import get_current_search_settings
 from onyx.document_index.document_index_utils import (
     get_multipass_config,
@@ -905,13 +906,15 @@ def convert_slack_score(slack_score: float) -> float:
 def slack_retrieval(
     query: ChunkIndexRequest,
     access_token: str,
-    db_session: Session,
+    db_session: Session | None = None,
     connector: FederatedConnectorDetail | None = None,  # noqa: ARG001
     entities: dict[str, Any] | None = None,
     limit: int | None = None,
     slack_event_context: SlackContext | None = None,
     bot_token: str | None = None,  # Add bot token parameter
     team_id: str | None = None,
+    # Pre-fetched data — when provided, avoids DB query (no session needed)
+    search_settings: SearchSettings | None = None,
 ) -> list[InferenceChunk]:
     """
     Main entry point for Slack federated search with entity filtering.
@@ -925,7 +928,7 @@ def slack_retrieval(
     Args:
         query: Search query object
         access_token: User OAuth access token
-        db_session: Database session
+        db_session: Database session (optional if search_settings provided)
         connector: Federated connector detail (unused, kept for backwards compat)
         entities: Connector-level config (entity filtering configuration)
         limit: Maximum number of results
@@ -1153,7 +1156,10 @@ def slack_retrieval(
 
     # chunk index docs into doc aware chunks
     # a single index doc can get split into multiple chunks
-    search_settings = get_current_search_settings(db_session)
+    if search_settings is None:
+        if db_session is None:
+            raise ValueError("Either db_session or search_settings must be provided")
+        search_settings = get_current_search_settings(db_session)
     embedder = DefaultIndexingEmbedder.from_db_search_settings(
         search_settings=search_settings
     )

backend/onyx/context/search/models.py

@@ -72,6 +72,7 @@ class BaseFilters(BaseModel):
 class UserFileFilters(BaseModel):
     user_file_ids: list[UUID] | None = None
     project_id: int | None = None
+    persona_id: int | None = None
 
 
 class AssistantKnowledgeFilters(BaseModel):

backend/onyx/context/search/pipeline.py

@@ -18,8 +18,10 @@ from onyx.context.search.utils import inference_section_from_chunks
 from onyx.db.models import Persona
 from onyx.db.models import User
 from onyx.document_index.interfaces import DocumentIndex
+from onyx.federated_connectors.federated_retrieval import FederatedRetrievalInfo
 from onyx.llm.interfaces import LLM
 from onyx.natural_language_processing.english_stopwords import strip_stopwords
+from onyx.natural_language_processing.search_nlp_models import EmbeddingModel
 from onyx.secondary_llm_flows.source_filter import extract_source_filter
 from onyx.secondary_llm_flows.time_filter import extract_time_filter
 from onyx.utils.logger import setup_logger
@@ -38,10 +40,11 @@ def _build_index_filters(
     user_provided_filters: BaseFilters | None,
     user: User,  # Used for ACLs, anonymous users only see public docs
     project_id: int | None,
+    persona_id: int | None,
     user_file_ids: list[UUID] | None,
     persona_document_sets: list[str] | None,
     persona_time_cutoff: datetime | None,
-    db_session: Session,
+    db_session: Session | None = None,
     auto_detect_filters: bool = False,
     query: str | None = None,
     llm: LLM | None = None,
@@ -49,18 +52,19 @@ def _build_index_filters(
     # Assistant knowledge filters
     attached_document_ids: list[str] | None = None,
     hierarchy_node_ids: list[int] | None = None,
+    # Pre-fetched ACL filters (skips DB query when provided)
+    acl_filters: list[str] | None = None,
 ) -> IndexFilters:
     if auto_detect_filters and (llm is None or query is None):
         raise RuntimeError("LLM and query are required for auto detect filters")
 
     base_filters = user_provided_filters or BaseFilters()
 
-    if (
-        user_provided_filters
-        and user_provided_filters.document_set is None
-        and persona_document_sets is not None
-    ):
-        base_filters.document_set = persona_document_sets
+    document_set_filter = (
+        base_filters.document_set
+        if base_filters.document_set is not None
+        else persona_document_sets
+    )
 
     time_filter = base_filters.time_cutoff or persona_time_cutoff
     source_filter = base_filters.source_type
@@ -103,15 +107,21 @@ def _build_index_filters(
             source_filter = list(source_filter) + [DocumentSource.USER_FILE]
             logger.debug("Added USER_FILE to source_filter for user knowledge search")
 
-    user_acl_filters = (
-        None if bypass_acl else build_access_filters_for_user(user, db_session)
-    )
+    if bypass_acl:
+        user_acl_filters = None
+    elif acl_filters is not None:
+        user_acl_filters = acl_filters
+    else:
+        if db_session is None:
+            raise ValueError("Either db_session or acl_filters must be provided")
+        user_acl_filters = build_access_filters_for_user(user, db_session)
 
     final_filters = IndexFilters(
         user_file_ids=user_file_ids,
         project_id=project_id,
+        persona_id=persona_id,
         source_type=source_filter,
-        document_set=persona_document_sets,
+        document_set=document_set_filter,
         time_cutoff=time_filter,
         tags=base_filters.tags,
         access_control_list=user_acl_filters,
@@ -252,11 +262,17 @@ def search_pipeline(
     user: User,
     # Used for default filters and settings
     persona: Persona | None,
-    db_session: Session,
+    db_session: Session | None = None,
     auto_detect_filters: bool = False,
     llm: LLM | None = None,
     # If a project ID is provided, it will be exclusively scoped to that project
     project_id: int | None = None,
+    # If a persona_id is provided, search scopes to files attached to this persona
+    persona_id: int | None = None,
+    # Pre-fetched data — when provided, avoids DB queries (no session needed)
+    acl_filters: list[str] | None = None,
+    embedding_model: EmbeddingModel | None = None,
+    prefetched_federated_retrieval_infos: list[FederatedRetrievalInfo] | None = None,
 ) -> list[InferenceChunk]:
     user_uploaded_persona_files: list[UUID] | None = (
         [user_file.id for user_file in persona.user_files] if persona else None
@@ -287,6 +303,7 @@ def search_pipeline(
         user_provided_filters=chunk_search_request.user_selected_filters,
         user=user,
         project_id=project_id,
+        persona_id=persona_id,
         user_file_ids=user_uploaded_persona_files,
         persona_document_sets=persona_document_sets,
         persona_time_cutoff=persona_time_cutoff,
@@ -297,6 +314,7 @@ def search_pipeline(
         bypass_acl=chunk_search_request.bypass_acl,
         attached_document_ids=attached_document_ids,
         hierarchy_node_ids=hierarchy_node_ids,
+        acl_filters=acl_filters,
     )
 
     query_keywords = strip_stopwords(chunk_search_request.query)
@@ -315,6 +333,8 @@ def search_pipeline(
         user_id=user.id if user else None,
         document_index=document_index,
         db_session=db_session,
+        embedding_model=embedding_model,
+        prefetched_federated_retrieval_infos=prefetched_federated_retrieval_infos,
     )
 
     # For some specific connectors like Salesforce, a user that has access to an object doesn't mean

backend/onyx/context/search/retrieval/search_runner.py

@@ -14,9 +14,11 @@ from onyx.context.search.utils import get_query_embedding
 from onyx.context.search.utils import inference_section_from_chunks
 from onyx.document_index.interfaces import DocumentIndex
 from onyx.document_index.interfaces import VespaChunkRequest
+from onyx.federated_connectors.federated_retrieval import FederatedRetrievalInfo
 from onyx.federated_connectors.federated_retrieval import (
     get_federated_retrieval_functions,
 )
+from onyx.natural_language_processing.search_nlp_models import EmbeddingModel
 from onyx.utils.logger import setup_logger
 from onyx.utils.threadpool_concurrency import run_functions_tuples_in_parallel
 
@@ -50,9 +52,14 @@ def combine_retrieval_results(
 def _embed_and_search(
     query_request: ChunkIndexRequest,
     document_index: DocumentIndex,
-    db_session: Session,
+    db_session: Session | None = None,
+    embedding_model: EmbeddingModel | None = None,
 ) -> list[InferenceChunk]:
-    query_embedding = get_query_embedding(query_request.query, db_session)
+    query_embedding = get_query_embedding(
+        query_request.query,
+        db_session=db_session,
+        embedding_model=embedding_model,
+    )
 
     hybrid_alpha = query_request.hybrid_alpha or HYBRID_ALPHA
 
@@ -78,7 +85,9 @@ def search_chunks(
     query_request: ChunkIndexRequest,
     user_id: UUID | None,
     document_index: DocumentIndex,
-    db_session: Session,
+    db_session: Session | None = None,
+    embedding_model: EmbeddingModel | None = None,
+    prefetched_federated_retrieval_infos: list[FederatedRetrievalInfo] | None = None,
 ) -> list[InferenceChunk]:
     run_queries: list[tuple[Callable, tuple]] = []
 
@@ -88,14 +97,22 @@ def search_chunks(
         else None
     )
 
-    # Federated retrieval
-    federated_retrieval_infos = get_federated_retrieval_functions(
-        db_session=db_session,
-        user_id=user_id,
-        source_types=list(source_filters) if source_filters else None,
-        document_set_names=query_request.filters.document_set,
-        user_file_ids=query_request.filters.user_file_ids,
-    )
+    # Federated retrieval — use pre-fetched if available, otherwise query DB
+    if prefetched_federated_retrieval_infos is not None:
+        federated_retrieval_infos = prefetched_federated_retrieval_infos
+    else:
+        if db_session is None:
+            raise ValueError(
+                "Either db_session or prefetched_federated_retrieval_infos "
+                "must be provided"
+            )
+        federated_retrieval_infos = get_federated_retrieval_functions(
+            db_session=db_session,
+            user_id=user_id,
+            source_types=list(source_filters) if source_filters else None,
+            document_set_names=query_request.filters.document_set,
+            user_file_ids=query_request.filters.user_file_ids,
+        )
 
     federated_sources = set(
         federated_retrieval_info.source.to_non_federated_source()
@@ -114,7 +131,10 @@ def search_chunks(
 
     if normal_search_enabled:
         run_queries.append(
-            (_embed_and_search, (query_request, document_index, db_session))
+            (
+                _embed_and_search,
+                (query_request, document_index, db_session, embedding_model),
+            )
         )
 
     parallel_search_results = run_functions_tuples_in_parallel(run_queries)

backend/onyx/context/search/utils.py

@@ -64,23 +64,34 @@ def inference_section_from_single_chunk(
     )
 
 
-def get_query_embeddings(queries: list[str], db_session: Session) -> list[Embedding]:
-    search_settings = get_current_search_settings(db_session)
+def get_query_embeddings(
+    queries: list[str],
+    db_session: Session | None = None,
+    embedding_model: EmbeddingModel | None = None,
+) -> list[Embedding]:
+    if embedding_model is None:
+        if db_session is None:
+            raise ValueError("Either db_session or embedding_model must be provided")
+        search_settings = get_current_search_settings(db_session)
+        embedding_model = EmbeddingModel.from_db_model(
+            search_settings=search_settings,
+            server_host=MODEL_SERVER_HOST,
+            server_port=MODEL_SERVER_PORT,
+        )
 
-    model = EmbeddingModel.from_db_model(
-        search_settings=search_settings,
-        # The below are globally set, this flow always uses the indexing one
-        server_host=MODEL_SERVER_HOST,
-        server_port=MODEL_SERVER_PORT,
-    )
-
-    query_embedding = model.encode(queries, text_type=EmbedTextType.QUERY)
+    query_embedding = embedding_model.encode(queries, text_type=EmbedTextType.QUERY)
     return query_embedding
 
 
 @log_function_time(print_only=True, debug_only=True)
-def get_query_embedding(query: str, db_session: Session) -> Embedding:
-    return get_query_embeddings([query], db_session)[0]
+def get_query_embedding(
+    query: str,
+    db_session: Session | None = None,
+    embedding_model: EmbeddingModel | None = None,
+) -> Embedding:
+    return get_query_embeddings(
+        [query], db_session=db_session, embedding_model=embedding_model
+    )[0]
 
 
 def convert_inference_sections_to_search_docs(

backend/onyx/db/api_key.py

@@ -4,6 +4,7 @@ from fastapi_users.password import PasswordHelper
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import joinedload
+from sqlalchemy.orm import selectinload
 from sqlalchemy.orm import Session
 
 from onyx.auth.api_key import ApiKeyDescriptor
@@ -54,6 +55,7 @@ async def fetch_user_for_api_key(
         select(User)
         .join(ApiKey, ApiKey.user_id == User.id)
         .where(ApiKey.hashed_api_key == hashed_api_key)
+        .options(selectinload(User.memories))
     )
 
 

backend/onyx/db/auth.py

@@ -13,6 +13,7 @@ from sqlalchemy import func
 from sqlalchemy import Select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.future import select
+from sqlalchemy.orm import selectinload
 from sqlalchemy.orm import Session
 
 from onyx.auth.schemas import UserRole
@@ -97,6 +98,11 @@ async def get_user_count(only_admin_users: bool = False) -> int:
 
 # Need to override this because FastAPI Users doesn't give flexibility for backend field creation logic in OAuth flow
 class SQLAlchemyUserAdminDB(SQLAlchemyUserDatabase[UP, ID]):
+    async def _get_user(self, statement: Select) -> UP | None:
+        statement = statement.options(selectinload(User.memories))
+        results = await self.session.execute(statement)
+        return results.unique().scalar_one_or_none()
+
     async def create(
         self,
         create_dict: Dict[str, Any],

backend/onyx/db/code_interpreter.py

@@ -0,0 +1,21 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from onyx.db.models import CodeInterpreterServer
+
+
+def fetch_code_interpreter_server(
+    db_session: Session,
+) -> CodeInterpreterServer:
+    server = db_session.scalars(select(CodeInterpreterServer)).one()
+    return server
+
+
+def update_code_interpreter_server_enabled(
+    db_session: Session,
+    enabled: bool,
+) -> CodeInterpreterServer:
+    server = db_session.scalars(select(CodeInterpreterServer)).one()
+    server.server_enabled = enabled
+    db_session.commit()
+    return server

backend/onyx/db/connector_credential_pair.py

@@ -116,12 +116,15 @@ def get_connector_credential_pairs_for_user(
     order_by_desc: bool = False,
     source: DocumentSource | None = None,
     processing_mode: ProcessingMode | None = ProcessingMode.REGULAR,
+    defer_connector_config: bool = False,
 ) -> list[ConnectorCredentialPair]:
     """Get connector credential pairs for a user.
 
     Args:
         processing_mode: Filter by processing mode. Defaults to REGULAR to hide
             FILE_SYSTEM connectors from standard admin UI. Pass None to get all.
+        defer_connector_config: If True, skips loading Connector.connector_specific_config
+            to avoid fetching large JSONB blobs when they aren't needed.
     """
     if eager_load_user:
         assert (
@@ -130,7 +133,10 @@ def get_connector_credential_pairs_for_user(
     stmt = select(ConnectorCredentialPair).distinct()
 
     if eager_load_connector:
-        stmt = stmt.options(selectinload(ConnectorCredentialPair.connector))
+        connector_load = selectinload(ConnectorCredentialPair.connector)
+        if defer_connector_config:
+            connector_load = connector_load.defer(Connector.connector_specific_config)
+        stmt = stmt.options(connector_load)
 
     if eager_load_credential:
         load_opts = selectinload(ConnectorCredentialPair.credential)
@@ -170,6 +176,7 @@ def get_connector_credential_pairs_for_user_parallel(
     order_by_desc: bool = False,
     source: DocumentSource | None = None,
     processing_mode: ProcessingMode | None = ProcessingMode.REGULAR,
+    defer_connector_config: bool = False,
 ) -> list[ConnectorCredentialPair]:
     with get_session_with_current_tenant() as db_session:
         return get_connector_credential_pairs_for_user(
@@ -183,6 +190,7 @@ def get_connector_credential_pairs_for_user_parallel(
             order_by_desc=order_by_desc,
             source=source,
             processing_mode=processing_mode,
+            defer_connector_config=defer_connector_config,
         )
 
 

backend/onyx/db/document_set.py

@@ -554,10 +554,19 @@ def fetch_all_document_sets_for_user(
     stmt = (
         select(DocumentSetDBModel)
         .distinct()
-        .options(selectinload(DocumentSetDBModel.federated_connectors))
+        .options(
+            selectinload(DocumentSetDBModel.connector_credential_pairs).selectinload(
+                ConnectorCredentialPair.connector
+            ),
+            selectinload(DocumentSetDBModel.users),
+            selectinload(DocumentSetDBModel.groups),
+            selectinload(DocumentSetDBModel.federated_connectors).selectinload(
+                FederatedConnector__DocumentSet.federated_connector
+            ),
+        )
     )
     stmt = _add_user_filters(stmt, user, get_editable=get_editable)
-    return db_session.scalars(stmt).all()
+    return db_session.scalars(stmt).unique().all()
 
 
 def fetch_documents_for_document_set_paginated(

backend/onyx/db/engine/tenant_utils.py

@@ -1,11 +1,102 @@
 from sqlalchemy import text
 
 from onyx.db.engine.sql_engine import get_session_with_shared_schema
+from onyx.db.engine.sql_engine import SqlEngine
 from shared_configs.configs import MULTI_TENANT
 from shared_configs.configs import POSTGRES_DEFAULT_SCHEMA
 from shared_configs.configs import TENANT_ID_PREFIX
 
 
+def get_schemas_needing_migration(
+    tenant_schemas: list[str], head_rev: str
+) -> list[str]:
+    """Return only schemas whose current alembic version is not at head.
+
+    Uses a server-side PL/pgSQL loop to collect each schema's alembic version
+    into a temp table one at a time. This avoids building a massive UNION ALL
+    query (which locks the DB and times out at 17k+ schemas) and instead
+    acquires locks sequentially, one schema per iteration.
+    """
+    if not tenant_schemas:
+        return []
+
+    engine = SqlEngine.get_engine()
+
+    with engine.connect() as conn:
+        # Populate a temp input table with exactly the schemas we care about.
+        # The DO block reads from this table so it only iterates the requested
+        # schemas instead of every tenant_% schema in the database.
+        conn.execute(text("DROP TABLE IF EXISTS _alembic_version_snapshot"))
+        conn.execute(text("DROP TABLE IF EXISTS _tenant_schemas_input"))
+        conn.execute(text("CREATE TEMP TABLE _tenant_schemas_input (schema_name text)"))
+        conn.execute(
+            text(
+                "INSERT INTO _tenant_schemas_input (schema_name) "
+                "SELECT unnest(CAST(:schemas AS text[]))"
+            ),
+            {"schemas": tenant_schemas},
+        )
+        conn.execute(
+            text(
+                "CREATE TEMP TABLE _alembic_version_snapshot "
+                "(schema_name text, version_num text)"
+            )
+        )
+
+        conn.execute(
+            text(
+                """
+                DO $$
+                DECLARE
+                    s        text;
+                    schemas  text[];
+                BEGIN
+                    SELECT array_agg(schema_name) INTO schemas
+                    FROM _tenant_schemas_input;
+
+                    IF schemas IS NULL THEN
+                        RAISE NOTICE 'No tenant schemas found.';
+                        RETURN;
+                    END IF;
+
+                    FOREACH s IN ARRAY schemas LOOP
+                        BEGIN
+                            EXECUTE format(
+                                'INSERT INTO _alembic_version_snapshot
+                                 SELECT %L, version_num FROM %I.alembic_version',
+                                s, s
+                            );
+                        EXCEPTION
+                            -- undefined_table: schema exists but has no alembic_version
+                            --   table yet (new tenant, not yet migrated).
+                            -- invalid_schema_name: tenant is registered but its
+                            --   PostgreSQL schema does not exist yet (e.g. provisioning
+                            --   incomplete). Both cases mean no version is available and
+                            --   the schema will be included in the migration list.
+                            WHEN undefined_table THEN NULL;
+                            WHEN invalid_schema_name THEN NULL;
+                        END;
+                    END LOOP;
+                END;
+                $$
+                """
+            )
+        )
+
+        rows = conn.execute(
+            text("SELECT schema_name, version_num FROM _alembic_version_snapshot")
+        )
+        version_by_schema = {row[0]: row[1] for row in rows}
+
+        conn.execute(text("DROP TABLE IF EXISTS _alembic_version_snapshot"))
+        conn.execute(text("DROP TABLE IF EXISTS _tenant_schemas_input"))
+
+    # Schemas missing from the snapshot have no alembic_version table yet and
+    # also need migration. version_by_schema.get(s) returns None for those,
+    # and None != head_rev, so they are included automatically.
+    return [s for s in tenant_schemas if version_by_schema.get(s) != head_rev]
+
+
 def get_all_tenant_ids() -> list[str]:
     """Returning [None] means the only tenant is the 'public' or self hosted tenant."""
 

backend/onyx/db/models.py

@@ -287,7 +287,7 @@ class User(SQLAlchemyBaseUserTableUUID, Base):
 
     # relationships
     credentials: Mapped[list["Credential"]] = relationship(
-        "Credential", back_populates="user", lazy="joined"
+        "Credential", back_populates="user"
     )
     chat_sessions: Mapped[list["ChatSession"]] = relationship(
         "ChatSession", back_populates="user"
@@ -321,7 +321,6 @@ class User(SQLAlchemyBaseUserTableUUID, Base):
         "Memory",
         back_populates="user",
         cascade="all, delete-orphan",
-        lazy="selectin",
         order_by="desc(Memory.id)",
     )
     oauth_user_tokens: Mapped[list["OAuthUserToken"]] = relationship(
@@ -4271,6 +4270,9 @@ class UserFile(Base):
     needs_project_sync: Mapped[bool] = mapped_column(
         Boolean, nullable=False, default=False
     )
+    needs_persona_sync: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False
+    )
     last_project_sync_at: Mapped[datetime.datetime | None] = mapped_column(
         DateTime(timezone=True), nullable=True
     )
@@ -4940,6 +4942,12 @@ class ScimUserMapping(Base):
     user_id: Mapped[UUID] = mapped_column(
         ForeignKey("user.id", ondelete="CASCADE"), unique=True, nullable=False
     )
+    scim_username: Mapped[str | None] = mapped_column(String, nullable=True)
+    department: Mapped[str | None] = mapped_column(String, nullable=True)
+    manager: Mapped[str | None] = mapped_column(String, nullable=True)
+    given_name: Mapped[str | None] = mapped_column(String, nullable=True)
+    family_name: Mapped[str | None] = mapped_column(String, nullable=True)
+    scim_emails_json: Mapped[str | None] = mapped_column(Text, nullable=True)
 
     created_at: Mapped[datetime.datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now(), nullable=False
@@ -4978,3 +4986,12 @@ class ScimGroupMapping(Base):
     user_group: Mapped[UserGroup] = relationship(
         "UserGroup", foreign_keys=[user_group_id]
     )
+
+
+class CodeInterpreterServer(Base):
+    """Details about the code interpreter server"""
+
+    __tablename__ = "code_interpreter_server"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    server_enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)

backend/onyx/db/pat.py

@@ -8,6 +8,7 @@ from uuid import UUID
 from sqlalchemy import select
 from sqlalchemy import update
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
 from sqlalchemy.orm import Session
 
 from onyx.auth.pat import build_displayable_pat
@@ -31,55 +32,61 @@ async def fetch_user_for_pat(
 
     NOTE: This is async since it's used during auth (which is necessarily async due to FastAPI Users).
     NOTE: Expired includes both naturally expired and user-revoked tokens (revocation sets expires_at=NOW()).
+
+    Uses select(User) as primary entity so that joined-eager relationships (e.g. oauth_accounts)
+    are loaded correctly — matching the pattern in fetch_user_for_api_key.
     """
-    # Single joined query with all filters pushed to database
     now = datetime.now(timezone.utc)
-    result = await async_db_session.execute(
-        select(PersonalAccessToken, User)
-        .join(User, PersonalAccessToken.user_id == User.id)
+
+    user = await async_db_session.scalar(
+        select(User)
+        .join(PersonalAccessToken, PersonalAccessToken.user_id == User.id)
         .where(PersonalAccessToken.hashed_token == hashed_token)
         .where(User.is_active)  # type: ignore
         .where(
             (PersonalAccessToken.expires_at.is_(None))
             | (PersonalAccessToken.expires_at > now)
         )
-        .limit(1)
+        .options(selectinload(User.memories))
     )
-    row = result.first()
-
-    if not row:
+    if not user:
         return None
 
-    pat, user = row
-
-    # Throttle last_used_at updates to reduce DB load (5-minute granularity sufficient for auditing)
-    # For request-level auditing, use application logs or a dedicated audit table
-    should_update = (
-        pat.last_used_at is None or (now - pat.last_used_at).total_seconds() > 300
-    )
-
-    if should_update:
-        # Update in separate session to avoid transaction coupling (fire-and-forget)
-        async def _update_last_used() -> None:
-            try:
-                tenant_id = get_current_tenant_id()
-                async with get_async_session_context_manager(
-                    tenant_id
-                ) as separate_session:
-                    await separate_session.execute(
-                        update(PersonalAccessToken)
-                        .where(PersonalAccessToken.hashed_token == hashed_token)
-                        .values(last_used_at=now)
-                    )
-                    await separate_session.commit()
-            except Exception as e:
-                logger.warning(f"Failed to update last_used_at for PAT: {e}")
-
-        asyncio.create_task(_update_last_used())
-
+    _schedule_pat_last_used_update(hashed_token, now)
     return user
 
 
+def _schedule_pat_last_used_update(hashed_token: str, now: datetime) -> None:
+    """Fire-and-forget update of last_used_at, throttled to 5-minute granularity."""
+
+    async def _update() -> None:
+        try:
+            tenant_id = get_current_tenant_id()
+            async with get_async_session_context_manager(tenant_id) as session:
+                pat = await session.scalar(
+                    select(PersonalAccessToken).where(
+                        PersonalAccessToken.hashed_token == hashed_token
+                    )
+                )
+                if not pat:
+                    return
+                if (
+                    pat.last_used_at is not None
+                    and (now - pat.last_used_at).total_seconds() <= 300
+                ):
+                    return
+                await session.execute(
+                    update(PersonalAccessToken)
+                    .where(PersonalAccessToken.hashed_token == hashed_token)
+                    .values(last_used_at=now)
+                )
+                await session.commit()
+        except Exception as e:
+            logger.warning(f"Failed to update last_used_at for PAT: {e}")
+
+    asyncio.create_task(_update())
+
+
 def create_pat(
     db_session: Session,
     user_id: UUID,

backend/onyx/db/persona.py

@@ -28,6 +28,7 @@ from onyx.db.document_access import get_accessible_documents_by_ids
 from onyx.db.models import ConnectorCredentialPair
 from onyx.db.models import Document
 from onyx.db.models import DocumentSet
+from onyx.db.models import FederatedConnector__DocumentSet
 from onyx.db.models import HierarchyNode
 from onyx.db.models import Persona
 from onyx.db.models import Persona__User
@@ -420,9 +421,16 @@ def get_minimal_persona_snapshots_for_user(
     stmt = stmt.options(
         selectinload(Persona.tools),
         selectinload(Persona.labels),
-        selectinload(Persona.document_sets)
-        .selectinload(DocumentSet.connector_credential_pairs)
-        .selectinload(ConnectorCredentialPair.connector),
+        selectinload(Persona.document_sets).options(
+            selectinload(DocumentSet.connector_credential_pairs).selectinload(
+                ConnectorCredentialPair.connector
+            ),
+            selectinload(DocumentSet.users),
+            selectinload(DocumentSet.groups),
+            selectinload(DocumentSet.federated_connectors).selectinload(
+                FederatedConnector__DocumentSet.federated_connector
+            ),
+        ),
         selectinload(Persona.hierarchy_nodes),
         selectinload(Persona.attached_documents).selectinload(
             Document.parent_hierarchy_node
@@ -453,7 +461,16 @@ def get_persona_snapshots_for_user(
             Document.parent_hierarchy_node
         ),
         selectinload(Persona.labels),
-        selectinload(Persona.document_sets),
+        selectinload(Persona.document_sets).options(
+            selectinload(DocumentSet.connector_credential_pairs).selectinload(
+                ConnectorCredentialPair.connector
+            ),
+            selectinload(DocumentSet.users),
+            selectinload(DocumentSet.groups),
+            selectinload(DocumentSet.federated_connectors).selectinload(
+                FederatedConnector__DocumentSet.federated_connector
+            ),
+        ),
         selectinload(Persona.user),
         selectinload(Persona.user_files),
         selectinload(Persona.users),
@@ -550,9 +567,16 @@ def get_minimal_persona_snapshots_paginated(
             Document.parent_hierarchy_node
         ),
         selectinload(Persona.labels),
-        selectinload(Persona.document_sets)
-        .selectinload(DocumentSet.connector_credential_pairs)
-        .selectinload(ConnectorCredentialPair.connector),
+        selectinload(Persona.document_sets).options(
+            selectinload(DocumentSet.connector_credential_pairs).selectinload(
+                ConnectorCredentialPair.connector
+            ),
+            selectinload(DocumentSet.users),
+            selectinload(DocumentSet.groups),
+            selectinload(DocumentSet.federated_connectors).selectinload(
+                FederatedConnector__DocumentSet.federated_connector
+            ),
+        ),
         selectinload(Persona.user),
     )
 
@@ -611,7 +635,16 @@ def get_persona_snapshots_paginated(
             Document.parent_hierarchy_node
         ),
         selectinload(Persona.labels),
-        selectinload(Persona.document_sets),
+        selectinload(Persona.document_sets).options(
+            selectinload(DocumentSet.connector_credential_pairs).selectinload(
+                ConnectorCredentialPair.connector
+            ),
+            selectinload(DocumentSet.users),
+            selectinload(DocumentSet.groups),
+            selectinload(DocumentSet.federated_connectors).selectinload(
+                FederatedConnector__DocumentSet.federated_connector
+            ),
+        ),
         selectinload(Persona.user),
         selectinload(Persona.user_files),
         selectinload(Persona.users),
@@ -732,6 +765,9 @@ def mark_persona_as_deleted(
 ) -> None:
     persona = get_persona_by_id(persona_id=persona_id, user=user, db_session=db_session)
     persona.deleted = True
+    affected_file_ids = [uf.id for uf in persona.user_files]
+    if affected_file_ids:
+        _mark_files_need_persona_sync(db_session, affected_file_ids)
     db_session.commit()
 
 
@@ -743,11 +779,13 @@ def mark_persona_as_not_deleted(
     persona = get_persona_by_id(
         persona_id=persona_id, user=user, db_session=db_session, include_deleted=True
     )
-    if persona.deleted:
-        persona.deleted = False
-        db_session.commit()
-    else:
+    if not persona.deleted:
         raise ValueError(f"Persona with ID {persona_id} is not deleted.")
+    persona.deleted = False
+    affected_file_ids = [uf.id for uf in persona.user_files]
+    if affected_file_ids:
+        _mark_files_need_persona_sync(db_session, affected_file_ids)
+    db_session.commit()
 
 
 def mark_delete_persona_by_name(
@@ -813,6 +851,20 @@ def update_personas_display_priority(
         db_session.commit()
 
 
+def _mark_files_need_persona_sync(
+    db_session: Session,
+    user_file_ids: list[UUID],
+) -> None:
+    """Flag the given UserFile rows so the background sync task picks them up
+    and updates their persona metadata in the vector DB."""
+    if not user_file_ids:
+        return
+    db_session.query(UserFile).filter(UserFile.id.in_(user_file_ids)).update(
+        {UserFile.needs_persona_sync: True},
+        synchronize_session=False,
+    )
+
+
 def upsert_persona(
     user: User | None,
     name: str,
@@ -1001,8 +1053,13 @@ def upsert_persona(
             existing_persona.tools = tools or []
 
         if user_file_ids is not None:
+            old_file_ids = {uf.id for uf in existing_persona.user_files}
+            new_file_ids = {uf.id for uf in (user_files or [])}
+            affected_file_ids = old_file_ids | new_file_ids
             existing_persona.user_files.clear()
             existing_persona.user_files = user_files or []
+            if affected_file_ids:
+                _mark_files_need_persona_sync(db_session, list(affected_file_ids))
 
         if hierarchy_node_ids is not None:
             existing_persona.hierarchy_nodes.clear()
@@ -1056,6 +1113,8 @@ def upsert_persona(
             attached_documents=attached_documents or [],
         )
         db_session.add(new_persona)
+        if user_files:
+            _mark_files_need_persona_sync(db_session, [uf.id for uf in user_files])
         persona = new_persona
     if commit:
         db_session.commit()

backend/onyx/db/seeding/chat_history_seeding.py

@@ -2,6 +2,7 @@ import random
 from datetime import datetime
 from datetime import timedelta
 from logging import getLogger
+from uuid import UUID
 
 from onyx.configs.constants import MessageType
 from onyx.db.chat import create_chat_session
@@ -13,18 +14,26 @@ from onyx.db.models import ChatSession
 logger = getLogger(__name__)
 
 
-def seed_chat_history(num_sessions: int, num_messages: int, days: int) -> None:
+def seed_chat_history(
+    num_sessions: int,
+    num_messages: int,
+    days: int,
+    user_id: UUID | None = None,
+    persona_id: int | None = None,
+) -> None:
     """Utility function to seed chat history for testing.
 
     num_sessions: the number of sessions to seed
     num_messages: the number of messages to seed per sessions
     days: the number of days looking backwards from the current time over which to randomize
     the times.
+    user_id: optional user to associate with sessions
+    persona_id: optional persona/assistant to associate with sessions
     """
     with get_session_with_current_tenant() as db_session:
         logger.info(f"Seeding {num_sessions} sessions.")
         for y in range(0, num_sessions):
-            create_chat_session(db_session, f"pytest_session_{y}", None, None)
+            create_chat_session(db_session, f"pytest_session_{y}", user_id, persona_id)
 
         # randomize all session times
         logger.info(f"Seeding {num_messages} messages per session.")

backend/onyx/db/user_file.py

@@ -3,6 +3,7 @@ from uuid import UUID
 
 from sqlalchemy import func
 from sqlalchemy import select
+from sqlalchemy.orm import selectinload
 from sqlalchemy.orm import Session
 
 from onyx.db.models import UserFile
@@ -64,6 +65,23 @@ def fetch_user_project_ids_for_user_files(
     }
 
 
+def fetch_persona_ids_for_user_files(
+    user_file_ids: list[str],
+    db_session: Session,
+) -> dict[str, list[int]]:
+    """Fetch persona (assistant) ids for specified user files."""
+    stmt = (
+        select(UserFile)
+        .where(UserFile.id.in_(user_file_ids))
+        .options(selectinload(UserFile.assistants))
+    )
+    results = db_session.execute(stmt).scalars().all()
+    return {
+        str(user_file.id): [persona.id for persona in user_file.assistants]
+        for user_file in results
+    }
+
+
 def update_last_accessed_at_for_user_files(
     user_file_ids: list[UUID],
     db_session: Session,

backend/onyx/document_index/interfaces.py

@@ -121,6 +121,7 @@ class VespaDocumentUserFields:
     """
 
     user_projects: list[int] | None = None
+    personas: list[int] | None = None
 
 
 @dataclass

backend/onyx/document_index/interfaces_new.py

@@ -148,6 +148,7 @@ class MetadataUpdateRequest(BaseModel):
     hidden: bool | None = None
     secondary_index_updated: bool | None = None
     project_ids: set[int] | None = None
+    persona_ids: set[int] | None = None
 
 
 class IndexRetrievalFilters(BaseModel):

backend/onyx/document_index/opensearch/opensearch_document_index.py

@@ -50,6 +50,7 @@ from onyx.document_index.opensearch.schema import DocumentSchema
 from onyx.document_index.opensearch.schema import get_opensearch_doc_chunk_id
 from onyx.document_index.opensearch.schema import GLOBAL_BOOST_FIELD_NAME
 from onyx.document_index.opensearch.schema import HIDDEN_FIELD_NAME
+from onyx.document_index.opensearch.schema import PERSONAS_FIELD_NAME
 from onyx.document_index.opensearch.schema import USER_PROJECTS_FIELD_NAME
 from onyx.document_index.opensearch.search import DocumentQuery
 from onyx.document_index.opensearch.search import (
@@ -215,6 +216,7 @@ def _convert_onyx_chunk_to_opensearch_document(
         # OpenSearch and it will not store any data at all for this field, which
         # is different from supplying an empty list.
         user_projects=chunk.user_project or None,
+        personas=chunk.personas or None,
         primary_owners=get_experts_stores_representations(
             chunk.source_document.primary_owners
         ),
@@ -362,6 +364,11 @@ class OpenSearchOldDocumentIndex(OldDocumentIndex):
                 if user_fields and user_fields.user_projects
                 else None
             ),
+            persona_ids=(
+                set(user_fields.personas)
+                if user_fields and user_fields.personas
+                else None
+            ),
         )
 
         try:
@@ -709,6 +716,10 @@ class OpenSearchDocumentIndex(DocumentIndex):
                 properties_to_update[USER_PROJECTS_FIELD_NAME] = list(
                     update_request.project_ids
                 )
+            if update_request.persona_ids is not None:
+                properties_to_update[PERSONAS_FIELD_NAME] = list(
+                    update_request.persona_ids
+                )
 
             if not properties_to_update:
                 if len(update_request.document_ids) > 1:

backend/onyx/document_index/opensearch/schema.py

@@ -41,6 +41,7 @@ IMAGE_FILE_ID_FIELD_NAME = "image_file_id"
 SOURCE_LINKS_FIELD_NAME = "source_links"
 DOCUMENT_SETS_FIELD_NAME = "document_sets"
 USER_PROJECTS_FIELD_NAME = "user_projects"
+PERSONAS_FIELD_NAME = "personas"
 DOCUMENT_ID_FIELD_NAME = "document_id"
 CHUNK_INDEX_FIELD_NAME = "chunk_index"
 MAX_CHUNK_SIZE_FIELD_NAME = "max_chunk_size"
@@ -156,6 +157,7 @@ class DocumentChunk(BaseModel):
 
     document_sets: list[str] | None = None
     user_projects: list[int] | None = None
+    personas: list[int] | None = None
     primary_owners: list[str] | None = None
     secondary_owners: list[str] | None = None
 
@@ -485,6 +487,7 @@ class DocumentSchema:
                 # Product-specific fields.
                 DOCUMENT_SETS_FIELD_NAME: {"type": "keyword"},
                 USER_PROJECTS_FIELD_NAME: {"type": "integer"},
+                PERSONAS_FIELD_NAME: {"type": "integer"},
                 PRIMARY_OWNERS_FIELD_NAME: {"type": "keyword"},
                 SECONDARY_OWNERS_FIELD_NAME: {"type": "keyword"},
                 # OpenSearch metadata fields.

backend/onyx/document_index/opensearch/search.py

@@ -28,6 +28,7 @@ from onyx.document_index.opensearch.schema import HIDDEN_FIELD_NAME
 from onyx.document_index.opensearch.schema import LAST_UPDATED_FIELD_NAME
 from onyx.document_index.opensearch.schema import MAX_CHUNK_SIZE_FIELD_NAME
 from onyx.document_index.opensearch.schema import METADATA_LIST_FIELD_NAME
+from onyx.document_index.opensearch.schema import PERSONAS_FIELD_NAME
 from onyx.document_index.opensearch.schema import PUBLIC_FIELD_NAME
 from onyx.document_index.opensearch.schema import set_or_convert_timezone_to_utc
 from onyx.document_index.opensearch.schema import SOURCE_TYPE_FIELD_NAME
@@ -144,6 +145,7 @@ class DocumentQuery:
             document_sets=index_filters.document_set or [],
             user_file_ids=index_filters.user_file_ids or [],
             project_id=index_filters.project_id,
+            persona_id=index_filters.persona_id,
             time_cutoff=index_filters.time_cutoff,
             min_chunk_index=min_chunk_index,
             max_chunk_index=max_chunk_index,
@@ -202,6 +204,7 @@ class DocumentQuery:
             document_sets=[],
             user_file_ids=[],
             project_id=None,
+            persona_id=None,
             time_cutoff=None,
             min_chunk_index=None,
             max_chunk_index=None,
@@ -267,6 +270,7 @@ class DocumentQuery:
             document_sets=index_filters.document_set or [],
             user_file_ids=index_filters.user_file_ids or [],
             project_id=index_filters.project_id,
+            persona_id=index_filters.persona_id,
             time_cutoff=index_filters.time_cutoff,
             min_chunk_index=None,
             max_chunk_index=None,
@@ -334,6 +338,7 @@ class DocumentQuery:
             document_sets=index_filters.document_set or [],
             user_file_ids=index_filters.user_file_ids or [],
             project_id=index_filters.project_id,
+            persona_id=index_filters.persona_id,
             time_cutoff=index_filters.time_cutoff,
             min_chunk_index=None,
             max_chunk_index=None,
@@ -496,6 +501,7 @@ class DocumentQuery:
         document_sets: list[str],
         user_file_ids: list[UUID],
         project_id: int | None,
+        persona_id: int | None,
         time_cutoff: datetime | None,
         min_chunk_index: int | None,
         max_chunk_index: int | None,
@@ -530,6 +536,8 @@ class DocumentQuery:
                 retrieved.
             project_id: If not None, only documents with this project ID in user
                 projects will be retrieved.
+            persona_id: If not None, only documents whose personas array
+                contains this persona ID will be retrieved.
             time_cutoff: Time cutoff for the documents to retrieve. If not None,
                 Documents which were last updated before this date will not be
                 returned. For documents which do not have a value for their last
@@ -627,6 +635,9 @@ class DocumentQuery:
             )
             return user_project_filter
 
+        def _get_persona_filter(persona_id: int) -> dict[str, Any]:
+            return {"term": {PERSONAS_FIELD_NAME: {"value": persona_id}}}
+
         def _get_time_cutoff_filter(time_cutoff: datetime) -> dict[str, Any]:
             # Convert to UTC if not already so the cutoff is comparable to the
             # document data.
@@ -780,6 +791,9 @@ class DocumentQuery:
             # document's user projects list.
             filter_clauses.append(_get_user_project_filter(project_id))
 
+        if persona_id is not None:
+            filter_clauses.append(_get_persona_filter(persona_id))
+
         if time_cutoff is not None:
             # If a time cutoff is provided, the caller will only retrieve
             # documents where the document was last updated at or after the time

backend/onyx/document_index/vespa/app_config/schemas/danswer_chunk.sd.jinja

@@ -181,6 +181,11 @@ schema {{ schema_name }} {
             rank: filter
             attribute: fast-search
         }
+        field personas type array<int> {
+            indexing: summary | attribute
+            rank: filter
+            attribute: fast-search
+        }
     }
 
     # If using different tokenization settings, the fieldset has to be removed, and the field must

backend/onyx/document_index/vespa/index.py

@@ -689,6 +689,9 @@ class VespaIndex(DocumentIndex):
         project_ids: set[int] | None = None
         if user_fields is not None and user_fields.user_projects is not None:
             project_ids = set(user_fields.user_projects)
+        persona_ids: set[int] | None = None
+        if user_fields is not None and user_fields.personas is not None:
+            persona_ids = set(user_fields.personas)
         update_request = MetadataUpdateRequest(
             document_ids=[doc_id],
             doc_id_to_chunk_cnt={
@@ -699,6 +702,7 @@ class VespaIndex(DocumentIndex):
             boost=fields.boost if fields is not None else None,
             hidden=fields.hidden if fields is not None else None,
             project_ids=project_ids,
+            persona_ids=persona_ids,
         )
 
         vespa_document_index.update([update_request])

backend/onyx/document_index/vespa/indexing_utils.py

@@ -46,6 +46,7 @@ from onyx.document_index.vespa_constants import METADATA
 from onyx.document_index.vespa_constants import METADATA_LIST
 from onyx.document_index.vespa_constants import METADATA_SUFFIX
 from onyx.document_index.vespa_constants import NUM_THREADS
+from onyx.document_index.vespa_constants import PERSONAS
 from onyx.document_index.vespa_constants import PRIMARY_OWNERS
 from onyx.document_index.vespa_constants import SECONDARY_OWNERS
 from onyx.document_index.vespa_constants import SECTION_CONTINUATION
@@ -218,6 +219,7 @@ def _index_vespa_chunk(
         # still called `image_file_name` in Vespa for backwards compatibility
         IMAGE_FILE_NAME: chunk.image_file_id,
         USER_PROJECT: chunk.user_project if chunk.user_project is not None else [],
+        PERSONAS: chunk.personas if chunk.personas is not None else [],
         BOOST: chunk.boost,
         AGGREGATED_CHUNK_BOOST_FACTOR: chunk.aggregated_chunk_boost_factor,
     }

backend/onyx/document_index/vespa/shared_utils/vespa_request_builders.py

@@ -12,6 +12,7 @@ from onyx.document_index.vespa_constants import DOCUMENT_ID
 from onyx.document_index.vespa_constants import DOCUMENT_SETS
 from onyx.document_index.vespa_constants import HIDDEN
 from onyx.document_index.vespa_constants import METADATA_LIST
+from onyx.document_index.vespa_constants import PERSONAS
 from onyx.document_index.vespa_constants import SOURCE_TYPE
 from onyx.document_index.vespa_constants import TENANT_ID
 from onyx.document_index.vespa_constants import USER_PROJECT
@@ -149,6 +150,18 @@ def build_vespa_filters(
         # Vespa YQL 'contains' expects a string literal; quote the integer
         return f'({USER_PROJECT} contains "{pid}") and '
 
+    def _build_persona_filter(
+        persona_id: int | None,
+    ) -> str:
+        if persona_id is None:
+            return ""
+        try:
+            pid = int(persona_id)
+        except Exception:
+            logger.warning(f"Invalid persona ID: {persona_id}")
+            return ""
+        return f'({PERSONAS} contains "{pid}") and '
+
     # Start building the filter string
     filter_str = f"!({HIDDEN}=true) and " if not include_hidden else ""
 
@@ -192,6 +205,9 @@ def build_vespa_filters(
     # User project filter (array<int> attribute membership)
     filter_str += _build_user_project_filter(filters.project_id)
 
+    # Persona filter (array<int> attribute membership)
+    filter_str += _build_persona_filter(filters.persona_id)
+
     # Time filter
     filter_str += _build_time_filter(filters.time_cutoff)
 

backend/onyx/document_index/vespa/vespa_document_index.py

@@ -183,6 +183,10 @@ def _update_single_chunk(
         model_config = {"frozen": True}
         assign: list[int]
 
+    class _Personas(BaseModel):
+        model_config = {"frozen": True}
+        assign: list[int]
+
     class _VespaPutFields(BaseModel):
         model_config = {"frozen": True}
         # The names of these fields are based the Vespa schema. Changes to the
@@ -193,6 +197,7 @@ def _update_single_chunk(
         access_control_list: _AccessControl | None = None
         hidden: _Hidden | None = None
         user_project: _UserProjects | None = None
+        personas: _Personas | None = None
 
     class _VespaPutRequest(BaseModel):
         model_config = {"frozen": True}
@@ -227,6 +232,11 @@ def _update_single_chunk(
         if update_request.project_ids is not None
         else None
     )
+    personas_update: _Personas | None = (
+        _Personas(assign=list(update_request.persona_ids))
+        if update_request.persona_ids is not None
+        else None
+    )
 
     vespa_put_fields = _VespaPutFields(
         boost=boost_update,
@@ -234,6 +244,7 @@ def _update_single_chunk(
         access_control_list=access_update,
         hidden=hidden_update,
         user_project=user_projects_update,
+        personas=personas_update,
     )
 
     vespa_put_request = _VespaPutRequest(
@@ -554,10 +565,9 @@ class VespaDocumentIndex(DocumentIndex):
         num_to_retrieve: int,
     ) -> list[InferenceChunk]:
         vespa_where_clauses = build_vespa_filters(filters)
-        # Needs to be at least as much as the rerank-count value set in the
-        # Vespa schema config. Otherwise we would be getting fewer results than
-        # expected for reranking.
-        target_hits = max(10 * num_to_retrieve, RERANK_COUNT)
+        # Avoid over-fetching a very large candidate set for global-phase reranking.
+        # Keep enough headroom for quality while capping cost on larger indices.
+        target_hits = min(max(4 * num_to_retrieve, 100), RERANK_COUNT)
 
         yql = (
             YQL_BASE.format(index_name=self._index_name)

backend/onyx/document_index/vespa_constants.py

@@ -58,6 +58,7 @@ DOCUMENT_SETS = "document_sets"
 USER_FILE = "user_file"
 USER_FOLDER = "user_folder"
 USER_PROJECT = "user_project"
+PERSONAS = "personas"
 LARGE_CHUNK_REFERENCE_IDS = "large_chunk_reference_ids"
 METADATA = "metadata"
 METADATA_LIST = "metadata_list"

backend/onyx/image_gen/interfaces.py

@@ -20,7 +20,20 @@ class ImageGenerationProviderCredentials(BaseModel):
     custom_config: dict[str, str] | None = None
 
 
+class ReferenceImage(BaseModel):
+    data: bytes
+    mime_type: str
+
+
 class ImageGenerationProvider(abc.ABC):
+    @property
+    def supports_reference_images(self) -> bool:
+        return False
+
+    @property
+    def max_reference_images(self) -> int:
+        return 0
+
     @classmethod
     @abc.abstractmethod
     def validate_credentials(
@@ -63,6 +76,7 @@ class ImageGenerationProvider(abc.ABC):
         size: str,
         n: int,
         quality: str | None = None,
+        reference_images: list[ReferenceImage] | None = None,
         **kwargs: Any,
     ) -> ImageGenerationResponse:
         """Generates an image based on a prompt."""

backend/onyx/image_gen/providers/azure_img_gen.py

@@ -5,12 +5,16 @@ from typing import TYPE_CHECKING
 
 from onyx.image_gen.interfaces import ImageGenerationProvider
 from onyx.image_gen.interfaces import ImageGenerationProviderCredentials
+from onyx.image_gen.interfaces import ReferenceImage
 
 if TYPE_CHECKING:
     from onyx.image_gen.interfaces import ImageGenerationResponse
 
 
 class AzureImageGenerationProvider(ImageGenerationProvider):
+    _GPT_IMAGE_MODEL_PREFIX = "gpt-image-"
+    _DALL_E_2_MODEL_NAME = "dall-e-2"
+
     def __init__(
         self,
         api_key: str,
@@ -52,6 +56,25 @@ class AzureImageGenerationProvider(ImageGenerationProvider):
             deployment_name=credentials.deployment_name,
         )
 
+    @property
+    def supports_reference_images(self) -> bool:
+        return True
+
+    @property
+    def max_reference_images(self) -> int:
+        # Azure GPT image models support up to 16 input images for edits.
+        return 16
+
+    def _normalize_model_name(self, model: str) -> str:
+        return model.rsplit("/", 1)[-1]
+
+    def _model_supports_image_edits(self, model: str) -> bool:
+        normalized_model = self._normalize_model_name(model)
+        return (
+            normalized_model.startswith(self._GPT_IMAGE_MODEL_PREFIX)
+            or normalized_model == self._DALL_E_2_MODEL_NAME
+        )
+
     def generate_image(
         self,
         prompt: str,
@@ -59,13 +82,44 @@ class AzureImageGenerationProvider(ImageGenerationProvider):
         size: str,
         n: int,
         quality: str | None = None,
+        reference_images: list[ReferenceImage] | None = None,
         **kwargs: Any,
     ) -> ImageGenerationResponse:
-        from litellm import image_generation
-
         deployment = self._deployment_name or model
         model_name = f"azure/{deployment}"
 
+        if reference_images:
+            if not self._model_supports_image_edits(model):
+                raise ValueError(
+                    f"Model '{model}' does not support image edits with reference images."
+                )
+
+            normalized_model = self._normalize_model_name(model)
+            if (
+                normalized_model == self._DALL_E_2_MODEL_NAME
+                and len(reference_images) > 1
+            ):
+                raise ValueError(
+                    "Model 'dall-e-2' only supports a single reference image for edits."
+                )
+
+            from litellm import image_edit
+
+            return image_edit(
+                image=[image.data for image in reference_images],
+                prompt=prompt,
+                model=model_name,
+                api_key=self._api_key,
+                api_base=self._api_base,
+                api_version=self._api_version,
+                size=size,
+                n=n,
+                quality=quality,
+                **kwargs,
+            )
+
+        from litellm import image_generation
+
         return image_generation(
             prompt=prompt,
             model=model_name,

backend/onyx/image_gen/providers/openai_img_gen.py

@@ -5,12 +5,16 @@ from typing import TYPE_CHECKING
 
 from onyx.image_gen.interfaces import ImageGenerationProvider
 from onyx.image_gen.interfaces import ImageGenerationProviderCredentials
+from onyx.image_gen.interfaces import ReferenceImage
 
 if TYPE_CHECKING:
     from onyx.image_gen.interfaces import ImageGenerationResponse
 
 
 class OpenAIImageGenerationProvider(ImageGenerationProvider):
+    _GPT_IMAGE_MODEL_PREFIX = "gpt-image-"
+    _DALL_E_2_MODEL_NAME = "dall-e-2"
+
     def __init__(
         self,
         api_key: str,
@@ -38,6 +42,25 @@ class OpenAIImageGenerationProvider(ImageGenerationProvider):
             api_base=credentials.api_base,
         )
 
+    @property
+    def supports_reference_images(self) -> bool:
+        return True
+
+    @property
+    def max_reference_images(self) -> int:
+        # GPT image models support up to 16 input images for edits.
+        return 16
+
+    def _normalize_model_name(self, model: str) -> str:
+        return model.rsplit("/", 1)[-1]
+
+    def _model_supports_image_edits(self, model: str) -> bool:
+        normalized_model = self._normalize_model_name(model)
+        return (
+            normalized_model.startswith(self._GPT_IMAGE_MODEL_PREFIX)
+            or normalized_model == self._DALL_E_2_MODEL_NAME
+        )
+
     def generate_image(
         self,
         prompt: str,
@@ -45,8 +68,38 @@ class OpenAIImageGenerationProvider(ImageGenerationProvider):
         size: str,
         n: int,
         quality: str | None = None,
+        reference_images: list[ReferenceImage] | None = None,
         **kwargs: Any,
     ) -> ImageGenerationResponse:
+        if reference_images:
+            if not self._model_supports_image_edits(model):
+                raise ValueError(
+                    f"Model '{model}' does not support image edits with reference images."
+                )
+
+            normalized_model = self._normalize_model_name(model)
+            if (
+                normalized_model == self._DALL_E_2_MODEL_NAME
+                and len(reference_images) > 1
+            ):
+                raise ValueError(
+                    "Model 'dall-e-2' only supports a single reference image for edits."
+                )
+
+            from litellm import image_edit
+
+            return image_edit(
+                image=[image.data for image in reference_images],
+                prompt=prompt,
+                model=model,
+                api_key=self._api_key,
+                api_base=self._api_base,
+                size=size,
+                n=n,
+                quality=quality,
+                **kwargs,
+            )
+
         from litellm import image_generation
 
         return image_generation(

backend/onyx/image_gen/providers/vertex_img_gen.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+import base64
 import json
+from datetime import datetime
 from typing import Any
 from typing import TYPE_CHECKING
 
@@ -9,6 +11,7 @@ from pydantic import BaseModel
 from onyx.image_gen.exceptions import ImageProviderCredentialsError
 from onyx.image_gen.interfaces import ImageGenerationProvider
 from onyx.image_gen.interfaces import ImageGenerationProviderCredentials
+from onyx.image_gen.interfaces import ReferenceImage
 
 if TYPE_CHECKING:
     from onyx.image_gen.interfaces import ImageGenerationResponse
@@ -51,6 +54,15 @@ class VertexImageGenerationProvider(ImageGenerationProvider):
             vertex_credentials=vertex_credentials,
         )
 
+    @property
+    def supports_reference_images(self) -> bool:
+        return True
+
+    @property
+    def max_reference_images(self) -> int:
+        # Gemini image editing supports up to 14 input images.
+        return 14
+
     def generate_image(
         self,
         prompt: str,
@@ -58,8 +70,18 @@ class VertexImageGenerationProvider(ImageGenerationProvider):
         size: str,
         n: int,
         quality: str | None = None,
+        reference_images: list[ReferenceImage] | None = None,
         **kwargs: Any,
     ) -> ImageGenerationResponse:
+        if reference_images:
+            return self._generate_image_with_reference_images(
+                prompt=prompt,
+                model=model,
+                size=size,
+                n=n,
+                reference_images=reference_images,
+            )
+
         from litellm import image_generation
 
         return image_generation(
@@ -74,6 +96,99 @@ class VertexImageGenerationProvider(ImageGenerationProvider):
             **kwargs,
         )
 
+    def _generate_image_with_reference_images(
+        self,
+        prompt: str,
+        model: str,
+        size: str,
+        n: int,
+        reference_images: list[ReferenceImage],
+    ) -> ImageGenerationResponse:
+        from google import genai
+        from google.genai import types as genai_types
+        from google.oauth2 import service_account
+        from litellm.types.utils import ImageObject
+        from litellm.types.utils import ImageResponse
+
+        service_account_info = json.loads(self._vertex_credentials)
+        credentials = service_account.Credentials.from_service_account_info(
+            service_account_info,
+            scopes=["https://www.googleapis.com/auth/cloud-platform"],
+        )
+
+        client = genai.Client(
+            vertexai=True,
+            project=self._vertex_project,
+            location=self._vertex_location,
+            credentials=credentials,
+        )
+
+        parts: list[genai_types.Part] = [
+            genai_types.Part.from_bytes(data=image.data, mime_type=image.mime_type)
+            for image in reference_images
+        ]
+        parts.append(genai_types.Part.from_text(text=prompt))
+
+        config = genai_types.GenerateContentConfig(
+            response_modalities=["TEXT", "IMAGE"],
+            candidate_count=max(1, n),
+            image_config=genai_types.ImageConfig(
+                aspect_ratio=_map_size_to_aspect_ratio(size)
+            ),
+        )
+        model_name = model.replace("vertex_ai/", "")
+        response = client.models.generate_content(
+            model=model_name,
+            contents=genai_types.Content(
+                role="user",
+                parts=parts,
+            ),
+            config=config,
+        )
+
+        generated_data: list[ImageObject] = []
+        for candidate in response.candidates or []:
+            candidate_content = candidate.content
+            if not candidate_content:
+                continue
+
+            for part in candidate_content.parts or []:
+                inline_data = part.inline_data
+                if not inline_data or inline_data.data is None:
+                    continue
+
+                if isinstance(inline_data.data, bytes):
+                    b64_json = base64.b64encode(inline_data.data).decode("utf-8")
+                elif isinstance(inline_data.data, str):
+                    b64_json = inline_data.data
+                else:
+                    continue
+
+                generated_data.append(
+                    ImageObject(
+                        b64_json=b64_json,
+                        revised_prompt=prompt,
+                    )
+                )
+
+        if not generated_data:
+            raise RuntimeError("No image data returned from Vertex AI.")
+
+        return ImageResponse(
+            created=int(datetime.now().timestamp()),
+            data=generated_data,
+        )
+
+
+def _map_size_to_aspect_ratio(size: str) -> str:
+    return {
+        "1024x1024": "1:1",
+        "1792x1024": "16:9",
+        "1024x1792": "9:16",
+        "1536x1024": "3:2",
+        "1024x1536": "2:3",
+    }.get(size, "1:1")
+
 
 def _parse_to_vertex_credentials(
     credentials: ImageGenerationProviderCredentials,

backend/onyx/indexing/adapters/document_indexing_adapter.py

@@ -146,6 +146,7 @@ class DocumentIndexingBatchAdapter:
                     doc_id_to_document_set.get(chunk.source_document.id, [])
                 ),
                 user_project=[],
+                personas=[],
                 boost=(
                     context.id_to_boost_map[chunk.source_document.id]
                     if chunk.source_document.id in context.id_to_boost_map

backend/onyx/indexing/adapters/user_file_indexing_adapter.py

@@ -20,6 +20,7 @@ from onyx.db.models import Persona
 from onyx.db.models import UserFile
 from onyx.db.notification import create_notification
 from onyx.db.user_file import fetch_chunk_counts_for_user_files
+from onyx.db.user_file import fetch_persona_ids_for_user_files
 from onyx.db.user_file import fetch_user_project_ids_for_user_files
 from onyx.file_store.utils import store_user_file_plaintext
 from onyx.indexing.indexing_pipeline import DocumentBatchPrepareContext
@@ -119,6 +120,10 @@ class UserFileIndexingAdapter:
             user_file_ids=updatable_ids,
             db_session=self.db_session,
         )
+        user_file_id_to_persona_ids = fetch_persona_ids_for_user_files(
+            user_file_ids=updatable_ids,
+            db_session=self.db_session,
+        )
         user_file_id_to_access: dict[str, DocumentAccess] = get_access_for_user_files(
             user_file_ids=updatable_ids,
             db_session=self.db_session,
@@ -182,7 +187,7 @@ class UserFileIndexingAdapter:
                 user_project=user_file_id_to_project_ids.get(
                     chunk.source_document.id, []
                 ),
-                # we are going to index userfiles only once, so we just set the boost to the default
+                personas=user_file_id_to_persona_ids.get(chunk.source_document.id, []),
                 boost=DEFAULT_BOOST,
                 tenant_id=tenant_id,
                 aggregated_chunk_boost_factor=chunk_content_scores[chunk_num],

backend/onyx/indexing/indexing_pipeline.py

@@ -49,6 +49,7 @@ from onyx.indexing.embedder import IndexingEmbedder
 from onyx.indexing.models import DocAwareChunk
 from onyx.indexing.models import IndexingBatchAdapter
 from onyx.indexing.models import UpdatableChunkData
+from onyx.indexing.postgres_sanitization import sanitize_documents_for_postgres
 from onyx.indexing.vector_db_insertion import write_chunks_to_vector_db_with_backoff
 from onyx.llm.factory import get_default_llm_with_vision
 from onyx.llm.factory import get_llm_for_contextual_rag
@@ -228,6 +229,8 @@ def index_doc_batch_prepare(
 ) -> DocumentBatchPrepareContext | None:
     """Sets up the documents in the relational DB (source of truth) for permissions, metadata, etc.
     This preceeds indexing it into the actual document index."""
+    documents = sanitize_documents_for_postgres(documents)
+
     # Create a trimmed list of docs that don't have a newer updated at
     # Shortcuts the time-consuming flow on connector index retries
     document_ids: list[str] = [document.id for document in documents]

backend/onyx/indexing/models.py

@@ -112,6 +112,7 @@ class DocMetadataAwareIndexChunk(IndexChunk):
     access: "DocumentAccess"
     document_sets: set[str]
     user_project: list[int]
+    personas: list[int]
     boost: int
     aggregated_chunk_boost_factor: float
     # Full ancestor path from root hierarchy node to document's parent.
@@ -126,6 +127,7 @@ class DocMetadataAwareIndexChunk(IndexChunk):
         access: "DocumentAccess",
         document_sets: set[str],
         user_project: list[int],
+        personas: list[int],
         boost: int,
         aggregated_chunk_boost_factor: float,
         tenant_id: str,
@@ -137,6 +139,7 @@ class DocMetadataAwareIndexChunk(IndexChunk):
             access=access,
             document_sets=document_sets,
             user_project=user_project,
+            personas=personas,
             boost=boost,
             aggregated_chunk_boost_factor=aggregated_chunk_boost_factor,
             tenant_id=tenant_id,

backend/onyx/indexing/postgres_sanitization.py

@@ -0,0 +1,150 @@
+from typing import Any
+
+from onyx.access.models import ExternalAccess
+from onyx.connectors.models import BasicExpertInfo
+from onyx.connectors.models import Document
+from onyx.connectors.models import HierarchyNode
+
+
+def _sanitize_string(value: str) -> str:
+    return value.replace("\x00", "")
+
+
+def _sanitize_json_like(value: Any) -> Any:
+    if isinstance(value, str):
+        return _sanitize_string(value)
+
+    if isinstance(value, list):
+        return [_sanitize_json_like(item) for item in value]
+
+    if isinstance(value, tuple):
+        return tuple(_sanitize_json_like(item) for item in value)
+
+    if isinstance(value, dict):
+        sanitized: dict[Any, Any] = {}
+        for key, nested_value in value.items():
+            cleaned_key = _sanitize_string(key) if isinstance(key, str) else key
+            sanitized[cleaned_key] = _sanitize_json_like(nested_value)
+        return sanitized
+
+    return value
+
+
+def _sanitize_expert_info(expert: BasicExpertInfo) -> BasicExpertInfo:
+    return expert.model_copy(
+        update={
+            "display_name": (
+                _sanitize_string(expert.display_name)
+                if expert.display_name is not None
+                else None
+            ),
+            "first_name": (
+                _sanitize_string(expert.first_name)
+                if expert.first_name is not None
+                else None
+            ),
+            "middle_initial": (
+                _sanitize_string(expert.middle_initial)
+                if expert.middle_initial is not None
+                else None
+            ),
+            "last_name": (
+                _sanitize_string(expert.last_name)
+                if expert.last_name is not None
+                else None
+            ),
+            "email": (
+                _sanitize_string(expert.email) if expert.email is not None else None
+            ),
+        }
+    )
+
+
+def _sanitize_external_access(external_access: ExternalAccess) -> ExternalAccess:
+    return ExternalAccess(
+        external_user_emails={
+            _sanitize_string(email) for email in external_access.external_user_emails
+        },
+        external_user_group_ids={
+            _sanitize_string(group_id)
+            for group_id in external_access.external_user_group_ids
+        },
+        is_public=external_access.is_public,
+    )
+
+
+def sanitize_document_for_postgres(document: Document) -> Document:
+    cleaned_doc = document.model_copy(deep=True)
+
+    cleaned_doc.id = _sanitize_string(cleaned_doc.id)
+    cleaned_doc.semantic_identifier = _sanitize_string(cleaned_doc.semantic_identifier)
+    if cleaned_doc.title is not None:
+        cleaned_doc.title = _sanitize_string(cleaned_doc.title)
+    if cleaned_doc.parent_hierarchy_raw_node_id is not None:
+        cleaned_doc.parent_hierarchy_raw_node_id = _sanitize_string(
+            cleaned_doc.parent_hierarchy_raw_node_id
+        )
+
+    cleaned_doc.metadata = {
+        _sanitize_string(key): (
+            [_sanitize_string(item) for item in value]
+            if isinstance(value, list)
+            else _sanitize_string(value)
+        )
+        for key, value in cleaned_doc.metadata.items()
+    }
+
+    if cleaned_doc.doc_metadata is not None:
+        cleaned_doc.doc_metadata = _sanitize_json_like(cleaned_doc.doc_metadata)
+
+    if cleaned_doc.primary_owners is not None:
+        cleaned_doc.primary_owners = [
+            _sanitize_expert_info(expert) for expert in cleaned_doc.primary_owners
+        ]
+    if cleaned_doc.secondary_owners is not None:
+        cleaned_doc.secondary_owners = [
+            _sanitize_expert_info(expert) for expert in cleaned_doc.secondary_owners
+        ]
+
+    if cleaned_doc.external_access is not None:
+        cleaned_doc.external_access = _sanitize_external_access(
+            cleaned_doc.external_access
+        )
+
+    for section in cleaned_doc.sections:
+        if section.link is not None:
+            section.link = _sanitize_string(section.link)
+        if section.text is not None:
+            section.text = _sanitize_string(section.text)
+        if section.image_file_id is not None:
+            section.image_file_id = _sanitize_string(section.image_file_id)
+
+    return cleaned_doc
+
+
+def sanitize_documents_for_postgres(documents: list[Document]) -> list[Document]:
+    return [sanitize_document_for_postgres(document) for document in documents]
+
+
+def sanitize_hierarchy_node_for_postgres(node: HierarchyNode) -> HierarchyNode:
+    cleaned_node = node.model_copy(deep=True)
+
+    cleaned_node.raw_node_id = _sanitize_string(cleaned_node.raw_node_id)
+    cleaned_node.display_name = _sanitize_string(cleaned_node.display_name)
+    if cleaned_node.raw_parent_id is not None:
+        cleaned_node.raw_parent_id = _sanitize_string(cleaned_node.raw_parent_id)
+    if cleaned_node.link is not None:
+        cleaned_node.link = _sanitize_string(cleaned_node.link)
+
+    if cleaned_node.external_access is not None:
+        cleaned_node.external_access = _sanitize_external_access(
+            cleaned_node.external_access
+        )
+
+    return cleaned_node
+
+
+def sanitize_hierarchy_nodes_for_postgres(
+    nodes: list[HierarchyNode],
+) -> list[HierarchyNode]:
+    return [sanitize_hierarchy_node_for_postgres(node) for node in nodes]

backend/onyx/llm/model_metadata_enrichments.json

@@ -64,21 +64,6 @@
     "model_vendor": "anthropic",
     "model_version": "20241022-v2:0"
   },
-  "anthropic.claude-3-7-sonnet-20240620-v1:0": {
-    "display_name": "Claude Sonnet 3.7",
-    "model_vendor": "anthropic",
-    "model_version": "20240620-v1:0"
-  },
-  "anthropic.claude-3-7-sonnet-20250219-v1:0": {
-    "display_name": "Claude Sonnet 3.7",
-    "model_vendor": "anthropic",
-    "model_version": "20250219-v1:0"
-  },
-  "anthropic.claude-3-haiku-20240307-v1:0": {
-    "display_name": "Claude Haiku 3",
-    "model_vendor": "anthropic",
-    "model_version": "20240307-v1:0"
-  },
   "anthropic.claude-3-sonnet-20240229-v1:0": {
     "display_name": "Claude Sonnet 3",
     "model_vendor": "anthropic",
@@ -159,11 +144,6 @@
     "model_vendor": "anthropic",
     "model_version": "20241022-v2:0"
   },
-  "apac.anthropic.claude-3-haiku-20240307-v1:0": {
-    "display_name": "Claude Haiku 3",
-    "model_vendor": "anthropic",
-    "model_version": "20240307-v1:0"
-  },
   "apac.anthropic.claude-3-sonnet-20240229-v1:0": {
     "display_name": "Claude Sonnet 3",
     "model_vendor": "anthropic",
@@ -1320,11 +1300,6 @@
     "model_vendor": "anthropic",
     "model_version": "20240620-v1:0"
   },
-  "bedrock/us-gov-east-1/anthropic.claude-3-haiku-20240307-v1:0": {
-    "display_name": "Claude Haiku 3",
-    "model_vendor": "anthropic",
-    "model_version": "20240307-v1:0"
-  },
   "bedrock/us-gov-east-1/claude-sonnet-4-5-20250929-v1:0": {
     "display_name": "Claude Sonnet 4.5",
     "model_vendor": "anthropic",
@@ -1365,16 +1340,6 @@
     "model_vendor": "anthropic",
     "model_version": "20240620-v1:0"
   },
-  "bedrock/us-gov-west-1/anthropic.claude-3-7-sonnet-20250219-v1:0": {
-    "display_name": "Claude Sonnet 3.7",
-    "model_vendor": "anthropic",
-    "model_version": "20250219-v1:0"
-  },
-  "bedrock/us-gov-west-1/anthropic.claude-3-haiku-20240307-v1:0": {
-    "display_name": "Claude Haiku 3",
-    "model_vendor": "anthropic",
-    "model_version": "20240307-v1:0"
-  },
   "bedrock/us-gov-west-1/claude-sonnet-4-5-20250929-v1:0": {
     "display_name": "Claude Sonnet 4.5",
     "model_vendor": "anthropic",
@@ -1505,26 +1470,6 @@
     "model_vendor": "anthropic",
     "model_version": "latest"
   },
-  "claude-3-7-sonnet-20250219": {
-    "display_name": "Claude Sonnet 3.7",
-    "model_vendor": "anthropic",
-    "model_version": "20250219"
-  },
-  "claude-3-7-sonnet-latest": {
-    "display_name": "Claude Sonnet 3.7",
-    "model_vendor": "anthropic",
-    "model_version": "latest"
-  },
-  "claude-3-7-sonnet@20250219": {
-    "display_name": "Claude Sonnet 3.7",
-    "model_vendor": "anthropic",
-    "model_version": "20250219"
-  },
-  "claude-3-haiku-20240307": {
-    "display_name": "Claude Haiku 3",
-    "model_vendor": "anthropic",
-    "model_version": "20240307"
-  },
   "claude-4-opus-20250514": {
     "display_name": "Claude Opus 4",
     "model_vendor": "anthropic",
@@ -1705,16 +1650,6 @@
     "model_vendor": "anthropic",
     "model_version": "20241022-v2:0"
   },
-  "eu.anthropic.claude-3-7-sonnet-20250219-v1:0": {
-    "display_name": "Claude Sonnet 3.7",
-    "model_vendor": "anthropic",
-    "model_version": "20250219-v1:0"
-  },
-  "eu.anthropic.claude-3-haiku-20240307-v1:0": {
-    "display_name": "Claude Haiku 3",
-    "model_vendor": "anthropic",
-    "model_version": "20240307-v1:0"
-  },
   "eu.anthropic.claude-3-sonnet-20240229-v1:0": {
     "display_name": "Claude Sonnet 3",
     "model_vendor": "anthropic",
@@ -3226,15 +3161,6 @@
     "model_vendor": "anthropic",
     "model_version": "latest"
   },
-  "openrouter/anthropic/claude-3-haiku": {
-    "display_name": "Claude Haiku 3",
-    "model_vendor": "anthropic"
-  },
-  "openrouter/anthropic/claude-3-haiku-20240307": {
-    "display_name": "Claude Haiku 3",
-    "model_vendor": "anthropic",
-    "model_version": "20240307"
-  },
   "openrouter/anthropic/claude-3-sonnet": {
     "display_name": "Claude Sonnet 3",
     "model_vendor": "anthropic"
@@ -3249,16 +3175,6 @@
     "model_vendor": "anthropic",
     "model_version": "latest"
   },
-  "openrouter/anthropic/claude-3.7-sonnet": {
-    "display_name": "Claude Sonnet 3.7",
-    "model_vendor": "anthropic",
-    "model_version": "latest"
-  },
-  "openrouter/anthropic/claude-3.7-sonnet:beta": {
-    "display_name": "Claude Sonnet 3.7:beta",
-    "model_vendor": "anthropic",
-    "model_version": "latest"
-  },
   "openrouter/anthropic/claude-haiku-4.5": {
     "display_name": "Claude Haiku 4.5",
     "model_vendor": "anthropic",
@@ -3750,16 +3666,6 @@
     "model_vendor": "anthropic",
     "model_version": "20241022"
   },
-  "us.anthropic.claude-3-7-sonnet-20250219-v1:0": {
-    "display_name": "Claude Sonnet 3.7",
-    "model_vendor": "anthropic",
-    "model_version": "20250219"
-  },
-  "us.anthropic.claude-3-haiku-20240307-v1:0": {
-    "display_name": "Claude Haiku 3",
-    "model_vendor": "anthropic",
-    "model_version": "20240307"
-  },
   "us.anthropic.claude-3-sonnet-20240229-v1:0": {
     "display_name": "Claude Sonnet 3",
     "model_vendor": "anthropic",
@@ -3879,20 +3785,6 @@
     "model_vendor": "anthropic",
     "model_version": "20240620"
   },
-  "vertex_ai/claude-3-7-sonnet@20250219": {
-    "display_name": "Claude Sonnet 3.7",
-    "model_vendor": "anthropic",
-    "model_version": "20250219"
-  },
-  "vertex_ai/claude-3-haiku": {
-    "display_name": "Claude Haiku 3",
-    "model_vendor": "anthropic"
-  },
-  "vertex_ai/claude-3-haiku@20240307": {
-    "display_name": "Claude Haiku 3",
-    "model_vendor": "anthropic",
-    "model_version": "20240307"
-  },
   "vertex_ai/claude-3-sonnet": {
     "display_name": "Claude Sonnet 3",
     "model_vendor": "anthropic"

backend/onyx/llm/well_known_providers/llm_provider_options.py

@@ -1,5 +1,7 @@
 import json
 import pathlib
+import threading
+import time
 
 from onyx.llm.constants import LlmProviderNames
 from onyx.llm.constants import PROVIDER_DISPLAY_NAMES
@@ -23,6 +25,11 @@ from onyx.utils.logger import setup_logger
 
 logger = setup_logger()
 
+_RECOMMENDATIONS_CACHE_TTL_SECONDS = 300
+_recommendations_cache_lock = threading.Lock()
+_cached_recommendations: LLMRecommendations | None = None
+_cached_recommendations_time: float = 0.0
+
 
 def _get_provider_to_models_map() -> dict[str, list[str]]:
     """Lazy-load provider model mappings to avoid importing litellm at module level.
@@ -41,19 +48,40 @@ def _get_provider_to_models_map() -> dict[str, list[str]]:
     }
 
 
-def get_recommendations() -> LLMRecommendations:
-    """Get the recommendations from the GitHub config."""
-    recommendations_from_github = fetch_llm_recommendations_from_github()
-    if recommendations_from_github:
-        return recommendations_from_github
-
-    # Fall back to json bundled with code
+def _load_bundled_recommendations() -> LLMRecommendations:
     json_path = pathlib.Path(__file__).parent / "recommended-models.json"
     with open(json_path, "r") as f:
         json_config = json.load(f)
+    return LLMRecommendations.model_validate(json_config)
 
-    recommendations_from_json = LLMRecommendations.model_validate(json_config)
-    return recommendations_from_json
+
+def get_recommendations() -> LLMRecommendations:
+    """Get the recommendations, with an in-memory cache to avoid
+    hitting GitHub on every API request."""
+    global _cached_recommendations, _cached_recommendations_time
+
+    now = time.monotonic()
+    if (
+        _cached_recommendations is not None
+        and (now - _cached_recommendations_time) < _RECOMMENDATIONS_CACHE_TTL_SECONDS
+    ):
+        return _cached_recommendations
+
+    with _recommendations_cache_lock:
+        # Double-check after acquiring lock
+        if (
+            _cached_recommendations is not None
+            and (time.monotonic() - _cached_recommendations_time)
+            < _RECOMMENDATIONS_CACHE_TTL_SECONDS
+        ):
+            return _cached_recommendations
+
+        recommendations_from_github = fetch_llm_recommendations_from_github()
+        result = recommendations_from_github or _load_bundled_recommendations()
+
+        _cached_recommendations = result
+        _cached_recommendations_time = time.monotonic()
+        return result
 
 
 def is_obsolete_model(model_name: str, provider: str) -> bool:

backend/onyx/main.py

@@ -97,6 +97,9 @@ from onyx.server.features.web_search.api import router as web_search_router
 from onyx.server.federated.api import router as federated_router
 from onyx.server.kg.api import admin_router as kg_admin_router
 from onyx.server.manage.administrative import router as admin_router
+from onyx.server.manage.code_interpreter.api import (
+    admin_router as code_interpreter_admin_router,
+)
 from onyx.server.manage.discord_bot.api import router as discord_bot_router
 from onyx.server.manage.embedding.api import admin_router as embedding_admin_router
 from onyx.server.manage.embedding.api import basic_router as embedding_router
@@ -421,6 +424,9 @@ def get_application(lifespan_override: Lifespan | None = None) -> FastAPI:
     include_router_with_global_prefix_prepended(application, llm_admin_router)
     include_router_with_global_prefix_prepended(application, kg_admin_router)
     include_router_with_global_prefix_prepended(application, llm_router)
+    include_router_with_global_prefix_prepended(
+        application, code_interpreter_admin_router
+    )
     include_router_with_global_prefix_prepended(
         application, image_generation_admin_router
     )

backend/onyx/onyxbot/slack/blocks.py

@@ -592,11 +592,8 @@ def build_slack_response_blocks(
         )
 
     citations_blocks = []
-    document_blocks = []
     if answer.citation_info:
         citations_blocks = _build_citations_blocks(answer)
-    else:
-        document_blocks = _priority_ordered_documents_blocks(answer)
 
     citations_divider = [DividerBlock()] if citations_blocks else []
     buttons_divider = [DividerBlock()] if web_follow_up_block or follow_up_block else []
@@ -608,7 +605,6 @@ def build_slack_response_blocks(
         + ai_feedback_block
         + citations_divider
         + citations_blocks
-        + document_blocks
         + buttons_divider
         + web_follow_up_block
         + follow_up_block

backend/onyx/onyxbot/slack/formatting.py

@@ -1,20 +1,149 @@
+import re
+from collections.abc import Callable
 from typing import Any
 
 from mistune import create_markdown
 from mistune import HTMLRenderer
 
+# Tags that should be replaced with a newline (line-break and block-level elements)
+_HTML_NEWLINE_TAG_PATTERN = re.compile(
+    r"<br\s*/?>|</(?:p|div|li|h[1-6]|tr|blockquote|section|article)>",
+    re.IGNORECASE,
+)
+
+# Strips HTML tags but excludes autolinks like <https://...> and <mailto:...>
+_HTML_TAG_PATTERN = re.compile(
+    r"<(?!https?://|mailto:)/?[a-zA-Z][^>]*>",
+)
+
+# Matches fenced code blocks (``` ... ```) so we can skip sanitization inside them
+_FENCED_CODE_BLOCK_PATTERN = re.compile(r"```[\s\S]*?```")
+
+# Matches the start of any markdown link: [text]( or [[n]](
+# The inner group handles nested brackets for citation links like [[1]](.
+_MARKDOWN_LINK_PATTERN = re.compile(r"\[(?:[^\[\]]|\[[^\]]*\])*\]\(")
+
+# Matches Slack-style links <url|text> that LLMs sometimes output directly.
+# Mistune doesn't recognise this syntax, so text() would escape the angle
+# brackets and Slack would render them as literal text instead of links.
+_SLACK_LINK_PATTERN = re.compile(r"<(https?://[^|>]+)\|([^>]+)>")
+
+
+def _sanitize_html(text: str) -> str:
+    """Strip HTML tags from a text fragment.
+
+    Block-level closing tags and <br> are converted to newlines.
+    All other HTML tags are removed. Autolinks (<https://...>) are preserved.
+    """
+    text = _HTML_NEWLINE_TAG_PATTERN.sub("\n", text)
+    text = _HTML_TAG_PATTERN.sub("", text)
+    return text
+
+
+def _transform_outside_code_blocks(
+    message: str, transform: Callable[[str], str]
+) -> str:
+    """Apply *transform* only to text outside fenced code blocks."""
+    parts = _FENCED_CODE_BLOCK_PATTERN.split(message)
+    code_blocks = _FENCED_CODE_BLOCK_PATTERN.findall(message)
+
+    result: list[str] = []
+    for i, part in enumerate(parts):
+        result.append(transform(part))
+        if i < len(code_blocks):
+            result.append(code_blocks[i])
+
+    return "".join(result)
+
+
+def _extract_link_destination(message: str, start_idx: int) -> tuple[str, int | None]:
+    """Extract markdown link destination, allowing nested parentheses in the URL."""
+    depth = 0
+    i = start_idx
+
+    while i < len(message):
+        curr = message[i]
+        if curr == "\\":
+            i += 2
+            continue
+
+        if curr == "(":
+            depth += 1
+        elif curr == ")":
+            if depth == 0:
+                return message[start_idx:i], i
+            depth -= 1
+        i += 1
+
+    return message[start_idx:], None
+
+
+def _normalize_link_destinations(message: str) -> str:
+    """Wrap markdown link URLs in angle brackets so the parser handles special chars safely.
+
+    Markdown link syntax [text](url) breaks when the URL contains unescaped
+    parentheses, spaces, or other special characters. Wrapping the URL in angle
+    brackets — [text](<url>) — tells the parser to treat everything inside as
+    a literal URL. This applies to all links, not just citations.
+    """
+    if "](" not in message:
+        return message
+
+    normalized_parts: list[str] = []
+    cursor = 0
+
+    while match := _MARKDOWN_LINK_PATTERN.search(message, cursor):
+        normalized_parts.append(message[cursor : match.end()])
+        destination_start = match.end()
+        destination, end_idx = _extract_link_destination(message, destination_start)
+        if end_idx is None:
+            normalized_parts.append(message[destination_start:])
+            return "".join(normalized_parts)
+
+        already_wrapped = destination.startswith("<") and destination.endswith(">")
+        if destination and not already_wrapped:
+            destination = f"<{destination}>"
+
+        normalized_parts.append(destination)
+        normalized_parts.append(")")
+        cursor = end_idx + 1
+
+    normalized_parts.append(message[cursor:])
+    return "".join(normalized_parts)
+
+
+def _convert_slack_links_to_markdown(message: str) -> str:
+    """Convert Slack-style <url|text> links to standard markdown [text](url).
+
+    LLMs sometimes emit Slack mrkdwn link syntax directly. Mistune doesn't
+    recognise it, so the angle brackets would be escaped by text() and Slack
+    would render the link as literal text instead of a clickable link.
+    """
+    return _transform_outside_code_blocks(
+        message, lambda text: _SLACK_LINK_PATTERN.sub(r"[\2](\1)", text)
+    )
+
 
 def format_slack_message(message: str | None) -> str:
     if message is None:
         return ""
+    message = _transform_outside_code_blocks(message, _sanitize_html)
+    message = _convert_slack_links_to_markdown(message)
+    normalized_message = _normalize_link_destinations(message)
     md = create_markdown(renderer=SlackRenderer(), plugins=["strikethrough"])
-    result = md(message)
+    result = md(normalized_message)
     # With HTMLRenderer, result is always str (not AST list)
     assert isinstance(result, str)
-    return result
+    return result.rstrip("\n")
 
 
 class SlackRenderer(HTMLRenderer):
+    """Renders markdown as Slack mrkdwn format instead of HTML.
+
+    Overrides all HTMLRenderer methods that produce HTML tags to ensure
+    no raw HTML ever appears in Slack messages.
+    """
+
     SPECIALS: dict[str, str] = {"&": "&amp;", "<": "&lt;", ">": "&gt;"}
 
     def escape_special(self, text: str) -> str:
@@ -23,7 +152,7 @@ class SlackRenderer(HTMLRenderer):
         return text
 
     def heading(self, text: str, level: int, **attrs: Any) -> str:  # noqa: ARG002
-        return f"*{text}*\n"
+        return f"*{text}*\n\n"
 
     def emphasis(self, text: str) -> str:
         return f"_{text}_"
@@ -42,7 +171,7 @@ class SlackRenderer(HTMLRenderer):
                 count += 1
                 prefix = f"{count}. " if ordered else "• "
                 lines[i] = f"{prefix}{line[4:]}"
-        return "\n".join(lines)
+        return "\n".join(lines) + "\n"
 
     def list_item(self, text: str) -> str:
         return f"li: {text}\n"
@@ -64,7 +193,30 @@ class SlackRenderer(HTMLRenderer):
         return f"`{text}`"
 
     def block_code(self, code: str, info: str | None = None) -> str:  # noqa: ARG002
-        return f"```\n{code}\n```\n"
+        return f"```\n{code.rstrip(chr(10))}\n```\n\n"
+
+    def linebreak(self) -> str:
+        return "\n"
+
+    def thematic_break(self) -> str:
+        return "---\n\n"
+
+    def block_quote(self, text: str) -> str:
+        lines = text.strip().split("\n")
+        quoted = "\n".join(f">{line}" for line in lines)
+        return quoted + "\n\n"
+
+    def block_html(self, html: str) -> str:
+        return _sanitize_html(html) + "\n\n"
+
+    def block_error(self, text: str) -> str:
+        return f"```\n{text}\n```\n\n"
+
+    def text(self, text: str) -> str:
+        # Only escape the three entities Slack recognizes: & < >
+        # HTMLRenderer.text() also escapes " to &quot; which Slack renders
+        # as literal &quot; text since Slack doesn't recognize that entity.
+        return self.escape_special(text)
 
     def paragraph(self, text: str) -> str:
-        return f"{text}\n"
+        return f"{text}\n\n"

backend/onyx/prompts/tool_prompts.py

@@ -1,6 +1,6 @@
 # ruff: noqa: E501, W605 start
 # If there are any tools, this section is included, the sections below are for the available tools
-TOOL_SECTION_HEADER = "\n\n# Tools\n"
+TOOL_SECTION_HEADER = "\n# Tools\n\n"
 
 
 # This section is included if there are search type tools, currently internal_search and web_search
@@ -16,11 +16,10 @@ When searching for information, if the initial results cannot fully answer the u
 Do not repeat the same or very similar queries if it already has been run in the chat history.
 
 If it is unclear which tool to use, consider using multiple in parallel to be efficient with time.
-"""
+""".lstrip()
 
 
 INTERNAL_SEARCH_GUIDANCE = """
-
 ## internal_search
 Use the `internal_search` tool to search connected applications for information. Some examples of when to use `internal_search` include:
 - Internal information: any time where there may be some information stored in internal applications that could help better answer the query.
@@ -28,34 +27,31 @@ Use the `internal_search` tool to search connected applications for information.
 - Keyword Queries: queries that are heavily keyword based are often internal document search queries.
 - Ambiguity: questions about something that is not widely known or understood.
 Never provide more than 3 queries at once to `internal_search`.
-"""
+""".lstrip()
 
 
 WEB_SEARCH_GUIDANCE = """
-
 ## web_search
 Use the `web_search` tool to access up-to-date information from the web. Some examples of when to use `web_search` include:
 - Freshness: when the answer might be enhanced by up-to-date information on a topic. Very important for topics that are changing or evolving.
 - Accuracy: if the cost of outdated/inaccurate information is high.
 - Niche Information: when detailed info is not widely known or understood (but is likely found on the internet).{site_colon_disabled}
-"""
+""".lstrip()
 
 WEB_SEARCH_SITE_DISABLED_GUIDANCE = """
 Do not use the "site:" operator in your web search queries.
-""".rstrip()
+""".lstrip()
 
 
 OPEN_URLS_GUIDANCE = """
-
 ## open_url
 Use the `open_url` tool to read the content of one or more URLs. Use this tool to access the contents of the most promising web pages from your web searches or user specified URLs. \
 You can open many URLs at once by passing multiple URLs in the array if multiple pages seem promising. Prioritize the most promising pages and reputable sources. \
 Do not open URLs that are image files like .png, .jpg, etc.
 You should almost always use open_url after a web_search call. Use this tool when a user asks about a specific provided URL.
-"""
+""".lstrip()
 
 PYTHON_TOOL_GUIDANCE = """
-
 ## python
 Use the `python` tool to execute Python code in an isolated sandbox. The tool will respond with the output of the execution or time out after 60.0 seconds.
 Any files uploaded to the chat will be automatically be available in the execution environment's current directory. \
@@ -64,21 +60,21 @@ Use this to give the user a way to download the file OR to display generated ima
 Internet access for this session is disabled. Do not make external web requests or API calls as they will fail.
 Use `openpyxl` to read and write Excel files. You have access to libraries like numpy, pandas, scipy, matplotlib, and PIL.
 IMPORTANT: each call to this tool is independent. Variables from previous calls will NOT be available in the current call.
-"""
+""".lstrip()
 
 GENERATE_IMAGE_GUIDANCE = """
-
 ## generate_image
 NEVER use generate_image unless the user specifically requests an image.
-"""
+For edits/variations of a previously generated image, pass `reference_image_file_ids` with
+the `file_id` values returned by earlier `generate_image` tool results.
+""".lstrip()
 
 MEMORY_GUIDANCE = """
-
 ## add_memory
 Use the `add_memory` tool for facts shared by the user that should be remembered for future conversations. \
 Only add memories that are specific, likely to remain true, and likely to be useful later. \
 Focus on enduring preferences, long-term goals, stable constraints, and explicit "remember this" type requests.
-"""
+""".lstrip()
 
 TOOL_CALL_FAILURE_PROMPT = """
 LLM attempted to call a tool but failed. Most likely the tool name or arguments were misspelled.

backend/onyx/prompts/user_info.py

@@ -1,40 +1,36 @@
 # ruff: noqa: E501, W605 start
-USER_INFORMATION_HEADER = "\n\n# User Information\n"
+USER_INFORMATION_HEADER = "\n# User Information\n\n"
 
 BASIC_INFORMATION_PROMPT = """
-
 ## Basic Information
 User name: {user_name}
 User email: {user_email}{user_role}
-"""
+""".lstrip()
 
 # This line only shows up if the user has configured their role.
 USER_ROLE_PROMPT = """
 User role: {user_role}
-"""
+""".lstrip()
 
 # Team information should be a paragraph style description of the user's team.
 TEAM_INFORMATION_PROMPT = """
-
 ## Team Information
 {team_information}
-"""
+""".lstrip()
 
 # User preferences should be a paragraph style description of the user's preferences.
 USER_PREFERENCES_PROMPT = """
-
 ## User Preferences
 {user_preferences}
-"""
+""".lstrip()
 
 # User memories should look something like:
 # - Memory 1
 # - Memory 2
 # - Memory 3
 USER_MEMORIES_PROMPT = """
-
 ## User Memories
 {user_memories}
-"""
+""".lstrip()
 
 # ruff: noqa: E501, W605 end

backend/onyx/redis/redis_pool.py

@@ -109,6 +109,7 @@ class TenantRedis(redis.Redis):
             "unlock",
             "get",
             "set",
+            "setex",
             "delete",
             "exists",
             "incrby",

backend/onyx/server/documents/connector.py

@@ -103,6 +103,7 @@ from onyx.db.engine.sql_engine import get_session
 from onyx.db.enums import AccessType
 from onyx.db.enums import ConnectorCredentialPairStatus
 from onyx.db.enums import IndexingMode
+from onyx.db.enums import ProcessingMode
 from onyx.db.federated import fetch_all_federated_connectors_parallel
 from onyx.db.index_attempt import get_index_attempts_for_cc_pair
 from onyx.db.index_attempt import get_latest_index_attempts_by_status
@@ -987,6 +988,7 @@ def get_connector_status(
         user=user,
         eager_load_connector=True,
         eager_load_credential=True,
+        eager_load_user=True,
         get_editable=False,
     )
 
@@ -1000,11 +1002,23 @@ def get_connector_status(
             relationship.user_group_id
         )
 
+    # Pre-compute credential_ids per connector to avoid N+1 lazy loads
+    connector_to_credential_ids: dict[int, list[int]] = {}
+    for cc_pair in cc_pairs:
+        connector_to_credential_ids.setdefault(cc_pair.connector_id, []).append(
+            cc_pair.credential_id
+        )
+
     return [
         ConnectorStatus(
             cc_pair_id=cc_pair.id,
             name=cc_pair.name,
-            connector=ConnectorSnapshot.from_connector_db_model(cc_pair.connector),
+            connector=ConnectorSnapshot.from_connector_db_model(
+                cc_pair.connector,
+                credential_ids=connector_to_credential_ids.get(
+                    cc_pair.connector_id, []
+                ),
+            ),
             credential=CredentialSnapshot.from_credential_db_model(cc_pair.credential),
             access_type=cc_pair.access_type,
             groups=group_cc_pair_relationships_dict.get(cc_pair.id, []),
@@ -1059,15 +1073,27 @@ def get_connector_indexing_status(
     parallel_functions: list[tuple[CallableProtocol, tuple[Any, ...]]] = [
         # Get editable connector/credential pairs
         (
-            get_connector_credential_pairs_for_user_parallel,
-            (user, True, None, True, True, True, True, request.source),
+            lambda: get_connector_credential_pairs_for_user_parallel(
+                user, True, None, True, True, False, True, request.source
+            ),
+            (),
         ),
         # Get federated connectors
         (fetch_all_federated_connectors_parallel, ()),
         # Get most recent index attempts
-        (get_latest_index_attempts_parallel, (request.secondary_index, True, False)),
+        (
+            lambda: get_latest_index_attempts_parallel(
+                request.secondary_index, True, False
+            ),
+            (),
+        ),
         # Get most recent finished index attempts
-        (get_latest_index_attempts_parallel, (request.secondary_index, True, True)),
+        (
+            lambda: get_latest_index_attempts_parallel(
+                request.secondary_index, True, True
+            ),
+            (),
+        ),
     ]
 
     if user and user.role == UserRole.ADMIN:
@@ -1084,8 +1110,10 @@ def get_connector_indexing_status(
         parallel_functions.append(
             # Get non-editable connector/credential pairs
             (
-                get_connector_credential_pairs_for_user_parallel,
-                (user, False, None, True, True, True, True, request.source),
+                lambda: get_connector_credential_pairs_for_user_parallel(
+                    user, False, None, True, True, False, True, request.source
+                ),
+                (),
             ),
         )
 
@@ -1911,6 +1939,7 @@ Tenant ID: {tenant_id}
 class BasicCCPairInfo(BaseModel):
     has_successful_run: bool
     source: DocumentSource
+    status: ConnectorCredentialPairStatus
 
 
 @router.get("/connector-status", tags=PUBLIC_API_TAGS)
@@ -1924,13 +1953,17 @@ def get_basic_connector_indexing_status(
         get_editable=False,
         user=user,
     )
+
+    # NOTE: This endpoint excludes Craft connectors
     return [
         BasicCCPairInfo(
             has_successful_run=cc_pair.last_successful_index_time is not None,
             source=cc_pair.connector.source,
+            status=cc_pair.status,
         )
         for cc_pair in cc_pairs
         if cc_pair.connector.source != DocumentSource.INGESTION_API
+        and cc_pair.processing_mode == ProcessingMode.REGULAR
     ]
 
 

backend/onyx/server/documents/models.py

@@ -365,7 +365,8 @@ class CCPairFullInfo(BaseModel):
             in_repeated_error_state=cc_pair_model.in_repeated_error_state,
             num_docs_indexed=num_docs_indexed,
             connector=ConnectorSnapshot.from_connector_db_model(
-                cc_pair_model.connector
+                cc_pair_model.connector,
+                credential_ids=[cc_pair_model.credential_id],
             ),
             credential=CredentialSnapshot.from_credential_db_model(
                 cc_pair_model.credential

backend/onyx/server/features/build/api/sessions_api.py

@@ -762,6 +762,43 @@ def download_webapp(
     )
 
 
+@router.get("/{session_id}/download-directory/{path:path}")
+def download_directory(
+    session_id: UUID,
+    path: str,
+    user: User = Depends(current_user),
+    db_session: Session = Depends(get_session),
+) -> Response:
+    """
+    Download a directory as a zip file.
+
+    Returns the specified directory as a zip archive.
+    """
+    user_id: UUID = user.id
+    session_manager = SessionManager(db_session)
+
+    try:
+        result = session_manager.download_directory(session_id, user_id, path)
+    except ValueError as e:
+        error_message = str(e)
+        if "path traversal" in error_message.lower():
+            raise HTTPException(status_code=403, detail="Access denied")
+        raise HTTPException(status_code=400, detail=error_message)
+
+    if result is None:
+        raise HTTPException(status_code=404, detail="Directory not found")
+
+    zip_bytes, filename = result
+
+    return Response(
+        content=zip_bytes,
+        media_type="application/zip",
+        headers={
+            "Content-Disposition": f'attachment; filename="{filename}"',
+        },
+    )
+
+
 @router.post("/{session_id}/upload", response_model=UploadResponse)
 def upload_file_endpoint(
     session_id: UUID,

backend/onyx/server/features/build/db/user_library.py

@@ -107,27 +107,23 @@ def get_or_create_craft_connector(db_session: Session, user: User) -> tuple[int,
     )
 
     for cc_pair in cc_pairs:
-        if cc_pair.connector.source == DocumentSource.CRAFT_FILE:
+        if (
+            cc_pair.connector.source == DocumentSource.CRAFT_FILE
+            and cc_pair.creator_id == user.id
+        ):
             return cc_pair.connector.id, cc_pair.credential.id
 
-    # Check for orphaned connector (created but cc_pair creation failed previously)
+    # No cc_pair for this user — find or create the shared CRAFT_FILE connector
     existing_connectors = fetch_connectors(
         db_session, sources=[DocumentSource.CRAFT_FILE]
     )
-    orphaned_connector = None
+    connector_id: int | None = None
     for conn in existing_connectors:
-        if conn.name != USER_LIBRARY_CONNECTOR_NAME:
-            continue
-        if not conn.credentials:
-            orphaned_connector = conn
+        if conn.name == USER_LIBRARY_CONNECTOR_NAME:
+            connector_id = conn.id
             break
 
-    if orphaned_connector:
-        connector_id = orphaned_connector.id
-        logger.info(
-            f"Found orphaned User Library connector {connector_id}, completing setup"
-        )
-    else:
+    if connector_id is None:
         connector_data = ConnectorBase(
             name=USER_LIBRARY_CONNECTOR_NAME,
             source=DocumentSource.CRAFT_FILE,