pr comments

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
Co-authored-by: Jamison Lahman <jamison@lahman.dev>

.github/workflows/deployment.yml

@@ -704,6 +704,9 @@ jobs:
             NEXT_PUBLIC_FORGOT_PASSWORD_ENABLED=true
             NEXT_PUBLIC_INCLUDE_ERROR_POPUP_SUPPORT_LINK=true
             NODE_OPTIONS=--max-old-space-size=8192
+            SENTRY_RELEASE=${{ github.sha }}
+          secrets: |
+            sentry_auth_token=${{ secrets.SENTRY_AUTH_TOKEN }}
           cache-from: |
             type=registry,ref=${{ env.RUNS_ON_ECR_CACHE }}:cloudweb-cache-amd64
             type=registry,ref=${{ env.REGISTRY_IMAGE }}:latest
@@ -786,6 +789,9 @@ jobs:
             NEXT_PUBLIC_FORGOT_PASSWORD_ENABLED=true
             NEXT_PUBLIC_INCLUDE_ERROR_POPUP_SUPPORT_LINK=true
             NODE_OPTIONS=--max-old-space-size=8192
+            SENTRY_RELEASE=${{ github.sha }}
+          secrets: |
+            sentry_auth_token=${{ secrets.SENTRY_AUTH_TOKEN }}
           cache-from: |
             type=registry,ref=${{ env.RUNS_ON_ECR_CACHE }}:cloudweb-cache-arm64
             type=registry,ref=${{ env.REGISTRY_IMAGE }}:latest
@@ -1503,232 +1509,105 @@ jobs:
             $(printf '%s\n' "${META_TAGS}" | xargs -I {} echo -t {}) \
             $IMAGES
 
-  trivy-scan-web:
+  trivy-scan:
     needs:
       - determine-builds
       - merge-web
-    if: needs.merge-web.result == 'success'
-    runs-on:
-      - runs-on
-      - runner=2cpu-linux-arm64
-      - run-id=${{ github.run_id }}-trivy-scan-web
-      - extras=ecr-cache
-    timeout-minutes: 90
-    environment: release
-    env:
-      REGISTRY_IMAGE: onyxdotapp/onyx-web-server
-    steps:
-      - uses: runs-on/action@cd2b598b0515d39d78c38a02d529db87d2196d1e # ratchet:runs-on/action@v2
-
-      - name: Configure AWS credentials
-        uses: aws-actions/configure-aws-credentials@8df5847569e6427dd6c4fb1cf565c83acfa8afa7
-        with:
-          role-to-assume: ${{ secrets.AWS_OIDC_ROLE_ARN }}
-          aws-region: us-east-2
-
-      - name: Get AWS Secrets
-        uses: aws-actions/aws-secretsmanager-get-secrets@a9a7eb4e2f2871d30dc5b892576fde60a2ecc802
-        with:
-          secret-ids: |
-            DOCKER_USERNAME, deploy/docker-username
-            DOCKER_TOKEN, deploy/docker-token
-          parse-json-secrets: true
-
-      - name: Run Trivy vulnerability scanner
-        uses: nick-fields/retry@ce71cc2ab81d554ebbe88c79ab5975992d79ba08 # ratchet:nick-fields/retry@v3
-        with:
-          timeout_minutes: 30
-          max_attempts: 3
-          retry_wait_seconds: 10
-          command: |
-            if [ "${{ needs.determine-builds.outputs.is-test-run }}" == "true" ]; then
-              SCAN_IMAGE="${{ env.RUNS_ON_ECR_CACHE }}:web-${{ needs.determine-builds.outputs.sanitized-tag }}"
-            else
-              SCAN_IMAGE="docker.io/${{ env.REGISTRY_IMAGE }}:${{ github.ref_name }}"
-            fi
-            docker run --rm -v $HOME/.cache/trivy:/root/.cache/trivy \
-              -e TRIVY_DB_REPOSITORY="public.ecr.aws/aquasecurity/trivy-db:2" \
-              -e TRIVY_JAVA_DB_REPOSITORY="public.ecr.aws/aquasecurity/trivy-java-db:1" \
-              -e TRIVY_USERNAME="${{ env.DOCKER_USERNAME }}" \
-              -e TRIVY_PASSWORD="${{ env.DOCKER_TOKEN }}" \
-              aquasec/trivy@sha256:a22415a38938a56c379387a8163fcb0ce38b10ace73e593475d3658d578b2436 \
-              image \
-              --skip-version-check \
-              --timeout 20m \
-              --severity CRITICAL,HIGH \
-              ${SCAN_IMAGE}
-
-  trivy-scan-web-cloud:
-    needs:
-      - determine-builds
       - merge-web-cloud
-    if: needs.merge-web-cloud.result == 'success'
-    runs-on:
-      - runs-on
-      - runner=2cpu-linux-arm64
-      - run-id=${{ github.run_id }}-trivy-scan-web-cloud
-      - extras=ecr-cache
-    timeout-minutes: 90
-    environment: release
-    env:
-      REGISTRY_IMAGE: onyxdotapp/onyx-web-server-cloud
-    steps:
-      - uses: runs-on/action@cd2b598b0515d39d78c38a02d529db87d2196d1e # ratchet:runs-on/action@v2
-
-      - name: Configure AWS credentials
-        uses: aws-actions/configure-aws-credentials@8df5847569e6427dd6c4fb1cf565c83acfa8afa7
-        with:
-          role-to-assume: ${{ secrets.AWS_OIDC_ROLE_ARN }}
-          aws-region: us-east-2
-
-      - name: Get AWS Secrets
-        uses: aws-actions/aws-secretsmanager-get-secrets@a9a7eb4e2f2871d30dc5b892576fde60a2ecc802
-        with:
-          secret-ids: |
-            DOCKER_USERNAME, deploy/docker-username
-            DOCKER_TOKEN, deploy/docker-token
-          parse-json-secrets: true
-
-      - name: Run Trivy vulnerability scanner
-        uses: nick-fields/retry@ce71cc2ab81d554ebbe88c79ab5975992d79ba08 # ratchet:nick-fields/retry@v3
-        with:
-          timeout_minutes: 30
-          max_attempts: 3
-          retry_wait_seconds: 10
-          command: |
-            if [ "${{ needs.determine-builds.outputs.is-test-run }}" == "true" ]; then
-              SCAN_IMAGE="${{ env.RUNS_ON_ECR_CACHE }}:web-cloud-${{ needs.determine-builds.outputs.sanitized-tag }}"
-            else
-              SCAN_IMAGE="docker.io/${{ env.REGISTRY_IMAGE }}:${{ github.ref_name }}"
-            fi
-            docker run --rm -v $HOME/.cache/trivy:/root/.cache/trivy \
-              -e TRIVY_DB_REPOSITORY="public.ecr.aws/aquasecurity/trivy-db:2" \
-              -e TRIVY_JAVA_DB_REPOSITORY="public.ecr.aws/aquasecurity/trivy-java-db:1" \
-              -e TRIVY_USERNAME="${{ env.DOCKER_USERNAME }}" \
-              -e TRIVY_PASSWORD="${{ env.DOCKER_TOKEN }}" \
-              aquasec/trivy@sha256:a22415a38938a56c379387a8163fcb0ce38b10ace73e593475d3658d578b2436 \
-              image \
-              --skip-version-check \
-              --timeout 20m \
-              --severity CRITICAL,HIGH \
-              ${SCAN_IMAGE}
-
-  trivy-scan-backend:
-    needs:
-      - determine-builds
       - merge-backend
-    if: needs.merge-backend.result == 'success'
+      - merge-model-server
+    if: >-
+      always() && !cancelled() &&
+      (needs.merge-web.result == 'success' ||
+       needs.merge-web-cloud.result == 'success' ||
+       needs.merge-backend.result == 'success' ||
+       needs.merge-model-server.result == 'success')
     runs-on:
       - runs-on
       - runner=2cpu-linux-arm64
-      - run-id=${{ github.run_id }}-trivy-scan-backend
+      - run-id=${{ github.run_id }}-trivy-scan-${{ matrix.component }}
       - extras=ecr-cache
-    timeout-minutes: 90
-    environment: release
-    env:
-      REGISTRY_IMAGE: ${{ contains(github.ref_name, 'cloud') && 'onyxdotapp/onyx-backend-cloud' || 'onyxdotapp/onyx-backend' }}
+    permissions:
+      security-events: write # needed for SARIF uploads
+    timeout-minutes: 10
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - component: web
+            registry-image: onyxdotapp/onyx-web-server
+          - component: web-cloud
+            registry-image: onyxdotapp/onyx-web-server-cloud
+          - component: backend
+            registry-image: ${{ contains(github.ref_name, 'cloud') && 'onyxdotapp/onyx-backend-cloud' || 'onyxdotapp/onyx-backend' }}
+            trivyignore: backend/.trivyignore
+          - component: model-server
+            registry-image: ${{ contains(github.ref_name, 'cloud') && 'onyxdotapp/onyx-model-server-cloud' || 'onyxdotapp/onyx-model-server' }}
     steps:
+      - name: Check if this scan should run
+        id: should-run
+        run: |
+          case "$COMPONENT" in
+            web) RESULT="$MERGE_WEB" ;;
+            web-cloud) RESULT="$MERGE_WEB_CLOUD" ;;
+            backend) RESULT="$MERGE_BACKEND" ;;
+            model-server) RESULT="$MERGE_MODEL_SERVER" ;;
+          esac
+          if [ "$RESULT" == "success" ]; then
+            echo "run=true" >> "$GITHUB_OUTPUT"
+          else
+            echo "run=false" >> "$GITHUB_OUTPUT"
+          fi
+        env:
+          COMPONENT: ${{ matrix.component }}
+          MERGE_WEB: ${{ needs.merge-web.result }}
+          MERGE_WEB_CLOUD: ${{ needs.merge-web-cloud.result }}
+          MERGE_BACKEND: ${{ needs.merge-backend.result }}
+          MERGE_MODEL_SERVER: ${{ needs.merge-model-server.result }}
+
       - uses: runs-on/action@cd2b598b0515d39d78c38a02d529db87d2196d1e # ratchet:runs-on/action@v2
+        if: steps.should-run.outputs.run == 'true'
 
       - name: Checkout
+        if: steps.should-run.outputs.run == 'true' && matrix.trivyignore != ''
         uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # ratchet:actions/checkout@v6
         with:
           persist-credentials: false
 
-      - name: Configure AWS credentials
-        uses: aws-actions/configure-aws-credentials@8df5847569e6427dd6c4fb1cf565c83acfa8afa7
-        with:
-          role-to-assume: ${{ secrets.AWS_OIDC_ROLE_ARN }}
-          aws-region: us-east-2
-
-      - name: Get AWS Secrets
-        uses: aws-actions/aws-secretsmanager-get-secrets@a9a7eb4e2f2871d30dc5b892576fde60a2ecc802
-        with:
-          secret-ids: |
-            DOCKER_USERNAME, deploy/docker-username
-            DOCKER_TOKEN, deploy/docker-token
-          parse-json-secrets: true
+      - name: Determine scan image
+        if: steps.should-run.outputs.run == 'true'
+        id: scan-image
+        run: |
+          if [ "$IS_TEST_RUN" == "true" ]; then
+            echo "image=${RUNS_ON_ECR_CACHE}:${TAG_PREFIX}-${SANITIZED_TAG}" >> "$GITHUB_OUTPUT"
+          else
+            echo "image=docker.io/${REGISTRY_IMAGE}:${REF_NAME}" >> "$GITHUB_OUTPUT"
+          fi
+        env:
+          IS_TEST_RUN: ${{ needs.determine-builds.outputs.is-test-run }}
+          TAG_PREFIX: ${{ matrix.component }}
+          SANITIZED_TAG: ${{ needs.determine-builds.outputs.sanitized-tag }}
+          REGISTRY_IMAGE: ${{ matrix.registry-image }}
+          REF_NAME: ${{ github.ref_name }}
 
       - name: Run Trivy vulnerability scanner
-        uses: nick-fields/retry@ce71cc2ab81d554ebbe88c79ab5975992d79ba08 # ratchet:nick-fields/retry@v3
+        if: steps.should-run.outputs.run == 'true'
+        uses: aquasecurity/trivy-action@57a97c7e7821a5776cebc9bb87c984fa69cba8f1 # ratchet:aquasecurity/trivy-action@v0.35.0
         with:
-          timeout_minutes: 30
-          max_attempts: 3
-          retry_wait_seconds: 10
-          command: |
-            if [ "${{ needs.determine-builds.outputs.is-test-run }}" == "true" ]; then
-              SCAN_IMAGE="${{ env.RUNS_ON_ECR_CACHE }}:backend-${{ needs.determine-builds.outputs.sanitized-tag }}"
-            else
-              SCAN_IMAGE="docker.io/${{ env.REGISTRY_IMAGE }}:${{ github.ref_name }}"
-            fi
-            docker run --rm -v $HOME/.cache/trivy:/root/.cache/trivy \
-              -v ${{ github.workspace }}/backend/.trivyignore:/tmp/.trivyignore:ro \
-              -e TRIVY_DB_REPOSITORY="public.ecr.aws/aquasecurity/trivy-db:2" \
-              -e TRIVY_JAVA_DB_REPOSITORY="public.ecr.aws/aquasecurity/trivy-java-db:1" \
-              -e TRIVY_USERNAME="${{ env.DOCKER_USERNAME }}" \
-              -e TRIVY_PASSWORD="${{ env.DOCKER_TOKEN }}" \
-              aquasec/trivy@sha256:a22415a38938a56c379387a8163fcb0ce38b10ace73e593475d3658d578b2436 \
-              image \
-              --skip-version-check \
-              --timeout 20m \
-              --severity CRITICAL,HIGH \
-              --ignorefile /tmp/.trivyignore \
-              ${SCAN_IMAGE}
+          image-ref: ${{ steps.scan-image.outputs.image }}
+          severity: CRITICAL,HIGH
+          format: "sarif"
+          output: "trivy-results.sarif"
+          trivyignores: ${{ matrix.trivyignore }}
+        env:
+          TRIVY_USERNAME: ${{ secrets.DOCKER_USERNAME }}
+          TRIVY_PASSWORD: ${{ secrets.DOCKER_TOKEN }}
 
-  trivy-scan-model-server:
-    needs:
-      - determine-builds
-      - merge-model-server
-    if: needs.merge-model-server.result == 'success'
-    runs-on:
-      - runs-on
-      - runner=2cpu-linux-arm64
-      - run-id=${{ github.run_id }}-trivy-scan-model-server
-      - extras=ecr-cache
-    timeout-minutes: 90
-    environment: release
-    env:
-      REGISTRY_IMAGE: ${{ contains(github.ref_name, 'cloud') && 'onyxdotapp/onyx-model-server-cloud' || 'onyxdotapp/onyx-model-server' }}
-    steps:
-      - uses: runs-on/action@cd2b598b0515d39d78c38a02d529db87d2196d1e # ratchet:runs-on/action@v2
-
-      - name: Configure AWS credentials
-        uses: aws-actions/configure-aws-credentials@8df5847569e6427dd6c4fb1cf565c83acfa8afa7
+      - name: Upload Trivy scan results to GitHub Security tab
+        if: steps.should-run.outputs.run == 'true'
+        uses: github/codeql-action/upload-sarif@ba454b8ab46733eb6145342877cd148270bb77ab
         with:
-          role-to-assume: ${{ secrets.AWS_OIDC_ROLE_ARN }}
-          aws-region: us-east-2
-
-      - name: Get AWS Secrets
-        uses: aws-actions/aws-secretsmanager-get-secrets@a9a7eb4e2f2871d30dc5b892576fde60a2ecc802
-        with:
-          secret-ids: |
-            DOCKER_USERNAME, deploy/docker-username
-            DOCKER_TOKEN, deploy/docker-token
-          parse-json-secrets: true
-
-      - name: Run Trivy vulnerability scanner
-        uses: nick-fields/retry@ce71cc2ab81d554ebbe88c79ab5975992d79ba08 # ratchet:nick-fields/retry@v3
-        with:
-          timeout_minutes: 30
-          max_attempts: 3
-          retry_wait_seconds: 10
-          command: |
-            if [ "${{ needs.determine-builds.outputs.is-test-run }}" == "true" ]; then
-              SCAN_IMAGE="${{ env.RUNS_ON_ECR_CACHE }}:model-server-${{ needs.determine-builds.outputs.sanitized-tag }}"
-            else
-              SCAN_IMAGE="docker.io/${{ env.REGISTRY_IMAGE }}:${{ github.ref_name }}"
-            fi
-            docker run --rm -v $HOME/.cache/trivy:/root/.cache/trivy \
-              -e TRIVY_DB_REPOSITORY="public.ecr.aws/aquasecurity/trivy-db:2" \
-              -e TRIVY_JAVA_DB_REPOSITORY="public.ecr.aws/aquasecurity/trivy-java-db:1" \
-              -e TRIVY_USERNAME="${{ env.DOCKER_USERNAME }}" \
-              -e TRIVY_PASSWORD="${{ env.DOCKER_TOKEN }}" \
-              aquasec/trivy@sha256:a22415a38938a56c379387a8163fcb0ce38b10ace73e593475d3658d578b2436 \
-              image \
-              --skip-version-check \
-              --timeout 20m \
-              --severity CRITICAL,HIGH \
-              ${SCAN_IMAGE}
+          sarif_file: "trivy-results.sarif"
 
   notify-slack-on-failure:
     needs:

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

@@ -47,7 +47,8 @@ jobs:
           done
 
       - name: Publish Helm charts to gh-pages
-        uses: stefanprodan/helm-gh-pages@0ad2bb377311d61ac04ad9eb6f252fb68e207260 # ratchet:stefanprodan/helm-gh-pages@v1.7.0
+        # NOTE: HEAD of https://github.com/stefanprodan/helm-gh-pages/pull/43
+        uses: stefanprodan/helm-gh-pages@ad32ad3b8720abfeaac83532fd1e9bdfca5bbe27 # zizmor: ignore[impostor-commit]
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
           charts_dir: deployment/helm/charts

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

@@ -35,6 +35,7 @@ jobs:
     needs: [provider-chat-test]
     if: failure() && github.event_name == 'schedule'
     runs-on: ubuntu-slim
+    environment: ci-protected
     timeout-minutes: 5
     steps:
       - name: Checkout

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

@@ -183,6 +183,7 @@ jobs:
       - cherry-pick-to-latest-release
     if: needs.resolve-cherry-pick-request.outputs.should_cherrypick == 'true' && needs.resolve-cherry-pick-request.result == 'success' && needs.cherry-pick-to-latest-release.result == 'success'
     runs-on: ubuntu-slim
+    environment: ci-protected
     timeout-minutes: 10
     steps:
       - name: Checkout
@@ -232,6 +233,7 @@ jobs:
       - cherry-pick-to-latest-release
     if: always() && needs.resolve-cherry-pick-request.outputs.should_cherrypick == 'true' && (needs.resolve-cherry-pick-request.result == 'failure' || needs.cherry-pick-to-latest-release.result == 'failure')
     runs-on: ubuntu-slim
+    environment: ci-protected
     timeout-minutes: 10
     steps:
       - name: Checkout

.github/workflows/pr-desktop-build.yml

@@ -63,7 +63,7 @@ jobs:
           targets: ${{ matrix.target }}
 
       - name: Cache Cargo registry and build
-        uses: actions/cache@cdf6c1fa76f9f475f3d7449005a359c84ca0f306 # zizmor: ignore[cache-poisoning]
+        uses: actions/cache@668228422ae6a00e4ad889ee87cd7109ec5666a7 # zizmor: ignore[cache-poisoning]
         with:
           path: |
             ~/.cargo/bin/

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

@@ -41,7 +41,7 @@ jobs:
           version: v3.19.0
 
       - name: Set up chart-testing
-        uses: helm/chart-testing-action@b5eebdd9998021f29756c53432f48dab66394810
+        uses: helm/chart-testing-action@2e2940618cb426dce2999631d543b53cdcfc8527
         with:
           uv_version: "0.9.9"
 

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

@@ -284,7 +284,7 @@ jobs:
 
       - name: Cache playwright cache
         # zizmor: ignore[cache-poisoning] ephemeral runners; no release artifacts
-        uses: runs-on/cache@50350ad4242587b6c8c2baa2e740b1bc11285ff4 # ratchet:runs-on/cache@v4
+        uses: runs-on/cache@a5f51d6f3fece787d03b7b4e981c82538a0654ed # ratchet:runs-on/cache@v4
         with:
           path: ~/.cache/ms-playwright
           key: ${{ runner.os }}-playwright-npm-${{ hashFiles('web/package-lock.json') }}
@@ -626,7 +626,7 @@ jobs:
 
       - name: Cache playwright cache
         # zizmor: ignore[cache-poisoning] ephemeral runners; no release artifacts
-        uses: runs-on/cache@50350ad4242587b6c8c2baa2e740b1bc11285ff4 # ratchet:runs-on/cache@v4
+        uses: runs-on/cache@a5f51d6f3fece787d03b7b4e981c82538a0654ed # ratchet:runs-on/cache@v4
         with:
           path: ~/.cache/ms-playwright
           key: ${{ runner.os }}-playwright-npm-${{ hashFiles('web/package-lock.json') }}

.github/workflows/pr-python-checks.yml

@@ -56,7 +56,7 @@ jobs:
 
       - name: Cache mypy cache
         if: ${{ vars.DISABLE_MYPY_CACHE != 'true' }}
-        uses: runs-on/cache@50350ad4242587b6c8c2baa2e740b1bc11285ff4 # ratchet:runs-on/cache@v4
+        uses: runs-on/cache@a5f51d6f3fece787d03b7b4e981c82538a0654ed # ratchet:runs-on/cache@v4
         with:
           path: .mypy_cache
           key: mypy-${{ runner.os }}-${{ github.base_ref || github.event.merge_group.base_ref || 'main' }}-${{ hashFiles('**/*.py', '**/*.pyi', 'pyproject.toml') }}

.github/workflows/pr-python-connector-tests.yml

@@ -22,132 +22,40 @@ on:
     - cron: "0 16 * * *"
 
 permissions:
+  id-token: write # Required for OIDC-based AWS credential exchange
   contents: read
 
 env:
-  # AWS
-  AWS_ACCESS_KEY_ID_DAILY_CONNECTOR_TESTS: ${{ secrets.AWS_ACCESS_KEY_ID_DAILY_CONNECTOR_TESTS }}
-  AWS_SECRET_ACCESS_KEY_DAILY_CONNECTOR_TESTS: ${{ secrets.AWS_SECRET_ACCESS_KEY_DAILY_CONNECTOR_TESTS }}
-
-  # Cloudflare R2
+  PYTHONPATH: ./backend
+  DISABLE_TELEMETRY: "true"
   R2_ACCOUNT_ID_DAILY_CONNECTOR_TESTS: ${{ vars.R2_ACCOUNT_ID_DAILY_CONNECTOR_TESTS }}
-  R2_ACCESS_KEY_ID_DAILY_CONNECTOR_TESTS: ${{ secrets.R2_ACCESS_KEY_ID_DAILY_CONNECTOR_TESTS }}
-  R2_SECRET_ACCESS_KEY_DAILY_CONNECTOR_TESTS: ${{ secrets.R2_SECRET_ACCESS_KEY_DAILY_CONNECTOR_TESTS }}
-
-  # Google Cloud Storage
-  GCS_ACCESS_KEY_ID_DAILY_CONNECTOR_TESTS: ${{ secrets.GCS_ACCESS_KEY_ID_DAILY_CONNECTOR_TESTS }}
-  GCS_SECRET_ACCESS_KEY_DAILY_CONNECTOR_TESTS: ${{ secrets.GCS_SECRET_ACCESS_KEY_DAILY_CONNECTOR_TESTS }}
-
-  # Confluence
   CONFLUENCE_TEST_SPACE_URL: ${{ vars.CONFLUENCE_TEST_SPACE_URL }}
   CONFLUENCE_TEST_SPACE: ${{ vars.CONFLUENCE_TEST_SPACE }}
-  CONFLUENCE_TEST_PAGE_ID: ${{ secrets.CONFLUENCE_TEST_PAGE_ID }}
   CONFLUENCE_USER_NAME: ${{ vars.CONFLUENCE_USER_NAME }}
-  CONFLUENCE_ACCESS_TOKEN: ${{ secrets.CONFLUENCE_ACCESS_TOKEN }}
-  CONFLUENCE_ACCESS_TOKEN_SCOPED: ${{ secrets.CONFLUENCE_ACCESS_TOKEN_SCOPED }}
-
-  # Jira
-  JIRA_BASE_URL: ${{ secrets.JIRA_BASE_URL }}
-  JIRA_USER_EMAIL: ${{ secrets.JIRA_USER_EMAIL }}
-  JIRA_API_TOKEN: ${{ secrets.JIRA_API_TOKEN }}
-  JIRA_API_TOKEN_SCOPED: ${{ secrets.JIRA_API_TOKEN_SCOPED }}
-
-  # Gong
-  GONG_ACCESS_KEY: ${{ secrets.GONG_ACCESS_KEY }}
-  GONG_ACCESS_KEY_SECRET: ${{ secrets.GONG_ACCESS_KEY_SECRET }}
-
-  # Google
-  GOOGLE_DRIVE_SERVICE_ACCOUNT_JSON_STR: ${{ secrets.GOOGLE_DRIVE_SERVICE_ACCOUNT_JSON_STR }}
-  GOOGLE_DRIVE_OAUTH_CREDENTIALS_JSON_STR_TEST_USER_1: ${{ secrets.GOOGLE_DRIVE_OAUTH_CREDENTIALS_JSON_STR_TEST_USER_1 }}
-  GOOGLE_DRIVE_OAUTH_CREDENTIALS_JSON_STR: ${{ secrets.GOOGLE_DRIVE_OAUTH_CREDENTIALS_JSON_STR }}
-  GOOGLE_GMAIL_SERVICE_ACCOUNT_JSON_STR: ${{ secrets.GOOGLE_GMAIL_SERVICE_ACCOUNT_JSON_STR }}
-  GOOGLE_GMAIL_OAUTH_CREDENTIALS_JSON_STR: ${{ secrets.GOOGLE_GMAIL_OAUTH_CREDENTIALS_JSON_STR }}
-
-  # Slab
-  SLAB_BOT_TOKEN: ${{ secrets.SLAB_BOT_TOKEN }}
-
-  # Zendesk
-  ZENDESK_SUBDOMAIN: ${{ secrets.ZENDESK_SUBDOMAIN }}
-  ZENDESK_EMAIL: ${{ secrets.ZENDESK_EMAIL }}
-  ZENDESK_TOKEN: ${{ secrets.ZENDESK_TOKEN }}
-
-  # Salesforce
   SF_USERNAME: ${{ vars.SF_USERNAME }}
-  SF_PASSWORD: ${{ secrets.SF_PASSWORD }}
-  SF_SECURITY_TOKEN: ${{ secrets.SF_SECURITY_TOKEN }}
-
-  # Hubspot
-  HUBSPOT_ACCESS_TOKEN: ${{ secrets.HUBSPOT_ACCESS_TOKEN }}
-
-  # IMAP
   IMAP_HOST: ${{ vars.IMAP_HOST }}
   IMAP_USERNAME: ${{ vars.IMAP_USERNAME }}
-  IMAP_PASSWORD: ${{ secrets.IMAP_PASSWORD }}
   IMAP_MAILBOXES: ${{ vars.IMAP_MAILBOXES }}
-
-  # Airtable
   AIRTABLE_TEST_BASE_ID: ${{ vars.AIRTABLE_TEST_BASE_ID }}
   AIRTABLE_TEST_TABLE_ID: ${{ vars.AIRTABLE_TEST_TABLE_ID }}
   AIRTABLE_TEST_TABLE_NAME: ${{ vars.AIRTABLE_TEST_TABLE_NAME }}
-  AIRTABLE_ACCESS_TOKEN: ${{ secrets.AIRTABLE_ACCESS_TOKEN }}
-
-  # Sharepoint
   SHAREPOINT_CLIENT_ID: ${{ vars.SHAREPOINT_CLIENT_ID }}
-  SHAREPOINT_CLIENT_SECRET: ${{ secrets.SHAREPOINT_CLIENT_SECRET }}
   SHAREPOINT_CLIENT_DIRECTORY_ID: ${{ vars.SHAREPOINT_CLIENT_DIRECTORY_ID }}
   SHAREPOINT_SITE: ${{ vars.SHAREPOINT_SITE }}
-  PERM_SYNC_SHAREPOINT_CLIENT_ID: ${{ secrets.PERM_SYNC_SHAREPOINT_CLIENT_ID }}
-  PERM_SYNC_SHAREPOINT_PRIVATE_KEY: ${{ secrets.PERM_SYNC_SHAREPOINT_PRIVATE_KEY }}
-  PERM_SYNC_SHAREPOINT_CERTIFICATE_PASSWORD: ${{ secrets.PERM_SYNC_SHAREPOINT_CERTIFICATE_PASSWORD }}
-  PERM_SYNC_SHAREPOINT_DIRECTORY_ID: ${{ secrets.PERM_SYNC_SHAREPOINT_DIRECTORY_ID }}
-
-  # Github
-  ACCESS_TOKEN_GITHUB: ${{ secrets.ACCESS_TOKEN_GITHUB }}
-
-  # Gitlab
-  GITLAB_ACCESS_TOKEN: ${{ secrets.GITLAB_ACCESS_TOKEN }}
-
-  # Gitbook
-  GITBOOK_SPACE_ID: ${{ secrets.GITBOOK_SPACE_ID }}
-  GITBOOK_API_KEY: ${{ secrets.GITBOOK_API_KEY }}
-
-  # Notion
-  NOTION_INTEGRATION_TOKEN: ${{ secrets.NOTION_INTEGRATION_TOKEN }}
-
-  # Highspot
-  HIGHSPOT_KEY: ${{ secrets.HIGHSPOT_KEY }}
-  HIGHSPOT_SECRET: ${{ secrets.HIGHSPOT_SECRET }}
-
-  # Slack
-  SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}
-
-  # Discord
-  DISCORD_CONNECTOR_BOT_TOKEN: ${{ secrets.DISCORD_CONNECTOR_BOT_TOKEN }}
-
-  # Teams
-  TEAMS_APPLICATION_ID: ${{ secrets.TEAMS_APPLICATION_ID }}
-  TEAMS_DIRECTORY_ID: ${{ secrets.TEAMS_DIRECTORY_ID }}
-  TEAMS_SECRET: ${{ secrets.TEAMS_SECRET }}
-
-  # Bitbucket
-  BITBUCKET_WORKSPACE: ${{ secrets.BITBUCKET_WORKSPACE }}
-  BITBUCKET_REPOSITORIES: ${{ secrets.BITBUCKET_REPOSITORIES }}
-  BITBUCKET_PROJECTS: ${{ secrets.BITBUCKET_PROJECTS }}
   BITBUCKET_EMAIL: ${{ vars.BITBUCKET_EMAIL }}
-  BITBUCKET_API_TOKEN: ${{ secrets.BITBUCKET_API_TOKEN }}
-
-  # Fireflies
-  FIREFLIES_API_KEY: ${{ secrets.FIREFLIES_API_KEY }}
 
 jobs:
   connectors-check:
     # See https://runs-on.com/runners/linux/
-    runs-on: [runs-on, runner=8cpu-linux-x64, "run-id=${{ github.run_id }}-connectors-check", "extras=s3-cache"]
+    runs-on:
+      [
+        runs-on,
+        runner=8cpu-linux-x64,
+        "run-id=${{ github.run_id }}-connectors-check",
+        "extras=s3-cache",
+      ]
     timeout-minutes: 45
-
-    env:
-      PYTHONPATH: ./backend
-      DISABLE_TELEMETRY: "true"
+    environment: ci-protected
 
     steps:
       - uses: runs-on/action@cd2b598b0515d39d78c38a02d529db87d2196d1e # ratchet:runs-on/action@v2
@@ -188,6 +96,66 @@ jobs:
               - 'backend/onyx/file_processing/**'
               - 'uv.lock'
 
+      - name: Configure AWS credentials
+        uses: aws-actions/configure-aws-credentials@8df5847569e6427dd6c4fb1cf565c83acfa8afa7 # ratchet:aws-actions/configure-aws-credentials@v4
+        with:
+          role-to-assume: ${{ secrets.AWS_OIDC_ROLE_ARN }}
+          aws-region: us-east-2
+
+      - name: Get connector test secrets from AWS Secrets Manager
+        uses: aws-actions/aws-secretsmanager-get-secrets@a9a7eb4e2f2871d30dc5b892576fde60a2ecc802 # ratchet:aws-actions/aws-secretsmanager-get-secrets@v2
+        with:
+          parse-json-secrets: false
+          secret-ids: |
+            AWS_ACCESS_KEY_ID_DAILY_CONNECTOR_TESTS, test/aws-access-key-id
+            AWS_SECRET_ACCESS_KEY_DAILY_CONNECTOR_TESTS, test/aws-secret-access-key
+            R2_ACCESS_KEY_ID_DAILY_CONNECTOR_TESTS, test/r2-access-key-id
+            R2_SECRET_ACCESS_KEY_DAILY_CONNECTOR_TESTS, test/r2-secret-access-key
+            GCS_ACCESS_KEY_ID_DAILY_CONNECTOR_TESTS, test/gcs-access-key-id
+            GCS_SECRET_ACCESS_KEY_DAILY_CONNECTOR_TESTS, test/gcs-secret-access-key
+            CONFLUENCE_ACCESS_TOKEN, test/confluence-access-token
+            CONFLUENCE_ACCESS_TOKEN_SCOPED, test/confluence-access-token-scoped
+            JIRA_BASE_URL, test/jira-base-url
+            JIRA_USER_EMAIL, test/jira-user-email
+            JIRA_API_TOKEN, test/jira-api-token
+            JIRA_API_TOKEN_SCOPED, test/jira-api-token-scoped
+            GONG_ACCESS_KEY, test/gong-access-key
+            GONG_ACCESS_KEY_SECRET, test/gong-access-key-secret
+            GOOGLE_DRIVE_SERVICE_ACCOUNT_JSON_STR, test/google-drive-service-account-json
+            GOOGLE_DRIVE_OAUTH_CREDENTIALS_JSON_STR_TEST_USER_1, test/google-drive-oauth-creds-test-user-1
+            GOOGLE_DRIVE_OAUTH_CREDENTIALS_JSON_STR, test/google-drive-oauth-creds
+            GOOGLE_GMAIL_SERVICE_ACCOUNT_JSON_STR, test/google-gmail-service-account-json
+            GOOGLE_GMAIL_OAUTH_CREDENTIALS_JSON_STR, test/google-gmail-oauth-creds
+            SLAB_BOT_TOKEN, test/slab-bot-token
+            ZENDESK_SUBDOMAIN, test/zendesk-subdomain
+            ZENDESK_EMAIL, test/zendesk-email
+            ZENDESK_TOKEN, test/zendesk-token
+            SF_PASSWORD, test/sf-password
+            SF_SECURITY_TOKEN, test/sf-security-token
+            HUBSPOT_ACCESS_TOKEN, test/hubspot-access-token
+            IMAP_PASSWORD, test/imap-password
+            AIRTABLE_ACCESS_TOKEN, test/airtable-access-token
+            SHAREPOINT_CLIENT_SECRET, test/sharepoint-client-secret
+            PERM_SYNC_SHAREPOINT_CLIENT_ID, test/perm-sync-sharepoint-client-id
+            PERM_SYNC_SHAREPOINT_PRIVATE_KEY, test/perm-sync-sharepoint-private-key
+            PERM_SYNC_SHAREPOINT_CERTIFICATE_PASSWORD, test/perm-sync-sharepoint-cert-password
+            PERM_SYNC_SHAREPOINT_DIRECTORY_ID, test/perm-sync-sharepoint-directory-id
+            ACCESS_TOKEN_GITHUB, test/github-access-token
+            GITLAB_ACCESS_TOKEN, test/gitlab-access-token
+            GITBOOK_SPACE_ID, test/gitbook-space-id
+            GITBOOK_API_KEY, test/gitbook-api-key
+            NOTION_INTEGRATION_TOKEN, test/notion-integration-token
+            HIGHSPOT_KEY, test/highspot-key
+            HIGHSPOT_SECRET, test/highspot-secret
+            SLACK_BOT_TOKEN, test/slack-bot-token
+            DISCORD_CONNECTOR_BOT_TOKEN, test/discord-bot-token
+            TEAMS_APPLICATION_ID, test/teams-application-id
+            TEAMS_DIRECTORY_ID, test/teams-directory-id
+            TEAMS_SECRET, test/teams-secret
+            BITBUCKET_WORKSPACE, test/bitbucket-workspace
+            BITBUCKET_API_TOKEN, test/bitbucket-api-token
+            FIREFLIES_API_KEY, test/fireflies-api-key
+
       - name: Run Tests (excluding HubSpot, Salesforce, GitHub, and Coda)
         shell: script -q -e -c "bash --noprofile --norc -eo pipefail {0}"
         run: |

