chore: release v3.71.0

* chore(otel): add metrics for shallow trace events and sdk batches

* push

.cursor/Dockerfile

@@ -0,0 +1,47 @@
+FROM node:20
+
+# ---------- System packages --------------------------------------------------
+# The buildpack-deps base already ships git, build-essential, python, etc.
+# Add a few extra tools handy during Langfuse development.
+RUN apt-get update && \
+    DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
+      openssl \
+      wget \
+      curl \
+      ca-certificates \
+      postgresql-client \
+      redis-tools \
+      less nano \
+    && rm -rf /var/lib/apt/lists/*
+
+# ---------- pnpm -------------------------------------------------------------
+# Langfuse monorepo relies on pnpm 9.5.0 (see CONTRIBUTING.md)
+ENV PNPM_HOME="/pnpm"
+ENV PATH="$PNPM_HOME:$PATH"
+RUN corepack enable && \
+    corepack prepare pnpm@9.5.0 --activate
+
+# ---------- golang-migrate ----------------------------------------------------
+# CLI used for database migrations during development.
+ENV MIGRATE_VERSION=4.18.2
+RUN wget -qO- "https://github.com/golang-migrate/migrate/releases/download/v${MIGRATE_VERSION}/migrate.linux-amd64.tar.gz" \
+    | tar -xz -C /usr/local/bin && \
+    chmod +x /usr/local/bin/migrate
+
+# ---------- Non-root user -----------------------------------------------------
+# Use root for convenience in development containers
+WORKDIR /workspace
+
+# Create non-root user
+RUN useradd -ms /bin/bash ubuntu
+
+# Pre-create pnpm store and set correct ownership to avoid first-run cost & permission issues
+RUN pnpm store path > /dev/null && \
+    chown -R ubuntu:ubuntu /pnpm
+
+USER ubuntu
+WORKDIR /home/ubuntu
+ENV HOME=/home/ubuntu
+
+# Container starts with a bash shell ready for hacking.
+CMD ["bash"]

.cursor/environment.json

@@ -0,0 +1,8 @@
+{
+  "install": "cp .env.dev.example .env && pnpm i",
+  "build": {
+    "context": ".",
+    "dockerfile": "Dockerfile"
+  },
+  "start": "pnpm dx-f"
+}

.cursor/rules/general-info.mdc

@@ -0,0 +1,8 @@
+---
+description: 
+globs: 
+alwaysApply: true
+---
+# General rules
+
+- Linting in this repo only works if the development server is running

.env.dev-redis-cluster.example

@@ -0,0 +1,84 @@
+# When adding additional environment variables, the schema in "/src/env.mjs"
+# should be updated accordingly.
+
+# Prisma
+# https://www.prisma.io/docs/reference/database-reference/connection-urls#env
+DIRECT_URL="postgresql://postgres:postgres@localhost:5432/postgres"
+DATABASE_URL="postgresql://postgres:postgres@localhost:5432/postgres"
+
+# Clickhouse
+CLICKHOUSE_MIGRATION_URL="clickhouse://localhost:9000"
+CLICKHOUSE_URL="http://localhost:8123"
+CLICKHOUSE_USER="clickhouse"
+CLICKHOUSE_PASSWORD="clickhouse"
+CLICKHOUSE_CLUSTER_ENABLED="false"
+
+# Next Auth
+# You can generate a new secret on the command line with:
+# openssl rand -base64 32
+# https://next-auth.js.org/configuration/options#secret
+# NEXTAUTH_SECRET=""
+NEXTAUTH_URL="http://localhost:3000"
+NEXTAUTH_SECRET="secret"
+
+# Langfuse Cloud Environment
+NEXT_PUBLIC_LANGFUSE_CLOUD_REGION="DEV"
+
+# Langfuse experimental features
+LANGFUSE_ENABLE_EXPERIMENTAL_FEATURES="true"
+
+# Salt for API key hashing
+SALT="salt"
+
+# Email
+EMAIL_FROM_ADDRESS=""  # Defines the email address to use as the from address.
+SMTP_CONNECTION_URL=""  # Defines the connection url for smtp server.
+
+# S3 Batch Exports
+LANGFUSE_S3_BATCH_EXPORT_ENABLED=true
+LANGFUSE_S3_BATCH_EXPORT_BUCKET=langfuse
+LANGFUSE_S3_BATCH_EXPORT_ACCESS_KEY_ID=minio
+LANGFUSE_S3_BATCH_EXPORT_SECRET_ACCESS_KEY=miniosecret
+LANGFUSE_S3_BATCH_EXPORT_REGION=us-east-1
+LANGFUSE_S3_BATCH_EXPORT_ENDPOINT=http://localhost:9090
+## Necessary for minio compatibility
+LANGFUSE_S3_BATCH_EXPORT_FORCE_PATH_STYLE=true
+LANGFUSE_S3_BATCH_EXPORT_PREFIX=exports/
+
+# S3 Media Upload LOCAL
+LANGFUSE_S3_MEDIA_UPLOAD_BUCKET=langfuse
+LANGFUSE_S3_MEDIA_UPLOAD_ACCESS_KEY_ID=minio
+LANGFUSE_S3_MEDIA_UPLOAD_SECRET_ACCESS_KEY=miniosecret
+LANGFUSE_S3_MEDIA_UPLOAD_REGION=us-east-1
+LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT=http://localhost:9090
+## Necessary for minio compatibility
+LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE=true
+LANGFUSE_S3_MEDIA_UPLOAD_PREFIX=media/
+
+# S3 Event Bucket Upload
+## Set to true to test uploading all events to S3
+LANGFUSE_S3_EVENT_UPLOAD_BUCKET=langfuse
+LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID=minio
+LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY=miniosecret
+LANGFUSE_S3_EVENT_UPLOAD_REGION=us-east-1
+LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT=http://localhost:9090
+## Necessary for minio compatibility
+LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE=true
+LANGFUSE_S3_EVENT_UPLOAD_PREFIX=events/
+
+# Set during docker build of application
+# Used to disable environment verification at build time
+# DOCKER_BUILD=1
+
+REDIS_HOST="127.0.0.1"
+REDIS_PORT=6379
+REDIS_AUTH="bitnami"
+REDIS_CLUSTER_ENABLED="true"
+REDIS_CLUSTER_NODES="127.0.0.1:6370,127.0.0.1:6371,127.0.0.1:6372,127.0.0.1:6373,127.0.0.1:6374,127.0.0.1:6375"
+LANGFUSE_INGESTION_QUEUE_SHARD_COUNT=8
+
+# openssl rand -hex 32 used only here
+ENCRYPTION_KEY=0000000000000000000000000000000000000000000000000000000000000000 
+
+# speeds up local development by not executing init scripts on server startup
+NEXT_PUBLIC_LANGFUSE_RUN_NEXT_INIT="false"

.env.prod.example

@@ -175,6 +175,10 @@ OTEL_SERVICE_NAME="langfuse"
 # REDIS_CONNECTION_STRING=
 # REDIS_ENABLE_AUTO_PIPELINING=
 
+# Redis Cluster configuration (optional)
+# REDIS_CLUSTER_ENABLED=false
+# REDIS_CLUSTER_NODES=redis-node1:6379,redis-node2:6379,redis-node3:6379
+
 # Cache configuration
 # LANGFUSE_CACHE_API_KEY_ENABLED=
 # LANGFUSE_CACHE_API_KEY_TTL_SECONDS=

.github/workflows/pipeline.yml

@@ -171,12 +171,12 @@ jobs:
     needs:
       - pre-job
     if: needs.pre-job.outputs.should_skip != 'true'
-    name: tests-web-async (node${{ matrix.node-version }}, pg${{ matrix.postgres-version }}, mode${{ matrix.blob-provider }})
+    name: tests-web-async (node${{ matrix.node-version }}, pg${{ matrix.postgres-version }}, mode${{ matrix.deploy-mode }})
     strategy:
       matrix:
         node-version: [20]
         postgres-version: [12, 15]
-        blob-provider: ["", "-azure"]
+        deploy-mode: ["", "-azure", "-redis-cluster"]
     steps:
       - name: Set Swap Space
         uses: pierotofy/set-swap-space@master
@@ -208,8 +208,8 @@ jobs:
           pnpm install
       - name: Load default env
         run: |
-          cp .env.dev${{ matrix.blob-provider }}.example .env
-          grep -v -e '^LANGFUSE_S3_BATCH_EXPORT_ENABLED=' -e '^NEXT_PUBLIC_LANGFUSE_RUN_NEXT_INIT=' .env.dev${{ matrix.blob-provider }}.example > .env
+          cp .env.dev${{ matrix.deploy-mode }}.example .env
+          grep -v -e '^LANGFUSE_S3_BATCH_EXPORT_ENABLED=' -e '^NEXT_PUBLIC_LANGFUSE_RUN_NEXT_INIT=' .env.dev${{ matrix.deploy-mode }}.example > .env
           echo "LANGFUSE_INGESTION_QUEUE_DELAY_MS=1" >> .env
           echo "LANGFUSE_INGESTION_CLICKHOUSE_WRITE_INTERVAL_MS=1" >> .env
           echo "LANGFUSE_TRACE_DELETE_CONCURRENCY=100" >> .env
@@ -217,7 +217,7 @@ jobs:
           echo "LANGFUSE_EE_LICENSE_KEY=langfuse_ee_test" >> .env
       - name: Run dev containers
         run: |
-          docker compose -f docker-compose.dev${{ matrix.blob-provider }}.yml up -d
+          docker compose -f docker-compose.dev${{ matrix.deploy-mode }}.yml up -d
           sleep 5 # Wait for PostgreSQL to accept connections
           docker compose ps
         env:
@@ -250,12 +250,12 @@ jobs:
     needs:
       - pre-job
     if: needs.pre-job.outputs.should_skip != 'true'
-    name: tests-worker (node${{ matrix.node-version }}, pg${{ matrix.postgres-version }}, mode${{ matrix.blob-provider }})
+    name: tests-worker (node${{ matrix.node-version }}, pg${{ matrix.postgres-version }}, mode${{ matrix.deploy-mode }})
     strategy:
       matrix:
         node-version: [20]
         postgres-version: [12, 15]
-        blob-provider: ["", "-azure"]
+        deploy-mode: ["", "-azure", "-redis-cluster"]
     steps:
       - name: Set Swap Space
         uses: pierotofy/set-swap-space@master
@@ -287,12 +287,12 @@ jobs:
           which migrate
       - name: Load default env
         run: |
-          cp .env.dev${{ matrix.blob-provider }}.example .env
-          cp .env.dev${{ matrix.blob-provider }}.example web/.env
-          cp .env.dev${{ matrix.blob-provider }}.example worker/.env
+          cp .env.dev${{ matrix.deploy-mode }}.example .env
+          cp .env.dev${{ matrix.deploy-mode }}.example web/.env
+          cp .env.dev${{ matrix.deploy-mode }}.example worker/.env
       - name: Run + migrate
         run: |
-          docker compose -f docker-compose.dev${{ matrix.blob-provider }}.yml up -d
+          docker compose -f docker-compose.dev${{ matrix.deploy-mode }}.yml up -d
           sleep 5 # Wait for PostgreSQL to accept connections
           docker compose ps
       - name: Ensure no unhealthy status

.gitignore

@@ -39,6 +39,7 @@ yarn-error.log*
 .env*
 !.env.dev.example
 !.env.dev-azure.example
+!.env.dev-redis-cluster.example
 !.env.prod.example
 
 # vercel

README.cn.md

@@ -213,8 +213,8 @@ LANGFUSE_HOST="https://cloud.langfuse.com" # 🇪🇺 欧盟区域
 
 创建示例代码（文件名：**main.py**）：
 
-````python:main.py
-from langfuse.decorators import observe
+```python:main.py
+from langfuse import observe
 from langfuse.openai import openai  # OpenAI 集成
 
 @observe()

README.ja.md

@@ -219,7 +219,7 @@ LANGFUSE_HOST="https://cloud.langfuse.com" # 🇪🇺 EUリージョン
 ```
 
 ```python:/@observe()/ /from langfuse.openai import openai/ filename="main.py"
-from langfuse.decorators import observe
+from langfuse import observe
 from langfuse.openai import openai  # OpenAI統合
 
 @observe()

README.kr.md

@@ -201,7 +201,7 @@ LANGFUSE_HOST="https://cloud.langfuse.com" # 🇪🇺 EU region
 ```
 
 ```python:main.py
-from langfuse.decorators import observe
+from langfuse import observe
 from langfuse.openai import openai # OpenAI integration
 
 @observe()

README.md

@@ -93,7 +93,7 @@ Langfuse is an **open source LLM engineering** platform. It helps teams collabor
 
 ### Langfuse Cloud
 
-Managed deployment by the Langfuse team, generous free-tier (hobby plan), no credit card required.
+Managed deployment by the Langfuse team, generous free-tier, no credit card required.
 
 <div align="center">
     <a href="https://cloud.langfuse.com" target="_blank">
@@ -115,12 +115,11 @@ Run Langfuse on your own infrastructure:
   # Run the langfuse docker compose
   docker compose up
   ```
-
-- [Kubernetes (Helm)](https://langfuse.com/self-hosting/kubernetes-helm): Run Langfuse on a Kubernetes cluster using Helm. This is the preferred production deployment.
 - [VM](https://langfuse.com/self-hosting/docker-compose): Run Langfuse on a single Virtual Machine using Docker Compose.
-- Planned: Cloud-specific deployment guides, please upvote and comment on the following threads: [AWS](https://github.com/orgs/langfuse/discussions/4645), [Google Cloud](https://github.com/orgs/langfuse/discussions/4646), [Azure](https://github.com/orgs/langfuse/discussions/4647).
+- [Kubernetes (Helm)](https://langfuse.com/self-hosting/kubernetes-helm): Run Langfuse on a Kubernetes cluster using Helm. This is the preferred production deployment.
+- Terraform Templates: [AWS](https://langfuse.com/self-hosting/aws), [Azure](https://langfuse.com/self-hosting/azure), [GCP](https://langfuse.com/self-hosting/gcp)
 
-See [self-hosting documentation](https://langfuse.com/self-hosting) to learn more about the architecture and configuration options.
+See [self-hosting documentation](https://langfuse.com/self-hosting) to learn more about architecture and configuration options.
 
 ## 🔌 Integrations
 
@@ -191,7 +190,7 @@ LANGFUSE_HOST="https://cloud.langfuse.com" # 🇪🇺 EU region
 ```
 
 ```python /@observe()/ /from langfuse.openai import openai/ filename="main.py"
-from langfuse.decorators import observe
+from langfuse import observe
 from langfuse.openai import openai # OpenAI integration
 
 @observe()

docker-compose.dev-redis-cluster.yml

@@ -0,0 +1,146 @@
+services:
+  clickhouse:
+    image: clickhouse/clickhouse-server:24.3
+    user: "101:101"
+    environment:
+      CLICKHOUSE_DB: default
+      CLICKHOUSE_USER: clickhouse
+      CLICKHOUSE_PASSWORD: clickhouse
+    volumes:
+      - langfuse_clickhouse_data:/var/lib/clickhouse
+      - langfuse_clickhouse_logs:/var/log/clickhouse-server
+    ports:
+      - 127.0.0.1:8123:8123
+      - 127.0.0.1:9000:9000
+    depends_on:
+      - postgres
+
+  minio:
+    image: minio/minio
+    entrypoint: sh
+    # create the 'langfuse' bucket before starting the service
+    command: -c 'mkdir -p /data/langfuse && minio server --address ":9000" --console-address ":9001" /data'
+    environment:
+      MINIO_ACCESS_KEY: minio
+      MINIO_SECRET_KEY: miniosecret
+    ports:
+      - 127.0.0.1:9090:9000
+      - 127.0.0.1:9091:9001
+    volumes:
+      - langfuse_minio_data:/data
+    healthcheck:
+      test: ["CMD", "mc", "ready", "local"]
+      interval: 1s
+      timeout: 5s
+      retries: 5
+      start_period: 1s
+
+  postgres:
+    image: postgres:${POSTGRES_VERSION:-latest}
+    restart: always
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U postgres"]
+      interval: 3s
+      timeout: 3s
+      retries: 10
+    command: ["postgres", "-c", "log_statement=all"]
+    environment:
+      - POSTGRES_USER=postgres
+      - POSTGRES_PASSWORD=postgres
+      - POSTGRES_DB=postgres
+    ports:
+      - 127.0.0.1:5432:5432
+    volumes:
+      - langfuse_postgres_data:/var/lib/postgresql/data
+
+  # Redis Cluster. Requires a Unix host to work for network_mode: host. I.e. tests will fail on windows and mac with this setup.
+  redis-node-0:
+    image: docker.io/bitnami/redis-cluster:8.0
+    network_mode: host
+    volumes:
+      - redis-cluster_data-0:/bitnami/redis/data
+    environment:
+      REDIS_PASSWORD: bitnami
+      REDIS_PORT_NUMBER: 6370
+      REDIS_NODES: 127.0.0.1:6370 127.0.0.1:6371 127.0.0.1:6372 127.0.0.1:6373 127.0.0.1:6374 127.0.0.1:6375
+
+  redis-node-1:
+    image: docker.io/bitnami/redis-cluster:8.0
+    network_mode: host
+    volumes:
+      - redis-cluster_data-1:/bitnami/redis/data
+    environment:
+      REDIS_PASSWORD: bitnami
+      REDIS_PORT_NUMBER: 6371
+      REDIS_NODES: 127.0.0.1:6370 127.0.0.1:6371 127.0.0.1:6372 127.0.0.1:6373 127.0.0.1:6374 127.0.0.1:6375
+
+  redis-node-2:
+    image: docker.io/bitnami/redis-cluster:8.0
+    network_mode: host
+    volumes:
+      - redis-cluster_data-2:/bitnami/redis/data
+    environment:
+      REDIS_PASSWORD: bitnami
+      REDIS_PORT_NUMBER: 6372
+      REDIS_NODES: 127.0.0.1:6370 127.0.0.1:6371 127.0.0.1:6372 127.0.0.1:6373 127.0.0.1:6374 127.0.0.1:6375
+
+  redis-node-3:
+    image: docker.io/bitnami/redis-cluster:8.0
+    network_mode: host
+    volumes:
+      - redis-cluster_data-3:/bitnami/redis/data
+    environment:
+      REDIS_PASSWORD: bitnami
+      REDIS_PORT_NUMBER: 6373
+      REDIS_NODES: 127.0.0.1:6370 127.0.0.1:6371 127.0.0.1:6372 127.0.0.1:6373 127.0.0.1:6374 127.0.0.1:6375
+
+  redis-node-4:
+    image: docker.io/bitnami/redis-cluster:8.0
+    network_mode: host
+    volumes:
+      - redis-cluster_data-4:/bitnami/redis/data
+    environment:
+      REDIS_PASSWORD: bitnami
+      REDIS_PORT_NUMBER: 6374
+      REDIS_NODES: 127.0.0.1:6370 127.0.0.1:6371 127.0.0.1:6372 127.0.0.1:6373 127.0.0.1:6374 127.0.0.1:6375
+
+  redis-node-5:
+    image: docker.io/bitnami/redis-cluster:8.0
+    network_mode: host
+    volumes:
+      - redis-cluster_data-5:/bitnami/redis/data
+    depends_on:
+      - redis-node-0
+      - redis-node-1
+      - redis-node-2
+      - redis-node-3
+      - redis-node-4
+    environment:
+      REDISCLI_AUTH: bitnami
+      REDIS_CLUSTER_REPLICAS: 1
+      REDIS_PASSWORD: bitnami
+      REDIS_PORT_NUMBER: 6375
+      REDIS_NODES: 127.0.0.1:6370 127.0.0.1:6371 127.0.0.1:6372 127.0.0.1:6373 127.0.0.1:6374 127.0.0.1:6375
+      REDIS_CLUSTER_CREATOR: yes
+
+volumes:
+  langfuse_postgres_data:
+    driver: local
+  langfuse_clickhouse_data:
+    driver: local
+  langfuse_clickhouse_logs:
+    driver: local
+  langfuse_minio_data:
+    driver: local
+  redis-cluster_data-0:
+    driver: local
+  redis-cluster_data-1:
+    driver: local
+  redis-cluster_data-2:
+    driver: local
+  redis-cluster_data-3:
+    driver: local
+  redis-cluster_data-4:
+    driver: local
+  redis-cluster_data-5:
+    driver: local

docker-compose.yml

@@ -29,6 +29,7 @@ services:
       CLICKHOUSE_USER: ${CLICKHOUSE_USER:-clickhouse}
       CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-clickhouse} # CHANGEME
       CLICKHOUSE_CLUSTER_ENABLED: ${CLICKHOUSE_CLUSTER_ENABLED:-false}
+      LANGFUSE_USE_AZURE_BLOB: ${LANGFUSE_USE_AZURE_BLOB:-false}
       LANGFUSE_S3_EVENT_UPLOAD_BUCKET: ${LANGFUSE_S3_EVENT_UPLOAD_BUCKET:-langfuse}
       LANGFUSE_S3_EVENT_UPLOAD_REGION: ${LANGFUSE_S3_EVENT_UPLOAD_REGION:-auto}
       LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID: ${LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID:-minio}

ee/package.json

@@ -28,9 +28,9 @@
     "@opentelemetry/api": ">=1.0.0 <1.10.0",
     "axios": "^1.8.2",
     "https-proxy-agent": "^7.0.6",
-    "next": "^14.2.26",
+    "next": "^14.2.30",
     "next-auth": "^4.24.11",
-    "zod": "^3.24.4"
+    "zod": "^3.25.62"
   },
   "devDependencies": {
     "@repo/eslint-config": "workspace:*",

ee/src/env.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 import { removeEmptyEnvVariables } from "@langfuse/shared";
 
 const EnvSchema = z.object({

fern/apis/server/definition/metrics.yml

@@ -28,7 +28,7 @@ service:
                 "metrics": [              // Required. At least one metric must be provided
                   {
                     "measure": string,    // What to measure, e.g. "count", "latency", "value"
-                    "aggregation": string // How to aggregate, e.g. "count", "sum", "avg", "p95"
+                    "aggregation": string // How to aggregate, e.g. "count", "sum", "avg", "p95", "histogram"
                   }
                 ],
                 "filters": [              // Optional. Default: []
@@ -50,7 +50,11 @@ service:
                     "field": string,      // Field to order by
                     "direction": string   // "asc" or "desc"
                   }
-                ]
+                ],
+                "config": {               // Optional. Query-specific configuration
+                  "bins": number,         // Optional. Number of bins for histogram (1-100), default: 10
+                  "row_limit": number     // Optional. Row limit for results (1-1000)
+                }
               }
               ```
       response: MetricsResponse
@@ -62,3 +66,4 @@ types:
         docs: |
           The metrics data. Each item in the list contains the metric values and dimensions requested in the query.
           Format varies based on the query parameters.
+          Histograms will return an array with [lower, upper, height] tuples.

package.json

@@ -1,6 +1,6 @@
 {
   "name": "langfuse",
-  "version": "3.65.2",
+  "version": "3.71.0",
   "author": "engineering@langfuse.com",
   "license": "MIT",
   "private": true,
@@ -18,7 +18,7 @@
     "db:seed:examples": "turbo run db:seed:examples",
     "nuke": "bash ./scripts/nuke.sh",
     "dx": "pnpm i && pnpm run infra:dev:prune && pnpm run infra:dev:up --pull always && pnpm --filter=shared run db:reset && pnpm --filter=shared run ch:reset && pnpm --filter=shared run db:seed:examples && pnpm run dev",
-    "dx-f": "pnpm i && pnpm run infra:dev:prune && pnpm run infra:dev:up --pull always && pnpm --filter=shared run db:reset -f && pnpm --filter=shared run ch:reset && pnpm --filter=shared run db:seed:examples && pnpm run dev",
+    "dx-f": "pnpm i && pnpm run infra:dev:prune && pnpm run infra:dev:up --pull always && pnpm --filter=shared run db:reset -f && SKIP_CONFIRM=1 pnpm --filter=shared run ch:reset && pnpm --filter=shared run db:seed:examples && pnpm run dev",
     "dx:skip-infra": "pnpm i && pnpm --filter=shared run db:reset && pnpm --filter=shared run ch:reset && pnpm --filter=shared run db:seed:examples && pnpm run dev",
     "build": "turbo run build",
     "start": "turbo run start",
@@ -31,12 +31,12 @@
     "prepare": "husky"
   },
   "devDependencies": {
-    "@release-it/bumper": "^7.0.1",
+    "@release-it/bumper": "^7.0.5",
     "braces": "3.0.3",
     "dotenv-cli": "^7.4.2",
     "husky": "^9.0.11",
     "prettier": "^3.3.3",
-    "release-it": "^18.1.2",
+    "release-it": "^19.0.3",
     "turbo": "^1.13.4"
   },
   "release-it": {

packages/shared/clickhouse/scripts/down.sh

@@ -36,8 +36,12 @@ if [ "$CLICKHOUSE_CLUSTER_ENABLED" == "false" ] ; then
       DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-migrations-table-engine=MergeTree"
   fi
 
-  # Execute the up command
-  migrate -source file://clickhouse/migrations/unclustered -database "$DATABASE_URL" down
+  # If SKIP_CONFIRM is set, automatically answer the confirmation prompt. Otherwise run interactively.
+  if [ "$SKIP_CONFIRM" = "1" ] || [ "$SKIP_CONFIRM" = "true" ]; then
+    printf 'y\n' | migrate -source file://clickhouse/migrations/unclustered -database "$DATABASE_URL" down
+  else
+    migrate -source file://clickhouse/migrations/unclustered -database "$DATABASE_URL" down
+  fi
 else
   if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
       DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-cluster-name=${CLICKHOUSE_CLUSTER_NAME}&x-migrations-table-engine=ReplicatedMergeTree"
@@ -45,6 +49,10 @@ else
       DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-cluster-name=${CLICKHOUSE_CLUSTER_NAME}&x-migrations-table-engine=ReplicatedMergeTree"
   fi
 
-  # Execute the up command
-  migrate -source file://clickhouse/migrations/clustered -database "$DATABASE_URL" down
+  # If SKIP_CONFIRM is set, automatically answer the confirmation prompt. Otherwise run interactively.
+  if [ "$SKIP_CONFIRM" = "1" ] || [ "$SKIP_CONFIRM" = "true" ]; then
+    printf 'y\n' | migrate -source file://clickhouse/migrations/clustered -database "$DATABASE_URL" down
+  else
+    migrate -source file://clickhouse/migrations/clustered -database "$DATABASE_URL" down
+  fi
 fi

packages/shared/package.json

@@ -60,18 +60,18 @@
     "@aws-sdk/lib-storage": "^3.675.0",
     "@aws-sdk/s3-request-presigner": "^3.679.0",
     "@azure/storage-blob": "^12.26.0",
-    "@clickhouse/client": "^1.11.1",
+    "@clickhouse/client": "^1.11.2",
     "@google-cloud/storage": "^7.15.2",
-    "@langchain/anthropic": "^0.3.12",
-    "@langchain/aws": "^0.1.3",
-    "@langchain/core": "^0.3.37",
-    "@langchain/google-genai": "^0.1.9",
-    "@langchain/google-vertexai": "^0.1.8",
-    "@langchain/openai": "^0.3.17",
+    "@langchain/anthropic": "^0.3.22",
+    "@langchain/aws": "^0.1.11",
+    "@langchain/core": "^0.3.58",
+    "@langchain/google-genai": "^0.2.12",
+    "@langchain/google-vertexai": "^0.2.12",
+    "@langchain/openai": "^0.5.13",
     "@opentelemetry/api": ">=1.0.0 <1.10.0",
     "@prisma/client": "^6.3.0",
-    "@react-email/components": "^0.0.19",
-    "@react-email/render": "^0.0.15",
+    "@react-email/components": "^0.0.42",
+    "@react-email/render": "^1.1.2",
     "@types/bcryptjs": "^2.4.6",
     "axios": "^1.8.2",
     "bcryptjs": "^2.4.3",
@@ -82,8 +82,8 @@
     "https-proxy-agent": "^7.0.6",
     "ioredis": "^5.4.1",
     "kysely": "^0.27.4",
-    "langchain": "^0.3.15",
-    "langfuse-langchain": "3.30.3",
+    "langchain": "^0.3.28",
+    "langfuse-langchain": "3.37.4",
     "lodash": "^4.17.21",
     "lossless-json": "^4.0.2",
     "next-auth": "^4.24.11",
@@ -91,7 +91,7 @@
     "prisma-extension-kysely": "^2.1.0",
     "uuid": "^9.0.1",
     "winston": "^3.15.0",
-    "zod": "^3.24.4",
+    "zod": "^3.25.62",
     "zod-to-json-schema": "^3.23.5"
   },
   "devDependencies": {
@@ -101,6 +101,7 @@
     "@types/node": "^20.11.29",
     "@types/nodemailer": "^6.4.16",
     "@types/pg": "^8.11.10",
+    "@types/react": "18.2.79",
     "@types/uuid": "^9.0.8",
     "@typescript-eslint/parser": "^7.12.0",
     "eslint": "^8.57.0",

packages/shared/prisma/generated/types.ts

@@ -110,7 +110,8 @@ export const DashboardWidgetChartType = {
     HORIZONTAL_BAR: "HORIZONTAL_BAR",
     VERTICAL_BAR: "VERTICAL_BAR",
     PIE: "PIE",
-    NUMBER: "NUMBER"
+    NUMBER: "NUMBER",
+    HISTOGRAM: "HISTOGRAM"
 } as const;
 export type DashboardWidgetChartType = (typeof DashboardWidgetChartType)[keyof typeof DashboardWidgetChartType];
 export type Account = {

packages/shared/prisma/migrations/20250604085536_add_histogram_chart_type/migration.sql

@@ -0,0 +1,2 @@
+-- AlterEnum
+ALTER TYPE "DashboardWidgetChartType" ADD VALUE 'HISTOGRAM';

packages/shared/prisma/schema.prisma

@@ -1156,6 +1156,7 @@ enum DashboardWidgetChartType {
   VERTICAL_BAR
   PIE
   NUMBER
+  HISTOGRAM
 }
 
 model DashboardWidget {

packages/shared/prisma/seed.ts

@@ -1133,6 +1133,33 @@ export const SEED_PROMPTS = [
     labels: ["production", "latest"],
     tags: ["tag1", "tag2"],
   },
+  {
+    id: `folder-customer-prompt-1`,
+    createdBy: "user-1",
+    prompt: "Folder prompt 1 content",
+    name: "folder/customer/prompt-1",
+    version: 1,
+    labels: ["production", "latest"],
+    tags: ["tag1", "tag2"],
+  },
+  {
+    id: `folder-customer-prompt-2`,
+    createdBy: "user-1",
+    prompt: "Folder prompt 2 content",
+    name: "folder/customer/prompt-2",
+    version: 1,
+    labels: ["production", "latest"],
+    tags: ["tag1", "tag2"],
+  },
+  {
+    id: `folder-prompt-1`,
+    createdBy: "user-1",
+    prompt: "Folder prompt 1 content",
+    name: "folder/prompt-1",
+    version: 1,
+    labels: ["production", "latest"],
+    tags: ["tag1", "tag2"],
+  },
 ];
 
 export const PROMPT_IDS: string[] = [];

packages/shared/src/domain/observations.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { jsonSchema } from "../utils/zod";
 import { MetadataDomain } from "./traces";
 

packages/shared/src/domain/scores.ts

@@ -1,5 +1,5 @@
 import { ScoreDataType } from "@prisma/client";
-import z from "zod";
+import z from "zod/v4";
 import { MetadataDomain } from "./traces";
 
 export const ScoreSource = {
@@ -31,7 +31,7 @@ export const ScoreSchema = z.object({
   queueId: z.string().nullable(),
   createdAt: z.date(),
   updatedAt: z.date(),
-  dataType: z.nativeEnum(ScoreDataType),
+  dataType: z.enum(ScoreDataType),
 });
 
 export type ScoreDomain = z.infer<typeof ScoreSchema>;

packages/shared/src/domain/table-view-presets.ts

@@ -1,6 +1,6 @@
 import { singleFilter } from "../interfaces/filters";
 import { orderBy } from "../interfaces/orderBy";
-import z from "zod";
+import z from "zod/v4";
 
 export enum TableViewPresetTableName {
   Traces = "traces",
@@ -17,7 +17,7 @@ const TableViewPresetDomainSchema = z.object({
   updatedAt: z.date(),
   createdBy: z.string().nullable(),
   name: z.string(),
-  tableName: z.nativeEnum(TableViewPresetTableName),
+  tableName: z.enum(TableViewPresetTableName),
   filters: z.array(singleFilter),
   columnOrder: z.array(z.string()),
   columnVisibility: z.record(z.string(), z.boolean()),

packages/shared/src/domain/traces.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { jsonSchema, jsonSchemaNullable } from "../utils/zod";
 
 export const MetadataDomain = z.record(

packages/shared/src/env.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 import { removeEmptyEnvVariables } from "./utils/environment";
 
 const EnvSchema = z.object({
@@ -9,10 +9,7 @@ const EnvSchema = z.object({
   NEXTAUTH_URL: z.string().url().optional(),
   REDIS_HOST: z.string().nullish(),
   REDIS_PORT: z.coerce
-    .number({
-      description:
-        ".env files convert numbers to strings, therefore we have to enforce them to be numbers",
-    })
+    .number() // .env files convert numbers to strings, therefore we have to enforce them to be numbers
     .positive()
     .max(65536, `options.port should be >= 0 and < 65536`)
     .default(6379)
@@ -24,6 +21,9 @@ const EnvSchema = z.object({
   REDIS_TLS_CERT_PATH: z.string().optional(),
   REDIS_TLS_KEY_PATH: z.string().optional(),
   REDIS_ENABLE_AUTO_PIPELINING: z.enum(["true", "false"]).default("true"),
+  // Redis Cluster Configuration
+  REDIS_CLUSTER_ENABLED: z.enum(["true", "false"]).default("false"),
+  REDIS_CLUSTER_NODES: z.string().optional(),
   ENCRYPTION_KEY: z
     .string()
     .length(
@@ -45,6 +45,10 @@ const EnvSchema = z.object({
     .number()
     .nonnegative()
     .default(15_000),
+  LANGFUSE_INGESTION_QUEUE_SHARD_COUNT: z.coerce
+    .number()
+    .positive()
+    .default(1),
   SALT: z.string().optional(), // used by components imported by web package
   LANGFUSE_LOG_LEVEL: z
     .enum(["trace", "debug", "info", "warn", "error", "fatal"])
@@ -60,9 +64,7 @@ const EnvSchema = z.object({
     .enum(["true", "false"])
     .default("false"),
   LANGFUSE_S3_CONCURRENT_WRITES: z.coerce.number().positive().default(50),
-  LANGFUSE_S3_EVENT_UPLOAD_BUCKET: z.string({
-    required_error: "Langfuse requires a bucket name for S3 Event Uploads.",
-  }),
+  LANGFUSE_S3_EVENT_UPLOAD_BUCKET: z.string(), // Langfuse requires a bucket name for S3 Event Uploads.
   LANGFUSE_S3_EVENT_UPLOAD_PREFIX: z.string().default(""),
   LANGFUSE_S3_EVENT_UPLOAD_REGION: z.string().optional(),
   LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT: z.string().optional(),

packages/shared/src/features/annotation/types.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { type ScoreDataType } from "../../db";
 
 const NUMERIC: ScoreDataType = "NUMERIC";

packages/shared/src/features/batchAction/types.ts

@@ -1,7 +1,8 @@
-import z from "zod";
+import z from "zod/v4";
 import { singleFilter } from "../../interfaces/filters";
 import { orderBy } from "../../interfaces/orderBy";
-import { BatchExportTableName } from "../batchExport/types";
+import { BatchTableNames } from "../../interfaces/tableNames";
+import { TracingSearchType } from "../../interfaces/search";
 
 /* eslint-disable no-unused-vars */
 export enum BatchActionType {
@@ -20,6 +21,8 @@ export type ActionId = z.infer<typeof ActionIdSchema>;
 export const BatchActionQuerySchema = z.object({
   filter: z.array(singleFilter).nullable(),
   orderBy,
+  searchQuery: z.string().optional(),
+  searchType: z.array(TracingSearchType).optional(),
 });
 
 export type BatchActionQuery = z.infer<typeof BatchActionQuerySchema>;
@@ -29,11 +32,11 @@ export const CreateBatchActionSchema = z.object({
   actionId: ActionIdSchema,
   targetId: z.string().optional(),
   query: BatchActionQuerySchema,
-  tableName: z.nativeEnum(BatchExportTableName),
+  tableName: z.enum(BatchTableNames),
 });
 
 export const GetIsBatchActionInProgressSchema = z.object({
   projectId: z.string(),
   actionId: ActionIdSchema,
-  tableName: z.nativeEnum(BatchExportTableName),
+  tableName: z.enum(BatchTableNames),
 });

packages/shared/src/features/batchExport/types.ts

@@ -1,9 +1,10 @@
-import z from "zod";
+import z from "zod/v4";
 
 import { BatchExport } from "@prisma/client";
 
 import { singleFilter } from "../../interfaces/filters";
 import { orderBy } from "../../interfaces/orderBy";
+import { BatchTableNames } from "../../interfaces/tableNames";
 
 export enum BatchExportStatus {
   QUEUED = "QUEUED",
@@ -18,13 +19,9 @@ export enum BatchExportFileFormat {
   JSONL = "JSONL",
 }
 
-export enum BatchExportTableName {
-  Scores = "scores",
-  Sessions = "sessions",
-  Traces = "traces",
-  Observations = "observations",
-  DatasetRunItems = "dataset_run_items",
-}
+// Use shared BatchTableNames enum for consistency across batch operations
+// Keep BatchExportTableName as alias for backward compatibility
+export { BatchTableNames as BatchExportTableName };
 
 export const exportOptions: Record<
   BatchExportFileFormat,
@@ -44,7 +41,7 @@ export const exportOptions: Record<
 } as const;
 
 export const BatchExportQuerySchema = z.object({
-  tableName: z.nativeEnum(BatchExportTableName),
+  tableName: z.enum(BatchTableNames),
   filter: z.array(singleFilter).nullable(),
   orderBy,
   limit: z.number().optional(),
@@ -57,7 +54,7 @@ export const CreateBatchExportSchema = z.object({
   projectId: z.string(),
   name: z.string(),
   query: BatchExportQuerySchema,
-  format: z.nativeEnum(BatchExportFileFormat),
+  format: z.enum(BatchExportFileFormat),
 });
 
 export const BatchExportSchema = z.object({
@@ -69,9 +66,9 @@ export const BatchExportSchema = z.object({
   finishedAt: z.date().nullable(),
   expiresAt: z.date().nullable(),
   name: z.string(),
-  status: z.nativeEnum(BatchExportStatus),
+  status: z.enum(BatchExportStatus),
   query: BatchExportQuerySchema,
-  format: z.nativeEnum(BatchExportFileFormat),
+  format: z.enum(BatchExportFileFormat),
   url: z.string().nullable(),
   log: z.string().nullable(),
 });

packages/shared/src/features/comments/types.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 
 const MAX_COMMENT_LENGTH = 3000;
 

packages/shared/src/features/entitlements/plans.ts

@@ -14,6 +14,7 @@ export const planLabels = {
   "cloud:pro": "Pro",
   "cloud:team": "Team",
   "cloud:enterprise": "Enterprise",
+  "self-hosted:pro": "Pro (self-hosted)",
   "self-hosted:enterprise": "Enterprise (self-hosted)",
 } as const;
 

packages/shared/src/features/evals/types.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 
 export const langfuseObjects = [
   "trace",

packages/shared/src/features/evals/utilities.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { variableMapping } from "./types";
 
 /**

packages/shared/src/features/prompts/parsePromptDependencyTags.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 
 export const PromptDependencyRegex = /@@@langfusePrompt:(.*?)@@@/g;
 

packages/shared/src/features/prompts/validation.ts

@@ -0,0 +1,15 @@
+import { z } from "zod/v4";
+
+/**
+ * Prompt name validation schema for API, tRPC and client
+ */
+export const PromptNameSchema = z
+  .string()
+  .min(1, "Enter a name")
+  .regex(/^[^|]*$/, "Prompt name cannot contain '|' character")
+  .regex(/^[^/]/, "Name cannot start with a slash")
+  .regex(/^(?!.*\/\/)/, "Name cannot contain consecutive slashes")
+  .regex(/^.*[^/]$/, "Name cannot end with a slash")
+  .transform((s) => s.trim())
+  .refine((s) => s.length > 0, "Name cannot be empty")
+  .refine((name) => name !== "new", "Prompt name cannot be 'new'");

packages/shared/src/features/scores/interfaces/api/shared.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 import { jsonSchema, publicApiPaginationZod } from "../../../../utils/zod";
 import { stringDateTime } from "../../../../utils/typeChecks";
 import { applyScoreValidation } from "../../../../utils/scores";

packages/shared/src/features/scores/interfaces/api/v1/endpoints.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 import { paginationMetaResponseZod } from "../../../../../utils/zod";
 import {
   DeleteScoreQuery,

packages/shared/src/features/scores/interfaces/api/v1/schemas.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 import {
   CategoricalData,
   NumericData,

packages/shared/src/features/scores/interfaces/api/v1/validation.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { GetScoreResponseDataV1 } from "./endpoints";
 
 /**

packages/shared/src/features/scores/interfaces/api/v2/endpoints.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { paginationMetaResponseZod } from "../../../../../utils/zod";
 import { GetScoreQuery, GetScoresQuery } from "../shared";
 import { APIScoreSchemaV2 } from "./schemas";

packages/shared/src/features/scores/interfaces/api/v2/schemas.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 import {
   CategoricalData,
   NumericData,

packages/shared/src/features/scores/interfaces/api/v2/validation.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { GetScoreResponseDataV2 } from "./endpoints";
 
 /**

packages/shared/src/features/scores/interfaces/application/validation.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { APIScoreSchemaV2, APIScoreV2 } from "../api/v2/schemas";
 import { APIScoreSchemaV1, APIScoreV1 } from "../api/v1/schemas";
 import { ScoreDomain } from "../../../../domain";

packages/shared/src/features/scores/interfaces/ingestion/validation.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { applyScoreValidation } from "../../../../utils/scores";
 import { PostScoreBodyFoundationSchema } from "../shared";
 import { isPresent } from "../../../../utils/typeChecks";

packages/shared/src/features/scores/interfaces/shared.ts

@@ -1,5 +1,5 @@
 import { jsonSchema } from "../../../utils/zod";
-import z from "zod";
+import z from "zod/v4";
 import { NonEmptyString } from "../../../utils/zod";
 
 /**

packages/shared/src/features/scores/scoreConfigTypes.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 
 import { ScoreConfig as ScoreConfigDbType } from "@prisma/client";
 
@@ -68,10 +68,10 @@ const CategoricalScoreConfig = z.object({
     const parseResult = Categories.safeParse(categories);
     if (!parseResult.success) {
       ctx.addIssue({
-        code: z.ZodIssueCode.custom,
+        code: "custom",
         message:
           "Category must be an array of objects with label value pairs, where labels and values are unique.",
-      });
+      } as z.core.$ZodIssueCustom);
       return;
     }
 
@@ -198,7 +198,7 @@ export const GetScoreConfigResponse = ValidatedScoreConfigSchema;
 
 // POST /score-configs
 export const PostScoreConfigBody = z
-  .union([
+  .discriminatedUnion("dataType", [
     ScoreConfigPostBase.merge(CategoricalScoreConfig),
     ScoreConfigPostBase.merge(NumericScoreConfig),
     ScoreConfigPostBase.merge(

packages/shared/src/index.ts

@@ -26,6 +26,7 @@ export * from "./features/evals/utilities";
 // table actions
 export * from "./features/batchExport/types";
 export * from "./features/batchAction/types";
+export { BatchTableNames } from "./interfaces/tableNames";
 
 // annotation
 export * from "./features/annotation/types";
@@ -41,6 +42,7 @@ export * from "./features/experiments/utils";
 
 // prompts
 export * from "./features/prompts/parsePromptDependencyTags";
+export * from "./features/prompts/validation";
 
 // export db types only
 export * from "@prisma/client";

packages/shared/src/interfaces/cloudConfigSchema.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 import { CloudConfigRateLimit } from "./rate-limits";
 import { cloudConfigPlans } from "../features/entitlements/plans";
 

packages/shared/src/interfaces/customLLMProviderConfigSchemas.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 
 export const BedrockConfigSchema = z.object({ region: z.string() });
 export type BedrockConfig = z.infer<typeof BedrockConfigSchema>;

packages/shared/src/interfaces/filters.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 
 export const filterOperators = {
   datetime: [">", "<", ">=", "<="],

packages/shared/src/interfaces/orderBy.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 
 export const orderBy = z
   .object({

packages/shared/src/interfaces/rate-limits.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { ApiAccessScope } from "../server";
 
 export const RateLimitResource = z.enum([

packages/shared/src/interfaces/search.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 
 export const TracingSearchType = z.enum(["id", "content"]);
 // id: for searching smaller columns like IDs, types, and other metadata

packages/shared/src/interfaces/tableNames.ts

@@ -0,0 +1,13 @@
+/**
+ * Shared table names used across batch operations (exports, actions, etc.)
+ * This enum provides a centralized definition of database table names
+ * to avoid coupling between different batch operation types.
+ */
+export enum BatchTableNames {
+  Scores = "scores",
+  Sessions = "sessions",
+  Traces = "traces",
+  Observations = "observations",
+  DatasetRunItems = "dataset_run_items",
+  AuditLogs = "audit_logs",
+}

packages/shared/src/server/auth/types.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { Plan, plans } from "../../features/entitlements/plans";
 import { CloudConfigRateLimit } from "../../interfaces/rate-limits";
 import { ApiKeyScope } from "../../";

packages/shared/src/server/filterToPrisma.ts

@@ -2,7 +2,7 @@ import { Prisma } from "@prisma/client";
 import { ColumnDefinition, type TableNames } from "../tableDefinitions";
 import { FilterState } from "../types";
 import { filterOperators, timeFilter } from "../interfaces/filters";
-import { z } from "zod";
+import { z } from "zod/v4";
 import { logger } from "./index";
 
 const operatorReplacements = {

packages/shared/src/server/ingestion/processEventBatch.ts

@@ -1,5 +1,5 @@
 import { randomUUID } from "crypto";
-import { z } from "zod";
+import { z } from "zod/v4";
 
 import { type Model } from "../../db";
 import { env } from "../../env";
@@ -229,10 +229,13 @@ export const processEventBatch = async (
     throw new Error("Redis not initialized, aborting event processing");
   }
 
-  const queue = IngestionQueue.getInstance();
   await Promise.all(
-    Object.keys(sortedBatchByEventBodyId).map(async (id) =>
-      queue
+    Object.keys(sortedBatchByEventBodyId).map(async (id) => {
+      const eventData = sortedBatchByEventBodyId[id];
+      const shardingKey = `${authCheck.scope.projectId}-${eventData.eventBodyId}`;
+      const queue = IngestionQueue.getInstance({ shardingKey });
+
+      return queue
         ? queue.add(
             QueueJobs.IngestionJob,
             {
@@ -241,14 +244,12 @@ export const processEventBatch = async (
               name: QueueJobs.IngestionJob as const,
               payload: {
                 data: {
-                  type: sortedBatchByEventBodyId[id].type,
-                  eventBodyId: sortedBatchByEventBodyId[id].eventBodyId,
-                  fileKey: sortedBatchByEventBodyId[id].key,
+                  type: eventData.type,
+                  eventBodyId: eventData.eventBodyId,
+                  fileKey: eventData.key,
                   skipS3List:
                     source === "otel" &&
-                    getClickhouseEntityType(
-                      sortedBatchByEventBodyId[id].type,
-                    ) === "observation",
+                    getClickhouseEntityType(eventData.type) === "observation",
                 },
                 authCheck: authCheck as {
                   validKey: true;
@@ -261,8 +262,8 @@ export const processEventBatch = async (
             },
             { delay: getDelay(delay) },
           )
-        : Promise.reject("Failed to instantiate queue"),
-    ),
+        : Promise.reject("Failed to instantiate queue");
+    }),
   );
 
   return aggregateBatchResult(

packages/shared/src/server/ingestion/types.ts

@@ -1,5 +1,5 @@
 import lodash from "lodash";
-import { z } from "zod";
+import { z } from "zod/v4";
 
 import { NonEmptyString, jsonSchema } from "../../utils/zod";
 import { ModelUsageUnit } from "../../constants";
@@ -20,7 +20,7 @@ export const Usage = z.object({
   input: z.number().int().nullish(),
   output: z.number().int().nullish(),
   total: z.number().int().nullish(),
-  unit: z.nativeEnum(ModelUsageUnit).nullish(),
+  unit: z.enum(ModelUsageUnit).nullish(),
   inputCost: z.number().nullish(),
   outputCost: z.number().nullish(),
   totalCost: z.number().nullish(),
@@ -30,7 +30,7 @@ const MixedUsage = z.object({
   input: z.number().int().nullish(),
   output: z.number().int().nullish(),
   total: z.number().int().nullish(),
-  unit: z.nativeEnum(ModelUsageUnit).nullish(),
+  unit: z.enum(ModelUsageUnit).nullish(),
   promptTokens: z.number().int().nullish(),
   completionTokens: z.number().int().nullish(),
   totalTokens: z.number().int().nullish(),
@@ -128,8 +128,7 @@ const OpenAICompletionUsageSchema = z
     }
 
     return result;
-  })
-  .pipe(RawUsageDetails);
+  });
 
 // The new OpenAI Response API uses a new Usage schema that departs from the Completion API Usage schema
 const OpenAIResponseUsageSchema = z
@@ -184,8 +183,7 @@ const OpenAIResponseUsageSchema = z
     }
 
     return result;
-  })
-  .pipe(RawUsageDetails);
+  });
 
 export const UsageDetails = z
   .union([
@@ -266,7 +264,7 @@ export const CreateGenerationBody = CreateSpanBody.extend({
           z.number(),
           z.boolean(),
           z.array(z.string()),
-          z.record(z.string()),
+          z.record(z.string(), z.string()),
         ])
         .nullish(),
     )
@@ -296,7 +294,7 @@ export const UpdateGenerationBody = UpdateSpanBody.extend({
           z.number(),
           z.boolean(),
           z.array(z.string()),
-          z.record(z.string()),
+          z.record(z.string(), z.string()),
         ])
         .nullish(),
     )

packages/shared/src/server/ingestion/validateAndInflateScore.ts

@@ -162,7 +162,7 @@ function validateConfigAgainstBody(
   });
 
   if (!rangeValidation.success) {
-    const errorDetails = rangeValidation.error.errors
+    const errorDetails = rangeValidation.error.issues
       .map((error) => `${error.path.join(".")} - ${error.message}`)
       .join(", ");
 

packages/shared/src/server/llm/fetchLLMCompletion.ts

@@ -1,4 +1,4 @@
-import { type ZodSchema } from "zod";
+import { type ZodSchema } from "zod/v4";
 
 import { ChatAnthropic } from "@langchain/anthropic";
 import { ChatVertexAI } from "@langchain/google-vertexai";
@@ -16,7 +16,7 @@ import {
   StringOutputParser,
 } from "@langchain/core/output_parsers";
 import { IterableReadableStream } from "@langchain/core/utils/stream";
-import { ChatOpenAI } from "@langchain/openai";
+import { ChatOpenAI, AzureChatOpenAI } from "@langchain/openai";
 import GCPServiceAccountKeySchema, {
   BedrockConfigSchema,
   BedrockCredentialSchema,
@@ -219,7 +219,7 @@ export async function fetchLLMCompletion(
       timeout: 1000 * 60 * 2, // 2 minutes timeout
     });
   } else if (modelParams.adapter === LLMAdapter.Azure) {
-    chatModel = new ChatOpenAI({
+    chatModel = new AzureChatOpenAI({
       azureOpenAIApiKey: apiKey,
       azureOpenAIBasePath: baseURL,
       azureOpenAIApiDeploymentName: modelParams.model,

packages/shared/src/server/llm/types.ts

@@ -1,5 +1,5 @@
 import { LlmApiKeys } from "@prisma/client";
-import z from "zod";
+import z from "zod/v4";
 import { BedrockConfigSchema } from "../../interfaces/customLLMProviderConfigSchemas";
 import { TokenCountDelegate } from "../ingestion/processEventBatch";
 import { AuthHeaderValidVerificationResult } from "../auth/types";
@@ -28,7 +28,7 @@ export const JSONSchemaFormSchema = z
     z
       .object({
         type: z.literal("object"),
-        properties: z.record(z.any()),
+        properties: z.record(z.string(), z.any()),
         required: z.array(z.string()).optional(),
         additionalProperties: z.boolean().optional(),
       })
@@ -168,7 +168,7 @@ export const ToolResultMessageSchema = z.object({
 });
 export type ToolResultMessage = z.infer<typeof ToolResultMessageSchema>;
 
-export const ChatMessageDefaultRoleSchema = z.nativeEnum(ChatMessageRole);
+export const ChatMessageDefaultRoleSchema = z.enum(ChatMessageRole);
 export const ChatMessageSchema = z.union([
   SystemMessageSchema,
   DeveloperMessageSchema,
@@ -370,7 +370,7 @@ export const LLMApiKeySchema = z
     projectId: z.string(),
     createdAt: z.date(),
     updatedAt: z.date(),
-    adapter: z.nativeEnum(LLMAdapter),
+    adapter: z.enum(LLMAdapter),
     provider: z.string(),
     displaySecretKey: z.string(),
     secretKey: z.string(),

packages/shared/src/server/llm/utils.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 
 import { decrypt } from "../../encryption";
 

packages/shared/src/server/orderByToPrisma.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 
 import { Prisma } from "@prisma/client";
 

packages/shared/src/server/queries/clickhouse-sql/factory.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { singleFilter } from "../../../interfaces/filters";
 import { FilterCondition } from "../../../types";
 import { isValidTableName } from "../../clickhouse/schemaUtils";

packages/shared/src/server/queries/clickhouse-sql/orderby-factory.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { OrderByState } from "../../../interfaces/orderBy";
 import { UiColumnMappings } from "../../../tableDefinitions";
 import { logger } from "../../logger";

packages/shared/src/server/queries/createSessionsAllQuery.ts

@@ -1,4 +1,4 @@
-import { z } from "zod";
+import { z } from "zod/v4";
 
 import { Prisma } from "@prisma/client";
 import { tableColumnsToSqlFilterAndPrefix } from "../filterToPrisma";

packages/shared/src/server/queries/types.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { singleFilter } from "../../interfaces/filters";
 import { orderBy } from "../../interfaces/orderBy";
 import { optionalPaginationZod } from "../../utils/zod";

packages/shared/src/server/queues.ts

@@ -1,15 +1,15 @@
 /* eslint-disable no-unused-vars */
-import { z } from "zod";
-import { eventTypes } from ".";
+import { z } from "zod/v4";
+import { eventTypes } from "./ingestion/types";
 import {
   BatchActionQuerySchema,
   BatchActionType,
 } from "../features/batchAction/types";
-import { BatchExportTableName } from "../features/batchExport/types";
+import { BatchTableNames } from "../interfaces/tableNames";
 
 export const IngestionEvent = z.object({
   data: z.object({
-    type: z.nativeEnum(eventTypes),
+    type: z.enum(Object.values(eventTypes)),
     eventBodyId: z.string(),
     fileKey: z.string().optional(),
     skipS3List: z.boolean().optional(),
@@ -76,28 +76,28 @@ export const BatchActionProcessingEventSchema = z.discriminatedUnion(
       actionId: z.literal("score-delete"),
       projectId: z.string(),
       query: BatchActionQuerySchema,
-      tableName: z.nativeEnum(BatchExportTableName),
+      tableName: z.enum(BatchTableNames),
       cutoffCreatedAt: z.date(),
       targetId: z.string().optional(),
-      type: z.nativeEnum(BatchActionType),
+      type: z.enum(BatchActionType),
     }),
     z.object({
       actionId: z.literal("trace-delete"),
       projectId: z.string(),
       query: BatchActionQuerySchema,
-      tableName: z.nativeEnum(BatchExportTableName),
+      tableName: z.enum(BatchTableNames),
       cutoffCreatedAt: z.date(),
       targetId: z.string().optional(),
-      type: z.nativeEnum(BatchActionType),
+      type: z.enum(BatchActionType),
     }),
     z.object({
       actionId: z.literal("trace-add-to-annotation-queue"),
       projectId: z.string(),
       query: BatchActionQuerySchema,
-      tableName: z.nativeEnum(BatchExportTableName),
+      tableName: z.enum(BatchTableNames),
       cutoffCreatedAt: z.date(),
       targetId: z.string().optional(),
-      type: z.nativeEnum(BatchActionType),
+      type: z.enum(BatchActionType),
     }),
     z.object({
       actionId: z.literal("eval-create"),

packages/shared/src/server/redis/batchActionQueue.ts

@@ -1,6 +1,10 @@
 import { Queue } from "bullmq";
 import { QueueName, TQueueJobTypes } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import {
+  createNewRedisInstance,
+  redisQueueRetryOptions,
+  getQueuePrefix,
+} from "./redis";
 import { logger } from "../logger";
 
 export class BatchActionQueue {
@@ -23,6 +27,7 @@ export class BatchActionQueue {
           QueueName.BatchActionQueue,
           {
             connection: newRedis,
+            prefix: getQueuePrefix(QueueName.BatchActionQueue),
             defaultJobOptions: {
               removeOnComplete: true,
               removeOnFail: 10_000,

packages/shared/src/server/redis/batchExport.ts

@@ -1,6 +1,6 @@
 import { Queue } from "bullmq";
 import { QueueName, TQueueJobTypes } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 
 export class BatchExportQueue {
@@ -22,6 +22,7 @@ export class BatchExportQueue {
           QueueName.BatchExport,
           {
             connection: newRedis,
+            prefix: getQueuePrefix(QueueName.BatchExport),
             defaultJobOptions: {
               removeOnComplete: true,
               removeOnFail: 10_000,

packages/shared/src/server/redis/blobStorageIntegrationProcessingQueue.ts

@@ -1,6 +1,6 @@
 import { Queue } from "bullmq";
 import { QueueName } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 
 export class BlobStorageIntegrationProcessingQueue {
@@ -18,7 +18,8 @@ export class BlobStorageIntegrationProcessingQueue {
 
     BlobStorageIntegrationProcessingQueue.instance = newRedis
       ? new Queue(QueueName.BlobStorageIntegrationProcessingQueue, {
-          connection: newRedis,
+            connection: newRedis,
+            prefix: getQueuePrefix(QueueName.BlobStorageIntegrationProcessingQueue),
           defaultJobOptions: {
             removeOnComplete: true,
             removeOnFail: 100_000,

packages/shared/src/server/redis/blobStorageIntegrationQueue.ts

@@ -1,6 +1,10 @@
 import { Queue } from "bullmq";
 import { QueueName, QueueJobs } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import {
+  createNewRedisInstance,
+  redisQueueRetryOptions,
+  getQueuePrefix,
+} from "./redis";
 import { logger } from "../logger";
 
 export class BlobStorageIntegrationQueue {
@@ -19,6 +23,7 @@ export class BlobStorageIntegrationQueue {
     BlobStorageIntegrationQueue.instance = newRedis
       ? new Queue(QueueName.BlobStorageIntegrationQueue, {
           connection: newRedis,
+          prefix: getQueuePrefix(QueueName.BlobStorageIntegrationQueue),
           defaultJobOptions: {
             removeOnComplete: true,
             removeOnFail: 100,

packages/shared/src/server/redis/cloudUsageMeteringQueue.ts

@@ -1,7 +1,7 @@
 import { Queue } from "bullmq";
 import { env } from "../../env";
 import { QueueName, QueueJobs } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 
 export class CloudUsageMeteringQueue {
@@ -24,6 +24,7 @@ export class CloudUsageMeteringQueue {
     CloudUsageMeteringQueue.instance = newRedis
       ? new Queue(QueueName.CloudUsageMeteringQueue, {
           connection: newRedis,
+          prefix: getQueuePrefix(QueueName.CloudUsageMeteringQueue),
           defaultJobOptions: {
             removeOnComplete: true,
             removeOnFail: 100,

packages/shared/src/server/redis/coreDataS3ExportQueue.ts

@@ -1,6 +1,6 @@
 import { Queue } from "bullmq";
 import { QueueName, QueueJobs } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 import { env } from "../../env";
 
@@ -23,7 +23,8 @@ export class CoreDataS3ExportQueue {
 
     CoreDataS3ExportQueue.instance = newRedis
       ? new Queue(QueueName.CoreDataS3ExportQueue, {
-          connection: newRedis,
+            connection: newRedis,
+            prefix: getQueuePrefix(QueueName.CoreDataS3ExportQueue),
           defaultJobOptions: {
             removeOnComplete: true,
             removeOnFail: 100,

packages/shared/src/server/redis/createEvalQueue.ts

@@ -1,6 +1,6 @@
 import { QueueName, TQueueJobTypes } from "../queues";
 import { Queue } from "bullmq";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 
 export class CreateEvalQueue {
@@ -23,6 +23,7 @@ export class CreateEvalQueue {
           QueueName.CreateEvalQueue,
           {
             connection: newRedis,
+            prefix: getQueuePrefix(QueueName.CreateEvalQueue),
             defaultJobOptions: {
               removeOnComplete: 100, // Important: If not true, new jobs for that ID would be ignored as jobs in the complete set are still considered as part of the queue
               removeOnFail: 100_000,

packages/shared/src/server/redis/dataRetentionProcessingQueue.ts

@@ -1,6 +1,6 @@
 import { Queue } from "bullmq";
 import { QueueName } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 
 export class DataRetentionProcessingQueue {
@@ -18,7 +18,8 @@ export class DataRetentionProcessingQueue {
 
     DataRetentionProcessingQueue.instance = newRedis
       ? new Queue(QueueName.DataRetentionProcessingQueue, {
-          connection: newRedis,
+            connection: newRedis,
+            prefix: getQueuePrefix(QueueName.DataRetentionProcessingQueue),
           defaultJobOptions: {
             removeOnComplete: true,
             removeOnFail: 10000,

packages/shared/src/server/redis/dataRetentionQueue.ts

@@ -1,6 +1,6 @@
 import { Queue } from "bullmq";
 import { QueueName, QueueJobs } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 
 export class DataRetentionQueue {
@@ -18,7 +18,8 @@ export class DataRetentionQueue {
 
     DataRetentionQueue.instance = newRedis
       ? new Queue(QueueName.DataRetentionQueue, {
-          connection: newRedis,
+            connection: newRedis,
+            prefix: getQueuePrefix(QueueName.DataRetentionQueue),
           defaultJobOptions: {
             removeOnComplete: true,
             removeOnFail: 100,

packages/shared/src/server/redis/datasetRunItemUpsert.ts

@@ -1,6 +1,6 @@
 import { QueueName, TQueueJobTypes } from "../queues";
 import { Queue } from "bullmq";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 
 export class DatasetRunItemUpsertQueue {
@@ -24,6 +24,7 @@ export class DatasetRunItemUpsertQueue {
           QueueName.DatasetRunItemUpsert,
           {
             connection: newRedis,
+            prefix: getQueuePrefix(QueueName.DatasetRunItemUpsert),
             defaultJobOptions: {
               removeOnComplete: true,
               removeOnFail: 10_000,

packages/shared/src/server/redis/dlqRetryQueue.ts

@@ -1,6 +1,6 @@
 import { Queue } from "bullmq";
 import { QueueName, QueueJobs } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 
 export class DeadLetterRetryQueue {
@@ -18,7 +18,8 @@ export class DeadLetterRetryQueue {
 
     DeadLetterRetryQueue.instance = newRedis
       ? new Queue(QueueName.DeadLetterRetryQueue, {
-          connection: newRedis,
+            connection: newRedis,
+            prefix: getQueuePrefix(QueueName.DeadLetterRetryQueue),
           defaultJobOptions: {
             removeOnComplete: true,
             removeOnFail: 100,

packages/shared/src/server/redis/evalExecutionQueue.ts

@@ -1,7 +1,7 @@
 import { Queue } from "bullmq";
 import { logger } from "../logger";
 import { TQueueJobTypes, QueueName } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 
 export class EvalExecutionQueue {
   private static instance: Queue<
@@ -23,6 +23,7 @@ export class EvalExecutionQueue {
           QueueName.EvaluationExecution,
           {
             connection: newRedis,
+            prefix: getQueuePrefix(QueueName.EvaluationExecution),
             defaultJobOptions: {
               removeOnComplete: true,
               removeOnFail: 10_000,

packages/shared/src/server/redis/experimentCreateQueue.ts

@@ -1,7 +1,7 @@
 import { Queue } from "bullmq";
 import { logger } from "../logger";
 import { TQueueJobTypes, QueueName } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 
 export class ExperimentCreateQueue {
   private static instance: Queue<
@@ -23,6 +23,7 @@ export class ExperimentCreateQueue {
           QueueName.ExperimentCreate,
           {
             connection: newRedis,
+            prefix: getQueuePrefix(QueueName.ExperimentCreate),
             defaultJobOptions: {
               removeOnComplete: true,
               removeOnFail: 10_000,

packages/shared/src/server/redis/getQueue.ts

@@ -5,7 +5,7 @@ import { CloudUsageMeteringQueue } from "./cloudUsageMeteringQueue";
 import { DatasetRunItemUpsertQueue } from "./datasetRunItemUpsert";
 import { EvalExecutionQueue } from "./evalExecutionQueue";
 import { ExperimentCreateQueue } from "./experimentCreateQueue";
-import { IngestionQueue, SecondaryIngestionQueue } from "./ingestionQueue";
+import { SecondaryIngestionQueue } from "./ingestionQueue";
 import { TraceUpsertQueue } from "./traceUpsert";
 import { TraceDeleteQueue } from "./traceDelete";
 import { ProjectDeleteQueue } from "./projectDelete";
@@ -22,7 +22,11 @@ import { CreateEvalQueue } from "./createEvalQueue";
 import { ScoreDeleteQueue } from "./scoreDelete";
 import { DeadLetterRetryQueue } from "./dlqRetryQueue";
 
-export function getQueue(queueName: QueueName): Queue | null {
+// IngestionQueue is sharded and requires a sharding key
+// Use IngestionQueue.getInstance({ shardName: queueName }) directly instead
+export function getQueue(
+  queueName: Exclude<QueueName, QueueName.IngestionQueue>,
+): Queue | null {
   switch (queueName) {
     case QueueName.BatchExport:
       return BatchExportQueue.getInstance();
@@ -38,8 +42,6 @@ export function getQueue(queueName: QueueName): Queue | null {
       return TraceUpsertQueue.getInstance();
     case QueueName.TraceDelete:
       return TraceDeleteQueue.getInstance();
-    case QueueName.IngestionQueue:
-      return IngestionQueue.getInstance();
     case QueueName.ProjectDelete:
       return ProjectDeleteQueue.getInstance();
     case QueueName.PostHogIntegrationQueue:

packages/shared/src/server/redis/ingestionQueue.ts

@@ -1,46 +1,94 @@
 import { Queue } from "bullmq";
 import { QueueName, TQueueJobTypes } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import {
+  createNewRedisInstance,
+  redisQueueRetryOptions,
+  getQueuePrefix,
+} from "./redis";
 import { logger } from "../logger";
+import { getShardIndex } from "./sharding";
+import { env } from "../../env";
 
 export class IngestionQueue {
-  private static instance: Queue<
-    TQueueJobTypes[QueueName.IngestionQueue]
-  > | null = null;
+  private static instances: Map<
+    number,
+    Queue<TQueueJobTypes[QueueName.IngestionQueue]> | null
+  > = new Map();
 
-  public static getInstance(): Queue<
-    TQueueJobTypes[QueueName.IngestionQueue]
-  > | null {
-    if (IngestionQueue.instance) return IngestionQueue.instance;
+  public static getShardNames() {
+    return Array.from(
+      { length: env.LANGFUSE_INGESTION_QUEUE_SHARD_COUNT },
+      (_, i) => `${QueueName.IngestionQueue}${i > 0 ? `-${i}` : ""}`,
+    );
+  }
+
+  static getShardIndexFromShardName(
+    shardName: string | undefined,
+  ): number | null {
+    if (!shardName) return null;
+
+    // Extract shard index from shard name
+    const shardIndex =
+      shardName === QueueName.IngestionQueue
+        ? 0
+        : parseInt(shardName.replace(`${QueueName.IngestionQueue}-`, ""), 10);
+
+    if (isNaN(shardIndex)) return null;
+    return shardIndex;
+  }
+
+  /**
+   * Get the ingestion queue instance for the given sharding key or shard name.
+   * @param shardingKey - ShardingKey is being hashed and randomly allocated to a shard. Should be `projectId-eventBodyId`.
+   * @param shardName - Name of the shard. Should be `ingestion-queue-${shardIndex}` or plainly `ingestion-queue` for the first shard.
+   */
+  public static getInstance({
+    shardingKey,
+    shardName,
+  }: {
+    shardingKey?: string;
+    shardName?: string;
+  }): Queue<TQueueJobTypes[QueueName.IngestionQueue]> | null {
+    const shardIndex =
+      IngestionQueue.getShardIndexFromShardName(shardName) ??
+      (env.REDIS_CLUSTER_ENABLED === "true" && shardingKey
+        ? getShardIndex(shardingKey, env.LANGFUSE_INGESTION_QUEUE_SHARD_COUNT)
+        : 0);
+
+    // Check if we already have an instance for this shard
+    if (IngestionQueue.instances.has(shardIndex)) {
+      return IngestionQueue.instances.get(shardIndex) || null;
+    }
 
     const newRedis = createNewRedisInstance({
       enableOfflineQueue: false,
       ...redisQueueRetryOptions,
     });
 
-    IngestionQueue.instance = newRedis
-      ? new Queue<TQueueJobTypes[QueueName.IngestionQueue]>(
-          QueueName.IngestionQueue,
-          {
-            connection: newRedis,
-            defaultJobOptions: {
-              removeOnComplete: true,
-              removeOnFail: 100_000,
-              attempts: 5,
-              backoff: {
-                type: "exponential",
-                delay: 5000,
-              },
+    const name = `${QueueName.IngestionQueue}${shardIndex > 0 ? `-${shardIndex}` : ""}`;
+    const queueInstance = newRedis
+      ? new Queue<TQueueJobTypes[QueueName.IngestionQueue]>(name, {
+          connection: newRedis,
+          prefix: getQueuePrefix(name),
+          defaultJobOptions: {
+            removeOnComplete: true,
+            removeOnFail: 100_000,
+            attempts: 5,
+            backoff: {
+              type: "exponential",
+              delay: 5000,
             },
           },
-        )
+        })
       : null;
 
-    IngestionQueue.instance?.on("error", (err) => {
-      logger.error("IngestionQueue error", err);
+    queueInstance?.on("error", (err) => {
+      logger.error(`IngestionQueue shard ${shardIndex} error`, err);
     });
 
-    return IngestionQueue.instance;
+    IngestionQueue.instances.set(shardIndex, queueInstance);
+
+    return queueInstance;
   }
 }
 
@@ -65,6 +113,7 @@ export class SecondaryIngestionQueue {
           QueueName.IngestionSecondaryQueue,
           {
             connection: newRedis,
+            prefix: getQueuePrefix(QueueName.IngestionSecondaryQueue),
             defaultJobOptions: {
               removeOnComplete: true,
               removeOnFail: 100_000,

packages/shared/src/server/redis/meteringDataPostgresExportQueue.ts

@@ -1,6 +1,6 @@
 import { Queue } from "bullmq";
 import { QueueName, QueueJobs } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 import { env } from "../../env";
 
@@ -23,7 +23,8 @@ export class MeteringDataPostgresExportQueue {
 
     MeteringDataPostgresExportQueue.instance = newRedis
       ? new Queue(QueueName.MeteringDataPostgresExportQueue, {
-          connection: newRedis,
+            connection: newRedis,
+            prefix: getQueuePrefix(QueueName.MeteringDataPostgresExportQueue),
           defaultJobOptions: {
             removeOnComplete: true,
             removeOnFail: 100,

packages/shared/src/server/redis/postHogIntegrationProcessingQueue.ts

@@ -1,6 +1,6 @@
 import { Queue } from "bullmq";
 import { QueueName } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 
 export class PostHogIntegrationProcessingQueue {
@@ -18,7 +18,8 @@ export class PostHogIntegrationProcessingQueue {
 
     PostHogIntegrationProcessingQueue.instance = newRedis
       ? new Queue(QueueName.PostHogIntegrationProcessingQueue, {
-          connection: newRedis,
+            connection: newRedis,
+            prefix: getQueuePrefix(QueueName.PostHogIntegrationProcessingQueue),
           defaultJobOptions: {
             removeOnComplete: true,
             removeOnFail: 100_000,

packages/shared/src/server/redis/postHogIntegrationQueue.ts

@@ -1,6 +1,6 @@
 import { Queue } from "bullmq";
 import { QueueName, QueueJobs } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 
 export class PostHogIntegrationQueue {
@@ -18,7 +18,8 @@ export class PostHogIntegrationQueue {
 
     PostHogIntegrationQueue.instance = newRedis
       ? new Queue(QueueName.PostHogIntegrationQueue, {
-          connection: newRedis,
+            connection: newRedis,
+            prefix: getQueuePrefix(QueueName.PostHogIntegrationQueue),
           defaultJobOptions: {
             removeOnComplete: true,
             removeOnFail: 100,

packages/shared/src/server/redis/projectDelete.ts

@@ -1,6 +1,6 @@
 import { QueueName, TQueueJobTypes } from "../queues";
 import { Queue } from "bullmq";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 
 export class ProjectDeleteQueue {
@@ -23,6 +23,7 @@ export class ProjectDeleteQueue {
           QueueName.ProjectDelete,
           {
             connection: newRedis,
+            prefix: getQueuePrefix(QueueName.ProjectDelete),
             defaultJobOptions: {
               removeOnComplete: true,
               removeOnFail: 100_000,

packages/shared/src/server/redis/redis.ts

@@ -1,4 +1,4 @@
-import Redis, { RedisOptions } from "ioredis";
+import Redis, { RedisOptions, Cluster, ClusterOptions } from "ioredis";
 import fs from "fs";
 import { env } from "../../env";
 import { logger } from "../logger";
@@ -24,26 +24,99 @@ export const redisQueueRetryOptions: Partial<RedisOptions> = {
   },
 };
 
+/**
+ * Parse Redis cluster nodes from environment variable
+ * Format: "host1:port1,host2:port2,host3:port3"
+ */
+const parseClusterNodes = (
+  nodesString: string,
+): Array<{ host: string; port: number }> => {
+  return nodesString.split(",").map((node) => {
+    const [host, port] = node.trim().split(":");
+    if (!host || !port) {
+      throw new Error(
+        `Invalid cluster node format: ${node}. Expected format: host:port`,
+      );
+    }
+    return { host, port: parseInt(port, 10) };
+  });
+};
+
+const createRedisClusterInstance = (
+  additionalOptions: Partial<RedisOptions> = {},
+): Cluster | null => {
+  if (!env.REDIS_CLUSTER_NODES) {
+    logger.error(
+      "REDIS_CLUSTER_NODES is required when REDIS_CLUSTER_ENABLED is true",
+    );
+    return null;
+  }
+
+  const nodes = parseClusterNodes(env.REDIS_CLUSTER_NODES);
+  const tlsOptions =
+    env.REDIS_TLS_ENABLED === "true"
+      ? {
+          tls: {
+            ca: env.REDIS_TLS_CA_PATH
+              ? fs.readFileSync(env.REDIS_TLS_CA_PATH)
+              : undefined,
+            cert: env.REDIS_TLS_CERT_PATH
+              ? fs.readFileSync(env.REDIS_TLS_CERT_PATH)
+              : undefined,
+            key: env.REDIS_TLS_KEY_PATH
+              ? fs.readFileSync(env.REDIS_TLS_KEY_PATH)
+              : undefined,
+          },
+        }
+      : {};
+
+  const clusterOptions: ClusterOptions = {
+    // Return incoming addresses as-is - required for AWS ElastiCache Certificate resolution
+    dnsLookup: (address, callback) => {
+      callback(null, address);
+    },
+    redisOptions: {
+      password: env.REDIS_AUTH || undefined,
+      ...defaultRedisOptions,
+      ...additionalOptions,
+      ...tlsOptions,
+    },
+    // Retry configuration for cluster
+    retryDelayOnFailover: 100,
+  };
+
+  const cluster = new Cluster(nodes, clusterOptions);
+
+  cluster.on("error", (error) => {
+    logger.error("Redis cluster error", error);
+  });
+
+  return cluster;
+};
+
 export const createNewRedisInstance = (
   additionalOptions: Partial<RedisOptions> = {},
-) => {
-  const tlsEnabled = env.REDIS_TLS_ENABLED === "true";
+): Redis | Cluster | null => {
+  if (env.REDIS_CLUSTER_ENABLED === "true") {
+    return createRedisClusterInstance(additionalOptions);
+  }
 
-  const tlsOptions = tlsEnabled
-    ? {
-        tls: {
-          ca: env.REDIS_TLS_CA_PATH
-            ? fs.readFileSync(env.REDIS_TLS_CA_PATH)
-            : undefined,
-          cert: env.REDIS_TLS_CERT_PATH
-            ? fs.readFileSync(env.REDIS_TLS_CERT_PATH)
-            : undefined,
-          key: env.REDIS_TLS_KEY_PATH
-            ? fs.readFileSync(env.REDIS_TLS_KEY_PATH)
-            : undefined,
-        },
-      }
-    : {};
+  const tlsOptions =
+    env.REDIS_TLS_ENABLED === "true"
+      ? {
+          tls: {
+            ca: env.REDIS_TLS_CA_PATH
+              ? fs.readFileSync(env.REDIS_TLS_CA_PATH)
+              : undefined,
+            cert: env.REDIS_TLS_CERT_PATH
+              ? fs.readFileSync(env.REDIS_TLS_CERT_PATH)
+              : undefined,
+            key: env.REDIS_TLS_KEY_PATH
+              ? fs.readFileSync(env.REDIS_TLS_KEY_PATH)
+              : undefined,
+          },
+        }
+      : {};
 
   const instance = env.REDIS_CONNECTION_STRING
     ? new Redis(env.REDIS_CONNECTION_STRING, {
@@ -69,6 +142,20 @@ export const createNewRedisInstance = (
   return instance;
 };
 
+/**
+ * Get the queue prefix for BullMQ cluster compatibility
+ * In cluster mode, uses hash tags to ensure queue keys are on the same node
+ * In single-node mode, returns undefined (no prefix needed)
+ */
+export const getQueuePrefix = (queueName: string): string | undefined => {
+  if (env.REDIS_CLUSTER_ENABLED === "true") {
+    // Use hash tags for Redis cluster compatibility
+    // This ensures all keys for a queue are placed on the same hash slot
+    return `{${queueName}}`;
+  }
+  return undefined;
+};
+
 const createRedisClient = () => {
   try {
     return createNewRedisInstance();

packages/shared/src/server/redis/scoreDelete.ts

@@ -1,6 +1,6 @@
 import { QueueName, TQueueJobTypes } from "../queues";
 import { Queue } from "bullmq";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 
 export class ScoreDeleteQueue {
@@ -15,8 +15,11 @@ export class ScoreDeleteQueue {
     });
 
     ScoreDeleteQueue.instance = newRedis
-      ? new Queue<TQueueJobTypes[QueueName.ScoreDelete]>(QueueName.ScoreDelete, {
-          connection: newRedis,
+      ? new Queue<TQueueJobTypes[QueueName.ScoreDelete]>(
+          QueueName.ScoreDelete,
+          {
+            connection: newRedis,
+            prefix: getQueuePrefix(QueueName.ScoreDelete),
           defaultJobOptions: {
             removeOnComplete: true,
             removeOnFail: 100_000,

packages/shared/src/server/redis/sharding.ts

@@ -0,0 +1,20 @@
+import { createHash } from "crypto";
+
+/**
+ * Utility function to compute a consistent hash for a given key and map it to a shard index
+ * @param key - The key to hash
+ * @param shardCount - The number of shards to distribute across
+ * @returns A shard index between 0 and shardCount-1
+ */
+export function getShardIndex(key: string, shardCount: number): number {
+  if (shardCount <= 1) return 0;
+
+  // Create a consistent hash using SHA-256
+  const hash = createHash("sha256").update(key).digest("hex");
+
+  // Convert first 8 characters of hex to integer
+  const hashInt = parseInt(hash.substring(0, 8), 16);
+
+  // Map to shard index
+  return hashInt % shardCount;
+}

packages/shared/src/server/redis/traceDelete.ts

@@ -1,6 +1,6 @@
 import { QueueName, TQueueJobTypes } from "../queues";
 import { Queue } from "bullmq";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 
 export class TraceDeleteQueue {
@@ -22,6 +22,7 @@ export class TraceDeleteQueue {
           QueueName.TraceDelete,
           {
             connection: newRedis,
+            prefix: getQueuePrefix(QueueName.TraceDelete),
             defaultJobOptions: {
               removeOnComplete: true,
               removeOnFail: 100_000,

packages/shared/src/server/redis/traceUpsert.ts

@@ -1,6 +1,6 @@
 import { QueueName, TQueueJobTypes } from "../queues";
 import { Queue } from "bullmq";
-import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
 import { logger } from "../logger";
 
 export class TraceUpsertQueue {
@@ -22,6 +22,7 @@ export class TraceUpsertQueue {
           QueueName.TraceUpsert,
           {
             connection: newRedis,
+            prefix: getQueuePrefix(QueueName.TraceUpsert),
             defaultJobOptions: {
               removeOnComplete: 100, // Important: If not true, new jobs for that ID would be ignored as jobs in the complete set are still considered as part of the queue
               removeOnFail: 100_000,

packages/shared/src/server/repositories/definitions.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 
 export const clickhouseStringDateSchema = z
   .string()
@@ -38,7 +38,7 @@ export const observationRecordBaseSchema = z.object({
   parent_observation_id: z.string().nullish(),
   environment: z.string().default("default"),
   name: z.string().nullish(),
-  metadata: z.record(z.string()),
+  metadata: z.record(z.string(), z.string()),
   level: z.string().nullish(),
   status_message: z.string().nullish(),
   version: z.string().nullish(),
@@ -92,7 +92,7 @@ export const traceRecordBaseSchema = z.object({
   id: z.string(),
   name: z.string().nullish(),
   user_id: z.string().nullish(),
-  metadata: z.record(z.string()),
+  metadata: z.record(z.string(), z.string()),
   release: z.string().nullish(),
   version: z.string().nullish(),
   project_id: z.string(),
@@ -134,7 +134,7 @@ export const scoreRecordBaseSchema = z.object({
   value: z.number().nullish(),
   source: z.string(),
   comment: z.string().nullish(),
-  metadata: z.record(z.string()),
+  metadata: z.record(z.string(), z.string()),
   author_user_id: z.string().nullish(),
   config_id: z.string().nullish(),
   data_type: z.enum(["NUMERIC", "CATEGORICAL", "BOOLEAN"]).nullish(),

packages/shared/src/server/repositories/trace-sessions.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { prisma } from "../../db";
 import { singleFilter } from "../../interfaces/filters";
 

packages/shared/src/server/services/DashboardService/DashboardService.ts

@@ -14,7 +14,7 @@ import {
   WidgetDomainSchema,
   DashboardDefinitionSchema,
 } from "./types";
-import { z } from "zod";
+import { z } from "zod/v4";
 
 export class DashboardService {
   /**

packages/shared/src/server/services/DashboardService/types.ts

@@ -1,5 +1,5 @@
 import { DashboardWidgetChartType, DashboardWidgetViews } from "@prisma/client";
-import { z } from "zod";
+import { z } from "zod/v4";
 import { singleFilter } from "../../../";
 
 export const BaseTimeSeriesChartConfig = z.object({});
@@ -28,6 +28,11 @@ export const BigNumberChartConfig = BaseTotalValueChartConfig.extend({
   type: z.literal("NUMBER"),
 });
 
+export const HistogramChartConfig = BaseTotalValueChartConfig.extend({
+  type: z.literal("HISTOGRAM"),
+  bins: z.number().int().min(1).max(100).optional().default(10),
+});
+
 // Define dimension schema
 export const DimensionSchema = z.object({
   field: z.string(),
@@ -47,6 +52,7 @@ export const ChartConfigSchema = z.discriminatedUnion("type", [
   VerticalBarChartConfig,
   PieChartConfig,
   BigNumberChartConfig,
+  HistogramChartConfig,
 ]);
 
 export const DashboardDefinitionWidgetWidgetSchema = z.object({
@@ -99,11 +105,11 @@ export const WidgetDomainSchema = z.object({
   projectId: z.string().nullable(),
   name: z.string(),
   description: z.string(),
-  view: z.nativeEnum(DashboardWidgetViews),
+  view: z.enum(DashboardWidgetViews),
   dimensions: z.array(DimensionSchema),
   metrics: z.array(MetricSchema),
   filters: z.array(singleFilter),
-  chartType: z.nativeEnum(DashboardWidgetChartType),
+  chartType: z.enum(DashboardWidgetChartType),
   chartConfig: ChartConfigSchema,
   owner: OwnerEnum,
 });
@@ -112,11 +118,11 @@ export const WidgetDomainSchema = z.object({
 export const CreateWidgetInputSchema = z.object({
   name: z.string().min(1, "Widget name is required"),
   description: z.string(),
-  view: z.nativeEnum(DashboardWidgetViews),
+  view: z.enum(DashboardWidgetViews),
   dimensions: z.array(DimensionSchema),
   metrics: z.array(MetricSchema),
   filters: z.array(singleFilter),
-  chartType: z.nativeEnum(DashboardWidgetChartType),
+  chartType: z.enum(DashboardWidgetChartType),
   chartConfig: ChartConfigSchema,
 });
 

packages/shared/src/server/services/DefaultEvaluationModelService/DefaultEvalModelService.ts

@@ -1,4 +1,4 @@
-import z from "zod";
+import z from "zod/v4";
 import { prisma } from "../../../db";
 import { LangfuseNotFoundError } from "../../../errors";
 import { LLMApiKeySchema, ZodModelConfig } from "../../llm/types";
@@ -90,7 +90,7 @@ export class DefaultEvalModelService {
       const result = ZodModelConfig.safeParse(config.modelParams);
       if (!result.success) {
         errors.push(
-          ...result.error.errors.map(
+          ...result.error.issues.map(
             (err) => `Model parameter error: ${err.message}`,
           ),
         );

packages/shared/src/server/services/PromptService/index.ts

@@ -1,5 +1,5 @@
 import { Prompt, PrismaClient } from "@prisma/client";
-import { Redis } from "ioredis";
+import { Redis, Cluster } from "ioredis";
 import { env } from "../../../env";
 import { logger } from "../../logger";
 import { escapeRegex } from "./utils";
@@ -22,7 +22,7 @@ export class PromptService {
 
   constructor(
     private prisma: PrismaClient,
-    private redis: Redis | null,
+    private redis: Redis | Cluster | null,
     private metricIncrementer?: // used for otel metrics
     (name: string, value?: number) => void,
     cacheEnabled?: boolean, // used for testing
@@ -157,6 +157,12 @@ export class PromptService {
     }
   }
 
+  /**
+   * Lock the cache so reads will go to the database and not to Redis
+   *
+   * This is useful in order to return consistent data during the
+   * invalidation of the cache where we are looping through the relevant cache keys
+   */
   public async lockCache(
     params: Pick<PromptParams, "projectId" | "promptName">,
   ): Promise<void> {