.github/workflows/pr-python-model-tests.yml

@@ -31,6 +31,7 @@ jobs:
       - runner=4cpu-linux-arm64
       - "run-id=${{ github.run_id }}-model-check"
       - "extras=ecr-cache"
+    environment: ci-protected
     timeout-minutes: 45
 
     env:

.github/workflows/release-cli.yml

@@ -13,15 +13,6 @@ jobs:
     permissions:
       id-token: write
     timeout-minutes: 10
-    strategy:
-      matrix:
-        os-arch:
-          - { goos: "linux", goarch: "amd64" }
-          - { goos: "linux", goarch: "arm64" }
-          - { goos: "windows", goarch: "amd64" }
-          - { goos: "windows", goarch: "arm64" }
-          - { goos: "darwin", goarch: "amd64" }
-          - { goos: "darwin", goarch: "arm64" }
     steps:
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # ratchet:actions/checkout@v6
         with:
@@ -31,9 +22,11 @@ jobs:
           enable-cache: false
           version: "0.9.9"
       - run: |
-          GOOS="${{ matrix.os-arch.goos }}" \
-          GOARCH="${{ matrix.os-arch.goarch }}" \
-          uv build --wheel
+          for goos in linux windows darwin; do
+            for goarch in amd64 arm64; do
+              GOOS="$goos" GOARCH="$goarch" uv build --wheel
+            done
+          done
         working-directory: cli
       - run: uv publish
         working-directory: cli

.github/workflows/storybook-deploy.yml

@@ -25,6 +25,7 @@ permissions:
 jobs:
   Deploy-Storybook:
     runs-on: ubuntu-latest
+    environment: ci-protected
     timeout-minutes: 30
     steps:
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # ratchet:actions/checkout@v4
@@ -54,6 +55,7 @@ jobs:
     needs: Deploy-Storybook
     if: always() && needs.Deploy-Storybook.result == 'failure'
     runs-on: ubuntu-latest
+    environment: ci-protected
     timeout-minutes: 10
     steps:
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # ratchet:actions/checkout@v4

.github/workflows/sync_foss.yml

@@ -9,6 +9,7 @@ on:
 jobs:
   sync-foss:
     runs-on: ubuntu-latest
+    environment: ci-protected
     timeout-minutes: 45
     permissions:
       contents: read

.github/workflows/tag-nightly.yml

@@ -11,6 +11,7 @@ permissions:
 jobs:
   create-and-push-tag:
     runs-on: ubuntu-slim
+    environment: ci-protected
     timeout-minutes: 45
 
     steps:

.greptile/rules.md

@@ -6,7 +6,7 @@ Use explicit type annotations for variables to enhance code clarity, especially
 
 ## Best Practices
 
-Use `contributing_guides/best_practices.md` as core review context. Prefer consistency with existing patterns, fix issues in code you touch, avoid tacking new features onto muddy interfaces, fail loudly instead of silently swallowing errors, keep code strictly typed, preserve clear state boundaries, remove duplicate or dead logic, break up overly long functions, avoid hidden import-time side effects, respect module boundaries, and favor correctness-by-construction over relying on callers to use an API correctly.
+Use the "Engineering Best Practices" section of `CONTRIBUTING.md` as core review context. Prefer consistency with existing patterns, fix issues in code you touch, avoid tacking new features onto muddy interfaces, fail loudly instead of silently swallowing errors, keep code strictly typed, preserve clear state boundaries, remove duplicate or dead logic, break up overly long functions, avoid hidden import-time side effects, respect module boundaries, and favor correctness-by-construction over relying on callers to use an API correctly.
 
 ## TODOs
 
@@ -24,6 +24,21 @@ When hardcoding a boolean variable to a constant value, remove the variable enti
 
 Code changes must consider both multi-tenant and single-tenant deployments. In multi-tenant mode, preserve tenant isolation, ensure tenant context is propagated correctly, and avoid assumptions that only hold for a single shared schema or globally shared state. In single-tenant mode, avoid introducing unnecessary tenant-specific requirements or cloud-only control-plane dependencies.
 
+## Nginx Routing — New Backend Routes
+
+Whenever a new backend route is added that does NOT start with `/api`, it must also be explicitly added to ALL nginx configs:
+
+- `deployment/helm/charts/onyx/templates/nginx-conf.yaml` (Helm/k8s)
+- `deployment/data/nginx/app.conf.template` (docker-compose dev)
+- `deployment/data/nginx/app.conf.template.prod` (docker-compose prod)
+- `deployment/data/nginx/app.conf.template.no-letsencrypt` (docker-compose no-letsencrypt)
+
+Routes not starting with `/api` are not caught by the existing `^/(api|openapi\.json)` location block and will fall through to `location /`, which proxies to the Next.js web server and returns an HTML 404. The new location block must be placed before the `/api` block. Examples of routes that need this treatment: `/scim`, `/mcp`.
+
 ## Full vs Lite Deployments
 
 Code changes must consider both regular Onyx deployments and Onyx lite deployments. Lite deployments disable the vector DB, Redis, model servers, and background workers by default, use PostgreSQL-backed cache/auth/file storage, and rely on the API server to handle background work. Do not assume those services are available unless the code path is explicitly limited to full deployments.
+
+## SWR Cache Keys — Always Use SWR_KEYS Registry
+
+All `useSWR()` calls and `mutate()` calls in the frontend must reference the centralized `SWR_KEYS` registry in `web/src/lib/swr-keys.ts` instead of inline endpoint strings or local string constants. Never write `useSWR("/api/some/endpoint", ...)` or `mutate("/api/some/endpoint")` — always use the corresponding `SWR_KEYS.someEndpoint` constant. If the endpoint does not yet exist in the registry, add it there first. This applies to all variants of an endpoint (e.g. query-string variants like `?get_editable=true` must also be registered as their own key).

.pre-commit-config.yaml

@@ -122,7 +122,7 @@ repos:
     rev: 5d1e709b7be35cb2025444e19de266b056b7b7ee # frozen: v2.10.1
     hooks:
       - id: golangci-lint
-        language_version: "1.26.0"
+        language_version: "1.26.1"
         entry: bash -c "find . -name go.mod -not -path './.venv/*' -print0 | xargs -0 -I{} bash -c 'cd \"$(dirname {})\" && golangci-lint run ./...'"
 
   - repo: https://github.com/astral-sh/ruff-pre-commit

AGENTS.md

@@ -357,5 +357,5 @@ raise OnyxError(OnyxErrorCode.BAD_GATEWAY, detail, status_code_override=e.respon
 ## 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.
+to the codebase can be found in the "Engineering Best Practices" section of
+`CONTRIBUTING.md`. Understand its contents and follow them.

CONTRIBUTING.md

@@ -1,32 +1,487 @@
 # Contributing to Onyx
+
 Hey there! We are so excited that you're interested in Onyx.
 
+## Table of Contents
+
+- [Contribution Opportunities](#contribution-opportunities)
+- [Contribution Process](#contribution-process)
+- [Development Setup](#development-setup)
+  - [Prerequisites](#prerequisites)
+  - [Backend: Python Requirements](#backend-python-requirements)
+  - [Frontend: Node Dependencies](#frontend-node-dependencies)
+  - [Formatting and Linting](#formatting-and-linting)
+- [Running the Application](#running-the-application)
+  - [VSCode Debugger (Recommended)](#vscode-debugger-recommended)
+  - [Manually Running for Development](#manually-running-for-development)
+  - [Running in Docker](#running-in-docker)
+- [macOS-Specific Notes](#macos-specific-notes)
+- [Engineering Best Practices](#engineering-best-practices)
+  - [Principles and Collaboration](#principles-and-collaboration)
+  - [Style and Maintainability](#style-and-maintainability)
+  - [Performance and Correctness](#performance-and-correctness)
+  - [Repository Conventions](#repository-conventions)
+- [Release Process](#release-process)
+- [Getting Help](#getting-help)
+- [Enterprise Edition Contributions](#enterprise-edition-contributions)
+
+---
 
 ## Contribution Opportunities
+
 The [GitHub Issues](https://github.com/onyx-dot-app/onyx/issues) page is a great place to look for and share contribution ideas.
 
-If you have your own feature that you would like to build please create an issue and community members can provide feedback and
-thumb it up if they feel a common need. 
+If you have your own feature that you would like to build, please create an issue and community members can provide feedback and upvote if they feel a common need.
 
+---
 
-## Contributing Code
-Please reference the documents in contributing_guides folder to ensure that the code base is kept to a high standard.
-1. dev_setup.md (start here): gives you a guide to setting up a local development environment.
-2. contribution_process.md: how to ensure you are building valuable features that will get reviewed and merged.
-3. best_practices.md: before asking for reviews, ensure your changes meet the repo code quality standards.
+## Contribution Process
 
 To contribute, please follow the
 ["fork and pull request"](https://docs.github.com/en/get-started/quickstart/contributing-to-projects) workflow.
 
+### 1. Get the feature or enhancement approved
+
+Create a GitHub issue and see if there are upvotes. If you feel the feature is sufficiently value-additive and you would like approval to contribute it to the repo, tag [Yuhong](https://github.com/yuhongsun96) to review.
+
+If you do not get a response within a week, feel free to email yuhong@onyx.app and include the issue in the message.
+
+Not all small features and enhancements will be accepted as there is a balance between feature richness and bloat. We strive to provide the best user experience possible so we have to be intentional about what we include in the app.
+
+### 2. Get the design approved
+
+The Onyx team will either provide a design doc and PRD for the feature or request one from you, the contributor. The scope and detail of the design will depend on the individual feature.
+
+### 3. IP attribution for EE contributions
+
+If you are contributing features to Onyx Enterprise Edition, you are required to sign the [IP Assignment Agreement](contributor_ip_assignment/EE_Contributor_IP_Assignment_Agreement.md).
+
+### 4. Review and testing
+
+Your features must pass all tests and all comments must be addressed prior to merging.
+
+### Implicit agreements
+
+If we approve an issue, we are promising you the following:
+- Your work will receive timely attention and we will put aside other important items to ensure you are not blocked.
+- You will receive necessary coaching on eng quality, system design, etc. to ensure the feature is completed well.
+- The Onyx team will pull resources and bandwidth from design, PM, and engineering to ensure that you have all the resources to build the feature to the quality required for merging.
+
+Because this is a large investment from our team, we ask that you:
+- Thoroughly read all the requirements of the design docs, engineering best practices, and try to minimize overhead for the Onyx team.
+- Complete the feature in a timely manner to reduce context switching and an ongoing resource pull from the Onyx team.
+
+---
+
+## Development Setup
+
+Onyx being a fully functional app, relies on some external software, specifically:
+
+- [Postgres](https://www.postgresql.org/) (Relational DB)
+- [OpenSearch](https://opensearch.org/) (Vector DB/Search Engine)
+- [Redis](https://redis.io/) (Cache)
+- [MinIO](https://min.io/) (File Store)
+- [Nginx](https://nginx.org/) (Not needed for development flows generally)
+
+> **Note:**
+> This guide provides instructions to build and run Onyx locally from source with Docker containers providing the above external software.
+> We believe this combination is easier for development purposes. If you prefer to use pre-built container images, see [Running in Docker](#running-in-docker) below.
+
+### Prerequisites
+
+- **Python 3.11** — If using a lower version, modifications will have to be made to the code. Higher versions may have library compatibility issues.
+- **Docker** — Required for running external services (Postgres, OpenSearch, Redis, MinIO).
+- **Node.js v22** — We recommend using [nvm](https://github.com/nvm-sh/nvm) to manage Node installations.
+
+### Backend: Python Requirements
+
+We use [uv](https://docs.astral.sh/uv/) and recommend creating a [virtual environment](https://docs.astral.sh/uv/pip/environments/#using-a-virtual-environment).
+
+```bash
+uv venv .venv --python 3.11
+source .venv/bin/activate
+```
+
+_For Windows, activate the virtual environment using Command Prompt:_
+
+```bash
+.venv\Scripts\activate
+```
+
+If using PowerShell, the command slightly differs:
+
+```powershell
+.venv\Scripts\Activate.ps1
+```
+
+Install the required Python dependencies:
+
+```bash
+uv sync --all-extras
+```
+
+Install Playwright for Python (headless browser required by the Web Connector):
+
+```bash
+uv run playwright install
+```
+
+### Frontend: Node Dependencies
+
+```bash
+nvm install 22 && nvm use 22
+node -v # verify your active version
+```
+
+Navigate to `onyx/web` and run:
+
+```bash
+npm i
+```
+
+### Formatting and Linting
+
+#### Backend
+
+Set up pre-commit hooks (black / reorder-python-imports):
+
+```bash
+uv run pre-commit install
+```
+
+We also use `mypy` for static type checking. Onyx is fully type-annotated, and we want to keep it that way! To run the mypy checks manually:
+
+```bash
+uv run mypy .  # from onyx/backend
+```
+
+#### Frontend
+
+We use `prettier` for formatting. The desired version will be installed via `npm i` from the `onyx/web` directory. To run the formatter:
+
+```bash
+npx prettier --write .  # from onyx/web
+```
+
+Pre-commit will also run prettier automatically on files you've recently touched. If re-formatted, your commit will fail. Re-stage your changes and commit again.
+
+---
+
+## Running the Application
+
+### VSCode Debugger (Recommended)
+
+We highly recommend using VSCode's debugger for development.
+
+#### Initial Setup
+
+1. Copy `.vscode/env_template.txt` to `.vscode/.env`
+2. Fill in the necessary environment variables in `.vscode/.env`
+
+#### Using the Debugger
+
+Before starting, make sure the Docker Daemon is running.
+
+1. Open the Debug view in VSCode (Cmd+Shift+D on macOS)
+2. From the dropdown at the top, select "Clear and Restart External Volumes and Containers" and press the green play button
+3. From the dropdown at the top, select "Run All Onyx Services" and press the green play button
+4. Navigate to http://localhost:3000 in your browser to start using the app
+5. Set breakpoints by clicking to the left of line numbers to help debug while the app is running
+6. Use the debug toolbar to step through code, inspect variables, etc.
+
+> **Note:** "Clear and Restart External Volumes and Containers" will reset your Postgres and OpenSearch (relational-db and index). Only run this if you are okay with wiping your data.
+
+**Features:**
+- Hot reload is enabled for the web server and API servers
+- Python debugging is configured with debugpy
+- Environment variables are loaded from `.vscode/.env`
+- Console output is organized in the integrated terminal with labeled tabs
+
+### Manually Running for Development
+
+#### Docker containers for external software
+
+You will need Docker installed to run these containers.
+
+Navigate to `onyx/deployment/docker_compose`, then start up Postgres/OpenSearch/Redis/MinIO with:
+
+```bash
+docker compose -f docker-compose.yml -f docker-compose.dev.yml up -d index relational_db cache minio
+```
+
+(index refers to OpenSearch, relational_db refers to Postgres, and cache refers to Redis)
+
+#### Running Onyx locally
+
+To start the frontend, navigate to `onyx/web` and run:
+
+```bash
+npm run dev
+```
+
+Next, start the model server which runs the local NLP models. Navigate to `onyx/backend` and run:
+
+```bash
+uvicorn model_server.main:app --reload --port 9000
+```
+
+_For Windows (for compatibility with both PowerShell and Command Prompt):_
+
+```bash
+powershell -Command "uvicorn model_server.main:app --reload --port 9000"
+```
+
+The first time running Onyx, you will need to run the DB migrations for Postgres. After the first time, this is no longer required unless the DB models change.
+
+Navigate to `onyx/backend` and with the venv active, run:
+
+```bash
+alembic upgrade head
+```
+
+Next, start the task queue which orchestrates the background jobs. Still in `onyx/backend`, run:
+
+```bash
+python ./scripts/dev_run_background_jobs.py
+```
+
+To run the backend API server, navigate back to `onyx/backend` and run:
+
+```bash
+AUTH_TYPE=basic uvicorn onyx.main:app --reload --port 8080
+```
+
+_For Windows (for compatibility with both PowerShell and Command Prompt):_
+
+```bash
+powershell -Command "
+    $env:AUTH_TYPE='basic'
+    uvicorn onyx.main:app --reload --port 8080
+"
+```
+
+> **Note:** If you need finer logging, add the additional environment variable `LOG_LEVEL=DEBUG` to the relevant services.
+
+#### Wrapping up
+
+You should now have 4 servers running:
+
+- Web server
+- Backend API
+- Model server
+- Background jobs
+
+Now, visit http://localhost:3000 in your browser. You should see the Onyx onboarding wizard where you can connect your external LLM provider to Onyx.
+
+You've successfully set up a local Onyx instance!
+
+### Running in Docker
+
+You can run the full Onyx application stack from pre-built images including all external software dependencies.
+
+Navigate to `onyx/deployment/docker_compose` and run:
+
+```bash
+docker compose up -d
+```
+
+After Docker pulls and starts these containers, navigate to http://localhost:3000 to use Onyx.
+
+If you want to make changes to Onyx and run those changes in Docker, you can also build a local version of the Onyx container images that incorporates your changes:
+
+```bash
+docker compose up -d --build
+```
+
+---
+
+## macOS-Specific Notes
+
+### Setting up Python
+
+Ensure [Homebrew](https://brew.sh/) is already set up, then install Python 3.11:
+
+```bash
+brew install python@3.11
+```
+
+Add Python 3.11 to your path by adding the following line to `~/.zshrc`:
+
+```
+export PATH="$(brew --prefix)/opt/python@3.11/libexec/bin:$PATH"
+```
+
+> **Note:** You will need to open a new terminal for the path change above to take effect.
+
+### Setting up Docker
+
+On macOS, you will need to install [Docker Desktop](https://www.docker.com/products/docker-desktop/) and ensure it is running before continuing with the docker commands.
+
+### Formatting and Linting
+
+macOS will likely require you to remove some quarantine attributes on some of the hooks for them to execute properly. After installing pre-commit, run the following command:
+
+```bash
+sudo xattr -r -d com.apple.quarantine ~/.cache/pre-commit
+```
+
+---
+
+## Engineering Best Practices
+
+> These are also what we adhere to as a team internally, we love to build in the open and to uplevel our community and each other through being transparent.
+
+### Principles and Collaboration
+
+- **Use 1-way vs 2-way doors.** For 2-way doors, move faster and iterate. For 1-way doors, be more deliberate.
+- **Consistency > being "right."** Prefer consistent patterns across the codebase. If something is truly bad, fix it everywhere.
+- **Fix what you touch (selectively).**
+  - Don't feel obligated to fix every best-practice issue you notice.
+  - Don't introduce new bad practices.
+  - If your change touches code that violates best practices, fix it as part of the change.
+- **Don't tack features on.** When adding functionality, restructure logically as needed to avoid muddying interfaces and accumulating tech debt.
+
+### Style and Maintainability
+
+#### Comments and readability
+Add clear comments:
+- At logical boundaries (e.g., interfaces) so the reader doesn't need to dig 10 layers deeper.
+- Wherever assumptions are made or something non-obvious/unexpected is done.
+- For complicated flows/functions.
+- Wherever it saves time (e.g., nontrivial regex patterns).
+
+#### Errors and exceptions
+- **Fail loudly** rather than silently skipping work.
+  - Example: raise and let exceptions propagate instead of silently dropping a document.
+- **Don't overuse `try/except`.**
+  - Put `try/except` at the correct logical level.
+  - Do not mask exceptions unless it is clearly appropriate.
+
+#### Typing
+- Everything should be **as strictly typed as possible**.
+- Use `cast` for annoying/loose-typed interfaces (e.g., results of `run_functions_tuples_in_parallel`).
+  - Only `cast` when the type checker sees `Any` or types are too loose.
+- Prefer types that are easy to read.
+  - Avoid dense types like `dict[tuple[str, str], list[list[float]]]`.
+  - Prefer domain models, e.g.:
+    - `EmbeddingModel(provider_name, model_name)` as a Pydantic model
+    - `dict[EmbeddingModel, list[EmbeddingVector]]`
+
+#### State, objects, and boundaries
+- Keep **clear logical boundaries** for state containers and objects.
+- A **config** object should never contain things like a `db_session`.
+- Avoid state containers that are overly nested, or huge + flat (use judgment).
+- Prefer **composition and functional style** over inheritance/OOP.
+- Prefer **no mutation** unless there's a strong reason.
+- State objects should be **intentional and explicit**, ideally nonmutating.
+- Use interfaces/objects to create clear separation of responsibility.
+- Prefer simplicity when there's no clear gain.
+  - Avoid overcomplicated mechanisms like semaphores.
+  - Prefer **hash maps (dicts)** over tree structures unless there's a strong reason.
+
+#### Naming
+- Name variables carefully and intentionally.
+- Prefer long, explicit names when undecided.
+- Avoid single-character variables except for small, self-contained utilities (or not at all).
+- Keep the same object/name consistent through the call stack and within functions when reasonable.
+  - Good: `for token in tokens:`
+  - Bad: `for msg in tokens:` (if iterating tokens)
+- Function names should bias toward **long + descriptive** for codebase search.
+  - IntelliSense can miss call sites; search works best with unique names.
+
+#### Correctness by construction
+- Prefer self-contained correctness — don't rely on callers to "use it right" if you can make misuse hard.
+- Avoid redundancies: if a function takes an arg, it shouldn't also take a state object that contains that same arg.
+- No dead code (unless there's a very good reason).
+- No commented-out code in main or feature branches (unless there's a very good reason).
+- No duplicate logic:
+  - Don't copy/paste into branches when shared logic can live above the conditional.
+  - If you're afraid to touch the original, you don't understand it well enough.
+  - LLMs often create subtle duplicate logic — review carefully and remove it.
+  - Avoid "nearly identical" objects that confuse when to use which.
+- Avoid extremely long functions with chained logic:
+  - Encapsulate steps into helpers for readability, even if not reused.
+  - "Pythonic" multi-step expressions are OK in moderation; don't trade clarity for cleverness.
+
+### Performance and Correctness
+
+- Avoid holding resources for extended periods (DB sessions, locks/semaphores).
+- Validate objects on creation and right before use.
+- Connector code (data to Onyx documents):
+  - Any in-memory structure that can grow without bound based on input must be periodically size-checked.
+  - If a connector is OOMing (often shows up as "missing celery tasks"), this is a top thing to check retroactively.
+- Async and event loops:
+  - Never introduce new async/event loop Python code, and try to make existing async code synchronous when possible if it makes sense.
+  - Writing async code without 100% understanding the code and having a concrete reason to do so is likely to introduce bugs and not add any meaningful performance gains.
+
+### Repository Conventions
+
+#### Where code lives
+- Pydantic + data models: `models.py` files.
+- DB interface functions (excluding lazy loading): `db/` directory.
+- LLM prompts: `prompts/` directory, roughly mirroring the code layout that uses them.
+- API routes: `server/` directory.
+
+#### Pydantic and modeling
+- Prefer **Pydantic** over dataclasses.
+- If absolutely required, use `allow_arbitrary_types`.
+
+#### Data conventions
+- Prefer explicit `None` over sentinel empty strings (usually; depends on intent).
+- Prefer explicit identifiers: use string enums instead of integer codes.
+- Avoid magic numbers (co-location is good when necessary). **Always avoid magic strings.**
+
+#### Logging
+- Log messages where they are created.
+- Don't propagate log messages around just to log them elsewhere.
+
+#### Encapsulation
+- Don't use private attributes/methods/properties from other classes/modules.
+- "Private" is private — respect that boundary.
+
+#### SQLAlchemy guidance
+- Lazy loading is often bad at scale, especially across multiple list relationships.
+- Be careful when accessing SQLAlchemy object attributes:
+  - It can help avoid redundant DB queries,
+  - but it can also fail if accessed outside an active session,
+  - and lazy loading can add hidden DB dependencies to otherwise "simple" functions.
+- Reference: https://www.reddit.com/r/SQLAlchemy/comments/138f248/joinedload_vs_selectinload/
+
+#### Trunk-based development and feature flags
+- **PRs should contain no more than 500 lines of real change.**
+- **Merge to main frequently.** Avoid long-lived feature branches — they create merge conflicts and integration pain.
+- **Use feature flags for incremental rollout.**
+  - Large features should be merged in small, shippable increments behind a flag.
+  - This allows continuous integration without exposing incomplete functionality.
+- **Keep flags short-lived.** Once a feature is fully rolled out, remove the flag and dead code paths promptly.
+- **Flag at the right level.** Prefer flagging at API/UI entry points rather than deep in business logic.
+- **Test both flag states.** Ensure the codebase works correctly with the flag on and off.
+
+#### Miscellaneous
+- Any TODOs you add in the code must be accompanied by either the name/username of the owner of that TODO, or an issue number for an issue referencing that piece of work.
+- Avoid module-level logic that runs on import, which leads to import-time side effects. Essentially every piece of meaningful logic should exist within some function that has to be explicitly invoked. Acceptable exceptions may include loading environment variables or setting up loggers.
+  - If you find yourself needing something like this, you may want that logic to exist in a file dedicated for manual execution (contains `if __name__ == "__main__":`) which should not be imported by anything else.
+- Do not conflate Python scripts you intend to run from the command line (contains `if __name__ == "__main__":`) with modules you intend to import from elsewhere. If for some unlikely reason they have to be the same file, any logic specific to executing the file (including imports) should be contained in the `if __name__ == "__main__":` block.
+  - Generally these executable files exist in `backend/scripts/`.
+
+---
+
+## Release Process
+
+Onyx loosely follows the SemVer versioning standard.
+A set of Docker containers will be pushed automatically to DockerHub with every tag.
+You can see the containers [here](https://hub.docker.com/search?q=onyx%2F).
+
+---
+
+## Getting Help
 
-## Getting Help 🙋
 We have support channels and generally interesting discussions on our [Discord](https://discord.gg/4NA5SbzrWb).
 
 See you there!
 
+---
 
-## Release Process
-Onyx loosely follows the SemVer versioning standard.
-Major changes are released with a "minor" version bump. Currently we use patch release versions to indicate small feature changes.
-A set of Docker containers will be pushed automatically to DockerHub with every tag.
-You can see the containers [here](https://hub.docker.com/search?q=onyx%2F).
+## Enterprise Edition Contributions
+
+If you are contributing features to Onyx Enterprise Edition (code under any `ee/` directory), you are required to sign the [IP Assignment Agreement](contributor_ip_assignment/EE_Contributor_IP_Assignment_Agreement.md) ([PDF version](contributor_ip_assignment/EE_Contributor_IP_Assignment_Agreement.pdf)).

README.md

@@ -4,8 +4,6 @@
     <a href="https://www.onyx.app/?utm_source=onyx_repo&utm_medium=github&utm_campaign=readme"> <img width="50%" src="https://github.com/onyx-dot-app/onyx/blob/logo/OnyxLogoCropped.jpg?raw=true" /></a>
 </h2>
 
-<p align="center">Open Source AI Platform</p>
-
 <p align="center">
     <a href="https://discord.gg/TDJ59cGV2X" target="_blank">
         <img src="https://img.shields.io/badge/discord-join-blue.svg?logo=discord&logoColor=white" alt="Discord" />
@@ -27,82 +25,94 @@
   </a>
 </p>
 
+# Onyx - The Open Source AI Platform
 
-**[Onyx](https://www.onyx.app/?utm_source=onyx_repo&utm_medium=github&utm_campaign=readme)** is a feature-rich, self-hostable Chat UI that works with any LLM. It is easy to deploy and can run in a completely airgapped environment.
+**[Onyx](https://www.onyx.app/?utm_source=onyx_repo&utm_medium=github&utm_campaign=readme)** is the application layer for LLMs - bringing a feature-rich interface that can be easily hosted by anyone.
+Onyx enables LLMs through advanced capabilities like RAG, web search, code execution, file creation, deep research and more.
 
-Onyx comes loaded with advanced features like Agents, Web Search, RAG, MCP, Deep Research, Connectors to 40+ knowledge sources, and more.
+Connect your applications with over 50+ indexing based connectors provided out of the box or via MCP.
 
 > [!TIP]
-> Run Onyx with one command (or see deployment section below):
+> Deploy with a single command:
 > ```
-> curl -fsSL https://raw.githubusercontent.com/onyx-dot-app/onyx/main/deployment/docker_compose/install.sh > install.sh && chmod +x install.sh && ./install.sh
+> curl -fsSL https://onyx.app/install_onyx.sh | bash
 > ```
 
-****
-
-![Onyx Chat Silent Demo](https://github.com/onyx-dot-app/onyx/releases/download/v0.21.1/OnyxChatSilentDemo.gif)
-
+![Onyx Chat Silent Demo](https://github.com/onyx-dot-app/onyx/releases/download/v3.0.0/Onyx.gif)
 
+---
 
 ## ⭐ Features
-- **🤖 Custom Agents:** Build AI Agents with unique instructions, knowledge and actions.
-- **🌍 Web Search:** Browse the web with Google PSE, Exa, and Serper as well as an in-house scraper or Firecrawl.
-- **🔍 RAG:** Best in class hybrid-search + knowledge graph for uploaded files and ingested documents from connectors. 
-- **🔄 Connectors:** Pull knowledge, metadata, and access information from over 40 applications.
-- **🔬 Deep Research:** Get in depth answers with an agentic multi-step search.
-- **▶️ Actions & MCP:** Give AI Agents the ability to interact with external systems.
-- **💻 Code Interpreter:** Execute code to analyze data, render graphs and create files.
+
+- **🔍 Agentic RAG:** Get best in class search and answer quality based on hybrid index + AI Agents for information retrieval
+  - Benchmark to release soon!
+- **🔬 Deep Research:** Get in depth reports with a multi-step research flow.
+  - Top of [leaderboard](https://github.com/onyx-dot-app/onyx_deep_research_bench) as of Feb 2026.
+- **🤖 Custom Agents:** Build AI Agents with unique instructions, knowledge, and actions.
+- **🌍 Web Search:** Browse the web to get up to date information.
+  - Supports Serper, Google PSE, Brave, SearXNG, and others.
+  - Comes with an in house web crawler and support for Firecrawl/Exa.
+- **📄 Artifacts:** Generate documents, graphics, and other downloadable artifacts.
+- **▶️ Actions & MCP:** Let Onyx agents interact with external applications, comes with flexible Auth options.
+- **💻 Code Execution:** Execute code in a sandbox to analyze data, render graphs, or modify files.
+- **🎙️ Voice Mode:** Chat with Onyx via text-to-speech and speech-to-text.
 - **🎨 Image Generation:** Generate images based on user prompts.
-- **👥 Collaboration:** Chat sharing, feedback gathering, user management, usage analytics, and more.
 
-Onyx works with all LLMs (like OpenAI, Anthropic, Gemini, etc.) and self-hosted LLMs (like Ollama, vLLM, etc.)
+Onyx supports all major LLM providers, both self-hosted (like Ollama, LiteLLM, vLLM, etc.) and proprietary (like Anthropic, OpenAI, Gemini, etc.).
 
-To learn more about the features, check out our [documentation](https://docs.onyx.app/welcome?utm_source=onyx_repo&utm_medium=github&utm_campaign=readme)!
+To learn more - check out our [docs](https://docs.onyx.app/welcome?utm_source=onyx_repo&utm_medium=github&utm_campaign=readme)!
 
+---
 
+## 🚀 Deployment Modes
 
-## 🚀 Deployment
-Onyx supports deployments in Docker, Kubernetes, Terraform, along with guides for major cloud providers.
+> Onyx supports deployments in Docker, Kubernetes, Helm/Terraform and provides guides for major cloud providers.
+> Detailed deployment guides found [here](https://docs.onyx.app/deployment/overview).
 
-See guides below:
-- [Docker](https://docs.onyx.app/deployment/local/docker?utm_source=onyx_repo&utm_medium=github&utm_campaign=readme) or [Quickstart](https://docs.onyx.app/deployment/getting_started/quickstart?utm_source=onyx_repo&utm_medium=github&utm_campaign=readme) (best for most users)
-- [Kubernetes](https://docs.onyx.app/deployment/local/kubernetes?utm_source=onyx_repo&utm_medium=github&utm_campaign=readme) (best for large teams)
-- [Terraform](https://docs.onyx.app/deployment/local/terraform?utm_source=onyx_repo&utm_medium=github&utm_campaign=readme) (best for teams already using Terraform)
-- Cloud specific guides (best if specifically using [AWS EKS](https://docs.onyx.app/deployment/cloud/aws/eks?utm_source=onyx_repo&utm_medium=github&utm_campaign=readme), [Azure VMs](https://docs.onyx.app/deployment/cloud/azure?utm_source=onyx_repo&utm_medium=github&utm_campaign=readme), etc.)
+Onyx supports two separate deployment options: standard and lite.
+
+#### Onyx Lite
+
+The Lite mode can be thought of as a lightweight Chat UI. It requires less resources (under 1GB memory) and runs a less complex stack.
+It is great for users who want to test out Onyx quickly or for teams who are only interested in the Chat UI and Agents functionalities.
+
+#### Standard Onyx
+
+The complete feature set of Onyx which is recommended for serious users and larger teams. Additional components not included in Lite mode:
+- Vector + Keyword index for RAG.
+- Background containers to run job queues and workers for syncing knowledge from connectors.
+- AI model inference servers to run deep learning models used during indexing and inference.
+- Performance optimizations for large scale use via in memory cache (Redis) and blob store (MinIO).
 
 > [!TIP]  
-> **To try Onyx for free without deploying, check out [Onyx Cloud](https://cloud.onyx.app/signup?utm_source=onyx_repo&utm_medium=github&utm_campaign=readme)**.
+> **To try Onyx for free without deploying, visit [Onyx Cloud](https://cloud.onyx.app/signup?utm_source=onyx_repo&utm_medium=github&utm_campaign=readme)**.
 
+---
 
+## 🏢 Onyx for Enterprise
 
-## 🔍 Other Notable Benefits
-Onyx is built for teams of all sizes, from individual users to the largest global enterprises.
-
-- **Enterprise Search**: far more than simple RAG, Onyx has custom indexing and retrieval that remains performant and accurate for scales of up to tens of millions of documents.
-- **Security**: SSO (OIDC/SAML/OAuth2), RBAC, encryption of credentials, etc.
-- **Management UI**: different user roles such as basic, curator, and admin.
-- **Document Permissioning**: mirrors user access from external apps for RAG use cases.
-
-
-
-## 🚧 Roadmap
-To see ongoing and upcoming projects, check out our [roadmap](https://github.com/orgs/onyx-dot-app/projects/2)!
-
-
+Onyx is built for teams of all sizes, from individual users to the largest global enterprises:
+- 👥 Collaboration: Share chats and agents with other members of your organization.
+- 🔐 Single Sign On: SSO via Google OAuth, OIDC, or SAML. Group syncing and user provisioning via SCIM.
+- 🛡️ Role Based Access Control: RBAC for sensitive resources like access to agents, actions, etc.
+- 📊 Analytics: Usage graphs broken down by teams, LLMs, or agents.
+- 🕵️ Query History: Audit usage to ensure safe adoption of AI in your organization.
+- 💻 Custom code: Run custom code to remove PII, reject sensitive queries, or to run custom analysis.
+- 🎨 Whitelabeling: Customize the look and feel of Onyx with custom naming, icons, banners, and more.
 
 ## 📚 Licensing
+
 There are two editions of Onyx:
 
-- Onyx Community Edition (CE) is available freely under the MIT license.
+- Onyx Community Edition (CE) is available freely under the MIT license and covers all of the core features for Chat, RAG, Agents, and Actions.
 - Onyx Enterprise Edition (EE) includes extra features that are primarily useful for larger organizations.
+
 For feature details, check out [our website](https://www.onyx.app/pricing?utm_source=onyx_repo&utm_medium=github&utm_campaign=readme).
 
-
-
 ## 👪 Community
+
 Join our open source community on **[Discord](https://discord.gg/TDJ59cGV2X)**!
 
-
-
 ## 💡 Contributing
+
 Looking to contribute? Please check out the [Contribution Guide](CONTRIBUTING.md) for more details.

backend/alembic/versions/8188861f4e92_csv_to_tabular_chat_file_type.py

@@ -0,0 +1,54 @@
+"""csv to tabular chat file type
+
+Revision ID: 8188861f4e92
+Revises: d8cdfee5df80
+Create Date: 2026-03-31 19:23:05.753184
+
+"""
+
+from alembic import op
+
+
+# revision identifiers, used by Alembic.
+revision = "8188861f4e92"
+down_revision = "d8cdfee5df80"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        UPDATE chat_message
+        SET files = (
+            SELECT jsonb_agg(
+                CASE
+                    WHEN elem->>'type' = 'csv'
+                    THEN jsonb_set(elem, '{type}', '"tabular"')
+                    ELSE elem
+                END
+            )
+            FROM jsonb_array_elements(files) AS elem
+        )
+        WHERE files::text LIKE '%"type": "csv"%'
+        """
+    )
+
+
+def downgrade() -> None:
+    op.execute(
+        """
+        UPDATE chat_message
+        SET files = (
+            SELECT jsonb_agg(
+                CASE
+                    WHEN elem->>'type' = 'tabular'
+                    THEN jsonb_set(elem, '{type}', '"csv"')
+                    ELSE elem
+                END
+            )
+            FROM jsonb_array_elements(files) AS elem
+        )
+        WHERE files::text LIKE '%"type": "tabular"%'
+        """
+    )

backend/alembic/versions/d8cdfee5df80_add_skipped_to_userfilestatus.py

@@ -0,0 +1,55 @@
+"""add skipped to userfilestatus
+
+Revision ID: d8cdfee5df80
+Revises: 1d78c0ca7853
+Create Date: 2026-04-01 10:47:12.593950
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "d8cdfee5df80"
+down_revision = "1d78c0ca7853"
+branch_labels = None
+depends_on = None
+
+
+TABLE = "user_file"
+COLUMN = "status"
+CONSTRAINT_NAME = "ck_user_file_status"
+
+OLD_VALUES = ("PROCESSING", "INDEXING", "COMPLETED", "FAILED", "CANCELED", "DELETING")
+NEW_VALUES = (
+    "PROCESSING",
+    "INDEXING",
+    "COMPLETED",
+    "SKIPPED",
+    "FAILED",
+    "CANCELED",
+    "DELETING",
+)
+
+
+def _drop_status_check_constraint() -> None:
+    inspector = sa.inspect(op.get_bind())
+    for constraint in inspector.get_check_constraints(TABLE):
+        if COLUMN in constraint.get("sqltext", ""):
+            constraint_name = constraint["name"]
+            if constraint_name is not None:
+                op.drop_constraint(constraint_name, TABLE, type_="check")
+
+
+def upgrade() -> None:
+    _drop_status_check_constraint()
+    in_clause = ", ".join(f"'{v}'" for v in NEW_VALUES)
+    op.create_check_constraint(CONSTRAINT_NAME, TABLE, f"{COLUMN} IN ({in_clause})")
+
+
+def downgrade() -> None:
+    op.execute(f"UPDATE {TABLE} SET {COLUMN} = 'COMPLETED' WHERE {COLUMN} = 'SKIPPED'")
+    _drop_status_check_constraint()
+    in_clause = ", ".join(f"'{v}'" for v in OLD_VALUES)
+    op.create_check_constraint(CONSTRAINT_NAME, TABLE, f"{COLUMN} IN ({in_clause})")

backend/ee/onyx/background/celery/apps/primary.py

@@ -5,6 +5,7 @@ from onyx.background.celery.apps.primary import celery_app
 celery_app.autodiscover_tasks(
     app_base.filter_task_modules(
         [
+            "ee.onyx.background.celery.tasks.hooks",
             "ee.onyx.background.celery.tasks.doc_permission_syncing",
             "ee.onyx.background.celery.tasks.external_group_syncing",
             "ee.onyx.background.celery.tasks.cloud",

backend/ee/onyx/background/celery/tasks/beat_schedule.py

@@ -55,6 +55,15 @@ ee_tasks_to_schedule: list[dict] = []
 
 if not MULTI_TENANT:
     ee_tasks_to_schedule = [
+        {
+            "name": "hook-execution-log-cleanup",
+            "task": OnyxCeleryTask.HOOK_EXECUTION_LOG_CLEANUP_TASK,
+            "schedule": timedelta(days=1),
+            "options": {
+                "priority": OnyxCeleryPriority.LOW,
+                "expires": BEAT_EXPIRES_DEFAULT,
+            },
+        },
         {
             "name": "autogenerate-usage-report",
             "task": OnyxCeleryTask.GENERATE_USAGE_REPORT_TASK,

backend/ee/onyx/background/celery/tasks/doc_permission_syncing/tasks.py

@@ -474,6 +474,8 @@ def connector_permission_sync_generator_task(
             cc_pair = get_connector_credential_pair_from_id(
                 db_session=db_session,
                 cc_pair_id=cc_pair_id,
+                eager_load_connector=True,
+                eager_load_credential=True,
             )
             if cc_pair is None:
                 raise ValueError(

backend/ee/onyx/background/celery/tasks/tenant_provisioning/tasks.py

@@ -13,6 +13,7 @@ from redis.lock import Lock as RedisLock
 from ee.onyx.server.tenants.provisioning import setup_tenant
 from ee.onyx.server.tenants.schema_management import create_schema_if_not_exists
 from ee.onyx.server.tenants.schema_management import get_current_alembic_version
+from ee.onyx.server.tenants.schema_management import run_alembic_migrations
 from onyx.background.celery.apps.app_base import task_logger
 from onyx.configs.app_configs import TARGET_AVAILABLE_TENANTS
 from onyx.configs.constants import ONYX_CLOUD_TENANT_ID
@@ -29,9 +30,10 @@ from shared_configs.configs import TENANT_ID_PREFIX
 # Each tenant takes ~80s (alembic migrations), so 5 tenants ≈ 7 minutes.
 _MAX_TENANTS_PER_RUN = 5
 
-# Time limits sized for worst-case batch: _MAX_TENANTS_PER_RUN × ~90s + buffer.
-_TENANT_PROVISIONING_SOFT_TIME_LIMIT = 60 * 10  # 10 minutes
-_TENANT_PROVISIONING_TIME_LIMIT = 60 * 15  # 15 minutes
+# Time limits sized for worst-case: provisioning up to _MAX_TENANTS_PER_RUN new tenants
+# (~90s each) plus migrating up to TARGET_AVAILABLE_TENANTS pool tenants (~90s each).
+_TENANT_PROVISIONING_SOFT_TIME_LIMIT = 60 * 20  # 20 minutes
+_TENANT_PROVISIONING_TIME_LIMIT = 60 * 25  # 25 minutes
 
 
 @shared_task(
@@ -91,8 +93,7 @@ def check_available_tenants(self: Task) -> None:  # noqa: ARG001
         batch_size = min(tenants_to_provision, _MAX_TENANTS_PER_RUN)
         if batch_size < tenants_to_provision:
             task_logger.info(
-                f"Capping batch to {batch_size} "
-                f"(need {tenants_to_provision}, will catch up next cycle)"
+                f"Capping batch to {batch_size} (need {tenants_to_provision}, will catch up next cycle)"
             )
 
         provisioned = 0
@@ -103,12 +104,14 @@ def check_available_tenants(self: Task) -> None:  # noqa: ARG001
                     provisioned += 1
             except Exception:
                 task_logger.exception(
-                    f"Failed to provision tenant {i + 1}/{batch_size}, "
-                    "continuing with remaining tenants"
+                    f"Failed to provision tenant {i + 1}/{batch_size}, continuing with remaining tenants"
                 )
 
         task_logger.info(f"Provisioning complete: {provisioned}/{batch_size} succeeded")
 
+        # Migrate any pool tenants that were provisioned before a new migration was deployed
+        _migrate_stale_pool_tenants()
+
     except Exception:
         task_logger.exception("Error in check_available_tenants task")
 
@@ -121,6 +124,46 @@ def check_available_tenants(self: Task) -> None:  # noqa: ARG001
             )
 
 
+def _migrate_stale_pool_tenants() -> None:
+    """
+    Run alembic upgrade head on all pool tenants. Since alembic upgrade head is
+    idempotent, tenants already at head are a fast no-op. This ensures pool
+    tenants are always current so that signup doesn't hit schema mismatches
+    (e.g. missing columns added after the tenant was pre-provisioned).
+    """
+    with get_session_with_shared_schema() as db_session:
+        pool_tenants = db_session.query(AvailableTenant).all()
+        tenant_ids = [t.tenant_id for t in pool_tenants]
+
+    if not tenant_ids:
+        return
+
+    task_logger.info(
+        f"Checking {len(tenant_ids)} pool tenant(s) for pending migrations"
+    )
+
+    for tenant_id in tenant_ids:
+        try:
+            run_alembic_migrations(tenant_id)
+            new_version = get_current_alembic_version(tenant_id)
+            with get_session_with_shared_schema() as db_session:
+                tenant = (
+                    db_session.query(AvailableTenant)
+                    .filter_by(tenant_id=tenant_id)
+                    .first()
+                )
+                if tenant and tenant.alembic_version != new_version:
+                    task_logger.info(
+                        f"Migrated pool tenant {tenant_id}: {tenant.alembic_version} -> {new_version}"
+                    )
+                    tenant.alembic_version = new_version
+                    db_session.commit()
+        except Exception:
+            task_logger.exception(
+                f"Failed to migrate pool tenant {tenant_id}, skipping"
+            )
+
+
 def pre_provision_tenant() -> bool:
     """
     Pre-provision a new tenant and store it in the NewAvailableTenant table.

backend/ee/onyx/configs/license_enforcement_config.py

@@ -69,5 +69,7 @@ EE_ONLY_PATH_PREFIXES: frozenset[str] = frozenset(
         "/admin/token-rate-limits",
         # Evals
         "/evals",
+        # Hook extensions
+        "/admin/hooks",
     }
 )

backend/ee/onyx/external_permissions/slack/doc_sync.py

@@ -8,6 +8,7 @@ from ee.onyx.external_permissions.slack.utils import fetch_user_id_to_email_map
 from onyx.access.models import DocExternalAccess
 from onyx.access.models import ExternalAccess
 from onyx.connectors.credentials_provider import OnyxDBCredentialsProvider
+from onyx.connectors.interfaces import SecondsSinceUnixEpoch
 from onyx.connectors.models import HierarchyNode
 from onyx.connectors.slack.connector import get_channels
 from onyx.connectors.slack.connector import make_paginated_slack_api_call
@@ -105,9 +106,11 @@ def _get_slack_document_access(
     slack_connector: SlackConnector,
     channel_permissions: dict[str, ExternalAccess],  # noqa: ARG001
     callback: IndexingHeartbeatInterface | None,
+    indexing_start: SecondsSinceUnixEpoch | None = None,
 ) -> Generator[DocExternalAccess, None, None]:
     slim_doc_generator = slack_connector.retrieve_all_slim_docs_perm_sync(
-        callback=callback
+        callback=callback,
+        start=indexing_start,
     )
 
     for doc_metadata_batch in slim_doc_generator:
@@ -180,9 +183,15 @@ def slack_doc_sync(
 
     slack_connector = SlackConnector(**cc_pair.connector.connector_specific_config)
     slack_connector.set_credentials_provider(provider)
+    indexing_start_ts: SecondsSinceUnixEpoch | None = (
+        cc_pair.connector.indexing_start.timestamp()
+        if cc_pair.connector.indexing_start is not None
+        else None
+    )
 
     yield from _get_slack_document_access(
-        slack_connector,
+        slack_connector=slack_connector,
         channel_permissions=channel_permissions,
         callback=callback,
+        indexing_start=indexing_start_ts,
     )

backend/ee/onyx/external_permissions/utils.py

@@ -6,6 +6,7 @@ from onyx.access.models import ElementExternalAccess
 from onyx.access.models import ExternalAccess
 from onyx.access.models import NodeExternalAccess
 from onyx.configs.constants import DocumentSource
+from onyx.connectors.interfaces import SecondsSinceUnixEpoch
 from onyx.connectors.interfaces import SlimConnectorWithPermSync
 from onyx.connectors.models import HierarchyNode
 from onyx.db.models import ConnectorCredentialPair
@@ -40,10 +41,19 @@ def generic_doc_sync(
 
     logger.info(f"Starting {doc_source} doc sync for CC Pair ID: {cc_pair.id}")
 
+    indexing_start: SecondsSinceUnixEpoch | None = (
+        cc_pair.connector.indexing_start.timestamp()
+        if cc_pair.connector.indexing_start is not None
+        else None
+    )
+
     newly_fetched_doc_ids: set[str] = set()
 
     logger.info(f"Fetching all slim documents from {doc_source}")
-    for doc_batch in slim_connector.retrieve_all_slim_docs_perm_sync(callback=callback):
+    for doc_batch in slim_connector.retrieve_all_slim_docs_perm_sync(
+        start=indexing_start,
+        callback=callback,
+    ):
         logger.info(f"Got {len(doc_batch)} slim documents from {doc_source}")
 
         if callback:

backend/ee/onyx/hooks/executor.py

@@ -0,0 +1,385 @@
+"""Hook executor — calls a customer's external HTTP endpoint for a given hook point.
+
+Usage (Celery tasks and FastAPI handlers):
+    result = execute_hook(
+        db_session=db_session,
+        hook_point=HookPoint.QUERY_PROCESSING,
+        payload={"query": "...", "user_email": "...", "chat_session_id": "..."},
+        response_type=QueryProcessingResponse,
+    )
+
+    if isinstance(result, HookSkipped):
+        # no active hook configured — continue with original behavior
+        ...
+    elif isinstance(result, HookSoftFailed):
+        # hook failed but fail strategy is SOFT — continue with original behavior
+        ...
+    else:
+        # result is a validated Pydantic model instance (response_type)
+        ...
+
+is_reachable update policy
+--------------------------
+``is_reachable`` on the Hook row is updated selectively — only when the outcome
+carries meaningful signal about physical reachability:
+
+  NetworkError (DNS, connection refused)  → False  (cannot reach the server)
+  HTTP 401 / 403                          → False  (api_key revoked or invalid)
+  TimeoutException                        → None   (server may be slow, skip write)
+  Other HTTP errors (4xx / 5xx)           → None   (server responded, skip write)
+  Unknown exception                       → None   (no signal, skip write)
+  Non-JSON / non-dict response            → None   (server responded, skip write)
+  Success (2xx, valid dict)               → True   (confirmed reachable)
+
+None means "leave the current value unchanged" — no DB round-trip is made.
+
+DB session design
+-----------------
+The executor uses three sessions:
+
+  1. Caller's session (db_session) — used only for the hook lookup read. All
+     needed fields are extracted from the Hook object before the HTTP call, so
+     the caller's session is not held open during the external HTTP request.
+
+  2. Log session — a separate short-lived session opened after the HTTP call
+     completes to write the HookExecutionLog row on failure. Success runs are
+     not recorded. Committed independently of everything else.
+
+  3. Reachable session — a second short-lived session to update is_reachable on
+     the Hook. Kept separate from the log session so a concurrent hook deletion
+     (which causes update_hook__no_commit to raise OnyxError(NOT_FOUND)) cannot
+     prevent the execution log from being written. This update is best-effort.
+"""
+
+import json
+import time
+from typing import Any
+from typing import TypeVar
+
+import httpx
+from pydantic import BaseModel
+from pydantic import ValidationError
+from sqlalchemy.orm import Session
+
+from onyx.db.engine.sql_engine import get_session_with_current_tenant
+from onyx.db.enums import HookFailStrategy
+from onyx.db.enums import HookPoint
+from onyx.db.hook import create_hook_execution_log__no_commit
+from onyx.db.hook import get_non_deleted_hook_by_hook_point
+from onyx.db.hook import update_hook__no_commit
+from onyx.db.models import Hook
+from onyx.error_handling.error_codes import OnyxErrorCode
+from onyx.error_handling.exceptions import OnyxError
+from onyx.hooks.executor import HookSkipped
+from onyx.hooks.executor import HookSoftFailed
+from onyx.utils.logger import setup_logger
+from shared_configs.configs import MULTI_TENANT
+
+logger = setup_logger()
+
+
+T = TypeVar("T", bound=BaseModel)
+
+
+# ---------------------------------------------------------------------------
+# Private helpers
+# ---------------------------------------------------------------------------
+
+
+class _HttpOutcome(BaseModel):
+    """Structured result of an HTTP hook call, returned by _process_response."""
+
+    is_success: bool
+    updated_is_reachable: (
+        bool | None
+    )  # True/False = write to DB, None = unchanged (skip write)
+    status_code: int | None
+    error_message: str | None
+    response_payload: dict[str, Any] | None
+
+
+def _lookup_hook(
+    db_session: Session,
+    hook_point: HookPoint,
+) -> Hook | HookSkipped:
+    """Return the active Hook or HookSkipped if hooks are unavailable/unconfigured.
+
+    No HTTP call is made and no DB writes are performed for any HookSkipped path.
+    There is nothing to log and no reachability information to update.
+    """
+    if MULTI_TENANT:
+        return HookSkipped()
+    hook = get_non_deleted_hook_by_hook_point(
+        db_session=db_session, hook_point=hook_point
+    )
+    if hook is None or not hook.is_active:
+        return HookSkipped()
+    if not hook.endpoint_url:
+        return HookSkipped()
+    return hook
+
+
+def _process_response(
+    *,
+    response: httpx.Response | None,
+    exc: Exception | None,
+    timeout: float,
+) -> _HttpOutcome:
+    """Process the result of an HTTP call and return a structured outcome.
+
+    Called after the client.post() try/except. If post() raised, exc is set and
+    response is None. Otherwise response is set and exc is None. Handles
+    raise_for_status(), JSON decoding, and the dict shape check.
+    """
+    if exc is not None:
+        if isinstance(exc, httpx.NetworkError):
+            msg = f"Hook network error (endpoint unreachable): {exc}"
+            logger.warning(msg, exc_info=exc)
+            return _HttpOutcome(
+                is_success=False,
+                updated_is_reachable=False,
+                status_code=None,
+                error_message=msg,
+                response_payload=None,
+            )
+        if isinstance(exc, httpx.TimeoutException):
+            msg = f"Hook timed out after {timeout}s: {exc}"
+            logger.warning(msg, exc_info=exc)
+            return _HttpOutcome(
+                is_success=False,
+                updated_is_reachable=None,  # timeout doesn't indicate unreachability
+                status_code=None,
+                error_message=msg,
+                response_payload=None,
+            )
+        msg = f"Hook call failed: {exc}"
+        logger.exception(msg, exc_info=exc)
+        return _HttpOutcome(
+            is_success=False,
+            updated_is_reachable=None,  # unknown error — don't make assumptions
+            status_code=None,
+            error_message=msg,
+            response_payload=None,
+        )
+
+    if response is None:
+        raise ValueError(
+            "exactly one of response or exc must be non-None; both are None"
+        )
+    status_code = response.status_code
+
+    try:
+        response.raise_for_status()
+    except httpx.HTTPStatusError as e:
+        msg = f"Hook returned HTTP {e.response.status_code}: {e.response.text}"
+        logger.warning(msg, exc_info=e)
+        # 401/403 means the api_key has been revoked or is invalid — mark unreachable
+        # so the operator knows to update it. All other HTTP errors keep is_reachable
+        # as-is (server is up, the request just failed for application reasons).
+        auth_failed = e.response.status_code in (401, 403)
+        return _HttpOutcome(
+            is_success=False,
+            updated_is_reachable=False if auth_failed else None,
+            status_code=status_code,
+            error_message=msg,
+            response_payload=None,
+        )
+
+    try:
+        response_payload = response.json()
+    except (json.JSONDecodeError, httpx.DecodingError) as e:
+        msg = f"Hook returned non-JSON response: {e}"
+        logger.warning(msg, exc_info=e)
+        return _HttpOutcome(
+            is_success=False,
+            updated_is_reachable=None,  # server responded — reachability unchanged
+            status_code=status_code,
+            error_message=msg,
+            response_payload=None,
+        )
+
+    if not isinstance(response_payload, dict):
+        msg = f"Hook returned non-dict JSON (got {type(response_payload).__name__})"
+        logger.warning(msg)
+        return _HttpOutcome(
+            is_success=False,
+            updated_is_reachable=None,  # server responded — reachability unchanged
+            status_code=status_code,
+            error_message=msg,
+            response_payload=None,
+        )
+
+    return _HttpOutcome(
+        is_success=True,
+        updated_is_reachable=True,
+        status_code=status_code,
+        error_message=None,
+        response_payload=response_payload,
+    )
+
+
+def _persist_result(
+    *,
+    hook_id: int,
+    outcome: _HttpOutcome,
+    duration_ms: int,
+) -> None:
+    """Write the execution log on failure and optionally update is_reachable, each
+    in its own session so a failure in one does not affect the other."""
+    # Only write the execution log on failure — success runs are not recorded.
+    # Must not be skipped if the is_reachable update fails (e.g. hook concurrently
+    # deleted between the initial lookup and here).
+    if not outcome.is_success:
+        try:
+            with get_session_with_current_tenant() as log_session:
+                create_hook_execution_log__no_commit(
+                    db_session=log_session,
+                    hook_id=hook_id,
+                    is_success=False,
+                    error_message=outcome.error_message,
+                    status_code=outcome.status_code,
+                    duration_ms=duration_ms,
+                )
+                log_session.commit()
+        except Exception:
+            logger.exception(
+                f"Failed to persist hook execution log for hook_id={hook_id}"
+            )
+
+    # Update is_reachable separately — best-effort, non-critical.
+    # None means the value is unchanged (set by the caller to skip the no-op write).
+    # update_hook__no_commit can raise OnyxError(NOT_FOUND) if the hook was
+    # concurrently deleted, so keep this isolated from the log write above.
+    if outcome.updated_is_reachable is not None:
+        try:
+            with get_session_with_current_tenant() as reachable_session:
+                update_hook__no_commit(
+                    db_session=reachable_session,
+                    hook_id=hook_id,
+                    is_reachable=outcome.updated_is_reachable,
+                )
+                reachable_session.commit()
+        except Exception:
+            logger.warning(f"Failed to update is_reachable for hook_id={hook_id}")
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+
+def _execute_hook_inner(
+    hook: Hook,
+    payload: dict[str, Any],
+    response_type: type[T],
+) -> T | HookSoftFailed:
+    """Make the HTTP call, validate the response, and return a typed model.
+
+    Raises OnyxError on HARD failure. Returns HookSoftFailed on SOFT failure.
+    """
+    timeout = hook.timeout_seconds
+    hook_id = hook.id
+    fail_strategy = hook.fail_strategy
+    endpoint_url = hook.endpoint_url
+    current_is_reachable: bool | None = hook.is_reachable
+
+    if not endpoint_url:
+        raise ValueError(
+            f"hook_id={hook_id} is active but has no endpoint_url — "
+            "active hooks without an endpoint_url must be rejected by _lookup_hook"
+        )
+
+    start = time.monotonic()
+    response: httpx.Response | None = None
+    exc: Exception | None = None
+    try:
+        api_key: str | None = (
+            hook.api_key.get_value(apply_mask=False) if hook.api_key else None
+        )
+        headers: dict[str, str] = {"Content-Type": "application/json"}
+        if api_key:
+            headers["Authorization"] = f"Bearer {api_key}"
+        with httpx.Client(
+            timeout=timeout, follow_redirects=False
+        ) as client:  # SSRF guard: never follow redirects
+            response = client.post(endpoint_url, json=payload, headers=headers)
+    except Exception as e:
+        exc = e
+    duration_ms = int((time.monotonic() - start) * 1000)
+
+    outcome = _process_response(response=response, exc=exc, timeout=timeout)
+
+    # Validate the response payload against response_type.
+    # A validation failure downgrades the outcome to a failure so it is logged,
+    # is_reachable is left unchanged (server responded — just a bad payload),
+    # and fail_strategy is respected below.
+    validated_model: T | None = None
+    if outcome.is_success and outcome.response_payload is not None:
+        try:
+            validated_model = response_type.model_validate(outcome.response_payload)
+        except ValidationError as e:
+            msg = (
+                f"Hook response failed validation against {response_type.__name__}: {e}"
+            )
+            outcome = _HttpOutcome(
+                is_success=False,
+                updated_is_reachable=None,  # server responded — reachability unchanged
+                status_code=outcome.status_code,
+                error_message=msg,
+                response_payload=None,
+            )
+
+    # Skip the is_reachable write when the value would not change — avoids a
+    # no-op DB round-trip on every call when the hook is already in the expected state.
+    if outcome.updated_is_reachable == current_is_reachable:
+        outcome = outcome.model_copy(update={"updated_is_reachable": None})
+    _persist_result(hook_id=hook_id, outcome=outcome, duration_ms=duration_ms)
+
+    if not outcome.is_success:
+        if fail_strategy == HookFailStrategy.HARD:
+            raise OnyxError(
+                OnyxErrorCode.HOOK_EXECUTION_FAILED,
+                outcome.error_message or "Hook execution failed.",
+            )
+        logger.warning(
+            f"Hook execution failed (soft fail) for hook_id={hook_id}: {outcome.error_message}"
+        )
+        return HookSoftFailed()
+
+    if validated_model is None:
+        raise OnyxError(
+            OnyxErrorCode.INTERNAL_ERROR,
+            f"validated_model is None for successful hook call (hook_id={hook_id})",
+        )
+    return validated_model
+
+
+def _execute_hook_impl(
+    *,
+    db_session: Session,
+    hook_point: HookPoint,
+    payload: dict[str, Any],
+    response_type: type[T],
+) -> T | HookSkipped | HookSoftFailed:
+    """EE implementation — loaded by CE's execute_hook via fetch_versioned_implementation.
+
+    Returns HookSkipped if no active hook is configured, HookSoftFailed if the
+    hook failed with SOFT fail strategy, or a validated response model on success.
+    Raises OnyxError on HARD failure or if the hook is misconfigured.
+    """
+    hook = _lookup_hook(db_session, hook_point)
+    if isinstance(hook, HookSkipped):
+        return hook
+
+    fail_strategy = hook.fail_strategy
+    hook_id = hook.id
+
+    try:
+        return _execute_hook_inner(hook, payload, response_type)
+    except Exception:
+        if fail_strategy == HookFailStrategy.SOFT:
+            logger.exception(
+                f"Unexpected error in hook execution (soft fail) for hook_id={hook_id}"
+            )
+            return HookSoftFailed()
+        raise

backend/ee/onyx/main.py

@@ -15,6 +15,7 @@ from ee.onyx.server.enterprise_settings.api import (
     basic_router as enterprise_settings_router,
 )
 from ee.onyx.server.evals.api import router as evals_router
+from ee.onyx.server.features.hooks.api import router as hook_router
 from ee.onyx.server.license.api import router as license_router
 from ee.onyx.server.manage.standard_answer import router as standard_answer_router
 from ee.onyx.server.middleware.license_enforcement import (
@@ -138,6 +139,7 @@ def get_application() -> FastAPI:
     include_router_with_global_prefix_prepended(application, ee_oauth_router)
     include_router_with_global_prefix_prepended(application, ee_document_cc_pair_router)
     include_router_with_global_prefix_prepended(application, evals_router)
+    include_router_with_global_prefix_prepended(application, hook_router)
 
     # Enterprise-only global settings
     include_router_with_global_prefix_prepended(

backend/ee/onyx/server/features/hooks/api.py

@@ -123,9 +123,8 @@ def _validate_endpoint(
     (not reachable — indicates the api_key is invalid).
 
     Timeout handling:
-    - ConnectTimeout: TCP handshake never completed → cannot_connect.
-    - ReadTimeout / WriteTimeout: TCP was established, server responded slowly → timeout
-      (operator should consider increasing timeout_seconds).
+    - Any httpx.TimeoutException (ConnectTimeout, ReadTimeout, WriteTimeout, PoolTimeout) →
+      timeout (operator should consider increasing timeout_seconds).
     - All other exceptions → cannot_connect.
     """
     _check_ssrf_safety(endpoint_url)

backend/ee/onyx/server/tenants/provisioning.py

@@ -99,6 +99,26 @@ async def get_or_provision_tenant(
         tenant_id = await get_available_tenant()
 
         if tenant_id:
+            # Run migrations to ensure the pre-provisioned tenant schema is current.
+            # Pool tenants may have been created before a new migration was deployed.
+            # Capture as a non-optional local so mypy can type the lambda correctly.
+            _tenant_id: str = tenant_id
+            loop = asyncio.get_running_loop()
+            try:
+                await loop.run_in_executor(
+                    None, lambda: run_alembic_migrations(_tenant_id)
+                )
+            except Exception:
+                # The tenant was already dequeued from the pool — roll it back so
+                # it doesn't end up orphaned (schema exists, but not assigned to anyone).
+                logger.exception(
+                    f"Migration failed for pre-provisioned tenant {_tenant_id}; rolling back"
+                )
+                try:
+                    await rollback_tenant_provisioning(_tenant_id)
+                except Exception:
+                    logger.exception(f"Failed to rollback orphaned tenant {_tenant_id}")
+                raise
             # If we have a pre-provisioned tenant, assign it to the user
             await assign_tenant_to_user(tenant_id, email, referral_source)
             logger.info(f"Assigned pre-provisioned tenant {tenant_id} to user {email}")

backend/model_server/main.py

@@ -100,6 +100,7 @@ def get_model_app() -> FastAPI:
             dsn=SENTRY_DSN,
             integrations=[StarletteIntegration(), FastApiIntegration()],
             traces_sample_rate=0.1,
+            release=__version__,
         )
         logger.info("Sentry initialized")
     else:

backend/onyx/background/celery/apps/app_base.py

@@ -20,6 +20,7 @@ from sentry_sdk.integrations.celery import CeleryIntegration
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 
+from onyx import __version__
 from onyx.background.celery.apps.task_formatters import CeleryTaskColoredFormatter
 from onyx.background.celery.apps.task_formatters import CeleryTaskPlainFormatter
 from onyx.background.celery.celery_utils import celery_is_worker_primary
@@ -65,6 +66,7 @@ if SENTRY_DSN:
         dsn=SENTRY_DSN,
         integrations=[CeleryIntegration()],
         traces_sample_rate=0.1,
+        release=__version__,
     )
     logger.info("Sentry initialized")
 else:
@@ -515,7 +517,8 @@ def reset_tenant_id(
 
 
 def wait_for_vespa_or_shutdown(
-    sender: Any, **kwargs: Any  # noqa: ARG001
+    sender: Any,  # noqa: ARG001
+    **kwargs: Any,  # noqa: ARG001
 ) -> None:  # noqa: ARG001
     """Waits for Vespa to become ready subject to a timeout.
     Raises WorkerShutdown if the timeout is reached."""

backend/onyx/background/celery/apps/primary.py

@@ -317,7 +317,6 @@ celery_app.autodiscover_tasks(
             "onyx.background.celery.tasks.docprocessing",
             "onyx.background.celery.tasks.evals",
             "onyx.background.celery.tasks.hierarchyfetching",
-            "onyx.background.celery.tasks.hooks",
             "onyx.background.celery.tasks.periodic",
             "onyx.background.celery.tasks.pruning",
             "onyx.background.celery.tasks.shared",

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

@@ -14,7 +14,6 @@ from onyx.configs.constants import ONYX_CLOUD_CELERY_TASK_PREFIX
 from onyx.configs.constants import OnyxCeleryPriority
 from onyx.configs.constants import OnyxCeleryQueues
 from onyx.configs.constants import OnyxCeleryTask
-from onyx.hooks.utils import HOOKS_AVAILABLE
 from shared_configs.configs import MULTI_TENANT
 
 # choosing 15 minutes because it roughly gives us enough time to process many tasks
@@ -362,19 +361,6 @@ if not MULTI_TENANT:
 
     tasks_to_schedule.extend(beat_task_templates)
 
-if HOOKS_AVAILABLE:
-    tasks_to_schedule.append(
-        {
-            "name": "hook-execution-log-cleanup",
-            "task": OnyxCeleryTask.HOOK_EXECUTION_LOG_CLEANUP_TASK,
-            "schedule": timedelta(days=1),
-            "options": {
-                "priority": OnyxCeleryPriority.LOW,
-                "expires": BEAT_EXPIRES_DEFAULT,
-            },
-        }
-    )
-
 
 def generate_cloud_tasks(
     beat_tasks: list[dict], beat_templates: list[dict], beat_multiplier: float

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

@@ -9,6 +9,7 @@ from celery import Celery
 from celery import shared_task
 from celery import Task
 
+from onyx import __version__
 from onyx.background.celery.apps.app_base import task_logger
 from onyx.background.celery.memory_monitoring import emit_process_memory
 from onyx.background.celery.tasks.docprocessing.heartbeat import start_heartbeat
@@ -137,6 +138,7 @@ def _docfetching_task(
         sentry_sdk.init(
             dsn=SENTRY_DSN,
             traces_sample_rate=0.1,
+            release=__version__,
         )
         logger.info("Sentry initialized")
     else:

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

@@ -319,6 +319,11 @@ def monitor_indexing_attempt_progress(
     )
 
     current_db_time = get_db_current_time(db_session)
+    total_batches: int | str = (
+        coordination_status.total_batches
+        if coordination_status.total_batches is not None
+        else "?"
+    )
     if coordination_status.found:
         task_logger.info(
             f"Indexing attempt progress: "
@@ -326,7 +331,7 @@ def monitor_indexing_attempt_progress(
             f"cc_pair={attempt.connector_credential_pair_id} "
             f"search_settings={attempt.search_settings_id} "
             f"completed_batches={coordination_status.completed_batches} "
-            f"total_batches={coordination_status.total_batches or '?'} "
+            f"total_batches={total_batches} "
             f"total_docs={coordination_status.total_docs} "
             f"total_failures={coordination_status.total_failures}"
             f"elapsed={(current_db_time - attempt.time_created).seconds}"
@@ -410,7 +415,7 @@ def check_indexing_completion(
     logger.info(
         f"Indexing status: "
         f"indexing_completed={indexing_completed} "
-        f"batches_processed={batches_processed}/{batches_total or '?'} "
+        f"batches_processed={batches_processed}/{batches_total if batches_total is not None else '?'} "
         f"total_docs={coordination_status.total_docs} "
         f"total_chunks={coordination_status.total_chunks} "
         f"total_failures={coordination_status.total_failures}"

backend/onyx/chat/chat_utils.py

@@ -5,6 +5,7 @@ from typing import cast
 from uuid import UUID
 
 from fastapi.datastructures import Headers
+from pydantic import BaseModel
 from sqlalchemy.orm import Session
 
 from onyx.chat.models import ChatHistoryResult
@@ -51,6 +52,60 @@ logger = setup_logger()
 IMAGE_GENERATION_TOOL_NAME = "generate_image"
 
 
+class FileContextResult(BaseModel):
+    """Result of building a file's LLM context representation."""
+
+    message: ChatMessageSimple
+    tool_metadata: FileToolMetadata
+
+
+def build_file_context(
+    tool_file_id: str,
+    filename: str,
+    file_type: ChatFileType,
+    content_text: str | None = None,
+    token_count: int = 0,
+    approx_char_count: int | None = None,
+) -> FileContextResult:
+    """Build the LLM context representation for a single file.
+
+    Centralises how files should appear in the LLM prompt
+    — the ID that FileReaderTool accepts (``UserFile.id`` for user files).
+    """
+    if file_type.use_metadata_only():
+        message_text = (
+            f"File: {filename} (id={tool_file_id})\n"
+            "Use the file_reader or python tools to access "
+            "this file's contents."
+        )
+        message = ChatMessageSimple(
+            message=message_text,
+            token_count=max(1, len(message_text) // 4),
+            message_type=MessageType.USER,
+            file_id=tool_file_id,
+        )
+    else:
+        message_text = f"File: {filename}\n{content_text or ''}\nEnd of File"
+        message = ChatMessageSimple(
+            message=message_text,
+            token_count=token_count,
+            message_type=MessageType.USER,
+            file_id=tool_file_id,
+        )
+
+    metadata = FileToolMetadata(
+        file_id=tool_file_id,
+        filename=filename,
+        approx_char_count=(
+            approx_char_count
+            if approx_char_count is not None
+            else len(content_text or "")
+        ),
+    )
+
+    return FileContextResult(message=message, tool_metadata=metadata)
+
+
 def create_chat_session_from_request(
     chat_session_request: ChatSessionCreationRequest,
     user_id: UUID | None,
@@ -538,7 +593,7 @@ def convert_chat_history(
     for idx, chat_message in enumerate(chat_history):
         if chat_message.message_type == MessageType.USER:
             # Process files attached to this message
-            text_files: list[ChatLoadedFile] = []
+            text_files: list[tuple[ChatLoadedFile, FileDescriptor]] = []
             image_files: list[ChatLoadedFile] = []
 
             if chat_message.files:
@@ -549,34 +604,26 @@ def convert_chat_history(
                         if loaded_file.file_type == ChatFileType.IMAGE:
                             image_files.append(loaded_file)
                         else:
-                            # Text files (DOC, PLAIN_TEXT, CSV) are added as separate messages
-                            text_files.append(loaded_file)
+                            # Text files (DOC, PLAIN_TEXT, TABULAR) are added as separate messages
+                            text_files.append((loaded_file, file_descriptor))
 
             # Add text files as separate messages before the user message.
             # Each message is tagged with ``file_id`` so that forgotten files
             # can be detected after context-window truncation.
-            for text_file in text_files:
-                file_text = text_file.content_text or ""
-                filename = text_file.filename
-                message = (
-                    f"File: {filename}\n{file_text}\nEnd of File"
-                    if filename
-                    else file_text
-                )
-                simple_messages.append(
-                    ChatMessageSimple(
-                        message=message,
-                        token_count=text_file.token_count,
-                        message_type=MessageType.USER,
-                        image_files=None,
-                        file_id=text_file.file_id,
-                    )
-                )
-                all_injected_file_metadata[text_file.file_id] = FileToolMetadata(
-                    file_id=text_file.file_id,
-                    filename=filename or "unknown",
-                    approx_char_count=len(file_text),
+            for text_file, fd in text_files:
+                # Use user_file_id as the FileReaderTool accepts that.
+                # Fall back to the file-store path id.
+                tool_id = fd.get("user_file_id") or text_file.file_id
+                filename = text_file.filename or "unknown"
+                ctx = build_file_context(
+                    tool_file_id=tool_id,
+                    filename=filename,
+                    file_type=text_file.file_type,
+                    content_text=text_file.content_text,
+                    token_count=text_file.token_count,
                 )
+                simple_messages.append(ctx.message)
+                all_injected_file_metadata[tool_id] = ctx.tool_metadata
 
             # Sum token counts from image files (excluding project image files)
             image_token_count = (

backend/onyx/chat/process_message.py

@@ -18,6 +18,7 @@ from onyx.cache.interface import CacheBackend
 from onyx.chat.chat_processing_checker import set_processing_status
 from onyx.chat.chat_state import ChatStateContainer
 from onyx.chat.chat_state import run_chat_loop_with_state_containers
+from onyx.chat.chat_utils import build_file_context
 from onyx.chat.chat_utils import convert_chat_history
 from onyx.chat.chat_utils import create_chat_history_chain
 from onyx.chat.chat_utils import create_chat_session_from_request
@@ -90,6 +91,7 @@ from onyx.llm.request_context import reset_llm_mock_response
 from onyx.llm.request_context import set_llm_mock_response
 from onyx.llm.utils import litellm_exception_to_error_msg
 from onyx.onyxbot.slack.models import SlackContext
+from onyx.server.query_and_chat.chat_utils import mime_type_to_chat_file_type
 from onyx.server.query_and_chat.models import AUTO_PLACE_AFTER_LATEST_MESSAGE
 from onyx.server.query_and_chat.models import MessageResponseIDInfo
 from onyx.server.query_and_chat.models import SendMessageRequest
@@ -117,6 +119,8 @@ from shared_configs.contextvars import get_current_tenant_id
 logger = setup_logger()
 ERROR_TYPE_CANCELLED = "cancelled"
 
+APPROX_CHARS_PER_TOKEN = 4
+
 
 class _AvailableFiles(BaseModel):
     """Separated file IDs for the FileReaderTool so it knows which loader to use."""
@@ -301,16 +305,27 @@ def extract_context_files(
     if not user_files:
         return _empty_extracted_context_files()
 
-    aggregate_tokens = sum(uf.token_count or 0 for uf in user_files)
+    # Aggregate tokens for the file content that will be added
+    # Skip tokens for those with metadata only
+    aggregate_tokens = sum(
+        uf.token_count or 0
+        for uf in user_files
+        if not mime_type_to_chat_file_type(uf.file_type).use_metadata_only()
+    )
     max_actual_tokens = (
         llm_max_context_window - reserved_token_count
     ) * max_llm_context_percentage
 
     if aggregate_tokens >= max_actual_tokens:
-        tool_metadata = []
         use_as_search_filter = not DISABLE_VECTOR_DB
         if DISABLE_VECTOR_DB:
-            tool_metadata = _build_file_tool_metadata_for_user_files(user_files)
+            overflow_tool_metadata = [_build_tool_metadata(uf) for uf in user_files]
+        else:
+            overflow_tool_metadata = [
+                _build_tool_metadata(uf)
+                for uf in user_files
+                if mime_type_to_chat_file_type(uf.file_type).use_metadata_only()
+            ]
         return ExtractedContextFiles(
             file_texts=[],
             image_files=[],
@@ -318,11 +333,11 @@ def extract_context_files(
             total_token_count=0,
             file_metadata=[],
             uncapped_token_count=aggregate_tokens,
-            file_metadata_for_tool=tool_metadata,
+            file_metadata_for_tool=overflow_tool_metadata,
         )
 
     # Files fit — load them into context
-    user_file_map = {str(uf.id): uf for uf in user_files}
+    user_file_map = {uf.file_id: uf for uf in user_files}
     in_memory_files = load_in_memory_chat_files(
         user_file_ids=[uf.id for uf in user_files],
         db_session=db_session,
@@ -331,23 +346,38 @@ def extract_context_files(
     file_texts: list[str] = []
     image_files: list[ChatLoadedFile] = []
     file_metadata: list[ContextFileMetadata] = []
+    tool_metadata: list[FileToolMetadata] = []
     total_token_count = 0
 
     for f in in_memory_files:
         uf = user_file_map.get(str(f.file_id))
-        if f.file_type.is_text_file():
+        filename = f.filename or f"file_{f.file_id}"
+
+        if f.file_type.use_metadata_only():
+            # Metadata-only files are not injected as full text.
+            # Only the metadata is provided, with LLM using tools
+            if not uf:
+                logger.error(
+                    f"File with id={f.file_id} in metadata-only path with no associated user file"
+                )
+                continue
+            tool_metadata.append(_build_tool_metadata(uf))
+        elif f.file_type.is_text_file():
             text_content = _extract_text_from_in_memory_file(f)
             if not text_content:
                 continue
+            if not uf:
+                logger.warning(f"No user file for file_id={f.file_id}")
+                continue
             file_texts.append(text_content)
             file_metadata.append(
                 ContextFileMetadata(
-                    file_id=str(f.file_id),
-                    filename=f.filename or f"file_{f.file_id}",
+                    file_id=str(uf.id),
+                    filename=filename,
                     file_content=text_content,
                 )
             )
-            if uf and uf.token_count:
+            if uf.token_count:
                 total_token_count += uf.token_count
         elif f.file_type == ChatFileType.IMAGE:
             token_count = uf.token_count if uf and uf.token_count else 0
@@ -370,24 +400,22 @@ def extract_context_files(
         total_token_count=total_token_count,
         file_metadata=file_metadata,
         uncapped_token_count=aggregate_tokens,
+        file_metadata_for_tool=tool_metadata,
     )
 
 
-APPROX_CHARS_PER_TOKEN = 4
+def _build_tool_metadata(user_file: UserFile) -> FileToolMetadata:
+    """Build lightweight FileToolMetadata from a UserFile record.
 
-
-def _build_file_tool_metadata_for_user_files(
-    user_files: list[UserFile],
-) -> list[FileToolMetadata]:
-    """Build lightweight FileToolMetadata from a list of UserFile records."""
-    return [
-        FileToolMetadata(
-            file_id=str(uf.id),
-            filename=uf.name,
-            approx_char_count=(uf.token_count or 0) * APPROX_CHARS_PER_TOKEN,
-        )
-        for uf in user_files
-    ]
+    Delegates to ``build_file_context`` so that the file ID exposed to the
+    LLM is always consistent with what FileReaderTool expects.
+    """
+    return build_file_context(
+        tool_file_id=str(user_file.id),
+        filename=user_file.name,
+        file_type=mime_type_to_chat_file_type(user_file.file_type),
+        approx_char_count=(user_file.token_count or 0) * APPROX_CHARS_PER_TOKEN,
+    ).tool_metadata
 
 
 def determine_search_params(

backend/onyx/configs/app_configs.py

@@ -44,6 +44,31 @@ SEND_USER_METADATA_TO_LLM_PROVIDER = (
 # User Facing Features Configs
 #####
 BLURB_SIZE = 128  # Number Encoder Tokens included in the chunk blurb
+
+# Hard ceiling for the admin-configurable file upload size (in MB).
+# Self-hosted customers can raise or lower this via the environment variable.
+_raw_max_upload_size_mb = int(os.environ.get("MAX_ALLOWED_UPLOAD_SIZE_MB", "250"))
+if _raw_max_upload_size_mb < 0:
+    logger.warning(
+        "MAX_ALLOWED_UPLOAD_SIZE_MB=%d is negative; falling back to 250",
+        _raw_max_upload_size_mb,
+    )
+    _raw_max_upload_size_mb = 250
+MAX_ALLOWED_UPLOAD_SIZE_MB = _raw_max_upload_size_mb
+
+# Default fallback for the per-user file upload size limit (in MB) when no
+# admin-configured value exists.  Clamped to MAX_ALLOWED_UPLOAD_SIZE_MB at
+# runtime so this never silently exceeds the hard ceiling.
+_raw_default_upload_size_mb = int(
+    os.environ.get("DEFAULT_USER_FILE_MAX_UPLOAD_SIZE_MB", "100")
+)
+if _raw_default_upload_size_mb < 0:
+    logger.warning(
+        "DEFAULT_USER_FILE_MAX_UPLOAD_SIZE_MB=%d is negative; falling back to 100",
+        _raw_default_upload_size_mb,
+    )
+    _raw_default_upload_size_mb = 100
+DEFAULT_USER_FILE_MAX_UPLOAD_SIZE_MB = _raw_default_upload_size_mb
 GENERATIVE_MODEL_ACCESS_CHECK_FREQ = int(
     os.environ.get("GENERATIVE_MODEL_ACCESS_CHECK_FREQ") or 86400
 )  # 1 day
@@ -61,17 +86,6 @@ CACHE_BACKEND = CacheBackendType(
     os.environ.get("CACHE_BACKEND", CacheBackendType.REDIS)
 )
 
-# Maximum token count for a single uploaded file. Files exceeding this are rejected.
-# Defaults to 100k tokens (or 10M when vector DB is disabled).
-_DEFAULT_FILE_TOKEN_LIMIT = 10_000_000 if DISABLE_VECTOR_DB else 100_000
-FILE_TOKEN_COUNT_THRESHOLD = int(
-    os.environ.get("FILE_TOKEN_COUNT_THRESHOLD", str(_DEFAULT_FILE_TOKEN_LIMIT))
-)
-
-# Maximum upload size for a single user file (chat/projects) in MB.
-USER_FILE_MAX_UPLOAD_SIZE_MB = int(os.environ.get("USER_FILE_MAX_UPLOAD_SIZE_MB") or 50)
-USER_FILE_MAX_UPLOAD_SIZE_BYTES = USER_FILE_MAX_UPLOAD_SIZE_MB * 1024 * 1024
-
 # If set to true, will show extra/uncommon connectors in the "Other" category
 SHOW_EXTRA_CONNECTORS = os.environ.get("SHOW_EXTRA_CONNECTORS", "").lower() == "true"
 
@@ -791,6 +805,10 @@ MINI_CHUNK_SIZE = 150
 # This is the number of regular chunks per large chunk
 LARGE_CHUNK_RATIO = 4
 
+# The maximum number of chunks that can be held for 1 document processing batch
+# The purpose of this is to set an upper bound on memory usage
+MAX_CHUNKS_PER_DOC_BATCH = int(os.environ.get("MAX_CHUNKS_PER_DOC_BATCH") or 1000)
+
 # Include the document level metadata in each chunk. If the metadata is too long, then it is thrown out
 # We don't want the metadata to overwhelm the actual contents of the chunk
 SKIP_METADATA_IN_CHUNK = os.environ.get("SKIP_METADATA_IN_CHUNK", "").lower() == "true"
@@ -1061,7 +1079,6 @@ POD_NAMESPACE = os.environ.get("POD_NAMESPACE")
 
 DEV_MODE = os.environ.get("DEV_MODE", "").lower() == "true"
 
-HOOK_ENABLED = os.environ.get("HOOK_ENABLED", "").lower() == "true"
 
 INTEGRATION_TESTS_MODE = os.environ.get("INTEGRATION_TESTS_MODE", "").lower() == "true"
 

backend/onyx/configs/constants.py

@@ -212,6 +212,7 @@ class DocumentSource(str, Enum):
     PRODUCTBOARD = "productboard"
     FILE = "file"
     CODA = "coda"
+    CANVAS = "canvas"
     NOTION = "notion"
     ZULIP = "zulip"
     LINEAR = "linear"
@@ -672,6 +673,7 @@ DocumentSourceDescription: dict[DocumentSource, str] = {
     DocumentSource.SLAB: "slab data",
     DocumentSource.PRODUCTBOARD: "productboard data (boards, etc.)",
     DocumentSource.FILE: "files",
+    DocumentSource.CANVAS: "canvas lms - courses, pages, assignments, and announcements",
     DocumentSource.CODA: "coda - team workspace with docs, tables, and pages",
     DocumentSource.NOTION: "notion data - a workspace that combines note-taking, \
 project management, and collaboration tools into a single, customizable platform",

backend/onyx/connectors/canvas/access.py

@@ -0,0 +1,32 @@
+"""
+Permissioning / AccessControl logic for Canvas courses.
+
+CE stub — returns None (no permissions). The EE implementation is loaded
+at runtime via ``fetch_versioned_implementation``.
+"""
+
+from collections.abc import Callable
+from typing import cast
+
+from onyx.access.models import ExternalAccess
+from onyx.connectors.canvas.client import CanvasApiClient
+from onyx.utils.variable_functionality import fetch_versioned_implementation
+from onyx.utils.variable_functionality import global_version
+
+
+def get_course_permissions(
+    canvas_client: CanvasApiClient,
+    course_id: int,
+) -> ExternalAccess | None:
+    if not global_version.is_ee_version():
+        return None
+
+    ee_get_course_permissions = cast(
+        Callable[[CanvasApiClient, int], ExternalAccess | None],
+        fetch_versioned_implementation(
+            "onyx.external_permissions.canvas.access",
+            "get_course_permissions",
+        ),
+    )
+
+    return ee_get_course_permissions(canvas_client, course_id)

backend/onyx/connectors/canvas/client.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import logging
 import re
+from collections.abc import Iterator
 from typing import Any
 from urllib.parse import urlparse
 
@@ -190,3 +191,22 @@ class CanvasApiClient:
         if clean_endpoint:
             final_url += "/" + clean_endpoint
         return final_url
+
+    def paginate(
+        self,
+        endpoint: str,
+        params: dict[str, Any] | None = None,
+    ) -> Iterator[list[Any]]:
+        """Yield each page of results, following Link-header pagination.
+
+        Makes the first request with endpoint + params, then follows
+        next_url from Link headers for subsequent pages.
+        """
+        response, next_url = self.get(endpoint, params=params)
+        while True:
+            if not response:
+                break
+            yield response
+            if not next_url:
+                break
+            response, next_url = self.get(full_url=next_url)

backend/onyx/connectors/canvas/connector.py

@@ -1,17 +1,82 @@
+from datetime import datetime
+from datetime import timezone
+from typing import Any
+from typing import cast
 from typing import Literal
+from typing import NoReturn
 from typing import TypeAlias
 
 from pydantic import BaseModel
+from retry import retry
+from typing_extensions import override
 
+from onyx.access.models import ExternalAccess
+from onyx.configs.app_configs import INDEX_BATCH_SIZE
+from onyx.configs.constants import DocumentSource
+from onyx.connectors.canvas.access import get_course_permissions
+from onyx.connectors.canvas.client import CanvasApiClient
+from onyx.connectors.exceptions import ConnectorValidationError
+from onyx.connectors.exceptions import CredentialExpiredError
+from onyx.connectors.exceptions import InsufficientPermissionsError
+from onyx.connectors.exceptions import UnexpectedValidationError
+from onyx.connectors.interfaces import CheckpointedConnectorWithPermSync
+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.models import ConnectorCheckpoint
+from onyx.connectors.models import ConnectorMissingCredentialError
+from onyx.connectors.models import Document
+from onyx.connectors.models import ImageSection
+from onyx.connectors.models import TextSection
+from onyx.error_handling.exceptions import OnyxError
+from onyx.file_processing.html_utils import parse_html_page_basic
+from onyx.indexing.indexing_heartbeat import IndexingHeartbeatInterface
+from onyx.utils.logger import setup_logger
+
+logger = setup_logger()
+
+
+def _handle_canvas_api_error(e: OnyxError) -> NoReturn:
+    """Map Canvas API errors to connector framework exceptions."""
+    if e.status_code == 401:
+        raise CredentialExpiredError(
+            "Canvas API token is invalid or expired (HTTP 401)."
+        )
+    elif e.status_code == 403:
+        raise InsufficientPermissionsError(
+            "Canvas API token does not have sufficient permissions (HTTP 403)."
+        )
+    elif e.status_code == 429:
+        raise ConnectorValidationError(
+            "Canvas rate-limit exceeded (HTTP 429). Please try again later."
+        )
+    elif e.status_code >= 500:
+        raise UnexpectedValidationError(
+            f"Unexpected Canvas HTTP error (status={e.status_code}): {e}"
+        )
+    else:
+        raise ConnectorValidationError(
+            f"Canvas API error (status={e.status_code}): {e}"
+        )
 
 
 class CanvasCourse(BaseModel):
     id: int
-    name: str
-    course_code: str
-    created_at: str
-    workflow_state: str
+    name: str | None = None
+    course_code: str | None = None
+    created_at: str | None = None
+    workflow_state: str | None = None
+
+    @classmethod
+    def from_api(cls, payload: dict[str, Any]) -> "CanvasCourse":
+        return cls(
+            id=payload["id"],
+            name=payload.get("name"),
+            course_code=payload.get("course_code"),
+            created_at=payload.get("created_at"),
+            workflow_state=payload.get("workflow_state"),
+        )
 
 
 class CanvasPage(BaseModel):
@@ -19,10 +84,22 @@ class CanvasPage(BaseModel):
     url: str
     title: str
     body: str | None = None
-    created_at: str
-    updated_at: str
+    created_at: str | None = None
+    updated_at: str | None = None
     course_id: int
 
+    @classmethod
+    def from_api(cls, payload: dict[str, Any], course_id: int) -> "CanvasPage":
+        return cls(
+            page_id=payload["page_id"],
+            url=payload["url"],
+            title=payload["title"],
+            body=payload.get("body"),
+            created_at=payload.get("created_at"),
+            updated_at=payload.get("updated_at"),
+            course_id=course_id,
+        )
+
 
 class CanvasAssignment(BaseModel):
     id: int
@@ -30,10 +107,23 @@ class CanvasAssignment(BaseModel):
     description: str | None = None
     html_url: str
     course_id: int
-    created_at: str
-    updated_at: str
+    created_at: str | None = None
+    updated_at: str | None = None
     due_at: str | None = None
 
+    @classmethod
+    def from_api(cls, payload: dict[str, Any], course_id: int) -> "CanvasAssignment":
+        return cls(
+            id=payload["id"],
+            name=payload["name"],
+            description=payload.get("description"),
+            html_url=payload["html_url"],
+            course_id=course_id,
+            created_at=payload.get("created_at"),
+            updated_at=payload.get("updated_at"),
+            due_at=payload.get("due_at"),
+        )
+
 
 class CanvasAnnouncement(BaseModel):
     id: int
@@ -43,6 +133,17 @@ class CanvasAnnouncement(BaseModel):
     posted_at: str | None = None
     course_id: int
 
+    @classmethod
+    def from_api(cls, payload: dict[str, Any], course_id: int) -> "CanvasAnnouncement":
+        return cls(
+            id=payload["id"],
+            title=payload["title"],
+            message=payload.get("message"),
+            html_url=payload["html_url"],
+            posted_at=payload.get("posted_at"),
+            course_id=course_id,
+        )
+
 
 CanvasStage: TypeAlias = Literal["pages", "assignments", "announcements"]
 
@@ -72,3 +173,286 @@ class CanvasConnectorCheckpoint(ConnectorCheckpoint):
         self.current_course_index += 1
         self.stage = "pages"
         self.next_url = None
+
+
+class CanvasConnector(
+    CheckpointedConnectorWithPermSync[CanvasConnectorCheckpoint],
+    SlimConnectorWithPermSync,
+):
+    def __init__(
+        self,
+        canvas_base_url: str,
+        batch_size: int = INDEX_BATCH_SIZE,
+    ) -> None:
+        self.canvas_base_url = canvas_base_url.rstrip("/").removesuffix("/api/v1")
+        self.batch_size = batch_size
+        self._canvas_client: CanvasApiClient | None = None
+        self._course_permissions_cache: dict[int, ExternalAccess | None] = {}
+
+    @property
+    def canvas_client(self) -> CanvasApiClient:
+        if self._canvas_client is None:
+            raise ConnectorMissingCredentialError("Canvas")
+        return self._canvas_client
+
+    def _get_course_permissions(self, course_id: int) -> ExternalAccess | None:
+        """Get course permissions with caching."""
+        if course_id not in self._course_permissions_cache:
+            self._course_permissions_cache[course_id] = get_course_permissions(
+                canvas_client=self.canvas_client,
+                course_id=course_id,
+            )
+        return self._course_permissions_cache[course_id]
+
+    @retry(tries=3, delay=1, backoff=2)
+    def _list_courses(self) -> list[CanvasCourse]:
+        """Fetch all courses accessible to the authenticated user."""
+        logger.debug("Fetching Canvas courses")
+
+        courses: list[CanvasCourse] = []
+        for page in self.canvas_client.paginate(
+            "courses", params={"per_page": "100", "state[]": "available"}
+        ):
+            courses.extend(CanvasCourse.from_api(c) for c in page)
+        return courses
+
+    @retry(tries=3, delay=1, backoff=2)
+    def _list_pages(self, course_id: int) -> list[CanvasPage]:
+        """Fetch all pages for a given course."""
+        logger.debug(f"Fetching pages for course {course_id}")
+
+        pages: list[CanvasPage] = []
+        for page in self.canvas_client.paginate(
+            f"courses/{course_id}/pages",
+            params={"per_page": "100", "include[]": "body", "published": "true"},
+        ):
+            pages.extend(CanvasPage.from_api(p, course_id=course_id) for p in page)
+        return pages
+
+    @retry(tries=3, delay=1, backoff=2)
+    def _list_assignments(self, course_id: int) -> list[CanvasAssignment]:
+        """Fetch all assignments for a given course."""
+        logger.debug(f"Fetching assignments for course {course_id}")
+
+        assignments: list[CanvasAssignment] = []
+        for page in self.canvas_client.paginate(
+            f"courses/{course_id}/assignments",
+            params={"per_page": "100", "published": "true"},
+        ):
+            assignments.extend(
+                CanvasAssignment.from_api(a, course_id=course_id) for a in page
+            )
+        return assignments
+
+    @retry(tries=3, delay=1, backoff=2)
+    def _list_announcements(self, course_id: int) -> list[CanvasAnnouncement]:
+        """Fetch all announcements for a given course."""
+        logger.debug(f"Fetching announcements for course {course_id}")
+
+        announcements: list[CanvasAnnouncement] = []
+        for page in self.canvas_client.paginate(
+            "announcements",
+            params={
+                "per_page": "100",
+                "context_codes[]": f"course_{course_id}",
+                "active_only": "true",
+            },
+        ):
+            announcements.extend(
+                CanvasAnnouncement.from_api(a, course_id=course_id) for a in page
+            )
+        return announcements
+
+    def _build_document(
+        self,
+        doc_id: str,
+        link: str,
+        text: str,
+        semantic_identifier: str,
+        doc_updated_at: datetime | None,
+        course_id: int,
+        doc_type: str,
+    ) -> Document:
+        """Build a Document with standard Canvas fields."""
+        return Document(
+            id=doc_id,
+            sections=cast(
+                list[TextSection | ImageSection],
+                [TextSection(link=link, text=text)],
+            ),
+            source=DocumentSource.CANVAS,
+            semantic_identifier=semantic_identifier,
+            doc_updated_at=doc_updated_at,
+            metadata={"course_id": str(course_id), "type": doc_type},
+        )
+
+    def _convert_page_to_document(self, page: CanvasPage) -> Document:
+        """Convert a Canvas page to a Document."""
+        link = f"{self.canvas_base_url}/courses/{page.course_id}/pages/{page.url}"
+
+        text_parts = [page.title]
+        body_text = parse_html_page_basic(page.body) if page.body else ""
+        if body_text:
+            text_parts.append(body_text)
+
+        doc_updated_at = (
+            datetime.fromisoformat(page.updated_at.replace("Z", "+00:00")).astimezone(
+                timezone.utc
+            )
+            if page.updated_at
+            else None
+        )
+
+        document = self._build_document(
+            doc_id=f"canvas-page-{page.course_id}-{page.page_id}",
+            link=link,
+            text="\n\n".join(text_parts),
+            semantic_identifier=page.title or f"Page {page.page_id}",
+            doc_updated_at=doc_updated_at,
+            course_id=page.course_id,
+            doc_type="page",
+        )
+        return document
+
+    def _convert_assignment_to_document(self, assignment: CanvasAssignment) -> Document:
+        """Convert a Canvas assignment to a Document."""
+        text_parts = [assignment.name]
+        desc_text = (
+            parse_html_page_basic(assignment.description)
+            if assignment.description
+            else ""
+        )
+        if desc_text:
+            text_parts.append(desc_text)
+        if assignment.due_at:
+            due_dt = datetime.fromisoformat(
+                assignment.due_at.replace("Z", "+00:00")
+            ).astimezone(timezone.utc)
+            text_parts.append(f"Due: {due_dt.strftime('%B %d, %Y %H:%M UTC')}")
+
+        doc_updated_at = (
+            datetime.fromisoformat(
+                assignment.updated_at.replace("Z", "+00:00")
+            ).astimezone(timezone.utc)
+            if assignment.updated_at
+            else None
+        )
+
+        document = self._build_document(
+            doc_id=f"canvas-assignment-{assignment.course_id}-{assignment.id}",
+            link=assignment.html_url,
+            text="\n\n".join(text_parts),
+            semantic_identifier=assignment.name or f"Assignment {assignment.id}",
+            doc_updated_at=doc_updated_at,
+            course_id=assignment.course_id,
+            doc_type="assignment",
+        )
+        return document
+
+    def _convert_announcement_to_document(
+        self, announcement: CanvasAnnouncement
+    ) -> Document:
+        """Convert a Canvas announcement to a Document."""
+        text_parts = [announcement.title]
+        msg_text = (
+            parse_html_page_basic(announcement.message) if announcement.message else ""
+        )
+        if msg_text:
+            text_parts.append(msg_text)
+
+        doc_updated_at = (
+            datetime.fromisoformat(
+                announcement.posted_at.replace("Z", "+00:00")
+            ).astimezone(timezone.utc)
+            if announcement.posted_at
+            else None
+        )
+
+        document = self._build_document(
+            doc_id=f"canvas-announcement-{announcement.course_id}-{announcement.id}",
+            link=announcement.html_url,
+            text="\n\n".join(text_parts),
+            semantic_identifier=announcement.title or f"Announcement {announcement.id}",
+            doc_updated_at=doc_updated_at,
+            course_id=announcement.course_id,
+            doc_type="announcement",
+        )
+        return document
+
+    @override
+    def load_credentials(self, credentials: dict[str, Any]) -> dict[str, Any] | None:
+        """Load and validate Canvas credentials."""
+        access_token = credentials.get("canvas_access_token")
+        if not access_token:
+            raise ConnectorMissingCredentialError("Canvas")
+
+        try:
+            client = CanvasApiClient(
+                bearer_token=access_token,
+                canvas_base_url=self.canvas_base_url,
+            )
+            client.get("courses", params={"per_page": "1"})
+        except ValueError as e:
+            raise ConnectorValidationError(f"Invalid Canvas base URL: {e}")
+        except OnyxError as e:
+            _handle_canvas_api_error(e)
+
+        self._canvas_client = client
+        return None
+
+    @override
+    def validate_connector_settings(self) -> None:
+        """Validate Canvas connector settings by testing API access."""
+        try:
+            self.canvas_client.get("courses", params={"per_page": "1"})
+            logger.info("Canvas connector settings validated successfully")
+        except OnyxError as e:
+            _handle_canvas_api_error(e)
+        except ConnectorMissingCredentialError:
+            raise
+        except Exception as exc:
+            raise UnexpectedValidationError(
+                f"Unexpected error during Canvas settings validation: {exc}"
+            )
+
+    @override
+    def load_from_checkpoint(
+        self,
+        start: SecondsSinceUnixEpoch,
+        end: SecondsSinceUnixEpoch,
+        checkpoint: CanvasConnectorCheckpoint,
+    ) -> CheckpointOutput[CanvasConnectorCheckpoint]:
+        # TODO(benwu408): implemented in PR3 (checkpoint)
+        raise NotImplementedError
+
+    @override
+    def load_from_checkpoint_with_perm_sync(
+        self,
+        start: SecondsSinceUnixEpoch,
+        end: SecondsSinceUnixEpoch,
+        checkpoint: CanvasConnectorCheckpoint,
+    ) -> CheckpointOutput[CanvasConnectorCheckpoint]:
+        # TODO(benwu408): implemented in PR3 (checkpoint)
+        raise NotImplementedError
+
+    @override
+    def build_dummy_checkpoint(self) -> CanvasConnectorCheckpoint:
+        # TODO(benwu408): implemented in PR3 (checkpoint)
+        raise NotImplementedError
+
+    @override
+    def validate_checkpoint_json(
+        self, checkpoint_json: str
+    ) -> CanvasConnectorCheckpoint:
+        # TODO(benwu408): implemented in PR3 (checkpoint)
+        raise NotImplementedError
+
+    @override
+    def retrieve_all_slim_docs_perm_sync(
+        self,
+        start: SecondsSinceUnixEpoch | None = None,
+        end: SecondsSinceUnixEpoch | None = None,
+        callback: IndexingHeartbeatInterface | None = None,
+    ) -> GenerateSlimDocumentOutput:
+        # TODO(benwu408): implemented in PR4 (perm sync)
+        raise NotImplementedError

backend/onyx/connectors/confluence/connector.py

@@ -890,8 +890,8 @@ class ConfluenceConnector(
 
     def _retrieve_all_slim_docs(
         self,
-        start: SecondsSinceUnixEpoch | None = None,  # noqa: ARG002
-        end: SecondsSinceUnixEpoch | None = None,  # noqa: ARG002
+        start: SecondsSinceUnixEpoch | None = None,
+        end: SecondsSinceUnixEpoch | None = None,
         callback: IndexingHeartbeatInterface | None = None,
         include_permissions: bool = True,
     ) -> GenerateSlimDocumentOutput:
@@ -915,8 +915,8 @@ class ConfluenceConnector(
                 self.confluence_client, doc_id, restrictions, ancestors
             ) or space_level_access_info.get(page_space_key)
 
-        # Query pages
-        page_query = self.base_cql_page_query + self.cql_label_filter
+        # Query pages (with optional time filtering for indexing_start)
+        page_query = self._construct_page_cql_query(start, end)
         for page in self.confluence_client.cql_paginate_all_expansions(
             cql=page_query,
             expand=restrictions_expand,
@@ -950,7 +950,9 @@ class ConfluenceConnector(
 
             # Query attachments for each page
             page_hierarchy_node_yielded = False
-            attachment_query = self._construct_attachment_query(_get_page_id(page))
+            attachment_query = self._construct_attachment_query(
+                _get_page_id(page), start, end
+            )
             for attachment in self.confluence_client.cql_paginate_all_expansions(
                 cql=attachment_query,
                 expand=restrictions_expand,

backend/onyx/connectors/discord/connector.py

@@ -11,11 +11,13 @@ from discord import Client
 from discord.channel import TextChannel
 from discord.channel import Thread
 from discord.enums import MessageType
+from discord.errors import LoginFailure
 from discord.flags import Intents
 from discord.message import Message as DiscordMessage
 
 from onyx.configs.app_configs import INDEX_BATCH_SIZE
 from onyx.configs.constants import DocumentSource
+from onyx.connectors.exceptions import CredentialInvalidError
 from onyx.connectors.interfaces import GenerateDocumentsOutput
 from onyx.connectors.interfaces import LoadConnector
 from onyx.connectors.interfaces import PollConnector
@@ -209,8 +211,19 @@ def _manage_async_retrieval(
         intents = Intents.default()
         intents.message_content = True
         async with Client(intents=intents) as discord_client:
-            asyncio.create_task(discord_client.start(token))
-            await discord_client.wait_until_ready()
+            start_task = asyncio.create_task(discord_client.start(token))
+            ready_task = asyncio.create_task(discord_client.wait_until_ready())
+
+            done, _ = await asyncio.wait(
+                {start_task, ready_task},
+                return_when=asyncio.FIRST_COMPLETED,
+            )
+
+            # start() runs indefinitely once connected, so it only lands
+            # in `done` when login/connection failed — propagate the error.
+            if start_task in done:
+                ready_task.cancel()
+                start_task.result()
 
             filtered_channels: list[TextChannel] = await _fetch_filtered_channels(
                 discord_client=discord_client,
@@ -276,6 +289,19 @@ class DiscordConnector(PollConnector, LoadConnector):
         self._discord_bot_token = credentials["discord_bot_token"]
         return None
 
+    def validate_connector_settings(self) -> None:
+        loop = asyncio.new_event_loop()
+        try:
+            client = Client(intents=Intents.default())
+            try:
+                loop.run_until_complete(client.login(self.discord_bot_token))
+            except LoginFailure as e:
+                raise CredentialInvalidError(f"Invalid Discord bot token: {e}")
+            finally:
+                loop.run_until_complete(client.close())
+        finally:
+            loop.close()
+
     def _manage_doc_batching(
         self,
         start: datetime | None = None,

backend/onyx/connectors/google_drive/connector.py

@@ -8,7 +8,6 @@ from collections.abc import Generator
 from collections.abc import Iterator
 from datetime import datetime
 from enum import Enum
-from functools import partial
 from typing import Any
 from typing import cast
 from typing import Protocol
@@ -43,6 +42,9 @@ from onyx.connectors.google_drive.file_retrieval import (
     get_all_files_in_my_drive_and_shared,
 )
 from onyx.connectors.google_drive.file_retrieval import get_external_access_for_folder
+from onyx.connectors.google_drive.file_retrieval import (
+    get_files_by_web_view_links_batch,
+)
 from onyx.connectors.google_drive.file_retrieval import get_files_in_shared_drive
 from onyx.connectors.google_drive.file_retrieval import get_folder_metadata
 from onyx.connectors.google_drive.file_retrieval import get_root_folder_id
@@ -71,11 +73,13 @@ from onyx.connectors.interfaces import CheckpointedConnectorWithPermSync
 from onyx.connectors.interfaces import CheckpointOutput
 from onyx.connectors.interfaces import GenerateSlimDocumentOutput
 from onyx.connectors.interfaces import NormalizationResult
+from onyx.connectors.interfaces import Resolver
 from onyx.connectors.interfaces import SecondsSinceUnixEpoch
 from onyx.connectors.interfaces import SlimConnectorWithPermSync
 from onyx.connectors.models import ConnectorFailure
 from onyx.connectors.models import ConnectorMissingCredentialError
 from onyx.connectors.models import Document
+from onyx.connectors.models import DocumentFailure
 from onyx.connectors.models import EntityFailure
 from onyx.connectors.models import HierarchyNode
 from onyx.connectors.models import SlimDocument
@@ -203,7 +207,9 @@ class DriveIdStatus(Enum):
 
 
 class GoogleDriveConnector(
-    SlimConnectorWithPermSync, CheckpointedConnectorWithPermSync[GoogleDriveCheckpoint]
+    SlimConnectorWithPermSync,
+    CheckpointedConnectorWithPermSync[GoogleDriveCheckpoint],
+    Resolver,
 ):
     def __init__(
         self,
@@ -1487,134 +1493,113 @@ class GoogleDriveConnector(
             end=end,
         )
 
-    def _extract_docs_from_google_drive(
+    def _convert_retrieved_files_to_documents(
         self,
+        drive_files_iter: Iterator[RetrievedDriveFile],
         checkpoint: GoogleDriveCheckpoint,
-        start: SecondsSinceUnixEpoch | None,
-        end: SecondsSinceUnixEpoch | None,
         include_permissions: bool,
     ) -> Iterator[Document | ConnectorFailure | HierarchyNode]:
         """
-        Retrieves and converts Google Drive files to documents.
-        Also yields HierarchyNode objects for ancestor folders.
+        Converts retrieved files to documents, yielding HierarchyNode
+        objects for ancestor folders before the converted documents.
         """
-        field_type = (
-            DriveFileFieldType.WITH_PERMISSIONS
-            if include_permissions or self.exclude_domain_link_only
-            else DriveFileFieldType.STANDARD
+        permission_sync_context = (
+            PermissionSyncContext(
+                primary_admin_email=self.primary_admin_email,
+                google_domain=self.google_domain,
+            )
+            if include_permissions
+            else None
         )
 
-        try:
-            # Build permission sync context if needed
-            permission_sync_context = (
-                PermissionSyncContext(
-                    primary_admin_email=self.primary_admin_email,
-                    google_domain=self.google_domain,
-                )
-                if include_permissions
-                else None
+        files_batch: list[RetrievedDriveFile] = []
+        for retrieved_file in drive_files_iter:
+            if self.exclude_domain_link_only and has_link_only_permission(
+                retrieved_file.drive_file
+            ):
+                continue
+            if retrieved_file.error is None:
+                files_batch.append(retrieved_file)
+                continue
+
+            failure_stage = retrieved_file.completion_stage.value
+            failure_message = f"retrieval failure during stage: {failure_stage},"
+            failure_message += f"user: {retrieved_file.user_email},"
+            failure_message += f"parent drive/folder: {retrieved_file.parent_id},"
+            failure_message += f"error: {retrieved_file.error}"
+            logger.error(failure_message)
+            yield ConnectorFailure(
+                failed_entity=EntityFailure(
+                    entity_id=retrieved_file.drive_file.get("id", failure_stage),
+                ),
+                failure_message=failure_message,
+                exception=retrieved_file.error,
             )
 
-            # Prepare a partial function with the credentials and admin email
-            convert_func = partial(
-                convert_drive_item_to_document,
+        new_ancestors = self._get_new_ancestors_for_files(
+            files=files_batch,
+            seen_hierarchy_node_raw_ids=checkpoint.seen_hierarchy_node_raw_ids,
+            fully_walked_hierarchy_node_raw_ids=checkpoint.fully_walked_hierarchy_node_raw_ids,
+            permission_sync_context=permission_sync_context,
+            add_prefix=True,
+        )
+        if new_ancestors:
+            logger.debug(f"Yielding {len(new_ancestors)} new hierarchy nodes")
+            yield from new_ancestors
+
+        func_with_args = [
+            (
+                self._convert_retrieved_file_to_document,
+                (retrieved_file, permission_sync_context),
+            )
+            for retrieved_file in files_batch
+        ]
+        raw_results = cast(
+            list[Document | ConnectorFailure | None],
+            run_functions_tuples_in_parallel(func_with_args, max_workers=8),
+        )
+
+        results: list[Document | ConnectorFailure] = [
+            r for r in raw_results if r is not None
+        ]
+        logger.debug(f"batch has {len(results)} docs or failures")
+        yield from results
+
+        checkpoint.retrieved_folder_and_drive_ids = self._retrieved_folder_and_drive_ids
+
+    def _convert_retrieved_file_to_document(
+        self,
+        retrieved_file: RetrievedDriveFile,
+        permission_sync_context: PermissionSyncContext | None,
+    ) -> Document | ConnectorFailure | None:
+        """
+        Converts a single retrieved file to a document.
+        """
+        try:
+            return convert_drive_item_to_document(
                 self.creds,
                 self.allow_images,
                 self.size_threshold,
                 permission_sync_context,
+                [retrieved_file.user_email, self.primary_admin_email]
+                + get_file_owners(retrieved_file.drive_file, self.primary_admin_email),
+                retrieved_file.drive_file,
             )
-            # Fetch files in batches
-            batches_complete = 0
-            files_batch: list[RetrievedDriveFile] = []
-
-            def _yield_batch(
-                files_batch: list[RetrievedDriveFile],
-            ) -> Iterator[Document | ConnectorFailure | HierarchyNode]:
-                nonlocal batches_complete
-
-                # First, yield any new ancestor hierarchy nodes
-                new_ancestors = self._get_new_ancestors_for_files(
-                    files=files_batch,
-                    seen_hierarchy_node_raw_ids=checkpoint.seen_hierarchy_node_raw_ids,
-                    fully_walked_hierarchy_node_raw_ids=checkpoint.fully_walked_hierarchy_node_raw_ids,
-                    permission_sync_context=permission_sync_context,
-                    add_prefix=True,  # Indexing path - prefix here
-                )
-                if new_ancestors:
-                    logger.debug(
-                        f"Yielding {len(new_ancestors)} new hierarchy nodes for batch {batches_complete}"
-                    )
-                    yield from new_ancestors
-
-                # Process the batch using run_functions_tuples_in_parallel
-                func_with_args = [
-                    (
-                        convert_func,
-                        (
-                            [file.user_email, self.primary_admin_email]
-                            + get_file_owners(
-                                file.drive_file, self.primary_admin_email
-                            ),
-                            file.drive_file,
-                        ),
-                    )
-                    for file in files_batch
-                ]
-                results = cast(
-                    list[Document | ConnectorFailure | None],
-                    run_functions_tuples_in_parallel(func_with_args, max_workers=8),
-                )
-                logger.debug(
-                    f"finished processing batch {batches_complete} with {len(results)} results"
-                )
-
-                docs_and_failures = [result for result in results if result is not None]
-                logger.debug(
-                    f"batch {batches_complete} has {len(docs_and_failures)} docs or failures"
-                )
-
-                if docs_and_failures:
-                    yield from docs_and_failures
-                    batches_complete += 1
-                logger.debug(f"finished yielding batch {batches_complete}")
-
-            for retrieved_file in self._fetch_drive_items(
-                field_type=field_type,
-                checkpoint=checkpoint,
-                start=start,
-                end=end,
-            ):
-                if self.exclude_domain_link_only and has_link_only_permission(
-                    retrieved_file.drive_file
-                ):
-                    continue
-                if retrieved_file.error is None:
-                    files_batch.append(retrieved_file)
-                    continue
-
-                # handle retrieval errors
-                failure_stage = retrieved_file.completion_stage.value
-                failure_message = f"retrieval failure during stage: {failure_stage},"
-                failure_message += f"user: {retrieved_file.user_email},"
-                failure_message += f"parent drive/folder: {retrieved_file.parent_id},"
-                failure_message += f"error: {retrieved_file.error}"
-                logger.error(failure_message)
-                yield ConnectorFailure(
-                    failed_entity=EntityFailure(
-                        entity_id=failure_stage,
-                    ),
-                    failure_message=failure_message,
-                    exception=retrieved_file.error,
-                )
-
-            yield from _yield_batch(files_batch)
-            checkpoint.retrieved_folder_and_drive_ids = (
-                self._retrieved_folder_and_drive_ids
-            )
-
         except Exception as e:
-            logger.exception(f"Error extracting documents from Google Drive: {e}")
-            raise e
+            logger.exception(
+                f"Error extracting document: "
+                f"{retrieved_file.drive_file.get('name')} from Google Drive"
+            )
+            return ConnectorFailure(
+                failed_entity=EntityFailure(
+                    entity_id=retrieved_file.drive_file.get("id", "unknown"),
+                ),
+                failure_message=(
+                    f"Error extracting document: "
+                    f"{retrieved_file.drive_file.get('name')}"
+                ),
+                exception=e,
+            )
 
     def _load_from_checkpoint(
         self,
@@ -1638,8 +1623,19 @@ class GoogleDriveConnector(
         checkpoint = copy.deepcopy(checkpoint)
         self._retrieved_folder_and_drive_ids = checkpoint.retrieved_folder_and_drive_ids
         try:
-            yield from self._extract_docs_from_google_drive(
-                checkpoint, start, end, include_permissions
+            field_type = (
+                DriveFileFieldType.WITH_PERMISSIONS
+                if include_permissions or self.exclude_domain_link_only
+                else DriveFileFieldType.STANDARD
+            )
+            drive_files_iter = self._fetch_drive_items(
+                field_type=field_type,
+                checkpoint=checkpoint,
+                start=start,
+                end=end,
+            )
+            yield from self._convert_retrieved_files_to_documents(
+                drive_files_iter, checkpoint, include_permissions
             )
         except Exception as e:
             if MISSING_SCOPES_ERROR_STR in str(e):
@@ -1676,6 +1672,82 @@ class GoogleDriveConnector(
             start, end, checkpoint, include_permissions=True
         )
 
+    @override
+    def resolve_errors(
+        self,
+        errors: list[ConnectorFailure],
+        include_permissions: bool = False,
+    ) -> Generator[Document | ConnectorFailure | HierarchyNode, None, None]:
+        if self._creds is None or self._primary_admin_email is None:
+            raise RuntimeError(
+                "Credentials missing, should not call this method before calling load_credentials"
+            )
+
+        logger.info(f"Resolving {len(errors)} errors")
+        doc_ids = [
+            failure.failed_document.document_id
+            for failure in errors
+            if failure.failed_document
+        ]
+        service = get_drive_service(self.creds, self.primary_admin_email)
+        field_type = (
+            DriveFileFieldType.WITH_PERMISSIONS
+            if include_permissions or self.exclude_domain_link_only
+            else DriveFileFieldType.STANDARD
+        )
+        batch_result = get_files_by_web_view_links_batch(service, doc_ids, field_type)
+
+        for doc_id, error in batch_result.errors.items():
+            yield ConnectorFailure(
+                failed_document=DocumentFailure(
+                    document_id=doc_id,
+                    document_link=doc_id,
+                ),
+                failure_message=f"Failed to retrieve file during error resolution: {error}",
+                exception=error,
+            )
+
+        permission_sync_context = (
+            PermissionSyncContext(
+                primary_admin_email=self.primary_admin_email,
+                google_domain=self.google_domain,
+            )
+            if include_permissions
+            else None
+        )
+
+        retrieved_files = [
+            RetrievedDriveFile(
+                drive_file=file,
+                user_email=self.primary_admin_email,
+                completion_stage=DriveRetrievalStage.DONE,
+            )
+            for file in batch_result.files.values()
+        ]
+
+        yield from self._get_new_ancestors_for_files(
+            files=retrieved_files,
+            seen_hierarchy_node_raw_ids=ThreadSafeSet(),
+            fully_walked_hierarchy_node_raw_ids=ThreadSafeSet(),
+            permission_sync_context=permission_sync_context,
+            add_prefix=True,
+        )
+
+        func_with_args = [
+            (
+                self._convert_retrieved_file_to_document,
+                (rf, permission_sync_context),
+            )
+            for rf in retrieved_files
+        ]
+        results = cast(
+            list[Document | ConnectorFailure | None],
+            run_functions_tuples_in_parallel(func_with_args, max_workers=8),
+        )
+        for result in results:
+            if result is not None:
+                yield result
+
     def _extract_slim_docs_from_google_drive(
         self,
         checkpoint: GoogleDriveCheckpoint,

backend/onyx/connectors/google_drive/file_retrieval.py

@@ -4,9 +4,12 @@ from datetime import datetime
 from datetime import timezone
 from enum import Enum
 from typing import cast
+from urllib.parse import parse_qs
+from urllib.parse import urlparse
 
 from googleapiclient.discovery import Resource  # type: ignore
 from googleapiclient.errors import HttpError  # type: ignore
+from googleapiclient.http import BatchHttpRequest  # type: ignore
 
 from onyx.access.models import ExternalAccess
 from onyx.connectors.google_drive.constants import DRIVE_FOLDER_TYPE
@@ -58,6 +61,8 @@ SLIM_FILE_FIELDS = (
 )
 FOLDER_FIELDS = "nextPageToken, files(id, name, permissions, modifiedTime, webViewLink, shortcutDetails)"
 
+MAX_BATCH_SIZE = 100
+
 HIERARCHY_FIELDS = "id, name, parents, webViewLink, mimeType, driveId"
 
 HIERARCHY_FIELDS_WITH_PERMISSIONS = (
@@ -214,7 +219,7 @@ def get_external_access_for_folder(
 
 
 def _get_fields_for_file_type(field_type: DriveFileFieldType) -> str:
-    """Get the appropriate fields string based on the field type enum"""
+    """Get the appropriate fields string for files().list() based on the field type enum."""
     if field_type == DriveFileFieldType.SLIM:
         return SLIM_FILE_FIELDS
     elif field_type == DriveFileFieldType.WITH_PERMISSIONS:
@@ -223,6 +228,25 @@ def _get_fields_for_file_type(field_type: DriveFileFieldType) -> str:
         return FILE_FIELDS
 
 
+def _extract_single_file_fields(list_fields: str) -> str:
+    """Convert a files().list() fields string to one suitable for files().get().
+
+    List fields look like "nextPageToken, files(field1, field2, ...)"
+    Single-file fields should be just "field1, field2, ..."
+    """
+    start = list_fields.find("files(")
+    if start == -1:
+        return list_fields
+    inner_start = start + len("files(")
+    inner_end = list_fields.rfind(")")
+    return list_fields[inner_start:inner_end]
+
+
+def _get_single_file_fields(field_type: DriveFileFieldType) -> str:
+    """Get the appropriate fields string for files().get() based on the field type enum."""
+    return _extract_single_file_fields(_get_fields_for_file_type(field_type))
+
+
 def _get_files_in_parent(
     service: Resource,
     parent_id: str,
@@ -496,3 +520,112 @@ def get_root_folder_id(service: Resource) -> str:
         .get(fileId="root", fields=GoogleFields.ID.value)
         .execute()[GoogleFields.ID.value]
     )
+
+
+def _extract_file_id_from_web_view_link(web_view_link: str) -> str:
+    parsed = urlparse(web_view_link)
+    path_parts = [part for part in parsed.path.split("/") if part]
+
+    if "d" in path_parts:
+        idx = path_parts.index("d")
+        if idx + 1 < len(path_parts):
+            return path_parts[idx + 1]
+
+    query_params = parse_qs(parsed.query)
+    for key in ("id", "fileId"):
+        value = query_params.get(key)
+        if value and value[0]:
+            return value[0]
+
+    raise ValueError(
+        f"Unable to extract Drive file id from webViewLink: {web_view_link}"
+    )
+
+
+def get_file_by_web_view_link(
+    service: GoogleDriveService,
+    web_view_link: str,
+    fields: str,
+) -> GoogleDriveFileType:
+    """Retrieve a Google Drive file using its webViewLink."""
+    file_id = _extract_file_id_from_web_view_link(web_view_link)
+    return (
+        service.files()
+        .get(
+            fileId=file_id,
+            supportsAllDrives=True,
+            fields=fields,
+        )
+        .execute()
+    )
+
+
+class BatchRetrievalResult:
+    """Result of a batch file retrieval, separating successes from errors."""
+
+    def __init__(self) -> None:
+        self.files: dict[str, GoogleDriveFileType] = {}
+        self.errors: dict[str, Exception] = {}
+
+
+def get_files_by_web_view_links_batch(
+    service: GoogleDriveService,
+    web_view_links: list[str],
+    field_type: DriveFileFieldType,
+) -> BatchRetrievalResult:
+    """Retrieve multiple Google Drive files by webViewLink using the batch API.
+
+    Returns a BatchRetrievalResult containing successful file retrievals
+    and errors for any files that could not be fetched.
+    Automatically splits into chunks of MAX_BATCH_SIZE.
+    """
+    fields = _get_single_file_fields(field_type)
+    if len(web_view_links) <= MAX_BATCH_SIZE:
+        return _get_files_by_web_view_links_batch(service, web_view_links, fields)
+
+    combined = BatchRetrievalResult()
+    for i in range(0, len(web_view_links), MAX_BATCH_SIZE):
+        chunk = web_view_links[i : i + MAX_BATCH_SIZE]
+        chunk_result = _get_files_by_web_view_links_batch(service, chunk, fields)
+        combined.files.update(chunk_result.files)
+        combined.errors.update(chunk_result.errors)
+    return combined
+
+
+def _get_files_by_web_view_links_batch(
+    service: GoogleDriveService,
+    web_view_links: list[str],
+    fields: str,
+) -> BatchRetrievalResult:
+    """Single-batch implementation."""
+
+    result = BatchRetrievalResult()
+
+    def callback(
+        request_id: str,
+        response: GoogleDriveFileType,
+        exception: Exception | None,
+    ) -> None:
+        if exception:
+            logger.warning(f"Error retrieving file {request_id}: {exception}")
+            result.errors[request_id] = exception
+        else:
+            result.files[request_id] = response
+
+    batch = cast(BatchHttpRequest, service.new_batch_http_request(callback=callback))
+
+    for web_view_link in web_view_links:
+        try:
+            file_id = _extract_file_id_from_web_view_link(web_view_link)
+            request = service.files().get(
+                fileId=file_id,
+                supportsAllDrives=True,
+                fields=fields,
+            )
+            batch.add(request, request_id=web_view_link)
+        except ValueError as e:
+            logger.warning(f"Failed to extract file ID from {web_view_link}: {e}")
+            result.errors[web_view_link] = e
+
+    batch.execute()
+    return result

backend/onyx/connectors/interfaces.py

@@ -298,6 +298,22 @@ class CheckpointedConnectorWithPermSync(CheckpointedConnector[CT]):
         raise NotImplementedError
 
 
+class Resolver(BaseConnector):
+    @abc.abstractmethod
+    def resolve_errors(
+        self,
+        errors: list[ConnectorFailure],
+        include_permissions: bool = False,
+    ) -> Generator[Document | ConnectorFailure | HierarchyNode, None, None]:
+        """Attempts to yield back ALL the documents described by the errors, no checkpointing.
+
+        Caller's responsibility is to delete the old ConnectorFailures and replace with the new ones.
+        If include_permissions is True, the documents will have permissions synced.
+        May also yield HierarchyNode objects for ancestor folders of resolved documents.
+        """
+        raise NotImplementedError
+
+
 class HierarchyConnector(BaseConnector):
     @abc.abstractmethod
     def load_hierarchy(

backend/onyx/connectors/registry.py

@@ -72,6 +72,10 @@ CONNECTOR_CLASS_MAP = {
         module_path="onyx.connectors.coda.connector",
         class_name="CodaConnector",
     ),
+    DocumentSource.CANVAS: ConnectorMapping(
+        module_path="onyx.connectors.canvas.connector",
+        class_name="CanvasConnector",
+    ),
     DocumentSource.NOTION: ConnectorMapping(
         module_path="onyx.connectors.notion.connector",
         class_name="NotionConnector",

backend/onyx/connectors/sharepoint/connector.py

@@ -1765,7 +1765,11 @@ class SharepointConnector(
         checkpoint.current_drive_delta_next_link = None
         checkpoint.seen_document_ids.clear()
 
-    def _fetch_slim_documents_from_sharepoint(self) -> GenerateSlimDocumentOutput:
+    def _fetch_slim_documents_from_sharepoint(
+        self,
+        start: datetime | None = None,
+        end: datetime | None = None,
+    ) -> GenerateSlimDocumentOutput:
         site_descriptors = self._filter_excluded_sites(
             self.site_descriptors or self.fetch_sites()
         )
@@ -1786,7 +1790,9 @@ class SharepointConnector(
             # Process site documents if flag is True
             if self.include_site_documents:
                 for driveitem, drive_name, drive_web_url in self._fetch_driveitems(
-                    site_descriptor=site_descriptor
+                    site_descriptor=site_descriptor,
+                    start=start,
+                    end=end,
                 ):
                     if self._is_driveitem_excluded(driveitem):
                         logger.debug(f"Excluding by path denylist: {driveitem.web_url}")
@@ -1841,7 +1847,9 @@ class SharepointConnector(
 
             # Process site pages if flag is True
             if self.include_site_pages:
-                site_pages = self._fetch_site_pages(site_descriptor)
+                site_pages = self._fetch_site_pages(
+                    site_descriptor, start=start, end=end
+                )
                 for site_page in site_pages:
                     logger.debug(
                         f"Processing site page: {site_page.get('webUrl', site_page.get('name', 'Unknown'))}"
@@ -2565,12 +2573,22 @@ class SharepointConnector(
 
     def retrieve_all_slim_docs_perm_sync(
         self,
-        start: SecondsSinceUnixEpoch | None = None,  # noqa: ARG002
-        end: SecondsSinceUnixEpoch | None = None,  # noqa: ARG002
+        start: SecondsSinceUnixEpoch | None = None,
+        end: SecondsSinceUnixEpoch | None = None,
         callback: IndexingHeartbeatInterface | None = None,  # noqa: ARG002
     ) -> GenerateSlimDocumentOutput:
-
-        yield from self._fetch_slim_documents_from_sharepoint()
+        start_dt = (
+            datetime.fromtimestamp(start, tz=timezone.utc)
+            if start is not None
+            else None
+        )
+        end_dt = (
+            datetime.fromtimestamp(end, tz=timezone.utc) if end is not None else None
+        )
+        yield from self._fetch_slim_documents_from_sharepoint(
+            start=start_dt,
+            end=end_dt,
+        )
 
 
 if __name__ == "__main__":

backend/onyx/connectors/slack/connector.py

@@ -516,6 +516,8 @@ def _get_all_doc_ids(
     ] = default_msg_filter,
     callback: IndexingHeartbeatInterface | None = None,
     workspace_url: str | None = None,
+    start: SecondsSinceUnixEpoch | None = None,
+    end: SecondsSinceUnixEpoch | None = None,
 ) -> GenerateSlimDocumentOutput:
     """
     Get all document ids in the workspace, channel by channel
@@ -546,6 +548,8 @@ def _get_all_doc_ids(
             client=client,
             channel=channel,
             callback=callback,
+            oldest=str(start) if start else None,  # 0.0 -> None intentionally
+            latest=str(end) if end is not None else None,
         )
 
         for message_batch in channel_message_batches:
@@ -847,8 +851,8 @@ class SlackConnector(
 
     def retrieve_all_slim_docs_perm_sync(
         self,
-        start: SecondsSinceUnixEpoch | None = None,  # noqa: ARG002
-        end: SecondsSinceUnixEpoch | None = None,  # noqa: ARG002
+        start: SecondsSinceUnixEpoch | None = None,
+        end: SecondsSinceUnixEpoch | None = None,
         callback: IndexingHeartbeatInterface | None = None,
     ) -> GenerateSlimDocumentOutput:
         if self.client is None:
@@ -861,6 +865,8 @@ class SlackConnector(
             msg_filter_func=self.msg_filter_func,
             callback=callback,
             workspace_url=self._workspace_url,
+            start=start,
+            end=end,
         )
 
     def _load_from_checkpoint(

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/enums.py

@@ -215,6 +215,7 @@ class UserFileStatus(str, PyEnum):
     PROCESSING = "PROCESSING"
     INDEXING = "INDEXING"
     COMPLETED = "COMPLETED"
+    SKIPPED = "SKIPPED"
     FAILED = "FAILED"
     CANCELED = "CANCELED"
     DELETING = "DELETING"

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
@@ -46,6 +47,7 @@ async def fetch_user_for_pat(
             (PersonalAccessToken.expires_at.is_(None))
             | (PersonalAccessToken.expires_at > now)
         )
+        .options(selectinload(User.memories))
     )
     if not user:
         return None

backend/onyx/db/projects.py

@@ -7,6 +7,7 @@ from fastapi import HTTPException
 from fastapi import UploadFile
 from pydantic import BaseModel
 from pydantic import ConfigDict
+from pydantic import Field
 from sqlalchemy import func
 from sqlalchemy.orm import Session
 from starlette.background import BackgroundTasks
@@ -17,6 +18,7 @@ from onyx.configs.constants import FileOrigin
 from onyx.configs.constants import OnyxCeleryPriority
 from onyx.configs.constants import OnyxCeleryQueues
 from onyx.configs.constants import OnyxCeleryTask
+from onyx.db.enums import UserFileStatus
 from onyx.db.models import Project__UserFile
 from onyx.db.models import User
 from onyx.db.models import UserFile
@@ -34,9 +36,19 @@ class CategorizedFilesResult(BaseModel):
     user_files: list[UserFile]
     rejected_files: list[RejectedFile]
     id_to_temp_id: dict[str, str]
+    # Filenames that should be stored but not indexed.
+    skip_indexing_filenames: set[str] = Field(default_factory=set)
     # Allow SQLAlchemy ORM models inside this result container
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
+    @property
+    def indexable_files(self) -> list[UserFile]:
+        return [
+            uf
+            for uf in self.user_files
+            if (uf.name or "") not in self.skip_indexing_filenames
+        ]
+
 
 def build_hashed_file_key(file: UploadFile) -> str:
     name_prefix = (file.filename or "")[:50]
@@ -70,6 +82,7 @@ def create_user_files(
         )
         if new_temp_id is not None:
             id_to_temp_id[str(new_id)] = new_temp_id
+        should_skip = (file.filename or "") in categorized_files.skip_indexing
         new_file = UserFile(
             id=new_id,
             user_id=user.id,
@@ -81,6 +94,7 @@ def create_user_files(
             link_url=link_url,
             content_type=file.content_type,
             file_type=file.content_type,
+            status=UserFileStatus.SKIPPED if should_skip else UserFileStatus.PROCESSING,
             last_accessed_at=datetime.datetime.now(datetime.timezone.utc),
         )
         # Persist the UserFile first to satisfy FK constraints for association table
@@ -98,6 +112,7 @@ def create_user_files(
         user_files=user_files,
         rejected_files=rejected_files,
         id_to_temp_id=id_to_temp_id,
+        skip_indexing_filenames=categorized_files.skip_indexing,
     )
 
 
@@ -123,6 +138,7 @@ def upload_files_to_user_files_with_indexing(
     user_files = categorized_files_result.user_files
     rejected_files = categorized_files_result.rejected_files
     id_to_temp_id = categorized_files_result.id_to_temp_id
+    indexable_files = categorized_files_result.indexable_files
     # Trigger per-file processing immediately for the current tenant
     tenant_id = get_current_tenant_id()
     for rejected_file in rejected_files:
@@ -134,12 +150,12 @@ def upload_files_to_user_files_with_indexing(
         from onyx.background.task_utils import drain_processing_loop
 
         background_tasks.add_task(drain_processing_loop, tenant_id)
-        for user_file in user_files:
+        for user_file in indexable_files:
             logger.info(f"Queued in-process processing for user_file_id={user_file.id}")
     else:
         from onyx.background.celery.versioned_apps.client import app as client_app
 
-        for user_file in user_files:
+        for user_file in indexable_files:
             task = client_app.send_task(
                 OnyxCeleryTask.PROCESS_SINGLE_USER_FILE,
                 kwargs={"user_file_id": user_file.id, "tenant_id": tenant_id},
@@ -155,6 +171,7 @@ def upload_files_to_user_files_with_indexing(
         user_files=user_files,
         rejected_files=rejected_files,
         id_to_temp_id=id_to_temp_id,
+        skip_indexing_filenames=categorized_files_result.skip_indexing_filenames,
     )
 
 

backend/onyx/db/user_preferences.py

@@ -229,9 +229,7 @@ def get_memories_for_user(
     user_id: UUID,
     db_session: Session,
 ) -> Sequence[Memory]:
-    return db_session.scalars(
-        select(Memory).where(Memory.user_id == user_id).order_by(Memory.id.desc())
-    ).all()
+    return db_session.scalars(select(Memory).where(Memory.user_id == user_id)).all()
 
 
 def update_user_pinned_assistants(

backend/onyx/document_index/disabled.py

@@ -5,6 +5,7 @@ accidentally reaches the vector DB layer will fail loudly instead of timing
 out against a nonexistent Vespa/OpenSearch instance.
 """
 
+from collections.abc import Iterable
 from typing import Any
 
 from onyx.context.search.models import IndexFilters
@@ -66,7 +67,7 @@ class DisabledDocumentIndex(DocumentIndex):
     # ------------------------------------------------------------------
     def index(
         self,
-        chunks: list[DocMetadataAwareIndexChunk],  # noqa: ARG002
+        chunks: Iterable[DocMetadataAwareIndexChunk],  # noqa: ARG002
         index_batch_params: IndexBatchParams,  # noqa: ARG002
     ) -> set[DocumentInsertionRecord]:
         raise RuntimeError(VECTOR_DB_DISABLED_ERROR)

backend/onyx/document_index/interfaces.py

@@ -1,4 +1,5 @@
 import abc
+from collections.abc import Iterable
 from dataclasses import dataclass
 from datetime import datetime
 from typing import Any
@@ -206,7 +207,7 @@ class Indexable(abc.ABC):
     @abc.abstractmethod
     def index(
         self,
-        chunks: list[DocMetadataAwareIndexChunk],
+        chunks: Iterable[DocMetadataAwareIndexChunk],
         index_batch_params: IndexBatchParams,
     ) -> set[DocumentInsertionRecord]:
         """
@@ -226,8 +227,8 @@ class Indexable(abc.ABC):
         it is done automatically outside of this code.
 
         Parameters:
-        - chunks: Document chunks with all of the information needed for indexing to the document
-                index.
+        - chunks: Document chunks with all of the information needed for
+                indexing to the document index.
         - tenant_id: The tenant id of the user whose chunks are being indexed
         - large_chunks_enabled: Whether large chunks are enabled
 

backend/onyx/document_index/interfaces_new.py

@@ -1,4 +1,5 @@
 import abc
+from collections.abc import Iterable
 from typing import Self
 
 from pydantic import BaseModel
@@ -209,10 +210,10 @@ class Indexable(abc.ABC):
     @abc.abstractmethod
     def index(
         self,
-        chunks: list[DocMetadataAwareIndexChunk],
+        chunks: Iterable[DocMetadataAwareIndexChunk],
         indexing_metadata: IndexingMetadata,
     ) -> list[DocumentInsertionRecord]:
-        """Indexes a list of document chunks into the document index.
+        """Indexes an iterable of document chunks into the document index.
 
         This is often a batch operation including chunks from multiple
         documents.

backend/onyx/document_index/opensearch/client.py

@@ -932,7 +932,7 @@ class OpenSearchIndexClient(OpenSearchClient):
     def search_for_document_ids(
         self,
         body: dict[str, Any],
-        search_type: OpenSearchSearchType = OpenSearchSearchType.DOCUMENT_IDS,
+        search_type: OpenSearchSearchType = OpenSearchSearchType.UNKNOWN,
     ) -> list[str]:
         """Searches the index and returns only document chunk IDs.
 

backend/onyx/document_index/opensearch/constants.py

@@ -60,8 +60,7 @@ class OpenSearchSearchType(str, Enum):
     KEYWORD = "keyword"
     SEMANTIC = "semantic"
     RANDOM = "random"
-    ID_RETRIEVAL = "id_retrieval"
-    DOCUMENT_IDS = "document_ids"
+    DOC_ID_RETRIEVAL = "doc_id_retrieval"
     UNKNOWN = "unknown"
 
 

backend/onyx/document_index/opensearch/opensearch_document_index.py

@@ -1,11 +1,12 @@
 import json
-from collections import defaultdict
+from collections.abc import Iterable
 from typing import Any
 
 import httpx
 from opensearchpy import NotFoundError
 
 from onyx.access.models import DocumentAccess
+from onyx.configs.app_configs import MAX_CHUNKS_PER_DOC_BATCH
 from onyx.configs.app_configs import VERIFY_CREATE_OPENSEARCH_INDEX_ON_INIT_MT
 from onyx.configs.chat_configs import NUM_RETURNED_HITS
 from onyx.configs.chat_configs import TITLE_CONTENT_RATIO
@@ -351,7 +352,7 @@ class OpenSearchOldDocumentIndex(OldDocumentIndex):
 
     def index(
         self,
-        chunks: list[DocMetadataAwareIndexChunk],
+        chunks: Iterable[DocMetadataAwareIndexChunk],
         index_batch_params: IndexBatchParams,
     ) -> set[OldDocumentInsertionRecord]:
         """
@@ -647,10 +648,10 @@ class OpenSearchDocumentIndex(DocumentIndex):
 
     def index(
         self,
-        chunks: list[DocMetadataAwareIndexChunk],
-        indexing_metadata: IndexingMetadata,  # noqa: ARG002
+        chunks: Iterable[DocMetadataAwareIndexChunk],
+        indexing_metadata: IndexingMetadata,
     ) -> list[DocumentInsertionRecord]:
-        """Indexes a list of document chunks into the document index.
+        """Indexes an iterable of document chunks into the document index.
 
         Groups chunks by document ID and for each document, deletes existing
         chunks and indexes the new chunks in bulk.
@@ -673,29 +674,34 @@ class OpenSearchDocumentIndex(DocumentIndex):
                 document is newly indexed or had already existed and was just
                 updated.
         """
-        # Group chunks by document ID.
-        doc_id_to_chunks: dict[str, list[DocMetadataAwareIndexChunk]] = defaultdict(
-            list
+        total_chunks = sum(
+            cc.new_chunk_cnt
+            for cc in indexing_metadata.doc_id_to_chunk_cnt_diff.values()
         )
-        for chunk in chunks:
-            doc_id_to_chunks[chunk.source_document.id].append(chunk)
         logger.debug(
-            f"[OpenSearchDocumentIndex] Indexing {len(chunks)} chunks from {len(doc_id_to_chunks)} "
+            f"[OpenSearchDocumentIndex] Indexing {total_chunks} chunks from {len(indexing_metadata.doc_id_to_chunk_cnt_diff)} "
             f"documents for index {self._index_name}."
         )
 
         document_indexing_results: list[DocumentInsertionRecord] = []
-        # Try to index per-document.
-        for _, chunks in doc_id_to_chunks.items():
+        deleted_doc_ids: set[str] = set()
+        # Buffer chunks per document as they arrive from the iterable.
+        # When the document ID changes flush the buffered chunks.
+        current_doc_id: str | None = None
+        current_chunks: list[DocMetadataAwareIndexChunk] = []
+
+        def _flush_chunks(doc_chunks: list[DocMetadataAwareIndexChunk]) -> None:
+            assert len(doc_chunks) > 0, "doc_chunks is empty"
+
             # Create a batch of OpenSearch-formatted chunks for bulk insertion.
-            # Do this before deleting existing chunks to reduce the amount of
-            # time the document index has no content for a given document, and
-            # to reduce the chance of entering a state where we delete chunks,
-            # then some error happens, and never successfully index new chunks.
+            # Since we are doing this in batches, an error occurring midway
+            # can result in a state where chunks are deleted and not all the
+            # new chunks have been indexed.
             chunk_batch: list[DocumentChunk] = [
-                _convert_onyx_chunk_to_opensearch_document(chunk) for chunk in chunks
+                _convert_onyx_chunk_to_opensearch_document(chunk)
+                for chunk in doc_chunks
             ]
-            onyx_document: Document = chunks[0].source_document
+            onyx_document: Document = doc_chunks[0].source_document
             # First delete the doc's chunks from the index. This is so that
             # there are no dangling chunks in the index, in the event that the
             # new document's content contains fewer chunks than the previous
@@ -704,22 +710,43 @@ class OpenSearchDocumentIndex(DocumentIndex):
             # if the chunk count has actually decreased. This assumes that
             # overlapping chunks are perfectly overwritten. If we can't
             # guarantee that then we need the code as-is.
-            num_chunks_deleted = self.delete(
-                onyx_document.id, onyx_document.chunk_count
-            )
-            # If we see that chunks were deleted we assume the doc already
-            # existed.
-            document_insertion_record = DocumentInsertionRecord(
-                document_id=onyx_document.id,
-                already_existed=num_chunks_deleted > 0,
-            )
+            if onyx_document.id not in deleted_doc_ids:
+                num_chunks_deleted = self.delete(
+                    onyx_document.id, onyx_document.chunk_count
+                )
+                deleted_doc_ids.add(onyx_document.id)
+                # If we see that chunks were deleted we assume the doc already
+                # existed. We record the result before bulk_index_documents
+                # runs. If indexing raises, this entire result list is discarded
+                # by the caller's retry logic, so early recording is safe.
+                document_indexing_results.append(
+                    DocumentInsertionRecord(
+                        document_id=onyx_document.id,
+                        already_existed=num_chunks_deleted > 0,
+                    )
+                )
             # Now index. This will raise if a chunk of the same ID exists, which
             # we do not expect because we should have deleted all chunks.
             self._client.bulk_index_documents(
                 documents=chunk_batch,
                 tenant_state=self._tenant_state,
             )
-            document_indexing_results.append(document_insertion_record)
+
+        for chunk in chunks:
+            doc_id = chunk.source_document.id
+            if doc_id != current_doc_id:
+                if current_chunks:
+                    _flush_chunks(current_chunks)
+                current_doc_id = doc_id
+                current_chunks = [chunk]
+            elif len(current_chunks) >= MAX_CHUNKS_PER_DOC_BATCH:
+                _flush_chunks(current_chunks)
+                current_chunks = [chunk]
+            else:
+                current_chunks.append(chunk)
+
+        if current_chunks:
+            _flush_chunks(current_chunks)
 
         return document_indexing_results
 
@@ -901,7 +928,7 @@ class OpenSearchDocumentIndex(DocumentIndex):
             search_hits = self._client.search(
                 body=query_body,
                 search_pipeline_id=None,
-                search_type=OpenSearchSearchType.ID_RETRIEVAL,
+                search_type=OpenSearchSearchType.DOC_ID_RETRIEVAL,
             )
             inference_chunks_uncleaned: list[InferenceChunkUncleaned] = [
                 _convert_retrieved_opensearch_chunk_to_inference_chunk_uncleaned(

backend/onyx/document_index/vespa/index.py

@@ -6,6 +6,7 @@ import re
 import time
 import urllib
 import zipfile
+from collections.abc import Iterable
 from dataclasses import dataclass
 from datetime import datetime
 from datetime import timedelta
@@ -461,7 +462,7 @@ class VespaIndex(DocumentIndex):
 
     def index(
         self,
-        chunks: list[DocMetadataAwareIndexChunk],
+        chunks: Iterable[DocMetadataAwareIndexChunk],
         index_batch_params: IndexBatchParams,
     ) -> set[OldDocumentInsertionRecord]:
         """

backend/onyx/document_index/vespa/vespa_document_index.py

@@ -1,6 +1,8 @@
 import concurrent.futures
 import logging
 import random
+from collections.abc import Generator
+from collections.abc import Iterable
 from typing import Any
 from uuid import UUID
 
@@ -8,6 +10,7 @@ import httpx
 from pydantic import BaseModel
 from retry import retry
 
+from onyx.configs.app_configs import MAX_CHUNKS_PER_DOC_BATCH
 from onyx.configs.app_configs import RECENCY_BIAS_MULTIPLIER
 from onyx.configs.app_configs import RERANK_COUNT
 from onyx.configs.chat_configs import DOC_TIME_DECAY
@@ -318,7 +321,7 @@ class VespaDocumentIndex(DocumentIndex):
 
     def index(
         self,
-        chunks: list[DocMetadataAwareIndexChunk],
+        chunks: Iterable[DocMetadataAwareIndexChunk],
         indexing_metadata: IndexingMetadata,
     ) -> list[DocumentInsertionRecord]:
         doc_id_to_chunk_cnt_diff = indexing_metadata.doc_id_to_chunk_cnt_diff
@@ -338,22 +341,31 @@ class VespaDocumentIndex(DocumentIndex):
 
         # Vespa has restrictions on valid characters, yet document IDs come from
         # external w.r.t. this class. We need to sanitize them.
-        cleaned_chunks: list[DocMetadataAwareIndexChunk] = [
-            clean_chunk_id_copy(chunk) for chunk in chunks
-        ]
-        assert len(cleaned_chunks) == len(
-            chunks
-        ), "Bug: Cleaned chunks and input chunks have different lengths."
+        #
+        # Instead of materializing all cleaned chunks upfront, we stream them
+        # through a generator that cleans IDs and builds the original-ID mapping
+        # incrementally as chunks flow into Vespa.
+        def _clean_and_track(
+            chunks_iter: Iterable[DocMetadataAwareIndexChunk],
+            id_map: dict[str, str],
+            seen_ids: set[str],
+        ) -> Generator[DocMetadataAwareIndexChunk, None, None]:
+            """Cleans chunk IDs and builds the original-ID mapping
+            incrementally as chunks flow through, avoiding a separate
+            materialization pass."""
+            for chunk in chunks_iter:
+                original_id = chunk.source_document.id
+                cleaned = clean_chunk_id_copy(chunk)
+                cleaned_id = cleaned.source_document.id
+                # Needed so the final DocumentInsertionRecord returned can have
+                # the original document ID. cleaned_chunks might not contain IDs
+                # exactly as callers supplied them.
+                id_map[cleaned_id] = original_id
+                seen_ids.add(cleaned_id)
+                yield cleaned
 
-        # Needed so the final DocumentInsertionRecord returned can have the
-        # original document ID. cleaned_chunks might not contain IDs exactly as
-        # callers supplied them.
-        new_document_id_to_original_document_id: dict[str, str] = dict()
-        for i, cleaned_chunk in enumerate(cleaned_chunks):
-            old_chunk = chunks[i]
-            new_document_id_to_original_document_id[
-                cleaned_chunk.source_document.id
-            ] = old_chunk.source_document.id
+        new_document_id_to_original_document_id: dict[str, str] = {}
+        all_cleaned_doc_ids: set[str] = set()
 
         existing_docs: set[str] = set()
 
@@ -409,8 +421,16 @@ class VespaDocumentIndex(DocumentIndex):
                     executor=executor,
                 )
 
-            # Insert new Vespa documents.
-            for chunk_batch in batch_generator(cleaned_chunks, BATCH_SIZE):
+            # Insert new Vespa documents, streaming through the cleaning
+            # pipeline so chunks are never fully materialized.
+            cleaned_chunks = _clean_and_track(
+                chunks,
+                new_document_id_to_original_document_id,
+                all_cleaned_doc_ids,
+            )
+            for chunk_batch in batch_generator(
+                cleaned_chunks, min(BATCH_SIZE, MAX_CHUNKS_PER_DOC_BATCH)
+            ):
                 batch_index_vespa_chunks(
                     chunks=chunk_batch,
                     index_name=self._index_name,
@@ -419,10 +439,6 @@ class VespaDocumentIndex(DocumentIndex):
                     executor=executor,
                 )
 
-        all_cleaned_doc_ids: set[str] = {
-            chunk.source_document.id for chunk in cleaned_chunks
-        }
-
         return [
             DocumentInsertionRecord(
                 document_id=new_document_id_to_original_document_id[cleaned_doc_id],

backend/onyx/file_processing/extract_file_text.py

@@ -44,6 +44,7 @@ KNOWN_OPENPYXL_BUGS = [
     "Value must be either numerical or a string containing a wildcard",
     "File contains no valid workbook part",
     "Unable to read workbook: could not read stylesheet from None",
+    "Colors must be aRGB hex values",
 ]
 
 

backend/onyx/file_processing/file_types.py

@@ -15,6 +15,7 @@ PLAIN_TEXT_MIME_TYPE = "text/plain"
 class OnyxMimeTypes:
     IMAGE_MIME_TYPES = {"image/jpg", "image/jpeg", "image/png", "image/webp"}
     CSV_MIME_TYPES = {"text/csv"}
+    TABULAR_MIME_TYPES = CSV_MIME_TYPES | {SPREADSHEET_MIME_TYPE}
     TEXT_MIME_TYPES = {
         PLAIN_TEXT_MIME_TYPE,
         "text/markdown",
@@ -34,13 +35,12 @@ class OnyxMimeTypes:
         PDF_MIME_TYPE,
         WORD_PROCESSING_MIME_TYPE,
         PRESENTATION_MIME_TYPE,
-        SPREADSHEET_MIME_TYPE,
         "message/rfc822",
         "application/epub+zip",
     }
 
     ALLOWED_MIME_TYPES = IMAGE_MIME_TYPES.union(
-        TEXT_MIME_TYPES, DOCUMENT_MIME_TYPES, CSV_MIME_TYPES
+        TEXT_MIME_TYPES, DOCUMENT_MIME_TYPES, TABULAR_MIME_TYPES
     )
 
     EXCLUDED_IMAGE_TYPES = {
@@ -53,6 +53,11 @@ class OnyxMimeTypes:
 
 
 class OnyxFileExtensions:
+    TABULAR_EXTENSIONS = {
+        ".csv",
+        ".tsv",
+        ".xlsx",
+    }
     PLAIN_TEXT_EXTENSIONS = {
         ".txt",
         ".md",

backend/onyx/file_store/models.py

@@ -13,15 +13,21 @@ class ChatFileType(str, Enum):
     DOC = "document"
     # Plain text only contain the text
     PLAIN_TEXT = "plain_text"
-    CSV = "csv"
+    # Tabular data files (CSV, XLSX)
+    TABULAR = "tabular"
 
     def is_text_file(self) -> bool:
         return self in (
             ChatFileType.PLAIN_TEXT,
             ChatFileType.DOC,
-            ChatFileType.CSV,
+            ChatFileType.TABULAR,
         )
 
+    def use_metadata_only(self) -> bool:
+        """File types where we can ignore the file content
+        and only use the metadata."""
+        return self in (ChatFileType.TABULAR,)
+
 
 class FileDescriptor(TypedDict):
     """NOTE: is a `TypedDict` so it can be used as a type hint for a JSONB column

backend/onyx/file_store/utils.py

@@ -110,16 +110,20 @@ def load_user_file(file_id: UUID, db_session: Session) -> InMemoryChatFile:
     # check for plain text normalized version first, then use original file otherwise
     try:
         file_io = file_store.read_file(plaintext_file_name, mode="b")
-        # For plaintext versions, use PLAIN_TEXT type (unless it's an image which doesn't have plaintext)
-        plaintext_chat_file_type = (
-            ChatFileType.PLAIN_TEXT
-            if chat_file_type != ChatFileType.IMAGE
-            else chat_file_type
-        )
-
-        # if we have plaintext for image (which happens when image extraction is enabled), we use PLAIN_TEXT type
-        if file_io is not None:
+        # Metadata-only file types preserve their original type so
+        # downstream injection paths can route them correctly.
+        if chat_file_type.use_metadata_only():
+            plaintext_chat_file_type = chat_file_type
+        elif file_io is not None:
+            # if we have plaintext for image (which happens when image
+            # extraction is enabled), we use PLAIN_TEXT type
             plaintext_chat_file_type = ChatFileType.PLAIN_TEXT
+        else:
+            plaintext_chat_file_type = (
+                ChatFileType.PLAIN_TEXT
+                if chat_file_type != ChatFileType.IMAGE
+                else chat_file_type
+            )
 
         chat_file = InMemoryChatFile(
             file_id=str(user_file.file_id),

backend/onyx/hooks/api_dependencies.py

@@ -1,4 +1,3 @@
-from onyx.configs.app_configs import HOOK_ENABLED
 from onyx.error_handling.error_codes import OnyxErrorCode
 from onyx.error_handling.exceptions import OnyxError
 from shared_configs.configs import MULTI_TENANT
@@ -7,10 +6,7 @@ from shared_configs.configs import MULTI_TENANT
 def require_hook_enabled() -> None:
     """FastAPI dependency that gates all hook management endpoints.
 
-    Hooks are only available in single-tenant / self-hosted deployments with
-    HOOK_ENABLED=true explicitly set. Two layers of protection:
-      1. MULTI_TENANT check — rejects even if HOOK_ENABLED is accidentally set true
-      2. HOOK_ENABLED flag — explicit opt-in by the operator
+    Hooks are only available in single-tenant / self-hosted EE deployments.
 
     Use as: Depends(require_hook_enabled)
     """
@@ -19,8 +15,3 @@ def require_hook_enabled() -> None:
             OnyxErrorCode.SINGLE_TENANT_ONLY,
             "Hooks are not available in multi-tenant deployments",
         )
-    if not HOOK_ENABLED:
-        raise OnyxError(
-            OnyxErrorCode.ENV_VAR_GATED,
-            "Hooks are not enabled. Set HOOK_ENABLED=true to enable.",
-        )

backend/onyx/hooks/executor.py

@@ -1,79 +1,22 @@
-"""Hook executor — calls a customer's external HTTP endpoint for a given hook point.
+"""CE hook executor.
 
-Usage (Celery tasks and FastAPI handlers):
-    result = execute_hook(
-        db_session=db_session,
-        hook_point=HookPoint.QUERY_PROCESSING,
-        payload={"query": "...", "user_email": "...", "chat_session_id": "..."},
-        response_type=QueryProcessingResponse,
-    )
+HookSkipped and HookSoftFailed are real classes kept here because
+process_message.py (CE code) uses isinstance checks against them.
 
-    if isinstance(result, HookSkipped):
-        # no active hook configured — continue with original behavior
-        ...
-    elif isinstance(result, HookSoftFailed):
-        # hook failed but fail strategy is SOFT — continue with original behavior
-        ...
-    else:
-        # result is a validated Pydantic model instance (response_type)
-        ...
-
-is_reachable update policy
---------------------------
-``is_reachable`` on the Hook row is updated selectively — only when the outcome
-carries meaningful signal about physical reachability:
-
-  NetworkError (DNS, connection refused)  → False  (cannot reach the server)
-  HTTP 401 / 403                          → False  (api_key revoked or invalid)
-  TimeoutException                        → None   (server may be slow, skip write)
-  Other HTTP errors (4xx / 5xx)           → None   (server responded, skip write)
-  Unknown exception                       → None   (no signal, skip write)
-  Non-JSON / non-dict response            → None   (server responded, skip write)
-  Success (2xx, valid dict)               → True   (confirmed reachable)
-
-None means "leave the current value unchanged" — no DB round-trip is made.
-
-DB session design
------------------
-The executor uses three sessions:
-
-  1. Caller's session (db_session) — used only for the hook lookup read. All
-     needed fields are extracted from the Hook object before the HTTP call, so
-     the caller's session is not held open during the external HTTP request.
-
-  2. Log session — a separate short-lived session opened after the HTTP call
-     completes to write the HookExecutionLog row on failure. Success runs are
-     not recorded. Committed independently of everything else.
-
-  3. Reachable session — a second short-lived session to update is_reachable on
-     the Hook. Kept separate from the log session so a concurrent hook deletion
-     (which causes update_hook__no_commit to raise OnyxError(NOT_FOUND)) cannot
-     prevent the execution log from being written. This update is best-effort.
+execute_hook is the public entry point. It dispatches to _execute_hook_impl
+via fetch_versioned_implementation so that:
+  - CE: onyx.hooks.executor._execute_hook_impl → no-op, returns HookSkipped()
+  - EE: ee.onyx.hooks.executor._execute_hook_impl → real HTTP call
 """
 
-import json
-import time
 from typing import Any
 from typing import TypeVar
 
-import httpx
 from pydantic import BaseModel
-from pydantic import ValidationError
 from sqlalchemy.orm import Session
 
-from onyx.db.engine.sql_engine import get_session_with_current_tenant
-from onyx.db.enums import HookFailStrategy
 from onyx.db.enums import HookPoint
-from onyx.db.hook import create_hook_execution_log__no_commit
-from onyx.db.hook import get_non_deleted_hook_by_hook_point
-from onyx.db.hook import update_hook__no_commit
-from onyx.db.models import Hook
-from onyx.error_handling.error_codes import OnyxErrorCode
-from onyx.error_handling.exceptions import OnyxError
-from onyx.hooks.utils import HOOKS_AVAILABLE
-from onyx.utils.logger import setup_logger
-
-logger = setup_logger()
+from onyx.utils.variable_functionality import fetch_versioned_implementation
 
 
 class HookSkipped:
@@ -87,277 +30,15 @@ class HookSoftFailed:
 T = TypeVar("T", bound=BaseModel)
 
 
-# ---------------------------------------------------------------------------
-# Private helpers
-# ---------------------------------------------------------------------------
-
-
-class _HttpOutcome(BaseModel):
-    """Structured result of an HTTP hook call, returned by _process_response."""
-
-    is_success: bool
-    updated_is_reachable: (
-        bool | None
-    )  # True/False = write to DB, None = unchanged (skip write)
-    status_code: int | None
-    error_message: str | None
-    response_payload: dict[str, Any] | None
-
-
-def _lookup_hook(
-    db_session: Session,
-    hook_point: HookPoint,
-) -> Hook | HookSkipped:
-    """Return the active Hook or HookSkipped if hooks are unavailable/unconfigured.
-
-    No HTTP call is made and no DB writes are performed for any HookSkipped path.
-    There is nothing to log and no reachability information to update.
-    """
-    if not HOOKS_AVAILABLE:
-        return HookSkipped()
-    hook = get_non_deleted_hook_by_hook_point(
-        db_session=db_session, hook_point=hook_point
-    )
-    if hook is None or not hook.is_active:
-        return HookSkipped()
-    if not hook.endpoint_url:
-        return HookSkipped()
-    return hook
-
-
-def _process_response(
+def _execute_hook_impl(
     *,
-    response: httpx.Response | None,
-    exc: Exception | None,
-    timeout: float,
-) -> _HttpOutcome:
-    """Process the result of an HTTP call and return a structured outcome.
-
-    Called after the client.post() try/except. If post() raised, exc is set and
-    response is None. Otherwise response is set and exc is None. Handles
-    raise_for_status(), JSON decoding, and the dict shape check.
-    """
-    if exc is not None:
-        if isinstance(exc, httpx.NetworkError):
-            msg = f"Hook network error (endpoint unreachable): {exc}"
-            logger.warning(msg, exc_info=exc)
-            return _HttpOutcome(
-                is_success=False,
-                updated_is_reachable=False,
-                status_code=None,
-                error_message=msg,
-                response_payload=None,
-            )
-        if isinstance(exc, httpx.TimeoutException):
-            msg = f"Hook timed out after {timeout}s: {exc}"
-            logger.warning(msg, exc_info=exc)
-            return _HttpOutcome(
-                is_success=False,
-                updated_is_reachable=None,  # timeout doesn't indicate unreachability
-                status_code=None,
-                error_message=msg,
-                response_payload=None,
-            )
-        msg = f"Hook call failed: {exc}"
-        logger.exception(msg, exc_info=exc)
-        return _HttpOutcome(
-            is_success=False,
-            updated_is_reachable=None,  # unknown error — don't make assumptions
-            status_code=None,
-            error_message=msg,
-            response_payload=None,
-        )
-
-    if response is None:
-        raise ValueError(
-            "exactly one of response or exc must be non-None; both are None"
-        )
-    status_code = response.status_code
-
-    try:
-        response.raise_for_status()
-    except httpx.HTTPStatusError as e:
-        msg = f"Hook returned HTTP {e.response.status_code}: {e.response.text}"
-        logger.warning(msg, exc_info=e)
-        # 401/403 means the api_key has been revoked or is invalid — mark unreachable
-        # so the operator knows to update it. All other HTTP errors keep is_reachable
-        # as-is (server is up, the request just failed for application reasons).
-        auth_failed = e.response.status_code in (401, 403)
-        return _HttpOutcome(
-            is_success=False,
-            updated_is_reachable=False if auth_failed else None,
-            status_code=status_code,
-            error_message=msg,
-            response_payload=None,
-        )
-
-    try:
-        response_payload = response.json()
-    except (json.JSONDecodeError, httpx.DecodingError) as e:
-        msg = f"Hook returned non-JSON response: {e}"
-        logger.warning(msg, exc_info=e)
-        return _HttpOutcome(
-            is_success=False,
-            updated_is_reachable=None,  # server responded — reachability unchanged
-            status_code=status_code,
-            error_message=msg,
-            response_payload=None,
-        )
-
-    if not isinstance(response_payload, dict):
-        msg = f"Hook returned non-dict JSON (got {type(response_payload).__name__})"
-        logger.warning(msg)
-        return _HttpOutcome(
-            is_success=False,
-            updated_is_reachable=None,  # server responded — reachability unchanged
-            status_code=status_code,
-            error_message=msg,
-            response_payload=None,
-        )
-
-    return _HttpOutcome(
-        is_success=True,
-        updated_is_reachable=True,
-        status_code=status_code,
-        error_message=None,
-        response_payload=response_payload,
-    )
-
-
-def _persist_result(
-    *,
-    hook_id: int,
-    outcome: _HttpOutcome,
-    duration_ms: int,
-) -> None:
-    """Write the execution log on failure and optionally update is_reachable, each
-    in its own session so a failure in one does not affect the other."""
-    # Only write the execution log on failure — success runs are not recorded.
-    # Must not be skipped if the is_reachable update fails (e.g. hook concurrently
-    # deleted between the initial lookup and here).
-    if not outcome.is_success:
-        try:
-            with get_session_with_current_tenant() as log_session:
-                create_hook_execution_log__no_commit(
-                    db_session=log_session,
-                    hook_id=hook_id,
-                    is_success=False,
-                    error_message=outcome.error_message,
-                    status_code=outcome.status_code,
-                    duration_ms=duration_ms,
-                )
-                log_session.commit()
-        except Exception:
-            logger.exception(
-                f"Failed to persist hook execution log for hook_id={hook_id}"
-            )
-
-    # Update is_reachable separately — best-effort, non-critical.
-    # None means the value is unchanged (set by the caller to skip the no-op write).
-    # update_hook__no_commit can raise OnyxError(NOT_FOUND) if the hook was
-    # concurrently deleted, so keep this isolated from the log write above.
-    if outcome.updated_is_reachable is not None:
-        try:
-            with get_session_with_current_tenant() as reachable_session:
-                update_hook__no_commit(
-                    db_session=reachable_session,
-                    hook_id=hook_id,
-                    is_reachable=outcome.updated_is_reachable,
-                )
-                reachable_session.commit()
-        except Exception:
-            logger.warning(f"Failed to update is_reachable for hook_id={hook_id}")
-
-
-# ---------------------------------------------------------------------------
-# Public API
-# ---------------------------------------------------------------------------
-
-
-def _execute_hook_inner(
-    hook: Hook,
-    payload: dict[str, Any],
-    response_type: type[T],
-) -> T | HookSoftFailed:
-    """Make the HTTP call, validate the response, and return a typed model.
-
-    Raises OnyxError on HARD failure. Returns HookSoftFailed on SOFT failure.
-    """
-    timeout = hook.timeout_seconds
-    hook_id = hook.id
-    fail_strategy = hook.fail_strategy
-    endpoint_url = hook.endpoint_url
-    current_is_reachable: bool | None = hook.is_reachable
-
-    if not endpoint_url:
-        raise ValueError(
-            f"hook_id={hook_id} is active but has no endpoint_url — "
-            "active hooks without an endpoint_url must be rejected by _lookup_hook"
-        )
-
-    start = time.monotonic()
-    response: httpx.Response | None = None
-    exc: Exception | None = None
-    try:
-        api_key: str | None = (
-            hook.api_key.get_value(apply_mask=False) if hook.api_key else None
-        )
-        headers: dict[str, str] = {"Content-Type": "application/json"}
-        if api_key:
-            headers["Authorization"] = f"Bearer {api_key}"
-        with httpx.Client(
-            timeout=timeout, follow_redirects=False
-        ) as client:  # SSRF guard: never follow redirects
-            response = client.post(endpoint_url, json=payload, headers=headers)
-    except Exception as e:
-        exc = e
-    duration_ms = int((time.monotonic() - start) * 1000)
-
-    outcome = _process_response(response=response, exc=exc, timeout=timeout)
-
-    # Validate the response payload against response_type.
-    # A validation failure downgrades the outcome to a failure so it is logged,
-    # is_reachable is left unchanged (server responded — just a bad payload),
-    # and fail_strategy is respected below.
-    validated_model: T | None = None
-    if outcome.is_success and outcome.response_payload is not None:
-        try:
-            validated_model = response_type.model_validate(outcome.response_payload)
-        except ValidationError as e:
-            msg = (
-                f"Hook response failed validation against {response_type.__name__}: {e}"
-            )
-            outcome = _HttpOutcome(
-                is_success=False,
-                updated_is_reachable=None,  # server responded — reachability unchanged
-                status_code=outcome.status_code,
-                error_message=msg,
-                response_payload=None,
-            )
-
-    # Skip the is_reachable write when the value would not change — avoids a
-    # no-op DB round-trip on every call when the hook is already in the expected state.
-    if outcome.updated_is_reachable == current_is_reachable:
-        outcome = outcome.model_copy(update={"updated_is_reachable": None})
-    _persist_result(hook_id=hook_id, outcome=outcome, duration_ms=duration_ms)
-
-    if not outcome.is_success:
-        if fail_strategy == HookFailStrategy.HARD:
-            raise OnyxError(
-                OnyxErrorCode.HOOK_EXECUTION_FAILED,
-                outcome.error_message or "Hook execution failed.",
-            )
-        logger.warning(
-            f"Hook execution failed (soft fail) for hook_id={hook_id}: {outcome.error_message}"
-        )
-        return HookSoftFailed()
-
-    if validated_model is None:
-        raise OnyxError(
-            OnyxErrorCode.INTERNAL_ERROR,
-            f"validated_model is None for successful hook call (hook_id={hook_id})",
-        )
-    return validated_model
+    db_session: Session,  # noqa: ARG001
+    hook_point: HookPoint,  # noqa: ARG001
+    payload: dict[str, Any],  # noqa: ARG001
+    response_type: type[T],  # noqa: ARG001
+) -> T | HookSkipped | HookSoftFailed:
+    """CE no-op — hooks are not available without EE."""
+    return HookSkipped()
 
 
 def execute_hook(
@@ -367,25 +48,15 @@ def execute_hook(
     payload: dict[str, Any],
     response_type: type[T],
 ) -> T | HookSkipped | HookSoftFailed:
-    """Execute the hook for the given hook point synchronously.
+    """Execute the hook for the given hook point.
 
-    Returns HookSkipped if no active hook is configured, HookSoftFailed if the
-    hook failed with SOFT fail strategy, or a validated response model on success.
-    Raises OnyxError on HARD failure or if the hook is misconfigured.
+    Dispatches to the versioned implementation so EE gets the real executor
+    and CE gets the no-op stub, without any changes at the call site.
     """
-    hook = _lookup_hook(db_session, hook_point)
-    if isinstance(hook, HookSkipped):
-        return hook
-
-    fail_strategy = hook.fail_strategy
-    hook_id = hook.id
-
-    try:
-        return _execute_hook_inner(hook, payload, response_type)
-    except Exception:
-        if fail_strategy == HookFailStrategy.SOFT:
-            logger.exception(
-                f"Unexpected error in hook execution (soft fail) for hook_id={hook_id}"
-            )
-            return HookSoftFailed()
-        raise
+    impl = fetch_versioned_implementation("onyx.hooks.executor", "_execute_hook_impl")
+    return impl(
+        db_session=db_session,
+        hook_point=hook_point,
+        payload=payload,
+        response_type=response_type,
+    )

backend/onyx/hooks/utils.py

@@ -1,5 +0,0 @@
-from onyx.configs.app_configs import HOOK_ENABLED
-from shared_configs.configs import MULTI_TENANT
-
-# True only when hooks are available: single-tenant deployment with HOOK_ENABLED=true.
-HOOKS_AVAILABLE: bool = HOOK_ENABLED and not MULTI_TENANT

backend/onyx/indexing/adapters/document_indexing_adapter.py

@@ -19,7 +19,8 @@ from onyx.db.document import update_docs_updated_at__no_commit
 from onyx.db.document_set import fetch_document_sets_for_documents
 from onyx.indexing.indexing_pipeline import DocumentBatchPrepareContext
 from onyx.indexing.indexing_pipeline import index_doc_batch_prepare
-from onyx.indexing.models import BuildMetadataAwareChunksResult
+from onyx.indexing.models import ChunkEnrichmentContext
+from onyx.indexing.models import DocAwareChunk
 from onyx.indexing.models import DocMetadataAwareIndexChunk
 from onyx.indexing.models import IndexChunk
 from onyx.indexing.models import UpdatableChunkData
@@ -85,14 +86,21 @@ class DocumentIndexingBatchAdapter:
         ) as transaction:
             yield transaction
 
-    def build_metadata_aware_chunks(
+    def prepare_enrichment(
         self,
-        chunks_with_embeddings: list[IndexChunk],
-        chunk_content_scores: list[float],
-        tenant_id: str,
         context: DocumentBatchPrepareContext,
-    ) -> BuildMetadataAwareChunksResult:
-        """Enrich chunks with access, document sets, boosts, token counts, and hierarchy."""
+        tenant_id: str,
+        chunks: list[DocAwareChunk],
+    ) -> "DocumentChunkEnricher":
+        """Do all DB lookups once and return a per-chunk enricher."""
+        updatable_ids = [doc.id for doc in context.updatable_docs]
+
+        doc_id_to_new_chunk_cnt: dict[str, int] = {
+            doc_id: 0 for doc_id in updatable_ids
+        }
+        for chunk in chunks:
+            if chunk.source_document.id in doc_id_to_new_chunk_cnt:
+                doc_id_to_new_chunk_cnt[chunk.source_document.id] += 1
 
         no_access = DocumentAccess.build(
             user_emails=[],
@@ -102,67 +110,30 @@ class DocumentIndexingBatchAdapter:
             is_public=False,
         )
 
-        updatable_ids = [doc.id for doc in context.updatable_docs]
-
-        doc_id_to_access_info = get_access_for_documents(
-            document_ids=updatable_ids, db_session=self.db_session
-        )
-        doc_id_to_document_set = {
-            document_id: document_sets
-            for document_id, document_sets in fetch_document_sets_for_documents(
+        return DocumentChunkEnricher(
+            doc_id_to_access_info=get_access_for_documents(
                 document_ids=updatable_ids, db_session=self.db_session
-            )
-        }
-
-        doc_id_to_previous_chunk_cnt: dict[str, int] = {
-            document_id: chunk_count
-            for document_id, chunk_count in fetch_chunk_counts_for_documents(
-                document_ids=updatable_ids,
-                db_session=self.db_session,
-            )
-        }
-
-        doc_id_to_new_chunk_cnt: dict[str, int] = {
-            doc_id: 0 for doc_id in updatable_ids
-        }
-        for chunk in chunks_with_embeddings:
-            if chunk.source_document.id in doc_id_to_new_chunk_cnt:
-                doc_id_to_new_chunk_cnt[chunk.source_document.id] += 1
-
-        # Get ancestor hierarchy node IDs for each document
-        doc_id_to_ancestor_ids = self._get_ancestor_ids_for_documents(
-            context.updatable_docs, tenant_id
-        )
-
-        access_aware_chunks = [
-            DocMetadataAwareIndexChunk.from_index_chunk(
-                index_chunk=chunk,
-                access=doc_id_to_access_info.get(chunk.source_document.id, no_access),
-                document_sets=set(
-                    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
-                    else DEFAULT_BOOST
-                ),
-                tenant_id=tenant_id,
-                aggregated_chunk_boost_factor=chunk_content_scores[chunk_num],
-                ancestor_hierarchy_node_ids=doc_id_to_ancestor_ids[
-                    chunk.source_document.id
-                ],
-            )
-            for chunk_num, chunk in enumerate(chunks_with_embeddings)
-        ]
-
-        return BuildMetadataAwareChunksResult(
-            chunks=access_aware_chunks,
-            doc_id_to_previous_chunk_cnt=doc_id_to_previous_chunk_cnt,
-            doc_id_to_new_chunk_cnt=doc_id_to_new_chunk_cnt,
-            user_file_id_to_raw_text={},
-            user_file_id_to_token_count={},
+            ),
+            doc_id_to_document_set={
+                document_id: document_sets
+                for document_id, document_sets in fetch_document_sets_for_documents(
+                    document_ids=updatable_ids, db_session=self.db_session
+                )
+            },
+            doc_id_to_ancestor_ids=self._get_ancestor_ids_for_documents(
+                context.updatable_docs, tenant_id
+            ),
+            id_to_boost_map=context.id_to_boost_map,
+            doc_id_to_previous_chunk_cnt={
+                document_id: chunk_count
+                for document_id, chunk_count in fetch_chunk_counts_for_documents(
+                    document_ids=updatable_ids,
+                    db_session=self.db_session,
+                )
+            },
+            doc_id_to_new_chunk_cnt=dict(doc_id_to_new_chunk_cnt),
+            no_access=no_access,
+            tenant_id=tenant_id,
         )
 
     def _get_ancestor_ids_for_documents(
@@ -203,7 +174,7 @@ class DocumentIndexingBatchAdapter:
         context: DocumentBatchPrepareContext,
         updatable_chunk_data: list[UpdatableChunkData],
         filtered_documents: list[Document],
-        result: BuildMetadataAwareChunksResult,
+        enrichment: ChunkEnrichmentContext,
     ) -> None:
         """Finalize DB updates, store plaintext, and mark docs as indexed."""
         updatable_ids = [doc.id for doc in context.updatable_docs]
@@ -227,7 +198,7 @@ class DocumentIndexingBatchAdapter:
 
         update_docs_chunk_count__no_commit(
             document_ids=updatable_ids,
-            doc_id_to_chunk_count=result.doc_id_to_new_chunk_cnt,
+            doc_id_to_chunk_count=enrichment.doc_id_to_new_chunk_cnt,
             db_session=self.db_session,
         )
 
@@ -249,3 +220,52 @@ class DocumentIndexingBatchAdapter:
         )
 
         self.db_session.commit()
+
+
+class DocumentChunkEnricher:
+    """Pre-computed metadata for per-chunk enrichment of connector documents."""
+
+    def __init__(
+        self,
+        doc_id_to_access_info: dict[str, DocumentAccess],
+        doc_id_to_document_set: dict[str, list[str]],
+        doc_id_to_ancestor_ids: dict[str, list[int]],
+        id_to_boost_map: dict[str, int],
+        doc_id_to_previous_chunk_cnt: dict[str, int],
+        doc_id_to_new_chunk_cnt: dict[str, int],
+        no_access: DocumentAccess,
+        tenant_id: str,
+    ) -> None:
+        self._doc_id_to_access_info = doc_id_to_access_info
+        self._doc_id_to_document_set = doc_id_to_document_set
+        self._doc_id_to_ancestor_ids = doc_id_to_ancestor_ids
+        self._id_to_boost_map = id_to_boost_map
+        self._no_access = no_access
+        self._tenant_id = tenant_id
+        self.doc_id_to_previous_chunk_cnt = doc_id_to_previous_chunk_cnt
+        self.doc_id_to_new_chunk_cnt = doc_id_to_new_chunk_cnt
+
+    def enrich_chunk(
+        self, chunk: IndexChunk, score: float
+    ) -> DocMetadataAwareIndexChunk:
+        return DocMetadataAwareIndexChunk.from_index_chunk(
+            index_chunk=chunk,
+            access=self._doc_id_to_access_info.get(
+                chunk.source_document.id, self._no_access
+            ),
+            document_sets=set(
+                self._doc_id_to_document_set.get(chunk.source_document.id, [])
+            ),
+            user_project=[],
+            personas=[],
+            boost=(
+                self._id_to_boost_map[chunk.source_document.id]
+                if chunk.source_document.id in self._id_to_boost_map
+                else DEFAULT_BOOST
+            ),
+            tenant_id=self._tenant_id,
+            aggregated_chunk_boost_factor=score,
+            ancestor_hierarchy_node_ids=self._doc_id_to_ancestor_ids[
+                chunk.source_document.id
+            ],
+        )

backend/onyx/indexing/adapters/user_file_indexing_adapter.py

@@ -1,6 +1,9 @@
+from __future__ import annotations
+
 import contextlib
 import datetime
 import time
+from collections import defaultdict
 from collections.abc import Generator
 from uuid import UUID
 
@@ -24,11 +27,13 @@ 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
-from onyx.indexing.models import BuildMetadataAwareChunksResult
+from onyx.indexing.models import ChunkEnrichmentContext
+from onyx.indexing.models import DocAwareChunk
 from onyx.indexing.models import DocMetadataAwareIndexChunk
 from onyx.indexing.models import IndexChunk
 from onyx.indexing.models import UpdatableChunkData
 from onyx.llm.factory import get_default_llm
+from onyx.natural_language_processing.utils import count_tokens
 from onyx.natural_language_processing.utils import get_tokenizer
 from onyx.utils.logger import setup_logger
 
@@ -101,13 +106,20 @@ class UserFileIndexingAdapter:
                 f"Failed to acquire locks after {_NUM_LOCK_ATTEMPTS} attempts for user files: {[doc.id for doc in documents]}"
             )
 
-    def build_metadata_aware_chunks(
+    def prepare_enrichment(
         self,
-        chunks_with_embeddings: list[IndexChunk],
-        chunk_content_scores: list[float],
-        tenant_id: str,
         context: DocumentBatchPrepareContext,
-    ) -> BuildMetadataAwareChunksResult:
+        tenant_id: str,
+        chunks: list[DocAwareChunk],
+    ) -> UserFileChunkEnricher:
+        """Do all DB lookups and pre-compute file metadata from chunks."""
+        updatable_ids = [doc.id for doc in context.updatable_docs]
+
+        doc_id_to_new_chunk_cnt: dict[str, int] = defaultdict(int)
+        content_by_file: dict[str, list[str]] = defaultdict(list)
+        for chunk in chunks:
+            doc_id_to_new_chunk_cnt[chunk.source_document.id] += 1
+            content_by_file[chunk.source_document.id].append(chunk.content)
 
         no_access = DocumentAccess.build(
             user_emails=[],
@@ -117,7 +129,6 @@ class UserFileIndexingAdapter:
             is_public=False,
         )
 
-        updatable_ids = [doc.id for doc in context.updatable_docs]
         user_file_id_to_project_ids = fetch_user_project_ids_for_user_files(
             user_file_ids=updatable_ids,
             db_session=self.db_session,
@@ -138,17 +149,6 @@ class UserFileIndexingAdapter:
             )
         }
 
-        user_file_id_to_new_chunk_cnt: dict[str, int] = {
-            user_file_id: len(
-                [
-                    chunk
-                    for chunk in chunks_with_embeddings
-                    if chunk.source_document.id == user_file_id
-                ]
-            )
-            for user_file_id in updatable_ids
-        }
-
         # Initialize tokenizer used for token count calculation
         try:
             llm = get_default_llm()
@@ -163,46 +163,30 @@ class UserFileIndexingAdapter:
         user_file_id_to_raw_text: dict[str, str] = {}
         user_file_id_to_token_count: dict[str, int | None] = {}
         for user_file_id in updatable_ids:
-            user_file_chunks = [
-                chunk
-                for chunk in chunks_with_embeddings
-                if chunk.source_document.id == user_file_id
-            ]
-            if user_file_chunks:
-                combined_content = " ".join(
-                    [chunk.content for chunk in user_file_chunks]
-                )
+            contents = content_by_file.get(user_file_id)
+            if contents:
+                combined_content = " ".join(contents)
                 user_file_id_to_raw_text[str(user_file_id)] = combined_content
-                token_count = (
-                    len(llm_tokenizer.encode(combined_content)) if llm_tokenizer else 0
+                token_count: int = (
+                    count_tokens(combined_content, llm_tokenizer)
+                    if llm_tokenizer
+                    else 0
                 )
                 user_file_id_to_token_count[str(user_file_id)] = token_count
             else:
                 user_file_id_to_raw_text[str(user_file_id)] = ""
                 user_file_id_to_token_count[str(user_file_id)] = None
 
-        access_aware_chunks = [
-            DocMetadataAwareIndexChunk.from_index_chunk(
-                index_chunk=chunk,
-                access=user_file_id_to_access.get(chunk.source_document.id, no_access),
-                document_sets=set(),
-                user_project=user_file_id_to_project_ids.get(
-                    chunk.source_document.id, []
-                ),
-                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],
-            )
-            for chunk_num, chunk in enumerate(chunks_with_embeddings)
-        ]
-
-        return BuildMetadataAwareChunksResult(
-            chunks=access_aware_chunks,
+        return UserFileChunkEnricher(
+            user_file_id_to_access=user_file_id_to_access,
+            user_file_id_to_project_ids=user_file_id_to_project_ids,
+            user_file_id_to_persona_ids=user_file_id_to_persona_ids,
             doc_id_to_previous_chunk_cnt=user_file_id_to_previous_chunk_cnt,
-            doc_id_to_new_chunk_cnt=user_file_id_to_new_chunk_cnt,
+            doc_id_to_new_chunk_cnt=dict(doc_id_to_new_chunk_cnt),
             user_file_id_to_raw_text=user_file_id_to_raw_text,
             user_file_id_to_token_count=user_file_id_to_token_count,
+            no_access=no_access,
+            tenant_id=tenant_id,
         )
 
     def _notify_assistant_owners_if_files_ready(
@@ -246,8 +230,9 @@ class UserFileIndexingAdapter:
         context: DocumentBatchPrepareContext,
         updatable_chunk_data: list[UpdatableChunkData],  # noqa: ARG002
         filtered_documents: list[Document],  # noqa: ARG002
-        result: BuildMetadataAwareChunksResult,
+        enrichment: ChunkEnrichmentContext,
     ) -> None:
+        assert isinstance(enrichment, UserFileChunkEnricher)
         user_file_ids = [doc.id for doc in context.updatable_docs]
 
         user_files = (
@@ -263,8 +248,10 @@ class UserFileIndexingAdapter:
             user_file.last_project_sync_at = datetime.datetime.now(
                 datetime.timezone.utc
             )
-            user_file.chunk_count = result.doc_id_to_new_chunk_cnt[str(user_file.id)]
-            user_file.token_count = result.user_file_id_to_token_count[
+            user_file.chunk_count = enrichment.doc_id_to_new_chunk_cnt.get(
+                str(user_file.id), 0
+            )
+            user_file.token_count = enrichment.user_file_id_to_token_count[
                 str(user_file.id)
             ]
 
@@ -276,8 +263,54 @@ class UserFileIndexingAdapter:
         # Store the plaintext in the file store for faster retrieval
         # NOTE: this creates its own session to avoid committing the overall
         # transaction.
-        for user_file_id, raw_text in result.user_file_id_to_raw_text.items():
+        for user_file_id, raw_text in enrichment.user_file_id_to_raw_text.items():
             store_user_file_plaintext(
                 user_file_id=UUID(user_file_id),
                 plaintext_content=raw_text,
             )
+
+
+class UserFileChunkEnricher:
+    """Pre-computed metadata for per-chunk enrichment of user-uploaded files."""
+
+    def __init__(
+        self,
+        user_file_id_to_access: dict[str, DocumentAccess],
+        user_file_id_to_project_ids: dict[str, list[int]],
+        user_file_id_to_persona_ids: dict[str, list[int]],
+        doc_id_to_previous_chunk_cnt: dict[str, int],
+        doc_id_to_new_chunk_cnt: dict[str, int],
+        user_file_id_to_raw_text: dict[str, str],
+        user_file_id_to_token_count: dict[str, int | None],
+        no_access: DocumentAccess,
+        tenant_id: str,
+    ) -> None:
+        self._user_file_id_to_access = user_file_id_to_access
+        self._user_file_id_to_project_ids = user_file_id_to_project_ids
+        self._user_file_id_to_persona_ids = user_file_id_to_persona_ids
+        self._no_access = no_access
+        self._tenant_id = tenant_id
+        self.doc_id_to_previous_chunk_cnt = doc_id_to_previous_chunk_cnt
+        self.doc_id_to_new_chunk_cnt = doc_id_to_new_chunk_cnt
+        self.user_file_id_to_raw_text = user_file_id_to_raw_text
+        self.user_file_id_to_token_count = user_file_id_to_token_count
+
+    def enrich_chunk(
+        self, chunk: IndexChunk, score: float
+    ) -> DocMetadataAwareIndexChunk:
+        return DocMetadataAwareIndexChunk.from_index_chunk(
+            index_chunk=chunk,
+            access=self._user_file_id_to_access.get(
+                chunk.source_document.id, self._no_access
+            ),
+            document_sets=set(),
+            user_project=self._user_file_id_to_project_ids.get(
+                chunk.source_document.id, []
+            ),
+            personas=self._user_file_id_to_persona_ids.get(
+                chunk.source_document.id, []
+            ),
+            boost=DEFAULT_BOOST,
+            tenant_id=self._tenant_id,
+            aggregated_chunk_boost_factor=score,
+        )

backend/onyx/indexing/chunk_batch_store.py

@@ -0,0 +1,89 @@
+import pickle
+import shutil
+import tempfile
+from collections.abc import Iterator
+from pathlib import Path
+
+from onyx.indexing.models import IndexChunk
+
+
+class ChunkBatchStore:
+    """Manages serialization of embedded chunks to a temporary directory.
+
+    Owns the temp directory lifetime and provides save/load/stream/scrub
+    operations.
+
+    Use as a context manager to ensure cleanup::
+
+        with ChunkBatchStore() as store:
+            store.save(chunks, batch_idx=0)
+            for chunk in store.stream():
+                ...
+    """
+
+    _EXT = ".pkl"
+
+    def __init__(self) -> None:
+        self._tmpdir: Path | None = None
+
+    # -- context manager -----------------------------------------------------
+
+    def __enter__(self) -> "ChunkBatchStore":
+        self._tmpdir = Path(tempfile.mkdtemp(prefix="onyx_embeddings_"))
+        return self
+
+    def __exit__(self, *_exc: object) -> None:
+        if self._tmpdir is not None:
+            shutil.rmtree(self._tmpdir, ignore_errors=True)
+            self._tmpdir = None
+
+    @property
+    def _dir(self) -> Path:
+        assert self._tmpdir is not None, "ChunkBatchStore used outside context manager"
+        return self._tmpdir
+
+    # -- storage primitives --------------------------------------------------
+
+    def save(self, chunks: list[IndexChunk], batch_idx: int) -> None:
+        """Serialize a batch of embedded chunks to disk."""
+        with open(self._dir / f"batch_{batch_idx}{self._EXT}", "wb") as f:
+            pickle.dump(chunks, f)
+
+    def _load(self, batch_file: Path) -> list[IndexChunk]:
+        """Deserialize a batch of embedded chunks from a file."""
+        with open(batch_file, "rb") as f:
+            return pickle.load(f)
+
+    def _batch_files(self) -> list[Path]:
+        """Return batch files sorted by numeric index."""
+        return sorted(
+            self._dir.glob(f"batch_*{self._EXT}"),
+            key=lambda p: int(p.stem.removeprefix("batch_")),
+        )
+
+    # -- higher-level operations ---------------------------------------------
+
+    def stream(self) -> Iterator[IndexChunk]:
+        """Yield all chunks across all batch files.
+
+        Each call returns a fresh generator, so the data can be iterated
+        multiple times (e.g. once per document index).
+        """
+        for batch_file in self._batch_files():
+            yield from self._load(batch_file)
+
+    def scrub_failed_docs(self, failed_doc_ids: set[str]) -> None:
+        """Remove chunks belonging to *failed_doc_ids* from all batch files.
+
+        When a document fails embedding in batch N, earlier batches may
+        already contain successfully embedded chunks for that document.
+        This ensures the output is all-or-nothing per document.
+        """
+        for batch_file in self._batch_files():
+            batch_chunks = self._load(batch_file)
+            cleaned = [
+                c for c in batch_chunks if c.source_document.id not in failed_doc_ids
+            ]
+            if len(cleaned) != len(batch_chunks):
+                with open(batch_file, "wb") as f:
+                    pickle.dump(cleaned, f)

backend/onyx/indexing/indexing_pipeline.py

@@ -1,5 +1,8 @@
 from collections import defaultdict
 from collections.abc import Callable
+from collections.abc import Generator
+from collections.abc import Iterator
+from contextlib import contextmanager
 from typing import Protocol
 
 from pydantic import BaseModel
@@ -9,6 +12,7 @@ from sqlalchemy.orm import Session
 from onyx.configs.app_configs import DEFAULT_CONTEXTUAL_RAG_LLM_NAME
 from onyx.configs.app_configs import DEFAULT_CONTEXTUAL_RAG_LLM_PROVIDER
 from onyx.configs.app_configs import ENABLE_CONTEXTUAL_RAG
+from onyx.configs.app_configs import MAX_CHUNKS_PER_DOC_BATCH
 from onyx.configs.app_configs import MAX_DOCUMENT_CHARS
 from onyx.configs.app_configs import MAX_TOKENS_FOR_FULL_INCLUSION
 from onyx.configs.app_configs import USE_CHUNK_SUMMARY
@@ -43,10 +47,12 @@ from onyx.document_index.interfaces import DocumentMetadata
 from onyx.document_index.interfaces import IndexBatchParams
 from onyx.file_processing.image_summarization import summarize_image_with_error_handling
 from onyx.file_store.file_store import get_default_file_store
+from onyx.indexing.chunk_batch_store import ChunkBatchStore
 from onyx.indexing.chunker import Chunker
 from onyx.indexing.embedder import embed_chunks_with_failure_handling
 from onyx.indexing.embedder import IndexingEmbedder
 from onyx.indexing.models import DocAwareChunk
+from onyx.indexing.models import DocMetadataAwareIndexChunk
 from onyx.indexing.models import IndexingBatchAdapter
 from onyx.indexing.models import UpdatableChunkData
 from onyx.indexing.vector_db_insertion import write_chunks_to_vector_db_with_backoff
@@ -63,6 +69,7 @@ from onyx.natural_language_processing.utils import tokenizer_trim_middle
 from onyx.prompts.contextual_retrieval import CONTEXTUAL_RAG_PROMPT1
 from onyx.prompts.contextual_retrieval import CONTEXTUAL_RAG_PROMPT2
 from onyx.prompts.contextual_retrieval import DOCUMENT_SUMMARY_PROMPT
+from onyx.utils.batching import batch_generator
 from onyx.utils.logger import setup_logger
 from onyx.utils.postgres_sanitization import sanitize_documents_for_postgres
 from onyx.utils.threadpool_concurrency import run_functions_tuples_in_parallel
@@ -91,6 +98,20 @@ class IndexingPipelineResult(BaseModel):
 
     failures: list[ConnectorFailure]
 
+    @classmethod
+    def empty(cls, total_docs: int) -> "IndexingPipelineResult":
+        return cls(
+            new_docs=0,
+            total_docs=total_docs,
+            total_chunks=0,
+            failures=[],
+        )
+
+
+class ChunkEmbeddingResult(BaseModel):
+    successful_chunk_ids: list[tuple[int, str]]  # (chunk_id, document_id)
+    connector_failures: list[ConnectorFailure]
+
 
 class IndexingPipelineProtocol(Protocol):
     def __call__(
@@ -139,6 +160,110 @@ def _upsert_documents_in_db(
         )
 
 
+def _get_failed_doc_ids(failures: list[ConnectorFailure]) -> set[str]:
+    """Extract document IDs from a list of connector failures."""
+    return {f.failed_document.document_id for f in failures if f.failed_document}
+
+
+def _embed_chunks_to_store(
+    chunks: list[DocAwareChunk],
+    embedder: IndexingEmbedder,
+    tenant_id: str,
+    request_id: str | None,
+    store: ChunkBatchStore,
+) -> ChunkEmbeddingResult:
+    """Embed chunks in batches, spilling each batch to *store*.
+
+    If a document fails embedding in any batch, its chunks are excluded from
+    all batches (including earlier ones already written) so that the output
+    is all-or-nothing per document.
+    """
+    successful_chunk_ids: list[tuple[int, str]] = []
+    all_embedding_failures: list[ConnectorFailure] = []
+    # Track failed doc IDs across all batches so that a failure in batch N
+    # causes chunks for that doc to be skipped in batch N+1 and stripped
+    # from earlier batches.
+    all_failed_doc_ids: set[str] = set()
+
+    for batch_idx, chunk_batch in enumerate(
+        batch_generator(chunks, MAX_CHUNKS_PER_DOC_BATCH)
+    ):
+        # Skip chunks belonging to documents that failed in earlier batches.
+        chunk_batch = [
+            c for c in chunk_batch if c.source_document.id not in all_failed_doc_ids
+        ]
+        if not chunk_batch:
+            continue
+
+        logger.debug(f"Embedding batch {batch_idx}: {len(chunk_batch)} chunks")
+
+        chunks_with_embeddings, embedding_failures = embed_chunks_with_failure_handling(
+            chunks=chunk_batch,
+            embedder=embedder,
+            tenant_id=tenant_id,
+            request_id=request_id,
+        )
+        all_embedding_failures.extend(embedding_failures)
+        all_failed_doc_ids.update(_get_failed_doc_ids(embedding_failures))
+
+        # Only keep successfully embedded chunks for non-failed docs.
+        chunks_with_embeddings = [
+            c
+            for c in chunks_with_embeddings
+            if c.source_document.id not in all_failed_doc_ids
+        ]
+
+        successful_chunk_ids.extend(
+            (c.chunk_id, c.source_document.id) for c in chunks_with_embeddings
+        )
+
+        store.save(chunks_with_embeddings, batch_idx)
+        del chunks_with_embeddings
+
+    # Scrub earlier batches for docs that failed in later batches.
+    if all_failed_doc_ids:
+        store.scrub_failed_docs(all_failed_doc_ids)
+        successful_chunk_ids = [
+            (chunk_id, doc_id)
+            for chunk_id, doc_id in successful_chunk_ids
+            if doc_id not in all_failed_doc_ids
+        ]
+
+    return ChunkEmbeddingResult(
+        successful_chunk_ids=successful_chunk_ids,
+        connector_failures=all_embedding_failures,
+    )
+
+
+@contextmanager
+def embed_and_stream(
+    chunks: list[DocAwareChunk],
+    embedder: IndexingEmbedder,
+    tenant_id: str,
+    request_id: str | None,
+) -> Generator[tuple[ChunkEmbeddingResult, ChunkBatchStore], None, None]:
+    """Embed chunks to disk and yield a ``(result, store)`` pair.
+
+    The store owns the temp directory — files are cleaned up when the context
+    manager exits.
+
+    Usage::
+
+        with embed_and_stream(chunks, embedder, tenant_id, req_id) as (result, store):
+            for chunk in store.stream():
+                ...
+    """
+    with ChunkBatchStore() as store:
+        result = _embed_chunks_to_store(
+            chunks=chunks,
+            embedder=embedder,
+            tenant_id=tenant_id,
+            request_id=request_id,
+            store=store,
+        )
+        yield result, store
+
+
 def get_doc_ids_to_update(
     documents: list[Document], db_docs: list[DBDocument]
 ) -> list[Document]:
@@ -637,6 +762,29 @@ def add_contextual_summaries(
     return chunks
 
 
+def _verify_indexing_completeness(
+    insertion_records: list[DocumentInsertionRecord],
+    write_failures: list[ConnectorFailure],
+    embedding_failed_doc_ids: set[str],
+    updatable_ids: list[str],
+    document_index_name: str,
+) -> None:
+    """Verify that every updatable document was either indexed or reported as failed."""
+    all_returned_doc_ids = (
+        {r.document_id for r in insertion_records}
+        | {f.failed_document.document_id for f in write_failures if f.failed_document}
+        | embedding_failed_doc_ids
+    )
+    if all_returned_doc_ids != set(updatable_ids):
+        raise RuntimeError(
+            f"Some documents were not successfully indexed. "
+            f"Updatable IDs: {updatable_ids}, "
+            f"Returned IDs: {all_returned_doc_ids}. "
+            f"This should never happen. "
+            f"This occured for document index {document_index_name}"
+        )
+
+
 @log_function_time(debug_only=True)
 def index_doc_batch(
     *,
@@ -672,12 +820,7 @@ def index_doc_batch(
     filtered_documents = filter_fnc(document_batch)
     context = adapter.prepare(filtered_documents, ignore_time_skip)
     if not context:
-        return IndexingPipelineResult(
-            new_docs=0,
-            total_docs=len(filtered_documents),
-            total_chunks=0,
-            failures=[],
-        )
+        return IndexingPipelineResult.empty(len(filtered_documents))
 
     # Convert documents to IndexingDocument objects with processed section
     # logger.debug("Processing image sections")
@@ -716,117 +859,99 @@ def index_doc_batch(
         )
 
     logger.debug("Starting embedding")
-    chunks_with_embeddings, embedding_failures = (
-        embed_chunks_with_failure_handling(
-            chunks=chunks,
-            embedder=embedder,
-            tenant_id=tenant_id,
-            request_id=request_id,
-        )
-        if chunks
-        else ([], [])
-    )
+    with embed_and_stream(chunks, embedder, tenant_id, request_id) as (
+        embedding_result,
+        chunk_store,
+    ):
+        updatable_ids = [doc.id for doc in context.updatable_docs]
+        updatable_chunk_data = [
+            UpdatableChunkData(
+                chunk_id=chunk_id,
+                document_id=document_id,
+                boost_score=1.0,
+            )
+            for chunk_id, document_id in embedding_result.successful_chunk_ids
+        ]
 
-    chunk_content_scores = [1.0] * len(chunks_with_embeddings)
-
-    updatable_ids = [doc.id for doc in context.updatable_docs]
-    updatable_chunk_data = [
-        UpdatableChunkData(
-            chunk_id=chunk.chunk_id,
-            document_id=chunk.source_document.id,
-            boost_score=score,
-        )
-        for chunk, score in zip(chunks_with_embeddings, chunk_content_scores)
-    ]
-
-    # Acquires a lock on the documents so that no other process can modify them
-    # NOTE: don't need to acquire till here, since this is when the actual race condition
-    # with Vespa can occur.
-    with adapter.lock_context(context.updatable_docs):
-        # we're concerned about race conditions where multiple simultaneous indexings might result
-        # in one set of metadata overwriting another one in vespa.
-        # we still write data here for the immediate and most likely correct sync, but
-        # to resolve this, an update of the last modified field at the end of this loop
-        # always triggers a final metadata sync via the celery queue
-        result = adapter.build_metadata_aware_chunks(
-            chunks_with_embeddings=chunks_with_embeddings,
-            chunk_content_scores=chunk_content_scores,
-            tenant_id=tenant_id,
-            context=context,
+        embedding_failed_doc_ids = _get_failed_doc_ids(
+            embedding_result.connector_failures
         )
 
-        short_descriptor_list = [chunk.to_short_descriptor() for chunk in result.chunks]
-        short_descriptor_log = str(short_descriptor_list)[:1024]
-        logger.debug(f"Indexing the following chunks: {short_descriptor_log}")
+        # Filter to only successfully embedded chunks so
+        # doc_id_to_new_chunk_cnt reflects what's actually written to Vespa.
+        embedded_chunks = [
+            c for c in chunks if c.source_document.id not in embedding_failed_doc_ids
+        ]
 
-        primary_doc_idx_insertion_records: list[DocumentInsertionRecord] | None = None
-        primary_doc_idx_vector_db_write_failures: list[ConnectorFailure] | None = None
-        for document_index in document_indices:
-            # A document will not be spread across different batches, so all the
-            # documents with chunks in this set, are fully represented by the chunks
-            # in this set
-            (
-                insertion_records,
-                vector_db_write_failures,
-            ) = write_chunks_to_vector_db_with_backoff(
-                document_index=document_index,
-                chunks=result.chunks,
-                index_batch_params=IndexBatchParams(
-                    doc_id_to_previous_chunk_cnt=result.doc_id_to_previous_chunk_cnt,
-                    doc_id_to_new_chunk_cnt=result.doc_id_to_new_chunk_cnt,
-                    tenant_id=tenant_id,
-                    large_chunks_enabled=chunker.enable_large_chunks,
-                ),
+        # Acquires a lock on the documents so that no other process can modify
+        # them.  Not needed until here, since this is when the actual race
+        # condition with vector db can occur.
+        with adapter.lock_context(context.updatable_docs):
+            enricher = adapter.prepare_enrichment(
+                context=context,
+                tenant_id=tenant_id,
+                chunks=embedded_chunks,
             )
 
-            all_returned_doc_ids: set[str] = (
-                {record.document_id for record in insertion_records}
-                .union(
-                    {
-                        record.failed_document.document_id
-                        for record in vector_db_write_failures
-                        if record.failed_document
-                    }
-                )
-                .union(
-                    {
-                        record.failed_document.document_id
-                        for record in embedding_failures
-                        if record.failed_document
-                    }
-                )
+            index_batch_params = IndexBatchParams(
+                doc_id_to_previous_chunk_cnt=enricher.doc_id_to_previous_chunk_cnt,
+                doc_id_to_new_chunk_cnt=enricher.doc_id_to_new_chunk_cnt,
+                tenant_id=tenant_id,
+                large_chunks_enabled=chunker.enable_large_chunks,
             )
-            if all_returned_doc_ids != set(updatable_ids):
-                raise RuntimeError(
-                    f"Some documents were not successfully indexed. "
-                    f"Updatable IDs: {updatable_ids}, "
-                    f"Returned IDs: {all_returned_doc_ids}. "
-                    "This should never happen."
-                    f"This occured for document index {document_index.__class__.__name__}"
-                )
-            # We treat the first document index we got as the primary one used
-            # for reporting the state of indexing.
-            if primary_doc_idx_insertion_records is None:
-                primary_doc_idx_insertion_records = insertion_records
-            if primary_doc_idx_vector_db_write_failures is None:
-                primary_doc_idx_vector_db_write_failures = vector_db_write_failures
 
-        adapter.post_index(
-            context=context,
-            updatable_chunk_data=updatable_chunk_data,
-            filtered_documents=filtered_documents,
-            result=result,
-        )
+            primary_doc_idx_insertion_records: list[DocumentInsertionRecord] | None = (
+                None
+            )
+            primary_doc_idx_vector_db_write_failures: list[ConnectorFailure] | None = (
+                None
+            )
+
+            for document_index in document_indices:
+
+                def _enriched_stream() -> Iterator[DocMetadataAwareIndexChunk]:
+                    for chunk in chunk_store.stream():
+                        yield enricher.enrich_chunk(chunk, 1.0)
+
+                insertion_records, write_failures = (
+                    write_chunks_to_vector_db_with_backoff(
+                        document_index=document_index,
+                        make_chunks=_enriched_stream,
+                        index_batch_params=index_batch_params,
+                    )
+                )
+
+                _verify_indexing_completeness(
+                    insertion_records=insertion_records,
+                    write_failures=write_failures,
+                    embedding_failed_doc_ids=embedding_failed_doc_ids,
+                    updatable_ids=updatable_ids,
+                    document_index_name=document_index.__class__.__name__,
+                )
+                # We treat the first document index we got as the primary one used
+                # for reporting the state of indexing.
+                if primary_doc_idx_insertion_records is None:
+                    primary_doc_idx_insertion_records = insertion_records
+                if primary_doc_idx_vector_db_write_failures is None:
+                    primary_doc_idx_vector_db_write_failures = write_failures
+
+            adapter.post_index(
+                context=context,
+                updatable_chunk_data=updatable_chunk_data,
+                filtered_documents=filtered_documents,
+                enrichment=enricher,
+            )
 
     assert primary_doc_idx_insertion_records is not None
     assert primary_doc_idx_vector_db_write_failures is not None
     return IndexingPipelineResult(
-        new_docs=len(
-            [r for r in primary_doc_idx_insertion_records if not r.already_existed]
+        new_docs=sum(
+            1 for r in primary_doc_idx_insertion_records if not r.already_existed
         ),
         total_docs=len(filtered_documents),
-        total_chunks=len(chunks_with_embeddings),
-        failures=primary_doc_idx_vector_db_write_failures + embedding_failures,
+        total_chunks=len(embedding_result.successful_chunk_ids),
+        failures=primary_doc_idx_vector_db_write_failures
+        + embedding_result.connector_failures,
     )
 
 

backend/onyx/indexing/models.py

@@ -235,12 +235,16 @@ class UpdatableChunkData(BaseModel):
     boost_score: float
 
 
-class BuildMetadataAwareChunksResult(BaseModel):
-    chunks: list[DocMetadataAwareIndexChunk]
+class ChunkEnrichmentContext(Protocol):
+    """Returned by prepare_enrichment. Holds pre-computed metadata lookups
+    and provides per-chunk enrichment."""
+
     doc_id_to_previous_chunk_cnt: dict[str, int]
     doc_id_to_new_chunk_cnt: dict[str, int]
-    user_file_id_to_raw_text: dict[str, str]
-    user_file_id_to_token_count: dict[str, int | None]
+
+    def enrich_chunk(
+        self, chunk: IndexChunk, score: float
+    ) -> DocMetadataAwareIndexChunk: ...
 
 
 class IndexingBatchAdapter(Protocol):
@@ -254,18 +258,24 @@ class IndexingBatchAdapter(Protocol):
     ) -> Generator[TransactionalContext, None, None]:
         """Provide a transaction/row-lock context for critical updates."""
 
-    def build_metadata_aware_chunks(
+    def prepare_enrichment(
         self,
-        chunks_with_embeddings: list[IndexChunk],
-        chunk_content_scores: list[float],
-        tenant_id: str,
         context: "DocumentBatchPrepareContext",
-    ) -> BuildMetadataAwareChunksResult: ...
+        tenant_id: str,
+        chunks: list[DocAwareChunk],
+    ) -> ChunkEnrichmentContext:
+        """Prepare per-chunk enrichment data (access, document sets, boost, etc.).
+
+        Precondition: ``chunks`` have already been through the embedding step
+        (i.e. they are ``IndexChunk`` instances with populated embeddings,
+        passed here as the base ``DocAwareChunk`` type).
+        """
+        ...
 
     def post_index(
         self,
         context: "DocumentBatchPrepareContext",
         updatable_chunk_data: list[UpdatableChunkData],
         filtered_documents: list[Document],
-        result: BuildMetadataAwareChunksResult,
+        enrichment: ChunkEnrichmentContext,
     ) -> None: ...

backend/onyx/indexing/vector_db_insertion.py

@@ -1,6 +1,9 @@
 import time
-from collections import defaultdict
+from collections.abc import Callable
+from collections.abc import Iterable
 from http import HTTPStatus
+from itertools import chain
+from itertools import groupby
 
 import httpx
 
@@ -28,22 +31,22 @@ def _log_insufficient_storage_error(e: Exception) -> None:
 
 def write_chunks_to_vector_db_with_backoff(
     document_index: DocumentIndex,
-    chunks: list[DocMetadataAwareIndexChunk],
+    make_chunks: Callable[[], Iterable[DocMetadataAwareIndexChunk]],
     index_batch_params: IndexBatchParams,
 ) -> tuple[list[DocumentInsertionRecord], list[ConnectorFailure]]:
     """Tries to insert all chunks in one large batch. If that batch fails for any reason,
     goes document by document to isolate the failure(s).
 
     IMPORTANT: must pass in whole documents at a time not individual chunks, since the
-    vector DB interface assumes that all chunks for a single document are present.
+    vector DB interface assumes that all chunks for a single document are present. The
+    chunks must also be in contiguous batches
     """
-
     # first try to write the chunks to the vector db
     try:
         return (
             list(
                 document_index.index(
-                    chunks=chunks,
+                    chunks=make_chunks(),
                     index_batch_params=index_batch_params,
                 )
             ),
@@ -60,14 +63,23 @@ def write_chunks_to_vector_db_with_backoff(
         # wait a couple seconds just to give the vector db a chance to recover
         time.sleep(2)
 
-    # try writing each doc one by one
-    chunks_for_docs: dict[str, list[DocMetadataAwareIndexChunk]] = defaultdict(list)
-    for chunk in chunks:
-        chunks_for_docs[chunk.source_document.id].append(chunk)
-
     insertion_records: list[DocumentInsertionRecord] = []
     failures: list[ConnectorFailure] = []
-    for doc_id, chunks_for_doc in chunks_for_docs.items():
+
+    def key(chunk: DocMetadataAwareIndexChunk) -> str:
+        return chunk.source_document.id
+
+    seen_doc_ids: set[str] = set()
+    for doc_id, chunks_for_doc in groupby(make_chunks(), key=key):
+        if doc_id in seen_doc_ids:
+            raise RuntimeError(
+                f"Doc chunks are not arriving in order. Current doc_id={doc_id}, seen_doc_ids={list(seen_doc_ids)}"
+            )
+        seen_doc_ids.add(doc_id)
+
+        first_chunk = next(chunks_for_doc)
+        chunks_for_doc = chain([first_chunk], chunks_for_doc)
+
         try:
             insertion_records.extend(
                 document_index.index(
@@ -87,9 +99,7 @@ def write_chunks_to_vector_db_with_backoff(
                 ConnectorFailure(
                     failed_document=DocumentFailure(
                         document_id=doc_id,
-                        document_link=(
-                            chunks_for_doc[0].get_link() if chunks_for_doc else None
-                        ),
+                        document_link=first_chunk.get_link(),
                     ),
                     failure_message=str(e),
                     exception=e,

backend/onyx/main.py

@@ -77,7 +77,6 @@ from onyx.server.features.default_assistant.api import (
 )
 from onyx.server.features.document_set.api import router as document_set_router
 from onyx.server.features.hierarchy.api import router as hierarchy_router
-from onyx.server.features.hooks.api import router as hook_router
 from onyx.server.features.input_prompt.api import (
     admin_router as admin_input_prompt_router,
 )
@@ -439,6 +438,7 @@ def get_application(lifespan_override: Lifespan | None = None) -> FastAPI:
             dsn=SENTRY_DSN,
             integrations=[StarletteIntegration(), FastApiIntegration()],
             traces_sample_rate=0.1,
+            release=__version__,
         )
         logger.info("Sentry initialized")
     else:
@@ -454,7 +454,6 @@ def get_application(lifespan_override: Lifespan | None = None) -> FastAPI:
 
     register_onyx_exception_handlers(application)
 
-    include_router_with_global_prefix_prepended(application, hook_router)
     include_router_with_global_prefix_prepended(application, password_router)
     include_router_with_global_prefix_prepended(application, chat_router)
     include_router_with_global_prefix_prepended(application, query_router)

backend/onyx/natural_language_processing/utils.py

@@ -175,6 +175,32 @@ def get_tokenizer(
     return _check_tokenizer_cache(provider_type, model_name)
 
 
+# Max characters per encode() call.
+_ENCODE_CHUNK_SIZE = 500_000
+
+
+def count_tokens(
+    text: str,
+    tokenizer: BaseTokenizer,
+    token_limit: int | None = None,
+) -> int:
+    """Count tokens, chunking the input to avoid tiktoken stack overflow.
+
+    If token_limit is provided and the text is large enough to require
+    multiple chunks (> 500k chars), stops early once the count exceeds it.
+    When early-exiting, the returned value exceeds token_limit but may be
+    less than the true full token count.
+    """
+    if len(text) <= _ENCODE_CHUNK_SIZE:
+        return len(tokenizer.encode(text))
+    total = 0
+    for start in range(0, len(text), _ENCODE_CHUNK_SIZE):
+        total += len(tokenizer.encode(text[start : start + _ENCODE_CHUNK_SIZE]))
+        if token_limit is not None and total > token_limit:
+            return total  # Already over — skip remaining chunks
+    return total
+
+
 def tokenizer_trim_content(
     content: str, desired_length: int, tokenizer: BaseTokenizer
 ) -> str:

backend/onyx/server/features/build/sandbox/kubernetes/docker/templates/outputs/web/package-lock.json

@@ -3844,9 +3844,9 @@
       }
     },
     "node_modules/@ts-morph/common/node_modules/brace-expansion": {
-      "version": "5.0.3",
-      "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-5.0.3.tgz",
-      "integrity": "sha512-fy6KJm2RawA5RcHkLa1z/ScpBeA762UF9KmZQxwIbDtRJrgLzM10depAiEQ+CXYcoiqW1/m96OAAoke2nE9EeA==",
+      "version": "5.0.5",
+      "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-5.0.5.tgz",
+      "integrity": "sha512-VZznLgtwhn+Mact9tfiwx64fA9erHH/MCXEUfB/0bX/6Fz6ny5EGTXYltMocqg4xFAQZtnO3DHWWXi8RiuN7cQ==",
       "license": "MIT",
       "dependencies": {
         "balanced-match": "^4.0.2"
@@ -4224,9 +4224,9 @@
       }
     },
     "node_modules/@typescript-eslint/typescript-estree/node_modules/brace-expansion": {
-      "version": "2.0.2",
-      "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-2.0.2.tgz",
-      "integrity": "sha512-Jt0vHyM+jmUBqojB7E1NIYadt0vI0Qxjxd2TErW94wDz+E2LAm5vKMXXwg6ZZBTHPuUlDgQHKXvjGBdfcF1ZDQ==",
+      "version": "2.0.3",
+      "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-2.0.3.tgz",
+      "integrity": "sha512-MCV/fYJEbqx68aE58kv2cA/kiky1G8vux3OR6/jbS+jIMe/6fJWa0DTzJU7dqijOWYwHi1t29FlfYI9uytqlpA==",
       "dev": true,
       "license": "MIT",
       "dependencies": {
@@ -5007,9 +5007,9 @@
       }
     },
     "node_modules/brace-expansion": {
-      "version": "1.1.12",
-      "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.12.tgz",
-      "integrity": "sha512-9T9UjW3r0UW5c1Q7GTwllptXwhvYmEzFhzMfZ9H7FQWt+uZePjZPjBP/W1ZEyZ1twGWom5/56TF4lPcqjnDHcg==",
+      "version": "1.1.13",
+      "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-1.1.13.tgz",
+      "integrity": "sha512-9ZLprWS6EENmhEOpjCYW2c8VkmOvckIJZfkr7rBW6dObmfgJ/L1GpSYW5Hpo9lDz4D1+n0Ckz8rU7FwHDQiG/w==",
       "dev": true,
       "license": "MIT",
       "dependencies": {

backend/onyx/server/features/projects/projects_file_utils.py

@@ -9,20 +9,15 @@ from pydantic import ConfigDict
 from pydantic import Field
 from sqlalchemy.orm import Session
 
-from onyx.configs.app_configs import FILE_TOKEN_COUNT_THRESHOLD
-from onyx.configs.app_configs import USER_FILE_MAX_UPLOAD_SIZE_BYTES
-from onyx.configs.app_configs import USER_FILE_MAX_UPLOAD_SIZE_MB
 from onyx.db.llm import fetch_default_llm_model
 from onyx.file_processing.extract_file_text import extract_file_text
 from onyx.file_processing.extract_file_text import get_file_ext
 from onyx.file_processing.file_types import OnyxFileExtensions
 from onyx.file_processing.password_validation import is_file_password_protected
+from onyx.natural_language_processing.utils import count_tokens
 from onyx.natural_language_processing.utils import get_tokenizer
+from onyx.server.settings.store import load_settings
 from onyx.utils.logger import setup_logger
-from shared_configs.configs import MULTI_TENANT
-from shared_configs.configs import SKIP_USERFILE_THRESHOLD
-from shared_configs.configs import SKIP_USERFILE_THRESHOLD_TENANT_LIST
-from shared_configs.contextvars import get_current_tenant_id
 
 
 logger = setup_logger()
@@ -81,11 +76,18 @@ class CategorizedFiles(BaseModel):
     acceptable: list[UploadFile] = Field(default_factory=list)
     rejected: list[RejectedFile] = Field(default_factory=list)
     acceptable_file_to_token_count: dict[str, int] = Field(default_factory=dict)
+    # Filenames within `acceptable` that should be stored but not indexed.
+    skip_indexing: set[str] = Field(default_factory=set)
 
     # Allow FastAPI UploadFile instances
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
 
+def _skip_token_threshold(extension: str) -> bool:
+    """Return True if this file extension should bypass the token limit."""
+    return extension.lower() in OnyxFileExtensions.TABULAR_EXTENSIONS
+
+
 def _apply_long_side_cap(width: int, height: int, cap: int) -> tuple[int, int]:
     if max(width, height) <= cap:
         return width, height
@@ -161,8 +163,8 @@ def categorize_uploaded_files(
       document formats (.pdf, .docx, …) and falls back to a text-detection
       heuristic for unknown extensions (.py, .js, .rs, …).
     - Uses default tokenizer to compute token length.
-    - If token length > threshold, reject file (unless threshold skip is enabled).
-    - If text cannot be extracted, reject file.
+    - If token length exceeds the admin-configured threshold, reject file.
+    - If extension unsupported or text cannot be extracted, reject file.
     - Otherwise marked as acceptable.
     """
 
@@ -173,36 +175,33 @@ def categorize_uploaded_files(
     provider_type = default_model.llm_provider.provider if default_model else None
     tokenizer = get_tokenizer(model_name=model_name, provider_type=provider_type)
 
-    # Check if threshold checks should be skipped
-    skip_threshold = False
-
-    # Check global skip flag (works for both single-tenant and multi-tenant)
-    if SKIP_USERFILE_THRESHOLD:
-        skip_threshold = True
-        logger.info("Skipping userfile threshold check (global setting)")
-    # Check tenant-specific skip list (only applicable in multi-tenant)
-    elif MULTI_TENANT and SKIP_USERFILE_THRESHOLD_TENANT_LIST:
-        try:
-            current_tenant_id = get_current_tenant_id()
-            skip_threshold = current_tenant_id in SKIP_USERFILE_THRESHOLD_TENANT_LIST
-            if skip_threshold:
-                logger.info(
-                    f"Skipping userfile threshold check for tenant: {current_tenant_id}"
-                )
-        except RuntimeError as e:
-            logger.warning(f"Failed to get current tenant ID: {str(e)}")
+    # Derive limits from admin-configurable settings.
+    # For upload size: load_settings() resolves 0/None to a positive default.
+    # For token threshold: 0 means "no limit" (converted to None below).
+    settings = load_settings()
+    max_upload_size_mb = (
+        settings.user_file_max_upload_size_mb
+    )  # always positive after load_settings()
+    max_upload_size_bytes = (
+        max_upload_size_mb * 1024 * 1024 if max_upload_size_mb else None
+    )
+    token_threshold_k = settings.file_token_count_threshold_k
+    token_threshold = (
+        token_threshold_k * 1000 if token_threshold_k else None
+    )  # 0 → None = no limit
 
     for upload in files:
         try:
             filename = get_safe_filename(upload)
 
-            # Size limit is a hard safety cap and is enforced even when token
-            # threshold checks are skipped via SKIP_USERFILE_THRESHOLD settings.
-            if is_upload_too_large(upload, USER_FILE_MAX_UPLOAD_SIZE_BYTES):
+            # Size limit is a hard safety cap.
+            if max_upload_size_bytes is not None and is_upload_too_large(
+                upload, max_upload_size_bytes
+            ):
                 results.rejected.append(
                     RejectedFile(
                         filename=filename,
-                        reason=f"Exceeds {USER_FILE_MAX_UPLOAD_SIZE_MB} MB file size limit",
+                        reason=f"Exceeds {max_upload_size_mb} MB file size limit",
                     )
                 )
                 continue
@@ -224,11 +223,11 @@ def categorize_uploaded_files(
                     )
                     continue
 
-                if not skip_threshold and token_count > FILE_TOKEN_COUNT_THRESHOLD:
+                if token_threshold is not None and token_count > token_threshold:
                     results.rejected.append(
                         RejectedFile(
                             filename=filename,
-                            reason=f"Exceeds {FILE_TOKEN_COUNT_THRESHOLD} token limit",
+                            reason=f"Exceeds {token_threshold_k}K token limit",
                         )
                     )
                 else:
@@ -269,12 +268,24 @@ def categorize_uploaded_files(
                     )
                     continue
 
-                token_count = len(tokenizer.encode(text_content))
-                if not skip_threshold and token_count > FILE_TOKEN_COUNT_THRESHOLD:
+                token_count = count_tokens(
+                    text_content, tokenizer, token_limit=token_threshold
+                )
+                exceeds_threshold = (
+                    token_threshold is not None and token_count > token_threshold
+                )
+                if exceeds_threshold and _skip_token_threshold(extension):
+                    # Exempt extensions (e.g. spreadsheets) are accepted
+                    # but flagged to skip indexing — only metadata is
+                    # injected into the LLM context.
+                    results.acceptable.append(upload)
+                    results.acceptable_file_to_token_count[filename] = token_count
+                    results.skip_indexing.add(filename)
+                elif exceeds_threshold:
                     results.rejected.append(
                         RejectedFile(
                             filename=filename,
-                            reason=f"Exceeds {FILE_TOKEN_COUNT_THRESHOLD} token limit",
+                            reason=f"Exceeds {token_threshold_k}K token limit",
                         )
                     )
                 else:

backend/onyx/server/manage/models.py

@@ -147,7 +147,6 @@ class UserInfo(BaseModel):
         is_anonymous_user: bool | None = None,
         tenant_info: TenantInfo | None = None,
         assistant_specific_configs: UserSpecificAssistantPreferences | None = None,
-        memories: list[MemoryItem] | None = None,
     ) -> "UserInfo":
         return cls(
             id=str(user.id),
@@ -192,7 +191,10 @@ class UserInfo(BaseModel):
                 role=user.personal_role or "",
                 use_memories=user.use_memories,
                 enable_memory_tool=user.enable_memory_tool,
-                memories=memories or [],
+                memories=[
+                    MemoryItem(id=memory.id, content=memory.memory_text)
+                    for memory in (user.memories or [])
+                ],
                 user_preferences=user.user_preferences or "",
             ),
         )

backend/onyx/server/manage/users.py

@@ -57,7 +57,6 @@ from onyx.db.user_preferences import activate_user
 from onyx.db.user_preferences import deactivate_user
 from onyx.db.user_preferences import get_all_user_assistant_specific_configs
 from onyx.db.user_preferences import get_latest_access_token_for_user
-from onyx.db.user_preferences import get_memories_for_user
 from onyx.db.user_preferences import update_assistant_preferences
 from onyx.db.user_preferences import update_user_assistant_visibility
 from onyx.db.user_preferences import update_user_auto_scroll
@@ -824,11 +823,6 @@ def verify_user_logged_in(
             [],
         ),
     )
-    memories = [
-        MemoryItem(id=memory.id, content=memory.memory_text)
-        for memory in get_memories_for_user(user.id, db_session)
-    ]
-
     user_info = UserInfo.from_model(
         user,
         current_token_created_at=token_created_at,
@@ -839,7 +833,6 @@ def verify_user_logged_in(
             new_tenant=new_tenant,
             invitation=tenant_invitation,
         ),
-        memories=memories,
     )
 
     return user_info
@@ -937,8 +930,7 @@ def update_user_personalization_api(
         else user.enable_memory_tool
     )
     existing_memories = [
-        MemoryItem(id=memory.id, content=memory.memory_text)
-        for memory in get_memories_for_user(user.id, db_session)
+        MemoryItem(id=memory.id, content=memory.memory_text) for memory in user.memories
     ]
     new_memories = (
         request.memories if request.memories is not None else existing_memories

backend/onyx/server/query_and_chat/chat_utils.py

@@ -9,8 +9,8 @@ def mime_type_to_chat_file_type(mime_type: str | None) -> ChatFileType:
     if mime_type in OnyxMimeTypes.IMAGE_MIME_TYPES:
         return ChatFileType.IMAGE
 
-    if mime_type in OnyxMimeTypes.CSV_MIME_TYPES:
-        return ChatFileType.CSV
+    if mime_type in OnyxMimeTypes.TABULAR_MIME_TYPES:
+        return ChatFileType.TABULAR
 
     if mime_type in OnyxMimeTypes.DOCUMENT_MIME_TYPES:
         return ChatFileType.DOC

backend/onyx/server/settings/api.py

@@ -9,7 +9,9 @@ from onyx import __version__ as onyx_version
 from onyx.auth.users import current_admin_user
 from onyx.auth.users import current_user
 from onyx.auth.users import is_user_admin
+from onyx.configs.app_configs import DEFAULT_USER_FILE_MAX_UPLOAD_SIZE_MB
 from onyx.configs.app_configs import DISABLE_VECTOR_DB
+from onyx.configs.app_configs import MAX_ALLOWED_UPLOAD_SIZE_MB
 from onyx.configs.constants import KV_REINDEX_KEY
 from onyx.configs.constants import NotificationType
 from onyx.db.engine.sql_engine import get_session
@@ -17,10 +19,15 @@ from onyx.db.models import User
 from onyx.db.notification import dismiss_all_notifications
 from onyx.db.notification import get_notifications
 from onyx.db.notification import update_notification_last_shown
-from onyx.hooks.utils import HOOKS_AVAILABLE
+from onyx.error_handling.error_codes import OnyxErrorCode
+from onyx.error_handling.exceptions import OnyxError
 from onyx.key_value_store.factory import get_kv_store
 from onyx.key_value_store.interface import KvKeyNotFoundError
 from onyx.server.features.build.utils import is_onyx_craft_enabled
+from onyx.server.settings.models import (
+    DEFAULT_FILE_TOKEN_COUNT_THRESHOLD_K_NO_VECTOR_DB,
+)
+from onyx.server.settings.models import DEFAULT_FILE_TOKEN_COUNT_THRESHOLD_K_VECTOR_DB
 from onyx.server.settings.models import Notification
 from onyx.server.settings.models import Settings
 from onyx.server.settings.models import UserSettings
@@ -30,6 +37,7 @@ from onyx.utils.logger import setup_logger
 from onyx.utils.variable_functionality import (
     fetch_versioned_implementation_with_fallback,
 )
+from shared_configs.configs import MULTI_TENANT
 
 logger = setup_logger()
 
@@ -41,6 +49,15 @@ basic_router = APIRouter(prefix="/settings")
 def admin_put_settings(
     settings: Settings, _: User = Depends(current_admin_user)
 ) -> None:
+    if (
+        settings.user_file_max_upload_size_mb is not None
+        and settings.user_file_max_upload_size_mb > 0
+        and settings.user_file_max_upload_size_mb > MAX_ALLOWED_UPLOAD_SIZE_MB
+    ):
+        raise OnyxError(
+            OnyxErrorCode.INVALID_INPUT,
+            f"File upload size limit cannot exceed {MAX_ALLOWED_UPLOAD_SIZE_MB} MB",
+        )
     store_settings(settings)
 
 
@@ -81,8 +98,18 @@ def fetch_settings(
         needs_reindexing=needs_reindexing,
         onyx_craft_enabled=onyx_craft_enabled_for_user,
         vector_db_enabled=not DISABLE_VECTOR_DB,
-        hooks_enabled=HOOKS_AVAILABLE,
+        hooks_enabled=not MULTI_TENANT,
         version=onyx_version,
+        max_allowed_upload_size_mb=MAX_ALLOWED_UPLOAD_SIZE_MB,
+        default_user_file_max_upload_size_mb=min(
+            DEFAULT_USER_FILE_MAX_UPLOAD_SIZE_MB,
+            MAX_ALLOWED_UPLOAD_SIZE_MB,
+        ),
+        default_file_token_count_threshold_k=(
+            DEFAULT_FILE_TOKEN_COUNT_THRESHOLD_K_NO_VECTOR_DB
+            if DISABLE_VECTOR_DB
+            else DEFAULT_FILE_TOKEN_COUNT_THRESHOLD_K_VECTOR_DB
+        ),
     )
 
 

backend/onyx/server/settings/models.py

@@ -2,12 +2,19 @@ from datetime import datetime
 from enum import Enum
 
 from pydantic import BaseModel
+from pydantic import Field
 
+from onyx.configs.app_configs import DEFAULT_USER_FILE_MAX_UPLOAD_SIZE_MB
+from onyx.configs.app_configs import DISABLE_VECTOR_DB
+from onyx.configs.app_configs import MAX_ALLOWED_UPLOAD_SIZE_MB
 from onyx.configs.constants import NotificationType
 from onyx.configs.constants import QueryHistoryType
 from onyx.db.models import Notification as NotificationDBModel
 from shared_configs.configs import POSTGRES_DEFAULT_SCHEMA
 
+DEFAULT_FILE_TOKEN_COUNT_THRESHOLD_K_VECTOR_DB = 200
+DEFAULT_FILE_TOKEN_COUNT_THRESHOLD_K_NO_VECTOR_DB = 10000
+
 
 class PageType(str, Enum):
     CHAT = "chat"
@@ -78,7 +85,12 @@ class Settings(BaseModel):
 
     # User Knowledge settings
     user_knowledge_enabled: bool | None = True
-    user_file_max_upload_size_mb: int | None = None
+    user_file_max_upload_size_mb: int | None = Field(
+        default=DEFAULT_USER_FILE_MAX_UPLOAD_SIZE_MB, ge=0
+    )
+    file_token_count_threshold_k: int | None = Field(
+        default=None, ge=0  # thousands of tokens; None = context-aware default
+    )
 
     # Connector settings
     show_extra_connectors: bool | None = True
@@ -104,7 +116,18 @@ class UserSettings(Settings):
     # False when DISABLE_VECTOR_DB is set — connectors, RAG search, and
     # document sets are unavailable.
     vector_db_enabled: bool = True
-    # True when hooks are available: single-tenant deployment with HOOK_ENABLED=true.
+    # True when hooks are available: single-tenant EE deployments only.
     hooks_enabled: bool = False
     # Application version, read from the ONYX_VERSION env var at startup.
     version: str | None = None
+    # Hard ceiling for user_file_max_upload_size_mb, derived from env var.
+    max_allowed_upload_size_mb: int = MAX_ALLOWED_UPLOAD_SIZE_MB
+    # Factory defaults so the frontend can show a "restore default" button.
+    default_user_file_max_upload_size_mb: int = DEFAULT_USER_FILE_MAX_UPLOAD_SIZE_MB
+    default_file_token_count_threshold_k: int = Field(
+        default_factory=lambda: (
+            DEFAULT_FILE_TOKEN_COUNT_THRESHOLD_K_NO_VECTOR_DB
+            if DISABLE_VECTOR_DB
+            else DEFAULT_FILE_TOKEN_COUNT_THRESHOLD_K_VECTOR_DB
+        )
+    )

backend/onyx/server/settings/store.py

@@ -1,13 +1,19 @@
 from onyx.cache.factory import get_cache_backend
+from onyx.configs.app_configs import DEFAULT_USER_FILE_MAX_UPLOAD_SIZE_MB
 from onyx.configs.app_configs import DISABLE_USER_KNOWLEDGE
+from onyx.configs.app_configs import DISABLE_VECTOR_DB
 from onyx.configs.app_configs import ENABLE_OPENSEARCH_INDEXING_FOR_ONYX
+from onyx.configs.app_configs import MAX_ALLOWED_UPLOAD_SIZE_MB
 from onyx.configs.app_configs import ONYX_QUERY_HISTORY_TYPE
 from onyx.configs.app_configs import SHOW_EXTRA_CONNECTORS
-from onyx.configs.app_configs import USER_FILE_MAX_UPLOAD_SIZE_MB
 from onyx.configs.constants import KV_SETTINGS_KEY
 from onyx.configs.constants import OnyxRedisLocks
 from onyx.key_value_store.factory import get_kv_store
 from onyx.key_value_store.interface import KvKeyNotFoundError
+from onyx.server.settings.models import (
+    DEFAULT_FILE_TOKEN_COUNT_THRESHOLD_K_NO_VECTOR_DB,
+)
+from onyx.server.settings.models import DEFAULT_FILE_TOKEN_COUNT_THRESHOLD_K_VECTOR_DB
 from onyx.server.settings.models import Settings
 from onyx.utils.logger import setup_logger
 
@@ -51,9 +57,36 @@ def load_settings() -> Settings:
     if DISABLE_USER_KNOWLEDGE:
         settings.user_knowledge_enabled = False
 
-    settings.user_file_max_upload_size_mb = USER_FILE_MAX_UPLOAD_SIZE_MB
     settings.show_extra_connectors = SHOW_EXTRA_CONNECTORS
     settings.opensearch_indexing_enabled = ENABLE_OPENSEARCH_INDEXING_FOR_ONYX
+
+    # Resolve context-aware defaults for token threshold.
+    # None = admin hasn't set a value yet → use context-aware default.
+    # 0 = admin explicitly chose "no limit" → preserve as-is.
+    if settings.file_token_count_threshold_k is None:
+        settings.file_token_count_threshold_k = (
+            DEFAULT_FILE_TOKEN_COUNT_THRESHOLD_K_NO_VECTOR_DB
+            if DISABLE_VECTOR_DB
+            else DEFAULT_FILE_TOKEN_COUNT_THRESHOLD_K_VECTOR_DB
+        )
+
+    # Upload size: 0 and None are treated as "unset" (not "no limit") →
+    # fall back to min(configured default, hard ceiling).
+    if not settings.user_file_max_upload_size_mb:
+        settings.user_file_max_upload_size_mb = min(
+            DEFAULT_USER_FILE_MAX_UPLOAD_SIZE_MB,
+            MAX_ALLOWED_UPLOAD_SIZE_MB,
+        )
+
+    # Clamp to env ceiling so stale KV values are capped even if the
+    # operator lowered MAX_ALLOWED_UPLOAD_SIZE_MB after a higher value
+    # was already saved (api.py only guards new writes).
+    if (
+        settings.user_file_max_upload_size_mb > 0
+        and settings.user_file_max_upload_size_mb > MAX_ALLOWED_UPLOAD_SIZE_MB
+    ):
+        settings.user_file_max_upload_size_mb = MAX_ALLOWED_UPLOAD_SIZE_MB
+
     return settings
 
 

backend/onyx/tools/tool_implementations/file_reader/file_reader_tool.py

@@ -1,3 +1,4 @@
+import io
 import json
 from typing import Any
 from typing import cast
@@ -9,6 +10,7 @@ from typing_extensions import override
 from onyx.chat.emitter import Emitter
 from onyx.configs.app_configs import DISABLE_VECTOR_DB
 from onyx.db.engine.sql_engine import get_session_with_current_tenant
+from onyx.file_processing.extract_file_text import extract_file_text
 from onyx.file_store.models import ChatFileType
 from onyx.file_store.models import InMemoryChatFile
 from onyx.file_store.utils import load_chat_file_by_id
@@ -169,10 +171,13 @@ class FileReaderTool(Tool[FileReaderToolOverrideKwargs]):
 
         chat_file = self._load_file(file_id)
 
-        # Only PLAIN_TEXT and CSV are guaranteed to contain actual text bytes.
+        # Only PLAIN_TEXT and TABULAR are guaranteed to contain actual text bytes.
         # DOC type in a loaded file means plaintext extraction failed and the
         # content is the original binary (e.g. raw PDF/DOCX bytes).
-        if chat_file.file_type not in (ChatFileType.PLAIN_TEXT, ChatFileType.CSV):
+        if chat_file.file_type not in (
+            ChatFileType.PLAIN_TEXT,
+            ChatFileType.TABULAR,
+        ):
             raise ToolCallException(
                 message=f"File {file_id} is not a text file (type={chat_file.file_type})",
                 llm_facing_message=(
@@ -181,7 +186,19 @@ class FileReaderTool(Tool[FileReaderToolOverrideKwargs]):
             )
 
         try:
-            full_text = chat_file.content.decode("utf-8", errors="replace")
+            if chat_file.file_type == ChatFileType.PLAIN_TEXT:
+                full_text = chat_file.content.decode("utf-8", errors="replace")
+            else:
+                full_text = (
+                    extract_file_text(
+                        file=io.BytesIO(chat_file.content),
+                        file_name=chat_file.filename or "",
+                        break_on_unprocessable=False,
+                    )
+                    or ""
+                )
+        except ToolCallException:
+            raise
         except Exception:
             raise ToolCallException(
                 message=f"Failed to decode file {file_id}",

backend/requirements/default.txt

@@ -14,7 +14,7 @@ aiofiles==25.1.0
     #   unstructured-client
 aiohappyeyeballs==2.6.1
     # via aiohttp
-aiohttp==3.13.3
+aiohttp==3.13.4
     # via
     #   aiobotocore
     #   discord-py
@@ -187,7 +187,7 @@ coloredlogs==15.0.1
     # via onnxruntime
 courlan==1.3.2
     # via trafilatura
-cryptography==46.0.5
+cryptography==46.0.6
     # via
     #   authlib
     #   google-auth
@@ -271,7 +271,7 @@ fastapi-users-db-sqlalchemy==7.0.0
     # via onyx
 fastavro==1.12.1
     # via cohere
-fastmcp==3.0.2
+fastmcp==3.2.0
     # via onyx
 fastuuid==0.14.0
     # via litellm
@@ -449,7 +449,7 @@ kombu==5.5.4
     # via celery
 kubernetes==31.0.0
     # via onyx
-langchain-core==1.2.11
+langchain-core==1.2.22
     # via onyx
 langdetect==1.0.9
     # via unstructured
@@ -735,7 +735,7 @@ pyee==13.0.0
     # via playwright
 pygithub==2.5.0
     # via onyx
-pygments==2.19.2
+pygments==2.20.0
     # via rich
 pyjwt==2.12.0
     # via
@@ -1102,6 +1102,8 @@ tzdata==2025.2
     #   tzlocal
 tzlocal==5.3.1
     # via dateparser
+uncalled-for==0.2.0
+    # via fastmcp
 unstructured==0.18.27
     # via onyx
 unstructured-client==0.42.6

backend/requirements/dev.txt

@@ -10,7 +10,7 @@ aiofiles==25.1.0
     # via aioboto3
 aiohappyeyeballs==2.6.1
     # via aiohttp
-aiohttp==3.13.3
+aiohttp==3.13.4
     # via
     #   aiobotocore
     #   discord-py
@@ -97,7 +97,7 @@ comm==0.2.3
     # via ipykernel
 contourpy==1.3.3
     # via matplotlib
-cryptography==46.0.5
+cryptography==46.0.6
     # via
     #   google-auth
     #   pyjwt
@@ -263,7 +263,7 @@ oauthlib==3.2.2
     # via
     #   kubernetes
     #   requests-oauthlib
-onyx-devtools==0.7.1
+onyx-devtools==0.7.2
     # via onyx
 openai==2.14.0
     # via
@@ -349,7 +349,7 @@ pydantic-core==2.33.2
     # via pydantic
 pydantic-settings==2.12.0
     # via mcp
-pygments==2.19.2
+pygments==2.20.0
     # via
     #   ipython
     #   ipython-pygments-lexers

backend/requirements/ee.txt

@@ -10,7 +10,7 @@ aiofiles==25.1.0
     # via aioboto3
 aiohappyeyeballs==2.6.1
     # via aiohttp
-aiohttp==3.13.3
+aiohttp==3.13.4
     # via
     #   aiobotocore
     #   discord-py
@@ -76,7 +76,7 @@ colorama==0.4.6 ; sys_platform == 'win32'
     # via
     #   click
     #   tqdm
-cryptography==46.0.5
+cryptography==46.0.6
     # via
     #   google-auth
     #   pyjwt

backend/requirements/model_server.txt

@@ -12,7 +12,7 @@ aiofiles==25.1.0
     # via aioboto3
 aiohappyeyeballs==2.6.1
     # via aiohttp
-aiohttp==3.13.3
+aiohttp==3.13.4
     # via
     #   aiobotocore
     #   discord-py
@@ -92,7 +92,7 @@ colorama==0.4.6 ; sys_platform == 'win32'
     # via
     #   click
     #   tqdm
-cryptography==46.0.5
+cryptography==46.0.6
     # via
     #   google-auth
     #   pyjwt