chore: release v3.80.0

* security: upgrade golang-migrate

* push

* push

* push

.claude/settings.local.json

@@ -0,0 +1,11 @@
+{
+  "permissions": {
+    "allow": [
+      "Bash(find:*)",
+      "Bash(rg:*)",
+      "Bash(grep:*)",
+      "Bash(pnpm run test:*)"
+    ],
+    "deny": []
+  }
+}

.cursor/Dockerfile

@@ -12,8 +12,15 @@ RUN apt-get update && \
       postgresql-client \
       redis-tools \
       less nano \
+      sudo \
     && rm -rf /var/lib/apt/lists/*
 
+# ---------- Docker -----------------------------------------------------------
+# Install Docker for background agents that need container capabilities
+RUN curl -fsSL https://get.docker.com -o get-docker.sh && \
+    sh get-docker.sh && \
+    rm get-docker.sh
+
 # ---------- pnpm -------------------------------------------------------------
 # Langfuse monorepo relies on pnpm 9.5.0 (see CONTRIBUTING.md)
 ENV PNPM_HOME="/pnpm"
@@ -23,17 +30,17 @@ RUN corepack enable && \
 
 # ---------- golang-migrate ----------------------------------------------------
 # CLI used for database migrations during development.
-ENV MIGRATE_VERSION=4.18.2
+ENV MIGRATE_VERSION=4.18.3
 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
+# Create non-root user with sudo privileges and docker group access
+RUN useradd -ms /bin/bash ubuntu && \
+    usermod -aG sudo ubuntu && \
+    usermod -aG docker ubuntu && \
+    echo "ubuntu ALL=(ALL) NOPASSWD:ALL" >> /etc/sudoers
 
 # Pre-create pnpm store and set correct ownership to avoid first-run cost & permission issues
 RUN pnpm store path > /dev/null && \

.cursor/environment.json

@@ -4,5 +4,11 @@
     "context": ".",
     "dockerfile": "Dockerfile"
   },
-  "start": "pnpm dx-f"
+  "start": "sudo service docker start",
+  "terminals": [
+    {
+      "name": "dev server",
+      "command": "pnpm run dx-f"
+    }
+  ]
 }

.cursor/rules/general-info.mdc

@@ -5,4 +5,5 @@ alwaysApply: true
 ---
 # General rules
 
-- Linting in this repo only works if the development server is running
+- Linting in this repo only works if the development server is running
+- Always run the full mono-repo via `pnpm run dx` (use `pnpm dx-f` when you run this in a background agent). Thereby the database will also be seeded.

.cursor/rules/global.mdc

@@ -0,0 +1,20 @@
+---
+description: 
+globs: 
+alwaysApply: true
+---
+## Project setup
+- This project is a Turborepo monorepo
+- We build two containers, web (./web) and worker (./worker)
+- We have shared code between these two in the shared package (./packages/shared). For that package, we have different entry points [package.json](mdc:packages/shared/package.json).
+
+
+## Domain layer
+The most important domain objects are in [observations.ts](mdc:packages/shared/src/domain/observations.ts), [traces.ts](mdc:packages/shared/src/domain/traces.ts), [scores.ts](mdc:packages/shared/src/domain/scores.ts).
+
+
+## Database schema
+We use Postgres and Clickhouse.
+- The postgres schema is in [schema.prisma](mdc:packages/shared/prisma/schema.prisma)
+- The clickhouse schema is in [0001_traces.up.sql](mdc:packages/shared/clickhouse/migrations/clustered/0001_traces.up.sql), [0002_observations.up.sql](mdc:packages/shared/clickhouse/migrations/clustered/0002_observations.up.sql), [0003_scores.up.sql](mdc:packages/shared/clickhouse/migrations/clustered/0003_scores.up.sql)
+

.cursor/rules/public-api.mdc

@@ -1,6 +1,6 @@
 ---
 description: How to create new public api routes for Langfuse
-globs: 
+globs:
 alwaysApply: false
 ---
 
@@ -13,4 +13,5 @@ alwaysApply: false
   - Add end-to-end test, similar to [datasets-api.servertest.ts](mdc:web/src/__tests__/async/datasets-api.servertest.ts)
   - Add fern configuration in /fern, learn more about structure here: https://buildwithfern.com/learn/api-definition/fern/overview
   - Prompt user to regenerate the OpenAPI spec via the fern CLI
-  - Pagination starts at 1, query typing defined in publicApiPaginationZod, return meta in paginationMetaResponseZod
+  - Pagination starts at 1, query typing defined in publicApiPaginationZod, return meta in paginationMetaResponseZod
+- For tests, please look at [this directory](mdc:web/src/__tests__/async/) for examples

.cursor/rules/tests.mdc

@@ -0,0 +1,9 @@
+---
+description: 
+globs: 
+alwaysApply: true
+---
+# Writing tests
+
+- When writing tests, focus on decoupling each `it` or `test` block to ensure that they can run independently and concurrently. Tests must never depend on the action or outcome of previous or subsequent tests.
+- When writing tests, especially in the __tests__/async directory, ensure that you avoid `pruneDatabase` calls.

.env.test.example

@@ -0,0 +1,12 @@
+# Test Environment Configuration
+# Copy this file to .env.test for test database isolation
+# Only overrides specific test variables - other values inherited from .env
+
+# PostgreSQL - Test Database
+DIRECT_URL="postgresql://postgres:postgres@localhost:5432/langfuse_test"
+DATABASE_URL="postgresql://postgres:postgres@localhost:5432/langfuse_test"
+
+# ClickHouse - Use Default Database for now, nothing set
+
+# Redis - Test Database (database 1 for isolation)
+REDIS_CONNECTION_STRING="redis://:myredissecret@127.0.0.1:6379/1"

.github/PULL_REQUEST_TEMPLATE.md

@@ -28,7 +28,7 @@ Fixes # (issue)
 <!-- Remove bullet points below that don't apply to you -->
 
 - I haven't read the [contributing guide](https://github.com/langfuse/langfuse/blob/main/CONTRIBUTING.md)
-- My code doesn't follow the style guidelines of this project (`npm run prettier`)
+- My code doesn't follow the style guidelines of this project (`pnpm run format`)
 - I haven't commented my code, particularly in hard-to-understand areas
 - I haven't checked if my PR needs changes to the documentation
 - I haven't checked if my changes generate no new warnings (`npm run lint`)

.github/workflows/pipeline.yml

@@ -52,6 +52,49 @@ jobs:
       - name: lint web
         run: pnpm run lint
 
+  prettier-check:
+    runs-on: ubuntu-latest
+    needs:
+      - pre-job
+    if: needs.pre-job.outputs.should_skip != 'true'
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+      - uses: pnpm/action-setup@v3
+        with:
+          version: 9.5.0
+      - uses: actions/setup-node@v4
+        with:
+          node-version: 20
+          cache: "pnpm"
+          cache-dependency-path: "pnpm-lock.yaml"
+      - name: install dependencies
+        run: |
+          pnpm i
+      - name: Load default env
+        run: |
+          cp .env.dev.example .env
+      - name: Check formatting on changed files
+        run: |
+          if [ "${{ github.event_name }}" = "pull_request" ]; then
+            BASE_SHA=${{ github.event.pull_request.base.sha }}
+          else
+            BASE_SHA=$(git merge-base origin/main HEAD)
+          fi
+          
+          echo "Checking files changed from $BASE_SHA to HEAD"
+          
+          # Get changed files
+          CHANGED_FILES=$(git diff --name-only $BASE_SHA HEAD -- '*.js' '*.jsx' '*.ts' '*.tsx' '*.css' | tr '\n' ' ')
+          
+          if [ -n "$CHANGED_FILES" ] && [ "$CHANGED_FILES" != " " ]; then
+            echo "Files to check: $CHANGED_FILES"
+            pnpm prettier --check --experimental-cli $CHANGED_FILES
+          else
+            echo "No JS/TS/CSS files changed - skipping prettier check"
+          fi
+
   test-docker-build:
     timeout-minutes: 20
     runs-on: ubuntu-latest
@@ -100,14 +143,10 @@ jobs:
         node-version: [20]
         postgres-version: [12, 15]
     steps:
-      - name: Set Swap Space
-        uses: pierotofy/set-swap-space@master
-        with:
-          swap-size-gb: 10
       - uses: actions/checkout@v4
       - name: Install golang-migrate for Clickhouse migrations
         run: |
-          curl -L https://github.com/golang-migrate/migrate/releases/download/v4.18.2/migrate.linux-amd64.tar.gz | tar xvz
+          curl -L https://github.com/golang-migrate/migrate/releases/download/v4.18.3/migrate.linux-amd64.tar.gz | tar xvz
           sudo mv migrate /usr/bin/migrate
           which migrate
       - uses: pnpm/action-setup@v3
@@ -178,14 +217,10 @@ jobs:
         postgres-version: [12, 15]
         deploy-mode: ["", "-azure", "-redis-cluster"]
     steps:
-      - name: Set Swap Space
-        uses: pierotofy/set-swap-space@master
-        with:
-          swap-size-gb: 10
       - uses: actions/checkout@v4
       - name: Install golang-migrate for Clickhouse migrations
         run: |
-          curl -L https://github.com/golang-migrate/migrate/releases/download/v4.18.2/migrate.linux-amd64.tar.gz | tar xvz
+          curl -L https://github.com/golang-migrate/migrate/releases/download/v4.18.3/migrate.linux-amd64.tar.gz | tar xvz
           sudo mv migrate /usr/bin/migrate
           which migrate
       - uses: pnpm/action-setup@v3
@@ -257,10 +292,6 @@ jobs:
         postgres-version: [12, 15]
         deploy-mode: ["", "-azure", "-redis-cluster"]
     steps:
-      - name: Set Swap Space
-        uses: pierotofy/set-swap-space@master
-        with:
-          swap-size-gb: 10
       - uses: actions/checkout@v4
       - uses: pnpm/action-setup@v3
         with:
@@ -282,7 +313,7 @@ jobs:
           pnpm install
       - name: Install golang-migrate for Clickhouse migrations
         run: |
-          curl -L https://github.com/golang-migrate/migrate/releases/download/v4.18.2/migrate.linux-amd64.tar.gz | tar xvz
+          curl -L https://github.com/golang-migrate/migrate/releases/download/v4.18.3/migrate.linux-amd64.tar.gz | tar xvz
           sudo mv migrate /usr/bin/migrate
           which migrate
       - name: Load default env
@@ -343,7 +374,7 @@ jobs:
           cp .env.dev.example web/.env
       - name: Install golang-migrate for Clickhouse migrations
         run: |
-          curl -L https://github.com/golang-migrate/migrate/releases/download/v4.18.2/migrate.linux-amd64.tar.gz | tar xvz
+          curl -L https://github.com/golang-migrate/migrate/releases/download/v4.18.3/migrate.linux-amd64.tar.gz | tar xvz
           sudo mv migrate /usr/bin/migrate
           which migrate
       - name: Run + migrate
@@ -397,7 +428,7 @@ jobs:
           pnpm install
       - name: Install golang-migrate for Clickhouse migrations
         run: |
-          curl -L https://github.com/golang-migrate/migrate/releases/download/v4.18.2/migrate.linux-amd64.tar.gz | tar xvz
+          curl -L https://github.com/golang-migrate/migrate/releases/download/v4.18.3/migrate.linux-amd64.tar.gz | tar xvz
           sudo mv migrate /usr/bin/migrate
           which migrate
       - name: Load default env
@@ -442,6 +473,7 @@ jobs:
     needs:
       [
         lint,
+        prettier-check,
         tests-web-sync,
         tests-worker,
         e2e-tests,

.gitignore

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

.husky/pre-commit

@@ -11,12 +11,13 @@ if [ "$current_branch" = "$protected_branch" ]; then
     echo "🚨 You are about to commit to the $protected_branch branch. Are you sure? (y/n)"
     read -r answer < /dev/tty
     if [ "$answer" != "${answer#[Yy]}" ]; then
-        exit 0 # Commit will proceed
+        # Commit approved, check formatting. On files changed, block commit
+        pnpm run format:check
     else
         echo "Commit to $protected_branch branch has been canceled."
         exit 1 # Commit will be blocked
     fi
 fi
 
-# If not the protected branch, proceed with the commit
-exit 0
+# If not the protected branch, check formatting (on changed files, block)
+pnpm run format:check

.prettierignore

@@ -0,0 +1,2 @@
+# Generated code
+packages/shared/prisma/generated

AGENTS.md

@@ -8,6 +8,8 @@ Langfuse is an **open source LLM engineering** platform for developing, monitori
 
 ## Tests
 - Codex cannot run the test suite because it depends on Docker-based infrastructure that is unavailable in this environment.
+- When writing tests, focus on decoupling each `it` or `test` block to ensure that they can run independently and concurrently. Tests must never depend on the action or outcome of previous or subsequent tests.
+- When writing tests, especially in the __tests__/async directory, ensure that you avoid `pruneDatabase` calls.
 
 ## Cursor Rules
 - Additional folder-specific rules live in `.cursor/rules/`.

CLAUDE.md

@@ -94,6 +94,7 @@ pnpm run test --filter=worker -- $TEST_FILE_NAME -t "$TEST_NAME"
 
 ### Utilities
 ```bash
+pnpm run format            # Format code across entire project
 pnpm run nuke              # Remove all node_modules, build files, wipe database, docker containers. **USE WITH CAUTION**
 ```
 
@@ -152,6 +153,8 @@ pnpm run nuke              # Remove all node_modules, build files, wipe database
 - Jest for API tests, Playwright for E2E tests
 - For backend/API changes, tests must pass before pushes
 - Add tests for new API endpoints and features
+- When writing tests, focus on decoupling each `it` or `test` block to ensure that they can run independently and concurrently. Tests must never depend on the action or outcome of previous or subsequent tests.
+- When writing tests, especially in the __tests__/async directory, ensure that you avoid `pruneDatabase` calls.
 
 ### Code Conventions
 - **Pages Router** (not App Router)
@@ -186,3 +189,6 @@ To get a project, use the `get_project` capability with the full project name as
 
 ## TypeScript Best Practices
 - In TypeScript, if possible, don't use the `any` type
+
+## General Coding Guidelines
+- For easier code reviews, prefer not to move functions etc around within a file unless necessary or instructed to do so

CONTRIBUTING.md

@@ -124,7 +124,14 @@ Requirements
    cd langfuse
    ```
 
-3. Create an env file
+3. Install dependencies and set up pre-commit hooks
+
+   ```bash
+   pnpm install
+   pnpm run prepare  # Sets up Husky pre-commit hooks for code formatting
+   ```
+
+4. Create an env file
 
    ```bash
     cp .env.dev.example .env
@@ -133,7 +140,7 @@ Requirements
 4. Run the entire infrastructure in dev mode. **Note**: if you have an existing database, this command wipes it.
 
    ```bash
-   pnpm run dx # first run only (resets db, node_modules, ...)
+   pnpm run dx # first run only (resets db, docker containers, etc...)
    pnpm run dev # any subsequent runs
    ```
 
@@ -149,8 +156,8 @@ Requirements
    - Username: `demo@langfuse.com`
    - Password: `password`
 
-
 To get comprehensive example data, you can use the `seed` command:
+
 ```sh
 pnpm run db:seed:examples
 ```
@@ -209,31 +216,55 @@ On the main branch, we adhere to the best practices of [conventional commits](ht
 All tests run in the CI and must pass before merging.
 All tests run against a running langfuse instance and **write/delete real data from the database**.
 
+### Test Database Setup
+
+Per default, the tests use the local development database. Therefore, wiping your data in the process.
+For proper test isolation, create a `.env.test` file in the root directory:
+
+```bash
+cp .env.test.example .env.test
+```
+
+Then, a different PostgreSQL and Redis are used for the tests.
+The `.env.test` file only overrides the set values and falls back on `.env` for all undefined values.
+
+- **PostgreSQL**: Uses separate `langfuse_test` database for isolation
+- **ClickHouse**: Uses shared `default` database for now
+- **Redis**: Uses database 1 instead of 0 for isolation (Redis data is not cleaned between tests)
+
+Tests automatically create the PostgreSQL test database if it doesn't exist and clean up data between runs.
+
 ### Tests in the `web` package (public API)
-We're using Jest with in the `web` package. Therefore, if you want to provide an argument to the test runner, do it directly without an intermittent ` -- `.
+
+We're using Jest with in the `web` package. Therefore, if you want to provide an argument to the test runner, do it directly without an intermittent `--`.
 
 There are three types of unit tests:
+
 - `test-sync`
 - `test-async`
 - `test-client`
 
 To run a specific test, for example the test: `"should handle special characters in prompt names"` in `prompts.v2.servertest.ts`, run:
+
 ```sh
 cd web  # or with --filter=web
 pnpm test-sync --testPathPattern="prompts\.v2\.servertest" --testNamePattern="should handle special characters in prompt names"
 ```
 
 To run all tests:
+
 ```sh
 pnpm run test
 ```
 
 Run interactively in watch mode (not recommended!)
+
 ```sh
 pnpm run test:watch
 ```
 
 ### Tests in the `worker` package
+
 For the `worker` package, we're using `vitest` to run unit tests.
 
 ```sh

docker-compose.build.yml

@@ -78,7 +78,7 @@ services:
       retries: 3
 
   clickhouse:
-    image: clickhouse/clickhouse-server
+    image: docker.io/clickhouse/clickhouse-server
     user: "101:101"
     environment:
       CLICKHOUSE_DB: default
@@ -98,7 +98,7 @@ services:
       start_period: 1s
 
   minio:
-    image: minio/minio
+    image: docker.io/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'
@@ -118,7 +118,7 @@ services:
       start_period: 1s
 
   redis:
-    image: redis:7
+    image: docker.io/redis:7
     restart: always
     command: >
       --requirepass ${REDIS_AUTH:-myredissecret}
@@ -131,7 +131,7 @@ services:
       retries: 10
 
   postgres:
-    image: postgres:${POSTGRES_VERSION:-latest}
+    image: docker.io/postgres:${POSTGRES_VERSION:-latest}
     restart: always
     healthcheck:
       test: ["CMD-SHELL", "pg_isready -U postgres"]

docker-compose.dev-azure.yml

@@ -1,6 +1,6 @@
 services:
   clickhouse:
-    image: clickhouse/clickhouse-server:24.3
+    image: docker.io/clickhouse/clickhouse-server:24.3
     user: "101:101"
     environment:
       CLICKHOUSE_DB: default
@@ -24,7 +24,7 @@ services:
       - langfuse_azurite_data:/data
 
   redis:
-    image: redis:7.2.4
+    image: docker.io/redis:7.2.4
     restart: always
     command: >
       --requirepass ${REDIS_AUTH:-myredissecret}
@@ -32,7 +32,7 @@ services:
       - 6379:6379
 
   postgres:
-    image: postgres:${POSTGRES_VERSION:-latest}
+    image: docker.io/postgres:${POSTGRES_VERSION:-latest}
     restart: always
     healthcheck:
       test: ["CMD-SHELL", "pg_isready -U postgres"]

docker-compose.dev-redis-cluster.yml

@@ -1,6 +1,6 @@
 services:
   clickhouse:
-    image: clickhouse/clickhouse-server:24.3
+    image: docker.io/clickhouse/clickhouse-server:24.3
     user: "101:101"
     environment:
       CLICKHOUSE_DB: default
@@ -16,7 +16,7 @@ services:
       - postgres
 
   minio:
-    image: minio/minio
+    image: docker.io/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'
@@ -36,7 +36,7 @@ services:
       start_period: 1s
 
   postgres:
-    image: postgres:${POSTGRES_VERSION:-latest}
+    image: docker.io/postgres:${POSTGRES_VERSION:-latest}
     restart: always
     healthcheck:
       test: ["CMD-SHELL", "pg_isready -U postgres"]

docker-compose.dev.yml

@@ -1,6 +1,6 @@
 services:
   clickhouse:
-    image: clickhouse/clickhouse-server:24.3
+    image: docker.io/clickhouse/clickhouse-server:24.3
     user: "101:101"
     environment:
       CLICKHOUSE_DB: default
@@ -16,7 +16,7 @@ services:
       - postgres
 
   minio:
-    image: minio/minio
+    image: docker.io/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'
@@ -36,7 +36,7 @@ services:
       start_period: 1s
 
   redis:
-    image: redis:7.2.4
+    image: docker.io/redis:7.2.4
     restart: always
     command: >
       --requirepass ${REDIS_AUTH:-myredissecret}
@@ -44,7 +44,7 @@ services:
       - 127.0.0.1:6379:6379
 
   postgres:
-    image: postgres:${POSTGRES_VERSION:-latest}
+    image: docker.io/postgres:${POSTGRES_VERSION:-latest}
     restart: always
     healthcheck:
       test: ["CMD-SHELL", "pg_isready -U postgres"]

docker-compose.yml

@@ -5,7 +5,7 @@
 # External connections from other machines will not be able to reach these services directly.
 services:
   langfuse-worker:
-    image: langfuse/langfuse-worker:3
+    image: docker.io/langfuse/langfuse-worker:3
     restart: always
     depends_on: &langfuse-depends-on
       postgres:
@@ -64,7 +64,7 @@ services:
       REDIS_TLS_KEY: ${REDIS_TLS_KEY:-/certs/redis.key}
 
   langfuse-web:
-    image: langfuse/langfuse:3
+    image: docker.io/langfuse/langfuse:3
     restart: always
     depends_on: *langfuse-depends-on
     ports:
@@ -84,7 +84,7 @@ services:
       LANGFUSE_INIT_USER_PASSWORD: ${LANGFUSE_INIT_USER_PASSWORD:-}
 
   clickhouse:
-    image: clickhouse/clickhouse-server
+    image: docker.io/clickhouse/clickhouse-server
     restart: always
     user: "101:101"
     environment:
@@ -105,7 +105,7 @@ services:
       start_period: 1s
 
   minio:
-    image: minio/minio
+    image: docker.io/minio/minio
     restart: always
     entrypoint: sh
     # create the 'langfuse' bucket before starting the service
@@ -126,7 +126,7 @@ services:
       start_period: 1s
 
   redis:
-    image: redis:7
+    image: docker.io/redis:7
     restart: always
     # CHANGEME: row below to secure redis password
     command: >
@@ -140,7 +140,7 @@ services:
       retries: 10
 
   postgres:
-    image: postgres:${POSTGRES_VERSION:-latest}
+    image: docker.io/postgres:${POSTGRES_VERSION:-latest}
     restart: always
     healthcheck:
       test: ["CMD-SHELL", "pg_isready -U postgres"]

ee/package.json

@@ -41,7 +41,7 @@
     "eslint-config-prettier": "^9.1.0",
     "eslint-config-standard": "^17.1.0",
     "eslint-plugin-prettier": "^5.1.3",
-    "prettier": "^3.3.3",
+    "prettier": "^3.6.2",
     "ts-node": "^10.9.2",
     "tsc-watch": "^6.2.0",
     "typescript": "^5.4.5"

fern/apis/server/definition/dataset-run-items.yml

@@ -27,7 +27,7 @@ service:
           limit:
             type: optional<integer>
             docs: limit of items per page
-          response: PaginatedDatasetRunItems
+      response: PaginatedDatasetRunItems
 
 types:
   CreateDatasetRunItemRequest:

package.json

@@ -1,6 +1,6 @@
 {
   "name": "langfuse",
-  "version": "3.78.0",
+  "version": "3.80.0",
   "author": "engineering@langfuse.com",
   "license": "MIT",
   "private": true,
@@ -17,9 +17,9 @@
     "db:seed": "turbo run db:seed",
     "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 && 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",
+    "dx": "pnpm i && pnpm run infra:dev:prune && pnpm run infra:dev:up --pull always && pnpm --filter=shared run db:reset:test && 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:test && 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:test && 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",
     "dev": "turbo run dev",
@@ -27,6 +27,8 @@
     "dev:web": "turbo run dev --filter=web",
     "dev:web-turbo": "turbo run dev --filter=web -- --turbo",
     "lint": "turbo run lint",
+    "format": "prettier --write \"**/*.{js,jsx,ts,tsx,css}\" --experimental-cli",
+    "format:check": "prettier --check \"**/*.{js,jsx,ts,tsx,css}\" --experimental-cli",
     "test": "turbo run test",
     "release": "dotenv -e ../.env -- release-it",
     "prepare": "husky"
@@ -36,7 +38,7 @@
     "braces": "3.0.3",
     "dotenv-cli": "^7.4.2",
     "husky": "^9.0.11",
-    "prettier": "^3.3.3",
+    "prettier": "^3.6.2",
     "release-it": "^19.0.3",
     "turbo": "^2.5.4"
   },

packages/config-eslint/library.js

@@ -35,5 +35,13 @@ module.exports = {
     {
       files: ["*.js?(x)", "*.ts?(x)"],
     },
+    {
+      files: ["*.ts", "*.mts", "*.cts", "*.tsx"],
+      // no-undef doesn't make sense in TS, see:
+      // https://typescript-eslint.io/troubleshooting/faqs/eslint/#i-get-errors-from-the-no-undef-rule-about-global-variables-not-being-defined-even-though-there-are-no-typescript-errors
+      rules: {
+        "no-undef": "off",
+      },
+    },
   ],
 };

packages/shared/package.json

@@ -33,11 +33,12 @@
   "scripts": {
     "build": "tsc",
     "dev": "tsc --watch",
-    "lint": "eslint . --ext .js,.jsx,.ts,.tsx --max-warnings 72",
+    "lint": "eslint . --ext .js,.jsx,.ts,.tsx --max-warnings 0",
     "lint:fix": "eslint . --ext .js,.jsx,.ts,.tsx --fix",
     "db:migrate": "DISABLE_ERD=false dotenv -e ../../.env -- npx prisma migrate dev",
     "db:push": "DISABLE_ERD=false dotenv -e ../../.env -- npx prisma db push",
     "db:reset": "dotenv -e ../../.env npx -- prisma migrate reset",
+    "db:reset:test": "if [ -f ../../.env.test ]; then dotenv -e ../../.env.test -e ../../.env -- npx prisma migrate reset --force; fi",
     "db:deploy": "dotenv -e ../../.env npx -- prisma migrate deploy",
     "db:seed": "dotenv -e ../../.env -- npx prisma db seed",
     "db:generate": "dotenv -e ../../.env -- npx prisma generate",
@@ -84,7 +85,7 @@
     "jsonpath-plus": "10.3.0",
     "kysely": "^0.27.4",
     "langchain": "^0.3.28",
-    "langfuse-langchain": "3.37.4",
+    "langfuse-langchain": "3.38.4",
     "lodash": "^4.17.21",
     "lossless-json": "^4.1.1",
     "next-auth": "^4.24.11",
@@ -111,7 +112,7 @@
     "eslint-plugin-prettier": "^5.1.3",
     "kysely-codegen": "^0.16.8",
     "nodemon": "^3.1.7",
-    "prettier": "^3.3.3",
+    "prettier": "^3.6.2",
     "prisma": "^6.10.1",
     "prisma-erd-generator": "^1.11.2",
     "prisma-kysely": "^1.8.0",

packages/shared/prisma/migrations/20250704170658_add_automations/migration.sql

@@ -0,0 +1,111 @@
+-- CreateEnum
+CREATE TYPE "ActionType" AS ENUM ('WEBHOOK');
+
+-- CreateEnum
+CREATE TYPE "ActionExecutionStatus" AS ENUM ('COMPLETED', 'ERROR', 'PENDING', 'CANCELLED');
+
+-- CreateTable
+CREATE TABLE "actions" (
+    "id" TEXT NOT NULL,
+    "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "project_id" TEXT NOT NULL,
+    "type" "ActionType" NOT NULL,
+    "config" JSONB NOT NULL,
+
+    CONSTRAINT "actions_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "triggers" (
+    "id" TEXT NOT NULL,
+    "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "project_id" TEXT NOT NULL,
+    "eventSource" TEXT NOT NULL,
+    "eventActions" TEXT[],
+    "filter" JSONB,
+    "status" "JobConfigState" NOT NULL DEFAULT 'ACTIVE',
+
+    CONSTRAINT "triggers_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "automations" (
+    "id" TEXT NOT NULL,
+    "name" TEXT NOT NULL,
+    "trigger_id" TEXT NOT NULL,
+    "action_id" TEXT NOT NULL,
+    "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "project_id" TEXT NOT NULL,
+
+    CONSTRAINT "automations_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "automation_executions" (
+    "id" TEXT NOT NULL,
+    "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "source_id" TEXT NOT NULL,
+    "automation_id" TEXT NOT NULL,
+    "trigger_id" TEXT NOT NULL,
+    "action_id" TEXT NOT NULL,
+    "project_id" TEXT NOT NULL,
+    "status" "ActionExecutionStatus" NOT NULL DEFAULT 'PENDING',
+    "input" JSONB NOT NULL,
+    "output" JSONB,
+    "started_at" TIMESTAMP(3),
+    "finished_at" TIMESTAMP(3),
+    "error" TEXT,
+
+    CONSTRAINT "automation_executions_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE INDEX "actions_project_id_idx" ON "actions"("project_id");
+
+-- CreateIndex
+CREATE INDEX "triggers_project_id_idx" ON "triggers"("project_id");
+
+-- CreateIndex
+CREATE INDEX "automations_project_id_action_id_trigger_id_idx" ON "automations"("project_id", "action_id", "trigger_id");
+
+-- CreateIndex
+CREATE INDEX "automations_project_id_name_idx" ON "automations"("project_id", "name");
+
+-- CreateIndex
+CREATE INDEX "automation_executions_trigger_id_idx" ON "automation_executions"("trigger_id");
+
+-- CreateIndex
+CREATE INDEX "automation_executions_action_id_idx" ON "automation_executions"("action_id");
+
+-- CreateIndex
+CREATE INDEX "automation_executions_project_id_idx" ON "automation_executions"("project_id");
+
+-- AddForeignKey
+ALTER TABLE "actions" ADD CONSTRAINT "actions_project_id_fkey" FOREIGN KEY ("project_id") REFERENCES "projects"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "triggers" ADD CONSTRAINT "triggers_project_id_fkey" FOREIGN KEY ("project_id") REFERENCES "projects"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "automations" ADD CONSTRAINT "automations_trigger_id_fkey" FOREIGN KEY ("trigger_id") REFERENCES "triggers"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "automations" ADD CONSTRAINT "automations_action_id_fkey" FOREIGN KEY ("action_id") REFERENCES "actions"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "automations" ADD CONSTRAINT "automations_project_id_fkey" FOREIGN KEY ("project_id") REFERENCES "projects"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "automation_executions" ADD CONSTRAINT "automation_executions_automation_id_fkey" FOREIGN KEY ("automation_id") REFERENCES "automations"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "automation_executions" ADD CONSTRAINT "automation_executions_trigger_id_fkey" FOREIGN KEY ("trigger_id") REFERENCES "triggers"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "automation_executions" ADD CONSTRAINT "automation_executions_action_id_fkey" FOREIGN KEY ("action_id") REFERENCES "actions"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "automation_executions" ADD CONSTRAINT "automation_executions_project_id_fkey" FOREIGN KEY ("project_id") REFERENCES "projects"("id") ON DELETE CASCADE ON UPDATE CASCADE;

packages/shared/prisma/migrations/20250709113103_add_blob_export_schedule_type/migration.sql

@@ -0,0 +1,6 @@
+-- CreateEnum
+CREATE TYPE "BlobStorageExportMode" AS ENUM ('FULL_HISTORY', 'FROM_TODAY', 'FROM_CUSTOM_DATE');
+
+-- AlterTable
+ALTER TABLE "blob_storage_integrations" ADD COLUMN     "export_mode" "BlobStorageExportMode" NOT NULL DEFAULT 'FULL_HISTORY',
+ADD COLUMN     "export_start_date" TIMESTAMP(3);

packages/shared/prisma/migrations/migration_lock.toml

@@ -1,3 +1,3 @@
 # Please do not edit this file manually
 # It should be added in your version-control system (e.g., Git)
-provider = "postgresql"
+provider = "postgresql"

packages/shared/prisma/schema.prisma

@@ -157,6 +157,10 @@ model Project {
   Dashboard              Dashboard[]
   DashboardWidget        DashboardWidget[]
   TableViewPreset        TableViewPreset[]
+  actions                Action[]
+  triggers               Trigger[]
+  automationExecutions   AutomationExecution[]
+  Automation             Automation[]
   DefaultLlmModel        DefaultLlmModel[]
 
   @@index([orgId])
@@ -963,6 +967,8 @@ model BlobStorageIntegration {
   enabled         Boolean
   exportFrequency String                         @map("export_frequency")
   fileType        BlobStorageIntegrationFileType @default(CSV) @map("file_type")
+  exportMode      BlobStorageExportMode          @default(FULL_HISTORY) @map("export_mode")
+  exportStartDate DateTime?                      @map("export_start_date")
 
   createdAt DateTime @default(now()) @map("created_at")
   updatedAt DateTime @default(now()) @updatedAt @map("updated_at")
@@ -986,6 +992,14 @@ enum BlobStorageIntegrationType {
   @@map("BlobStorageIntegrationType")
 }
 
+enum BlobStorageExportMode {
+  FULL_HISTORY
+  FROM_TODAY
+  FROM_CUSTOM_DATE
+
+  @@map("BlobStorageExportMode")
+}
+
 model BatchExport {
   id        String   @id @default(cuid())
   createdAt DateTime @default(now()) @map("created_at")
@@ -1214,3 +1228,109 @@ model TableViewPreset {
   @@unique([projectId, tableName, name])
   @@map("table_view_presets")
 }
+
+model Action {
+  id        String   @id @default(cuid())
+  createdAt DateTime @default(now()) @map("created_at")
+  updatedAt DateTime @default(now()) @updatedAt @map("updated_at")
+
+  projectId String  @map("project_id")
+  project   Project @relation(fields: [projectId], references: [id], onDelete: Cascade)
+
+  type ActionType
+
+  // Configuration specific to each action type
+  config Json // Structured JSON for different action types
+  // For WEBHOOK: { version: "1.0", url: "...", method: "POST", headers: {...}, secretId: "..." }
+  // For ANNOTATION_QUEUE: { version: "1.0", queueId: "..." }
+
+  automations          Automation[]
+  automationExecutions AutomationExecution[]
+
+  @@index([projectId])
+  @@map("actions")
+}
+
+model Trigger {
+  id        String   @id @default(cuid())
+  createdAt DateTime @default(now()) @map("created_at")
+  updatedAt DateTime @default(now()) @updatedAt @map("updated_at")
+
+  projectId String  @map("project_id")
+  project   Project @relation(fields: [projectId], references: [id], onDelete: Cascade)
+
+  // When should this trigger fire
+  eventSource  String // trace, prompt, etc.
+  eventActions String[] // created, updated, deleted
+  filter       Json? // Filter conditions (format: { field: "name", operator: "equals", value: "my_trace" })
+
+  // Additional attributes
+  status JobConfigState @default(ACTIVE) @map("status")
+
+  // Link to executions
+  automationExecutions AutomationExecution[]
+  automations          Automation[]
+
+  @@index([projectId])
+  @@map("triggers")
+}
+
+model Automation {
+  id                  String                @id @default(cuid())
+  name                String                @map("name")
+  trigger             Trigger               @relation(fields: [triggerId], references: [id], onDelete: Cascade)
+  triggerId           String                @map("trigger_id")
+  action              Action                @relation(fields: [actionId], references: [id], onDelete: Cascade)
+  actionId            String                @map("action_id")
+  createdAt           DateTime              @default(now()) @map("created_at")
+  projectId           String                @map("project_id")
+  project             Project               @relation(fields: [projectId], references: [id], onDelete: Cascade)
+  AutomationExecution AutomationExecution[]
+
+  @@index([projectId, actionId, triggerId])
+  @@index([projectId, name])
+  @@map("automations")
+}
+
+enum ActionType {
+  WEBHOOK
+  // More action types can be added as needed
+}
+
+enum ActionExecutionStatus {
+  COMPLETED
+  ERROR
+  PENDING
+  CANCELLED
+}
+
+model AutomationExecution {
+  id        String   @id @default(cuid())
+  createdAt DateTime @default(now()) @map("created_at")
+  updatedAt DateTime @default(now()) @updatedAt @map("updated_at")
+  sourceId  String   @map("source_id")
+
+  automationId String     @map("automation_id")
+  automation   Automation @relation(fields: [automationId], references: [id], onDelete: Cascade)
+
+  triggerId String  @map("trigger_id")
+  trigger   Trigger @relation(fields: [triggerId], references: [id], onDelete: Cascade)
+
+  actionId String @map("action_id")
+  action   Action @relation(fields: [actionId], references: [id], onDelete: Cascade)
+
+  projectId String  @map("project_id")
+  project   Project @relation(fields: [projectId], references: [id], onDelete: Cascade)
+
+  status     ActionExecutionStatus @default(PENDING) @map("status")
+  input      Json                  @map("input")
+  output     Json?                 @map("output")
+  startedAt  DateTime?             @map("started_at")
+  finishedAt DateTime?             @map("finished_at")
+  error      String?               @map("error")
+
+  @@index([triggerId])
+  @@index([actionId])
+  @@index([projectId])
+  @@map("automation_executions")
+}

packages/shared/scripts/seeder/seed-postgres.ts

@@ -4,15 +4,16 @@ import { hash } from "bcryptjs";
 import { v4 } from "uuid";
 import { encrypt } from "../../src/encryption";
 import {
-	type JobConfiguration,
-	JobExecutionStatus,
-	PrismaClient,
-	type Project,
-	ScoreDataType,
+  type JobConfiguration,
+  JobExecutionStatus,
+  PrismaClient,
+  type Project,
+  ScoreDataType,
 } from "../../src/index";
 import { getDisplaySecretKey, hashSecretKey, logger } from "../../src/server";
 import { redis } from "../../src/server/redis/redis";
-import {EVAL_TRACE_COUNT,
+import {
+  EVAL_TRACE_COUNT,
   FAILED_EVAL_TRACE_INTERVAL,
   SEED_CHAT_ML_PROMPTS,
   SEED_DATASETS,
@@ -178,7 +179,7 @@ async function main() {
 
   const seedApiKey = {
     id: "seed-api-key",
-    secret: process.env.SEED_SECRET_KEY ?? "sk-lf-1234567890",
+    secret: process.env.SEED_SECRET_KEY ?? "sk-lf-1234567890", // eslint-disable-line turbo/no-undeclared-env-vars
     public: "pk-lf-1234567890",
     note: "seeded key",
   };
@@ -241,7 +242,7 @@ async function main() {
 
     const secondKey = {
       id: "seed-api-key-2",
-      secret: process.env.SEED_SECRET_KEY ?? "sk-lf-asdfghjkl",
+      secret: process.env.SEED_SECRET_KEY ?? "sk-lf-asdfghjkl", // eslint-disable-line turbo/no-undeclared-env-vars
       public: "pk-lf-asdfghjkl",
       note: "seeded key 2",
     };
@@ -274,7 +275,7 @@ async function main() {
     await createTraceSessions(project1, project2);
 
     // If openai key is in environment, add it to the projects LLM API keys
-    const OPENAI_API_KEY = process.env.OPENAI_API_KEY;
+    const OPENAI_API_KEY = process.env.OPENAI_API_KEY; // eslint-disable-line turbo/no-undeclared-env-vars
 
     if (OPENAI_API_KEY) {
       await prisma.llmApiKeys.create({

packages/shared/scripts/seeder/utils/postgres-seed-constants.ts

@@ -271,6 +271,68 @@ export const SEED_TEXT_PROMPTS = [
     labels: ["production", "latest"],
     tags: ["tag1", "tag2"],
   },
+  {
+    id: `prompt-with-many-labels`,
+    createdBy: "user-1",
+    prompt:
+      "This is a comprehensive prompt for testing multiple label scenarios. It demonstrates how prompts can be tagged with numerous labels for organization, categorization, and filtering purposes. Use this prompt to understand how label management works at scale. Variables: {{input}}",
+    name: "prompt-with-many-labels",
+    version: 1,
+    labels: [
+      "production",
+      "latest",
+      "v1",
+      "v2",
+      "stable",
+      "beta",
+      "alpha",
+      "test",
+      "development",
+      "staging",
+      "experimental",
+      "feature",
+      "bugfix",
+      "hotfix",
+      "critical",
+      "high-priority",
+      "medium-priority",
+      "low-priority",
+      "urgent",
+      "customer-facing",
+      "internal",
+      "public",
+      "private",
+      "confidential",
+      "ai",
+      "nlp",
+      "chatbot",
+      "assistant",
+      "automation",
+      "ml",
+      "data",
+      "analytics",
+      "monitoring",
+      "logging",
+      "debug",
+      "performance",
+      "security",
+      "compliance",
+      "audit",
+      "review",
+      "approved",
+      "rejected",
+      "pending",
+      "archived",
+      "deprecated",
+      "legacy",
+      "migration",
+      "upgrade",
+      "downgrade",
+      "template",
+      "example",
+    ],
+    tags: [],
+  },
 ];
 
 export const SEED_CHAT_ML_PROMPTS = [

packages/shared/src/db.ts

@@ -88,6 +88,7 @@ declare const globalThis: {
   kyselyPrismaGlobal: { $kysely: Kysely<DB> } | undefined;
 } & typeof global;
 
+// eslint-disable-next-line turbo/no-undeclared-env-vars
 if (process.env.NODE_ENV === "development") {
   globalThis.prismaGlobal ??= createPrismaInstance(); // regular instantiation
   globalThis.kyselyPrismaGlobal ??= globalThis.prismaGlobal.$extends(

packages/shared/src/domain/automations.ts

@@ -0,0 +1,83 @@
+import { Action, Trigger } from "@prisma/client";
+import { FilterState } from "../types";
+import { z } from "zod/v4";
+
+export enum TriggerEventSource {
+  // eslint-disable-next-line no-unused-vars
+  Prompt = "prompt",
+}
+
+export const EventActionSchema = z.enum(["created", "updated", "deleted"]);
+
+export type TriggerEventAction = z.infer<typeof EventActionSchema>;
+
+export const TriggerEventSourceSchema = z.enum([TriggerEventSource.Prompt]);
+
+export type TriggerDomain = Omit<
+  Trigger,
+  "filter" | "eventSource" | "eventActions"
+> & {
+  filter: FilterState;
+  eventSource: TriggerEventSource;
+  eventActions: TriggerEventAction[];
+};
+
+export type AutomationDomain = {
+  id: string;
+  name: string;
+  trigger: TriggerDomain;
+  action: ActionDomain;
+};
+
+export type ActionDomain = Omit<Action, "config"> & {
+  config: SafeWebhookActionConfig;
+};
+
+export type ActionDomainWithSecrets = Omit<Action, "config"> & {
+  config: WebhookActionConfigWithSecrets;
+};
+
+export const ActionTypeSchema = z.enum(["WEBHOOK"]);
+
+export const AvailableWebhookApiSchema = z.record(
+  z.enum(["prompt"]),
+  z.enum(["v1"]),
+);
+
+export const WebhookActionConfigSchema = z.object({
+  type: z.literal("WEBHOOK"),
+  url: z.url(),
+  headers: z.record(z.string(), z.string()),
+  apiVersion: AvailableWebhookApiSchema,
+  secretKey: z.string(),
+  displaySecretKey: z.string(),
+});
+
+export const SafeWebhookActionConfigSchema = WebhookActionConfigSchema.omit({
+  secretKey: true,
+});
+
+export type SafeWebhookActionConfig = z.infer<
+  typeof SafeWebhookActionConfigSchema
+>;
+
+export const WebhookActionCreateSchema = WebhookActionConfigSchema.omit({
+  secretKey: true,
+  displaySecretKey: true,
+});
+
+export const ActionConfigSchema = z.discriminatedUnion("type", [
+  WebhookActionConfigSchema,
+]);
+
+export const ActionCreateSchema = z.discriminatedUnion("type", [
+  WebhookActionCreateSchema,
+]);
+
+export type ActionTypes = z.infer<typeof ActionTypeSchema>;
+export type ActionConfig = z.infer<typeof ActionConfigSchema>;
+export type ActionCreate = z.infer<typeof ActionCreateSchema>;
+
+export type WebhookActionConfigWithSecrets = z.infer<
+  typeof WebhookActionConfigSchema
+>;

packages/shared/src/domain/index.ts

@@ -2,3 +2,6 @@ export * from "./observations";
 export * from "./traces";
 export * from "./scores";
 export * from "./table-view-presets";
+export * from "./automations";
+export * from "./webhooks";
+export * from "./prompts";

packages/shared/src/domain/prompts.ts

@@ -0,0 +1,21 @@
+import { z } from "zod/v4";
+import { jsonSchemaNullable } from "../utils/zod";
+
+export const PromptDomainSchema = z.object({
+  id: z.string(),
+  name: z.string(),
+  version: z.number(),
+  createdAt: z.date(),
+  updatedAt: z.date(),
+  createdBy: z.string(),
+  isActive: z.boolean().nullable(),
+  type: z.string().default("text"),
+  tags: z.array(z.string()).default([]),
+  labels: z.array(z.string()).default([]),
+  prompt: jsonSchemaNullable,
+  config: jsonSchemaNullable,
+  projectId: z.string(),
+  commitMessage: z.string().nullable(),
+});
+
+export type PromptDomain = z.infer<typeof PromptDomainSchema>;

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

@@ -3,11 +3,11 @@ import { orderBy } from "../interfaces/orderBy";
 import z from "zod/v4";
 
 export enum TableViewPresetTableName {
-  Traces = "traces",
-  Observations = "observations",
-  Scores = "scores",
-  Sessions = "sessions",
-  Datasets = "datasets",
+  Traces = "traces", // eslint-disable-line no-unused-vars
+  Observations = "observations", // eslint-disable-line no-unused-vars
+  Scores = "scores", // eslint-disable-line no-unused-vars
+  Sessions = "sessions", // eslint-disable-line no-unused-vars
+  Datasets = "datasets", // eslint-disable-line no-unused-vars
 }
 
 const TableViewPresetDomainSchema = z.object({

packages/shared/src/domain/webhooks.ts

@@ -0,0 +1,37 @@
+import { z } from "zod/v4";
+import { jsonSchema } from "../utils/zod";
+import { EventActionSchema } from "./automations";
+
+export const WebhookDefaultHeaders = {
+  "Content-Type": "application/json",
+  "User-Agent": "Langfuse/1.0",
+};
+
+export const WebhookOutboundBaseSchema = z.object({
+  id: z.string(),
+  timestamp: z.coerce.date(),
+  type: z.literal("prompt-version"),
+  apiVersion: z.literal("v1"),
+  action: EventActionSchema,
+});
+
+export const PromptWebhookOutboundSchema = z
+  .object({
+    prompt: z.object({
+      id: z.string(),
+      name: z.string(),
+      version: z.number(),
+      projectId: z.string(),
+      labels: z.array(z.string()),
+      prompt: jsonSchema.nullable(),
+      type: z.string(),
+      config: z.record(z.string(), z.any()),
+      commitMessage: z.string().nullable(),
+      tags: z.array(z.string()),
+      createdAt: z.coerce.date(),
+      updatedAt: z.coerce.date(),
+    }),
+  })
+  .and(WebhookOutboundBaseSchema);
+
+export type PromptWebhookOutput = z.infer<typeof PromptWebhookOutboundSchema>;

packages/shared/src/encryption/encryption.ts

@@ -0,0 +1,60 @@
+import crypto from "crypto";
+import { env } from "../env";
+
+const ENCRYPTION_KEY: string | undefined = env.ENCRYPTION_KEY; // Must be 256 bits (32 bytes, 64 hex characters)
+const IV_LENGTH: number = 12; // For AES-GCM, this is always 12
+
+// Alternatively: openssl rand -hex 32
+export function keyGen() {
+  return crypto.randomBytes(32).toString("hex");
+}
+
+/**
+ * Encrypts the given plain text using AES-256-GCM algorithm.
+ *
+ * @param {string} plainText - The text to encrypt.
+ * @returns {string} The encrypted data in hex format, including IV and authentication tag.
+ */
+export function encrypt(plainText: string): string {
+  if (!ENCRYPTION_KEY) {
+    throw new Error("Missing environment variable: `ENCRYPTION_KEY`");
+  }
+  const iv = crypto.randomBytes(IV_LENGTH); // Directly use Buffer returned by randomBytes
+  const cipher = crypto.createCipheriv(
+    "aes-256-gcm",
+    Buffer.from(ENCRYPTION_KEY, "hex"),
+    iv,
+  );
+  let encrypted = cipher.update(plainText, "utf8", "hex");
+  encrypted += cipher.final("hex");
+  const authTag = cipher.getAuthTag();
+
+  // Return iv, encrypted data, and authTag as hex, combined in one line
+  return iv.toString("hex") + ":" + encrypted + ":" + authTag.toString("hex");
+}
+
+export function decrypt(text: string): string {
+  if (!ENCRYPTION_KEY) {
+    throw new Error("Missing environment variable: `ENCRYPTION_KEY`");
+  }
+  const [ivHex, encryptedHex, authTagHex] = text.split(":");
+  if (!ivHex || !encryptedHex || !authTagHex) {
+    throw new Error("Invalid or corrupted cipher format");
+  }
+
+  const iv = Buffer.from(ivHex, "hex");
+  const encryptedText = Buffer.from(encryptedHex, "hex");
+  const authTag = Buffer.from(authTagHex, "hex");
+
+  const decipher = crypto.createDecipheriv(
+    "aes-256-gcm",
+    Buffer.from(ENCRYPTION_KEY, "hex"),
+    iv,
+  );
+  decipher.setAuthTag(authTag);
+
+  let decrypted = decipher.update(encryptedText, undefined, "utf8");
+  decrypted += decipher.final("utf8");
+
+  return decrypted.toString();
+}

packages/shared/src/encryption/index.ts

@@ -1,60 +1,2 @@
-import crypto from "crypto";
-import { env } from "../env";
-
-const ENCRYPTION_KEY: string | undefined = env.ENCRYPTION_KEY; // Must be 256 bits (32 bytes, 64 hex characters)
-const IV_LENGTH: number = 12; // For AES-GCM, this is always 12
-
-// Alternatively: openssl rand -hex 32
-export function keyGen() {
-  return crypto.randomBytes(32).toString("hex");
-}
-
-/**
- * Encrypts the given plain text using AES-256-GCM algorithm.
- *
- * @param {string} plainText - The text to encrypt.
- * @returns {string} The encrypted data in hex format, including IV and authentication tag.
- */
-export function encrypt(plainText: string): string {
-  if (!ENCRYPTION_KEY) {
-    throw new Error("Missing environment variable: `ENCRYPTION_KEY`");
-  }
-  const iv = crypto.randomBytes(IV_LENGTH); // Directly use Buffer returned by randomBytes
-  const cipher = crypto.createCipheriv(
-    "aes-256-gcm",
-    Buffer.from(ENCRYPTION_KEY, "hex"),
-    iv,
-  );
-  let encrypted = cipher.update(plainText, "utf8", "hex");
-  encrypted += cipher.final("hex");
-  const authTag = cipher.getAuthTag();
-
-  // Return iv, encrypted data, and authTag as hex, combined in one line
-  return iv.toString("hex") + ":" + encrypted + ":" + authTag.toString("hex");
-}
-
-export function decrypt(text: string): string {
-  if (!ENCRYPTION_KEY) {
-    throw new Error("Missing environment variable: `ENCRYPTION_KEY`");
-  }
-  const [ivHex, encryptedHex, authTagHex] = text.split(":");
-  if (!ivHex || !encryptedHex || !authTagHex) {
-    throw new Error("Invalid or corrupted cipher format");
-  }
-
-  const iv = Buffer.from(ivHex, "hex");
-  const encryptedText = Buffer.from(encryptedHex, "hex");
-  const authTag = Buffer.from(authTagHex, "hex");
-
-  const decipher = crypto.createDecipheriv(
-    "aes-256-gcm",
-    Buffer.from(ENCRYPTION_KEY, "hex"),
-    iv,
-  );
-  decipher.setAuthTag(authTag);
-
-  let decrypted = decipher.update(encryptedText, undefined, "utf8");
-  decrypted += decipher.final("utf8");
-
-  return decrypted.toString();
-}
+export * from "./encryption";
+export * from "./signature";

packages/shared/src/encryption/signature.ts

@@ -0,0 +1,41 @@
+import crypto from "crypto";
+
+// Generate a webhook secret key
+export function generateWebhookSecret(): {
+  secretKey: string;
+  displaySecretKey: string;
+} {
+  // Generate 32 random bytes and encode as hex (64 characters)
+  const rawSecret = crypto.randomBytes(32).toString("hex");
+  const secretKey = `lf-whsec_${rawSecret}`;
+  return { secretKey, displaySecretKey: getDisplaySecretKey(secretKey) };
+}
+
+// Create display version of webhook secret
+export function getDisplaySecretKey(secretKey: string): string {
+  if (!secretKey || secretKey.length < 12) {
+    // whsec_ + at least 4 chars
+    return "****";
+  }
+
+  return `lf-whsec_...${secretKey.slice(-4)}`;
+}
+
+// Generate HMAC-SHA256 signature for webhook payload
+export function generateWebhookSignature(
+  payload: string,
+  timestamp: number,
+  secret: string,
+) {
+  const signedPayload = `${timestamp}.${payload}`;
+  return crypto
+    .createHmac("sha256", secret)
+    .update(signedPayload, "utf8")
+    .digest("hex");
+}
+
+export function createSignatureHeader(payload: string, secret: string): string {
+  const timestamp = Math.floor(Date.now() / 1000);
+  const signature = generateWebhookSignature(payload, timestamp, secret);
+  return `t=${timestamp},v1=${signature}`;
+}

packages/shared/src/env.ts

@@ -105,10 +105,35 @@ const EnvSchema = z.object({
     .number()
     .default(80e6), // 80MB
   LANGFUSE_CLICKHOUSE_DELETION_TIMEOUT_MS: z.coerce.number().default(240_000), // 4 minutes
+  LANGFUSE_CLICKHOUSE_QUERY_MAX_ATTEMPTS: z.coerce.number().default(3), // Maximum attempts for socket hang up errors
   LANGFUSE_SKIP_S3_LIST_FOR_OBSERVATIONS_PROJECT_IDS: z.string().optional(),
+
+  LANGFUSE_EXPERIMENT_COMPARE_READ_FROM_AGGREGATING_MERGE_TREES: z
+    .enum(["true", "false"])
+    .default("false"),
+  LANGFUSE_EXPERIMENT_ADD_QUERY_RESULT_TO_SPAN_PROJECT_IDS: z
+    .string()
+    .optional()
+    .transform((s) =>
+      s ? s.split(",").map((s) => s.toLowerCase().trim()) : [],
+    ),
+  LANGFUSE_EXPERIMENT_SAMPLING_RATE: z.coerce
+    .number()
+    .min(0)
+    .max(1)
+    .default(0.1),
+  LANGFUSE_EXPERIMENT_WHITELISTED_PROJECT_IDS: z
+    .string()
+    .optional()
+    .transform((s) =>
+      s ? s.split(",").map((s) => s.toLowerCase().trim()) : [],
+    ),
+  LANGFUSE_EXPERIMENT_RETURN_NEW_RESULT: z
+    .enum(["true", "false"])
+    .default("false"),
 });
 
 export const env: z.infer<typeof EnvSchema> =
-  process.env.DOCKER_BUILD === "1"
+  process.env.DOCKER_BUILD === "1" // eslint-disable-line turbo/no-undeclared-env-vars
     ? (process.env as any)
     : EnvSchema.parse(removeEmptyEnvVariables(process.env));

packages/shared/src/errors/BaseError.ts

@@ -7,7 +7,7 @@ export class BaseError extends Error {
     name: string,
     httpCode: number,
     description: string,
-    isOperational: boolean
+    isOperational: boolean,
   ) {
     super(description);
     Object.setPrototypeOf(this, new.target.prototype); // restore prototype chain

packages/shared/src/errors/index.ts

@@ -7,3 +7,4 @@ export { MethodNotAllowedError } from "./MethodNotAllowedError";
 export { ApiError } from "./ApiError";
 export { InternalServerError } from "./InternalServerError";
 export { LangfuseConflictError } from "./ConflictError";
+export { QUEUE_ERROR_MESSAGES } from "./utils/constants";

packages/shared/src/errors/utils/constants.ts

@@ -0,0 +1,10 @@
+export const QUEUE_ERROR_MESSAGES = {
+  API_KEY_ERROR: "API key for provider",
+  NO_DEFAULT_MODEL_ERROR: "No default model or custom model found for project",
+  MAPPED_DATA_ERROR:
+    "Please ensure the mapped data exists and consider extending the job delay.",
+  INVALID_JSON_ERROR: "is not valid JSON",
+  TOO_LOW_MAX_TOKENS_ERROR: "Error: Unterminated string in JSON at position",
+  OUTPUT_TOKENS_TOO_LONG_ERROR:
+    "Could not parse response content as the length limit was reached",
+};

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

@@ -8,16 +8,16 @@ import { BatchTableNames } from "../../interfaces/tableNames";
 import { TracingSearchType } from "../../interfaces/search";
 
 export enum BatchExportStatus {
-  QUEUED = "QUEUED",
-  PROCESSING = "PROCESSING",
-  COMPLETED = "COMPLETED",
-  FAILED = "FAILED",
+  QUEUED = "QUEUED", // eslint-disable-line no-unused-vars
+  PROCESSING = "PROCESSING", // eslint-disable-line no-unused-vars
+  COMPLETED = "COMPLETED", // eslint-disable-line no-unused-vars
+  FAILED = "FAILED", // eslint-disable-line no-unused-vars
 }
 
 export enum BatchExportFileFormat {
-  JSON = "JSON",
-  CSV = "CSV",
-  JSONL = "JSONL",
+  JSON = "JSON", // eslint-disable-line no-unused-vars
+  CSV = "CSV", // eslint-disable-line no-unused-vars
+  JSONL = "JSONL", // eslint-disable-line no-unused-vars
 }
 
 // Use shared BatchTableNames enum for consistency across batch operations

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

@@ -42,7 +42,7 @@ function parseJsonDefault(selectedColumn: unknown, jsonSelector: string) {
 export function extractValueFromObject(
   obj: Record<string, unknown>,
   mapping: z.infer<typeof variableMapping>,
-  parseJson?: (selectedColumn: unknown, jsonSelector: string) => unknown,
+  parseJson?: (selectedColumn: unknown, jsonSelector: string) => unknown, // eslint-disable-line no-unused-vars
 ): { value: string; error: Error | null } {
   const selectedColumn = obj[mapping.selectedColumnId];
   const jsonParser = parseJson || parseJsonDefault;

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

@@ -37,13 +37,13 @@ const ScorePropsAgainstConfigNumeric = z
     dataType: z.literal("NUMERIC"),
   })
   .superRefine((data, ctx) => {
-    if (isPresent(data.maxValue) && data.value >= data.maxValue) {
+    if (isPresent(data.maxValue) && data.value > data.maxValue) {
       ctx.addIssue({
         code: z.ZodIssueCode.custom,
         message: `Value exceeds maximum value of ${data.maxValue} defined in config`,
       });
     }
-    if (isPresent(data.minValue) && data.value <= data.minValue) {
+    if (isPresent(data.minValue) && data.value < data.minValue) {
       ctx.addIssue({
         code: z.ZodIssueCode.custom,
         message: `Value is below minimum value of ${data.minValue} defined in config`,

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

@@ -150,7 +150,7 @@ const ValidatedScoreConfigSchema = z
  */
 export const filterAndValidateDbScoreConfigList = (
   scoreConfigs: ScoreConfigDbType[],
-  onParseError?: (error: z.ZodError) => void,
+  onParseError?: (error: z.ZodError) => void, // eslint-disable-line no-unused-vars
 ): ValidatedScoreConfig[] =>
   scoreConfigs.reduce((acc, ts) => {
     const result = ValidatedScoreConfigSchema.safeParse(ts);

packages/shared/src/index.ts

@@ -13,9 +13,11 @@ export * from "./utils/json";
 export * from "./utils/stringChecks";
 export * from "./utils/objects";
 export * from "./utils/typeChecks";
+export * from "./utils/prompts";
 export * from "./features/entitlements/plans";
 export * from "./interfaces/rate-limits";
 export * from "./tableDefinitions/typeHelpers";
+export * from "./domain/webhooks";
 
 // llm api
 export * from "./server/llm/types";
@@ -45,7 +47,13 @@ export * from "./features/prompts/parsePromptDependencyTags";
 export * from "./features/prompts/validation";
 export * from "./features/prompts/types";
 export * from "./features/prompts/constants";
-export * from "./server/llm/compileChatMessages";
+export {
+  compileChatMessages,
+  compileChatMessagesWithIds,
+  isPlaceholder,
+  type MessagePlaceholderValues,
+  type PromptMessage as ServerPromptMessage,
+} from "./server/llm/compileChatMessages";
 
 // export db types only
 export * from "@prisma/client";

packages/shared/src/interfaces/tableNames.ts

@@ -4,11 +4,11 @@
  * to avoid coupling between different batch operation types.
  */
 export enum BatchTableNames {
-  Scores = "scores",
-  Sessions = "sessions",
-  Traces = "traces",
-  Observations = "observations",
-  DatasetRunItems = "dataset_run_items",
-  DatasetItems = "dataset_items",
-  AuditLogs = "audit_logs",
+  Scores = "scores", // eslint-disable-line no-unused-vars
+  Sessions = "sessions", // eslint-disable-line no-unused-vars
+  Traces = "traces", // eslint-disable-line no-unused-vars
+  Observations = "observations", // eslint-disable-line no-unused-vars
+  DatasetRunItems = "dataset_run_items", // eslint-disable-line no-unused-vars
+  DatasetItems = "dataset_items", // eslint-disable-line no-unused-vars
+  AuditLogs = "audit_logs", // eslint-disable-line no-unused-vars
 }

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

@@ -12,9 +12,8 @@ interface CustomSSOUser extends Record<string, any> {
   verified: boolean;
 }
 
-
 export function CustomSSOProvider<P extends CustomSSOUser>(
-  options: OAuthUserConfig<P>
+  options: OAuthUserConfig<P>,
 ): OAuthConfig<P> {
   return {
     id: "custom",

packages/shared/src/server/automations/webhooks.ts

@@ -0,0 +1,247 @@
+import { Job, Processor } from "bullmq";
+import { backOff } from "exponential-backoff";
+import {
+  ActionExecutionStatus,
+  JobConfigState,
+} from "../../../prisma/generated/types";
+import {
+  PromptWebhookOutboundSchema,
+  WebhookDefaultHeaders,
+} from "../../domain";
+import { prisma } from "../../db";
+import { TQueueJobTypes, QueueName, WebhookInput } from "../queues";
+import {
+  getActionByIdWithSecrets,
+  getAutomationById,
+  getConsecutiveAutomationFailures,
+} from "../repositories";
+import { logger } from "..";
+import { createSignatureHeader } from "../../encryption/signature";
+import { decrypt } from "../../encryption";
+import { InternalServerError, LangfuseNotFoundError } from "../../errors";
+
+export const webhookProcessor: Processor = async (
+  job: Job<TQueueJobTypes[QueueName.WebhookQueue]>,
+) => {
+  try {
+    return await executeWebhook(job.data.payload);
+  } catch (error) {
+    logger.error("Error executing WebhookJob", error);
+    throw error;
+  }
+};
+
+// TODO: Webhook outgoing API versioning
+export const executeWebhook = async (input: WebhookInput) => {
+  const executionStart = new Date();
+
+  const { projectId, automationId, executionId } = input;
+  let httpStatus: number | undefined;
+  let responseBody: string | undefined;
+
+  try {
+    logger.debug(`Executing webhook for automation ${automationId}`);
+
+    const automation = await getAutomationById({
+      projectId,
+      automationId,
+    });
+
+    if (!automation) {
+      throw new LangfuseNotFoundError(`Automation ${automationId} not found`);
+    }
+
+    const actionConfig = await getActionByIdWithSecrets({
+      projectId,
+      actionId: automation.action.id,
+    });
+
+    if (!actionConfig) {
+      throw new Error("Action config not found");
+    }
+
+    if (actionConfig.config.type !== "WEBHOOK") {
+      throw new InternalServerError("Action config is not a webhook");
+    }
+
+    // TypeScript now knows actionConfig.config is WebhookActionConfig
+    const webhookConfig = actionConfig.config;
+
+    const validatedPayload = PromptWebhookOutboundSchema.safeParse({
+      id: input.executionId,
+      timestamp: new Date(),
+      type: input.payload.type,
+      apiVersion: "v1",
+      action: input.payload.action,
+      prompt: input.payload.prompt,
+    });
+
+    if (!validatedPayload.success) {
+      throw new InternalServerError(
+        `Invalid webhook payload: ${validatedPayload.error.message}`,
+      );
+    }
+
+    // Prepare webhook payload with prompt always last
+    const { prompt, ...otherFields } = validatedPayload.data;
+    const webhookPayload = JSON.stringify({
+      ...otherFields,
+      prompt,
+    });
+
+    // Prepare headers with signature if secret exists
+    const requestHeaders: Record<string, string> = {
+      ...WebhookDefaultHeaders,
+      ...webhookConfig.headers,
+    };
+
+    if (!webhookConfig.secretKey) {
+      logger.warn(
+        `Webhook config for action ${automation.action.id} has no secret key, failing webhook execution`,
+      );
+      throw new InternalServerError(
+        "Webhook config has no secret key, failing webhook execution",
+      );
+    }
+
+    if (webhookConfig.secretKey) {
+      try {
+        const decryptedSecret = decrypt(webhookConfig.secretKey);
+
+        const signature = createSignatureHeader(
+          webhookPayload,
+          decryptedSecret,
+        );
+        requestHeaders["x-langfuse-signature"] = signature;
+      } catch (error) {
+        logger.error(
+          "Failed to decrypt webhook secret or generate signature",
+          error,
+        );
+        throw new InternalServerError("Failed to generate webhook signature");
+      }
+    }
+
+    await backOff(
+      async () => {
+        logger.debug(
+          `Sending webhook to ${webhookConfig.url} with payload ${JSON.stringify(
+            webhookPayload,
+          )} and headers ${JSON.stringify(requestHeaders)}`,
+        );
+        const res = await fetch(webhookConfig.url, {
+          method: "POST",
+          body: webhookPayload,
+          headers: requestHeaders,
+        });
+
+        httpStatus = res.status;
+        responseBody = await res.text();
+
+        if (res.status !== 200) {
+          logger.warn(
+            `Webhook does not return 200: failed with status ${res.status} for url ${webhookConfig.url} and project ${projectId}. Body: ${responseBody}`,
+          );
+          throw new Error(
+            `Webhook does not return 200: failed with status ${res.status} for url ${webhookConfig.url} and project ${projectId}`,
+          );
+        }
+      },
+      {
+        numOfAttempts: 4, // no retries for webhook calls via BullMQ
+      },
+    );
+
+    // Update action execution status on success
+    await prisma.automationExecution.update({
+      where: {
+        projectId,
+        triggerId: automation.trigger.id,
+        actionId: automation.action.id,
+        id: executionId,
+      },
+      data: {
+        status: ActionExecutionStatus.COMPLETED,
+        startedAt: executionStart,
+        finishedAt: new Date(),
+      },
+    });
+
+    logger.debug(
+      `Webhook executed successfully for action ${automation.action.id}`,
+    );
+  } catch (error) {
+    logger.error("Error executing webhook", error);
+
+    const automation = await getAutomationById({
+      projectId,
+      automationId,
+    });
+
+    if (!automation) {
+      throw new LangfuseNotFoundError(`Automation ${automationId} not found`);
+    }
+
+    const shouldRetryJob =
+      error instanceof LangfuseNotFoundError ||
+      error instanceof InternalServerError;
+
+    if (shouldRetryJob) {
+      logger.warn(
+        `Retrying bullmq for webhook job for action ${automation.action.id}`,
+      );
+      throw error;
+    }
+
+    // Update action execution status and check if we should disable trigger
+    await prisma.$transaction(async (tx) => {
+      // Update execution status
+      await tx.automationExecution.update({
+        where: {
+          id: executionId,
+          projectId,
+          triggerId: automation.trigger.id,
+          actionId: automation.action.id,
+        },
+        data: {
+          status: ActionExecutionStatus.ERROR,
+          startedAt: executionStart,
+          finishedAt: new Date(),
+          error: error instanceof Error ? error.message : "Unknown error",
+          output: httpStatus
+            ? {
+                httpStatus,
+                responseBody: responseBody?.substring(0, 1000),
+              }
+            : undefined,
+        },
+      });
+
+      // Check consecutive failures from execution history
+      const consecutiveFailures = await getConsecutiveAutomationFailures({
+        automationId,
+        projectId,
+      });
+
+      logger.info(
+        `Consecutive failures: ${consecutiveFailures} for trigger ${automation.trigger.id} in project ${projectId}`,
+      );
+
+      // Check if trigger should be disabled (this is the 5th failure, looking for 4 in the past.)
+      if (consecutiveFailures >= 4) {
+        await tx.trigger.update({
+          where: { id: automation.trigger.id, projectId },
+          data: { status: JobConfigState.INACTIVE },
+        });
+
+        logger.warn(
+          `Automation ${automation.trigger.id} disabled after ${consecutiveFailures} consecutive failures in project ${projectId}`,
+        );
+      }
+    });
+
+    logger.debug(
+      `Webhook failed for action ${automation.action.id} in project ${projectId}`,
+    );
+  }
+};

packages/shared/src/server/clickhouse/measureAndReturn.ts

@@ -0,0 +1,98 @@
+import { instrumentAsync } from "../instrumentation";
+import * as opentelemetry from "@opentelemetry/api";
+import { env } from "../../env";
+import { logger } from "../logger";
+
+const executionWrapper = async <T, Y>(
+  input: T,
+  fn: (input: T) => Promise<Y>, // eslint-disable-line no-unused-vars
+  span?: opentelemetry.Span,
+  attributePrefix?: string,
+): Promise<[Y, number]> => {
+  const startTime = Date.now();
+  const res = await fn(input);
+  const duration = Date.now() - startTime;
+  span?.setAttribute(
+    `langfuse.experiment.amts.${attributePrefix}-duration`,
+    duration,
+  );
+  return [res, duration];
+};
+
+export const measureAndReturn = async <T, Y>(args: {
+  operationName: string;
+  projectId: string;
+  input: T;
+  existingExecution: (input: T) => Promise<Y>; // eslint-disable-line no-unused-vars
+  newExecution: (input: T) => Promise<Y>; // eslint-disable-line no-unused-vars
+}): Promise<Y> => {
+  return instrumentAsync(
+    {
+      name: `experiment-${args.operationName}`,
+      spanKind: opentelemetry.SpanKind.CLIENT,
+    },
+    async (currentSpan) => {
+      const { input, existingExecution, newExecution } = args;
+
+      if (
+        env.LANGFUSE_EXPERIMENT_COMPARE_READ_FROM_AGGREGATING_MERGE_TREES !==
+        "true"
+      ) {
+        currentSpan.setAttribute(`langfuse.experiment.amts.run`, "disabled");
+        return existingExecution(input);
+      }
+
+      // If not whitelisted, apply sampling logic
+      if (
+        !env.LANGFUSE_EXPERIMENT_WHITELISTED_PROJECT_IDS.includes(
+          args.projectId,
+        ) &&
+        Math.random() > env.LANGFUSE_EXPERIMENT_SAMPLING_RATE
+      ) {
+        currentSpan.setAttribute(`langfuse.experiment.amts.run`, "sampled-out");
+        return existingExecution(input);
+      }
+
+      currentSpan.setAttribute(`langfuse.experiment.amts.run`, "true");
+
+      try {
+        const [[existingResult, existingDuration], [newResult, newDuration]] =
+          await Promise.all([
+            executionWrapper(input, existingExecution, currentSpan, "existing"),
+            executionWrapper(input, newExecution, currentSpan, "new"),
+          ]);
+        // Positive duration difference means new is faster
+        const durationDifference = existingDuration - newDuration;
+        currentSpan?.setAttribute(
+          "langfuse.experiment.amts.execution-time-difference",
+          durationDifference,
+        );
+
+        if (
+          env.LANGFUSE_EXPERIMENT_ADD_QUERY_RESULT_TO_SPAN_PROJECT_IDS.some(
+            (p) => p === args.projectId,
+          )
+        ) {
+          currentSpan?.setAttribute(
+            "langfuse.experiment.amts.existing-result",
+            JSON.stringify(existingResult),
+          );
+          currentSpan?.setAttribute(
+            "langfuse.experiment.amts.new-result",
+            JSON.stringify(newResult),
+          );
+        }
+
+        return env.LANGFUSE_EXPERIMENT_RETURN_NEW_RESULT === "true"
+          ? newResult
+          : existingResult;
+      } catch (e) {
+        logger.error(
+          "Failed to run experiment wrapper. Retrying existing query",
+          e,
+        );
+        return existingExecution(input);
+      }
+    },
+  );
+};

packages/shared/src/server/index.ts

@@ -3,6 +3,7 @@ export * from "./services/email/organizationInvitation/sendMembershipInvitationE
 export * from "./services/email/batchExportSuccess/sendBatchExportSuccessEmail";
 export * from "./services/email/passwordReset/sendResetPasswordVerificationRequest";
 export * from "./services/PromptService";
+export * from "./services/PromptService/types";
 export * from "./services/traces-ui-table-service";
 export * from "./services/InMemoryFilterService";
 export * from "./auth/apiKeys";
@@ -27,6 +28,7 @@ export * from "./redis/traceUpsert";
 export * from "./redis/createEvalQueue";
 export * from "./redis/cloudUsageMeteringQueue";
 export * from "./redis/getQueue";
+export * from "./redis/webhookQueue";
 export * from "./redis/traceDelete";
 export * from "./redis/projectDelete";
 export * from "./redis/scoreDelete";
@@ -44,6 +46,7 @@ export * from "./redis/coreDataS3ExportQueue";
 export * from "./redis/meteringDataPostgresExportQueue";
 export * from "./redis/experimentCreateQueue";
 export * from "./redis/dlqRetryQueue";
+export * from "./redis/entityChangeQueue";
 export * from "./auth/types";
 export * from "./queues";
 export * from "./orderByToPrisma";
@@ -59,9 +62,12 @@ export * from "./services/datasets-ui-table-service";
 export * from "./services/DashboardService";
 export * from "./services/TableViewService";
 export * from "./services/DefaultEvaluationModelService";
+export * from "./clickhouse/measureAndReturn";
 
 export * from "./data-deletion/ingestionFileDeletion";
 export * from "./s3";
 
+export * from "./automations/webhooks";
+
 // test utils
 export * from "./test-utils";

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

@@ -45,6 +45,7 @@ const getS3StorageServiceClient = (bucketName: string): StorageService => {
   return s3StorageServiceClient;
 };
 
+// eslint-disable-next-line no-unused-vars
 export type TokenCountDelegate = (p: {
   model: Model;
   text: unknown;

packages/shared/src/server/instrumentation/index.ts

@@ -22,7 +22,7 @@ export type SpanCtx = {
   traceContext?: TCarrier;
 };
 
-type AsyncCallbackFn<T> = (span: opentelemetry.Span) => Promise<T>;
+type AsyncCallbackFn<T> = (span: opentelemetry.Span) => Promise<T>; // eslint-disable-line no-unused-vars
 
 export async function instrumentAsync<T>(
   ctx: SpanCtx,
@@ -64,7 +64,7 @@ export async function instrumentAsync<T>(
   );
 }
 
-type SyncCallbackFn<T> = (span: opentelemetry.Span) => T;
+type SyncCallbackFn<T> = (span: opentelemetry.Span) => T; // eslint-disable-line no-unused-vars
 
 export function instrumentSync<T>(
   ctx: SpanCtx,

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

@@ -1,21 +1,26 @@
 import { z } from "zod/v4";
 import { v4 as uuidv4 } from "uuid";
-import { type ChatMessage, type PlaceholderMessage, ChatMessageType, type PromptChatMessageSchema, type ChatMessageWithId, type ChatMessageWithIdNoPlaceholders, ChatMessageSchema } from "./types";
+import {
+  type ChatMessage,
+  type PlaceholderMessage,
+  ChatMessageType,
+  type PromptChatMessageSchema,
+  type ChatMessageWithId,
+  type ChatMessageWithIdNoPlaceholders,
+} from "./types";
 
-export type MessagePlaceholderValues = Record<string, ChatMessage[]>;
+export type MessagePlaceholderValues = Record<string, unknown[]>;
 export type PromptMessage = z.infer<typeof PromptChatMessageSchema>;
 
-export function isPlaceholder(message: PromptMessage): message is PlaceholderMessage {
+export function isPlaceholder(
+  message: PromptMessage,
+): message is PlaceholderMessage {
   return "type" in message && message.type === ChatMessageType.Placeholder;
 }
 
-function validateMessage(message: unknown): message is ChatMessage {
-  return ChatMessageSchema.safeParse(message).success;
-}
-
 function replaceTextVariables(
   content: string,
-  textVariables: Record<string, string>
+  textVariables: Record<string, string>,
 ): string {
   let result = content;
   for (const [varName, varValue] of Object.entries(textVariables)) {
@@ -28,35 +33,44 @@ function replaceTextVariables(
 
 function expandPlaceholder(
   placeholder: PlaceholderMessage,
-  placeholderValues: MessagePlaceholderValues
+  placeholderValues: MessagePlaceholderValues,
 ): ChatMessage[] {
   const replacementMessages = placeholderValues[placeholder.name];
 
   if (!replacementMessages) {
-    throw new Error(`Missing value for message placeholder: ${placeholder.name}`);
+    throw new Error(
+      `Missing value for message placeholder: ${placeholder.name}`,
+    );
   }
 
   if (!Array.isArray(replacementMessages)) {
-    throw new Error(`Placeholder value for '${placeholder.name}' must be an array of messages`);
+    throw new Error(
+      `Placeholder value for '${placeholder.name}' must be an array of messages`,
+    );
   }
 
-  for (const replacementMsg of replacementMessages) {
-    if (!validateMessage(replacementMsg)) {
-      throw new Error(`Invalid message format in placeholder '${placeholder.name}': messages must have 'role' and 'content' properties`);
+  // Allow arbitrary objects - just pass them through as ChatMessage
+  // Users might want to use ChatML with placeholders for any message structure
+  return replacementMessages.map((replacementMsg) => {
+    if (typeof replacementMsg === "object" && replacementMsg !== null) {
+      return replacementMsg as ChatMessage;
     }
-  }
-  return replacementMessages;
+
+    throw new Error(
+      `Invalid message in placeholder '${placeholder.name}': expected object but got ${typeof replacementMsg}`,
+    );
+  });
 }
 
 export function compileChatMessages(
   messages: PromptMessage[],
   placeholderValues: MessagePlaceholderValues,
-  textVariables?: Record<string, string>
+  textVariables?: Record<string, string>,
 ): ChatMessage[] {
   const expandedMessages = messages.flatMap((message) =>
     isPlaceholder(message)
       ? expandPlaceholder(message, placeholderValues)
-      : [message as ChatMessage]
+      : [message as ChatMessage],
   );
 
   // substitute text variables
@@ -65,27 +79,27 @@ export function compileChatMessages(
   }
 
   return expandedMessages.map((message) => {
-    if (!message.content) {
+    if (!message.content || typeof message.content !== "string") {
       return message;
     }
 
     return {
       ...message,
-      content: replaceTextVariables(message.content, textVariables)
+      content: replaceTextVariables(message.content, textVariables),
     };
   });
 }
 
 export function compileChatMessagesWithIds(
   messages: ChatMessageWithId[],
-  placeholderValues: Record<string, ChatMessage[]>,
-  textVariables?: Record<string, string>
+  placeholderValues: MessagePlaceholderValues,
+  textVariables?: Record<string, string>,
 ): ChatMessageWithIdNoPlaceholders[] {
   // TODO: check, is it even important to retain the IDs?
   const expandedMessages = messages.flatMap((message) => {
     if (isPlaceholder(message)) {
       const expandedMsgs = expandPlaceholder(message, placeholderValues);
-      return expandedMsgs.map(msg => ({ ...msg, id: uuidv4() }));
+      return expandedMsgs.map((msg) => ({ ...msg, id: uuidv4() }));
     } else {
       // Preserve message IDs for already non-placeholder messages
       return [message as ChatMessageWithIdNoPlaceholders];
@@ -98,19 +112,22 @@ export function compileChatMessagesWithIds(
   }
 
   return expandedMessages.map((message) => {
-    if (!message.content) {
+    if (!message.content || typeof message.content !== "string") {
       return message;
     }
 
     return {
       ...message,
-      content: replaceTextVariables(message.content, textVariables)
+      content: replaceTextVariables(message.content, textVariables),
     };
   });
 }
 
 export function extractPlaceholderNames(messages: PromptMessage[]): string[] {
   return messages
-    .filter((msg): msg is PlaceholderMessage => "type" in msg && msg.type === ChatMessageType.Placeholder)
-    .map(msg => msg.name);
+    .filter(
+      (msg): msg is PlaceholderMessage =>
+        "type" in msg && msg.type === ChatMessageType.Placeholder,
+    )
+    .map((msg) => msg.name);
 }

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

@@ -63,6 +63,7 @@ type FetchLLMCompletionParams = LLMCompletionParams & {
 };
 
 export async function fetchLLMCompletion(
+  // eslint-disable-next-line no-unused-vars
   params: LLMCompletionParams & {
     streaming: true;
   },
@@ -72,6 +73,7 @@ export async function fetchLLMCompletion(
 }>;
 
 export async function fetchLLMCompletion(
+  // eslint-disable-next-line no-unused-vars
   params: LLMCompletionParams & {
     streaming: false;
   },
@@ -81,6 +83,7 @@ export async function fetchLLMCompletion(
 }>;
 
 export async function fetchLLMCompletion(
+  // eslint-disable-next-line no-unused-vars
   params: LLMCompletionParams & {
     streaming: false;
     structuredOutputSchema: ZodSchema;
@@ -91,6 +94,7 @@ export async function fetchLLMCompletion(
 }>;
 
 export async function fetchLLMCompletion(
+  // eslint-disable-next-line no-unused-vars
   params: LLMCompletionParams & {
     tools: LLMToolDefinition[];
     streaming: false;
@@ -154,28 +158,47 @@ export async function fetchLLMCompletion(
 
   finalCallbacks = finalCallbacks.length > 0 ? finalCallbacks : undefined;
 
+  // Helper function to safely stringify content
+  const safeStringify = (content: any): string => {
+    try {
+      return JSON.stringify(content);
+    } catch {
+      return "[Unserializable content]";
+    }
+  };
+
   let finalMessages: BaseMessage[];
   // VertexAI requires at least 1 user message
   if (modelParams.adapter === LLMAdapter.VertexAI && messages.length === 1) {
-    finalMessages = [new HumanMessage(messages[0].content)];
+    const safeContent =
+      typeof messages[0].content === "string"
+        ? messages[0].content
+        : JSON.stringify(messages[0].content);
+    finalMessages = [new HumanMessage(safeContent)];
   } else {
     finalMessages = messages.map((message) => {
+      // For arbitrary content types, convert to string safely
+      const safeContent =
+        typeof message.content === "string"
+          ? message.content
+          : safeStringify(message.content);
+
       if (message.role === ChatMessageRole.User)
-        return new HumanMessage(message.content);
+        return new HumanMessage(safeContent);
       if (
         message.role === ChatMessageRole.System ||
         message.role === ChatMessageRole.Developer
       )
-        return new SystemMessage(message.content);
+        return new SystemMessage(safeContent);
 
       if (message.type === ChatMessageType.ToolResult)
         return new ToolMessage({
-          content: message.content,
+          content: safeContent,
           tool_call_id: message.toolCallId,
         });
 
       return new AIMessage({
-        content: message.content,
+        content: safeContent,
         tool_calls:
           message.type === ChatMessageType.AssistantToolCall
             ? (message.toolCalls as any)
@@ -325,7 +348,7 @@ export async function fetchLLMCompletion(
 
     /*
   Workaround OpenAI reasoning models:
-  
+
   This is a temporary workaround to avoid sending unsupported parameters to OpenAI's O1 models.
   O1 models do not support:
   - system messages

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

@@ -176,7 +176,12 @@ export type ToolResultMessage = z.infer<typeof ToolResultMessageSchema>;
 
 export const PlaceholderMessageSchema = z.object({
   type: z.literal(ChatMessageType.Placeholder),
-  name: z.string().regex(/^[a-zA-Z][a-zA-Z0-9_]*$/, "Placeholder name must start with a letter and contain only alphanumeric characters and underscores"),
+  name: z
+    .string()
+    .regex(
+      /^[a-zA-Z][a-zA-Z0-9_]*$/,
+      "Placeholder name must start with a letter and contain only alphanumeric characters and underscores",
+    ),
 });
 export type PlaceholderMessage = z.infer<typeof PlaceholderMessageSchema>;
 
@@ -191,7 +196,7 @@ export const ChatMessageSchema = z.union([
   z
     .object({
       role: z.union([ChatMessageDefaultRoleSchema, z.string()]), // Users may ingest any string as role via API/SDK
-      content: z.string(),
+      content: z.union([z.string(), z.array(z.any()), z.any()]), // Support arbitrary content types for message placeholders
     })
     .transform((msg) => {
       return {
@@ -202,8 +207,10 @@ export const ChatMessageSchema = z.union([
 ]);
 
 export type ChatMessage = z.infer<typeof ChatMessageSchema>;
-export type ChatMessageWithId = (ChatMessage & { id: string }) | (PlaceholderMessage & { id: string });
-export type ChatMessageWithIdNoPlaceholders = (ChatMessage & { id: string });
+export type ChatMessageWithId =
+  | (ChatMessage & { id: string })
+  | (PlaceholderMessage & { id: string });
+export type ChatMessageWithIdNoPlaceholders = ChatMessage & { id: string };
 
 export const PromptChatMessageSchema = z.union([
   z.object({
@@ -335,9 +342,12 @@ export const anthropicModels = [
 
 // WARNING: The first entry in the array is chosen as the default model to add LLM API keys
 export const vertexAIModels = [
-  "gemini-2.0-flash",
+  "gemini-2.5-pro",
+  "gemini-2.5-flash",
+  "gemini-2.5-flash-lite-preview-06-17",
   "gemini-2.5-pro-preview-05-06",
   "gemini-2.5-flash-preview-05-20",
+  "gemini-2.0-flash",
   "gemini-2.0-pro-exp-02-05",
   "gemini-2.0-flash-001",
   "gemini-2.0-flash-lite-preview-02-05",
@@ -349,6 +359,9 @@ export const vertexAIModels = [
 
 // WARNING: The first entry in the array is chosen as the default model to add LLM API keys. Make sure it supports top_p, max_tokens and temperature.
 export const googleAIStudioModels = [
+  "gemini-2.5-flash",
+  "gemini-2.5-pro",
+  "gemini-2.5-flash-lite-preview-06-17",
   "gemini-2.5-pro-preview-05-06",
   "gemini-2.5-flash-preview-05-20",
   "gemini-2.0-flash",

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

@@ -432,18 +432,22 @@ export class FilterList {
     this.filters.push(...filter);
   }
 
+  // eslint-disable-next-line no-unused-vars
   find(predicate: (filter: Filter) => boolean) {
     return this.filters.find(predicate);
   }
 
+  // eslint-disable-next-line no-unused-vars
   filter(predicate: (filter: Filter) => boolean) {
     return new FilterList(this.filters.filter(predicate));
   }
 
+  // eslint-disable-next-line no-unused-vars
   some(predicate: (filter: Filter) => boolean) {
     return this.filters.some(predicate);
   }
 
+  // eslint-disable-next-line no-unused-vars
   forEach(callback: (filter: Filter) => void) {
     this.filters.forEach(callback);
   }

packages/shared/src/server/queues.ts

@@ -6,6 +6,8 @@ import {
   BatchActionType,
 } from "../features/batchAction/types";
 import { BatchTableNames } from "../interfaces/tableNames";
+import { EventActionSchema } from "../domain";
+import { PromptDomainSchema } from "../domain/prompts";
 
 export const IngestionEvent = z.object({
   data: z.object({
@@ -129,6 +131,33 @@ export const DeadLetterRetryQueueEventSchema = z.object({
   timestamp: z.date(),
 });
 
+export const WebhookOutboundEnvelopeSchema = z.object({
+  prompt: PromptDomainSchema,
+  action: EventActionSchema,
+  type: z.literal("prompt-version"),
+});
+
+export const WebhookInputSchema = z.object({
+  projectId: z.string(),
+  automationId: z.string(),
+  executionId: z.string(),
+  payload: WebhookOutboundEnvelopeSchema,
+});
+
+export const EntityChangeEventSchema = z.discriminatedUnion("entityType", [
+  z.object({
+    entityType: z.literal("prompt-version"),
+    projectId: z.string(),
+    promptId: z.string(),
+    action: EventActionSchema,
+    prompt: PromptDomainSchema,
+  }),
+  // Add other entity types here in the future
+]);
+
+export type WebhookInput = z.infer<typeof WebhookInputSchema>;
+export type EntityChangeEventType = z.infer<typeof EntityChangeEventSchema>;
+
 export type CreateEvalQueueEventType = z.infer<
   typeof CreateEvalQueueEventSchema
 >;
@@ -161,6 +190,8 @@ export type DeadLetterRetryQueueEventType = z.infer<
   typeof DeadLetterRetryQueueEventSchema
 >;
 
+export type WebhookQueueEventType = z.infer<typeof WebhookInputSchema>;
+
 export enum QueueName {
   TraceUpsert = "trace-upsert", // Ingestion pipeline adds events on each Trace upsert
   TraceDelete = "trace-delete",
@@ -184,6 +215,8 @@ export enum QueueName {
   CreateEvalQueue = "create-eval-queue",
   ScoreDelete = "score-delete",
   DeadLetterRetryQueue = "dead-letter-retry-queue",
+  WebhookQueue = "webhook-queue",
+  EntityChangeQueue = "entity-change-queue",
 }
 
 export enum QueueJobs {
@@ -209,6 +242,8 @@ export enum QueueJobs {
   CreateEvalJob = "create-eval-job",
   ScoreDelete = "score-delete",
   DeadLetterRetryJob = "dead-letter-retry-job",
+  WebhookJob = "webhook-job",
+  EntityChangeJob = "entity-change-job",
 }
 
 export type TQueueJobTypes = {
@@ -308,4 +343,16 @@ export type TQueueJobTypes = {
     payload: DeadLetterRetryQueueEventType;
     name: QueueJobs.DeadLetterRetryJob;
   };
+  [QueueName.WebhookQueue]: {
+    timestamp: Date;
+    id: string;
+    payload: WebhookInput;
+    name: QueueJobs.WebhookJob;
+  };
+  [QueueName.EntityChangeQueue]: {
+    timestamp: Date;
+    id: string;
+    payload: EntityChangeEventType;
+    name: QueueJobs.EntityChangeJob;
+  };
 };

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

@@ -1,6 +1,10 @@
 import { Queue } from "bullmq";
 import { QueueName, TQueueJobTypes } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
+import {
+  createNewRedisInstance,
+  redisQueueRetryOptions,
+  getQueuePrefix,
+} from "./redis";
 import { logger } from "../logger";
 
 export class BatchExportQueue {

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

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

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

@@ -1,7 +1,11 @@
 import { Queue } from "bullmq";
 import { env } from "../../env";
 import { QueueName, QueueJobs } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
+import {
+  createNewRedisInstance,
+  redisQueueRetryOptions,
+  getQueuePrefix,
+} from "./redis";
 import { logger } from "../logger";
 
 export class CloudUsageMeteringQueue {

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

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

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

@@ -1,6 +1,10 @@
 import { QueueName, TQueueJobTypes } from "../queues";
 import { Queue } from "bullmq";
-import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
+import {
+  createNewRedisInstance,
+  redisQueueRetryOptions,
+  getQueuePrefix,
+} from "./redis";
 import { logger } from "../logger";
 
 export class CreateEvalQueue {

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

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

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

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

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

@@ -1,6 +1,10 @@
 import { QueueName, TQueueJobTypes } from "../queues";
 import { Queue } from "bullmq";
-import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
+import {
+  createNewRedisInstance,
+  redisQueueRetryOptions,
+  getQueuePrefix,
+} from "./redis";
 import { logger } from "../logger";
 
 export class DatasetRunItemUpsertQueue {

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

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

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

@@ -0,0 +1,50 @@
+import { Queue } from "bullmq";
+import { QueueName, TQueueJobTypes } from "../queues";
+import {
+  createNewRedisInstance,
+  getQueuePrefix,
+  redisQueueRetryOptions,
+} from "./redis";
+import { logger } from "../logger";
+
+export class EntityChangeQueue {
+  private static instance: Queue<
+    TQueueJobTypes[QueueName.EntityChangeQueue]
+  > | null = null;
+
+  public static getInstance(): Queue<
+    TQueueJobTypes[QueueName.EntityChangeQueue]
+  > | null {
+    if (EntityChangeQueue.instance) return EntityChangeQueue.instance;
+
+    const newRedis = createNewRedisInstance({
+      enableOfflineQueue: false,
+      ...redisQueueRetryOptions,
+    });
+
+    EntityChangeQueue.instance = newRedis
+      ? new Queue<TQueueJobTypes[QueueName.EntityChangeQueue]>(
+          QueueName.EntityChangeQueue,
+          {
+            connection: newRedis,
+            prefix: getQueuePrefix(QueueName.EntityChangeQueue),
+            defaultJobOptions: {
+              removeOnComplete: true,
+              removeOnFail: 100_000,
+              attempts: 5,
+              backoff: {
+                type: "exponential",
+                delay: 5000,
+              },
+            },
+          },
+        )
+      : null;
+
+    EntityChangeQueue.instance?.on("error", (err) => {
+      logger.error("EntityChangeQueue error", err);
+    });
+
+    return EntityChangeQueue.instance;
+  }
+}

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

@@ -1,7 +1,11 @@
 import { Queue } from "bullmq";
 import { logger } from "../logger";
 import { TQueueJobTypes, QueueName } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
+import {
+  createNewRedisInstance,
+  redisQueueRetryOptions,
+  getQueuePrefix,
+} from "./redis";
 
 export class EvalExecutionQueue {
   private static instance: Queue<

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

@@ -1,7 +1,11 @@
 import { Queue } from "bullmq";
 import { logger } from "../logger";
 import { TQueueJobTypes, QueueName } from "../queues";
-import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
+import {
+  createNewRedisInstance,
+  redisQueueRetryOptions,
+  getQueuePrefix,
+} from "./redis";
 
 export class ExperimentCreateQueue {
   private static instance: Queue<

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

@@ -21,6 +21,8 @@ import { BatchActionQueue } from "./batchActionQueue";
 import { CreateEvalQueue } from "./createEvalQueue";
 import { ScoreDeleteQueue } from "./scoreDelete";
 import { DeadLetterRetryQueue } from "./dlqRetryQueue";
+import { WebhookQueue } from "./webhookQueue";
+import { EntityChangeQueue } from "./entityChangeQueue";
 
 // IngestionQueue is sharded and requires a sharding key
 // Use IngestionQueue.getInstance({ shardName: queueName }) directly instead
@@ -70,8 +72,14 @@ export function getQueue(
       return ScoreDeleteQueue.getInstance();
     case QueueName.DeadLetterRetryQueue:
       return DeadLetterRetryQueue.getInstance();
-    default:
+    case QueueName.WebhookQueue:
+      return WebhookQueue.getInstance();
+    case QueueName.EntityChangeQueue:
+      return EntityChangeQueue.getInstance();
+    default: {
+      // eslint-disable-next-line no-case-declarations, no-unused-vars
       const exhaustiveCheckDefault: never = queueName;
       throw new Error(`Queue ${queueName} not found`);
+    }
   }
 }

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

@@ -73,7 +73,7 @@ export class IngestionQueue {
           defaultJobOptions: {
             removeOnComplete: true,
             removeOnFail: 100_000,
-            attempts: 5,
+            attempts: 6,
             backoff: {
               type: "exponential",
               delay: 5000,

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

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

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

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

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

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

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

@@ -1,6 +1,10 @@
 import { QueueName, TQueueJobTypes } from "../queues";
 import { Queue } from "bullmq";
-import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
+import {
+  createNewRedisInstance,
+  redisQueueRetryOptions,
+  getQueuePrefix,
+} from "./redis";
 import { logger } from "../logger";
 
 export class ProjectDeleteQueue {

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

@@ -167,7 +167,7 @@ const createRedisClient = () => {
 
 declare global {
   // eslint-disable-next-line no-var
-  var redis: undefined | ReturnType<typeof createRedisClient>;
+  var redis: undefined | ReturnType<typeof createRedisClient>; // eslint-disable-line no-unused-vars
 }
 
 export const redis = globalThis.redis ?? createRedisClient();

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

@@ -1,12 +1,19 @@
 import { QueueName, TQueueJobTypes } from "../queues";
 import { Queue } from "bullmq";
-import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
+import {
+  createNewRedisInstance,
+  redisQueueRetryOptions,
+  getQueuePrefix,
+} from "./redis";
 import { logger } from "../logger";
 
 export class ScoreDeleteQueue {
-  private static instance: Queue<TQueueJobTypes[QueueName.ScoreDelete]> | null = null;
+  private static instance: Queue<TQueueJobTypes[QueueName.ScoreDelete]> | null =
+    null;
 
-  public static getInstance(): Queue<TQueueJobTypes[QueueName.ScoreDelete]> | null {
+  public static getInstance(): Queue<
+    TQueueJobTypes[QueueName.ScoreDelete]
+  > | null {
     if (ScoreDeleteQueue.instance) return ScoreDeleteQueue.instance;
 
     const newRedis = createNewRedisInstance({
@@ -20,16 +27,17 @@ export class ScoreDeleteQueue {
           {
             connection: newRedis,
             prefix: getQueuePrefix(QueueName.ScoreDelete),
-          defaultJobOptions: {
-            removeOnComplete: true,
-            removeOnFail: 100_000,
-            attempts: 2,
-            backoff: {
-              type: "exponential",
-              delay: 30_000,
+            defaultJobOptions: {
+              removeOnComplete: true,
+              removeOnFail: 100_000,
+              attempts: 2,
+              backoff: {
+                type: "exponential",
+                delay: 30_000,
+              },
             },
           },
-        })
+        )
       : null;
 
     ScoreDeleteQueue.instance?.on("error", (err) => {
@@ -38,4 +46,4 @@ export class ScoreDeleteQueue {
 
     return ScoreDeleteQueue.instance;
   }
-} 
+}

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

@@ -1,6 +1,10 @@
 import { QueueName, TQueueJobTypes } from "../queues";
 import { Queue } from "bullmq";
-import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
+import {
+  createNewRedisInstance,
+  redisQueueRetryOptions,
+  getQueuePrefix,
+} from "./redis";
 import { logger } from "../logger";
 
 export class TraceDeleteQueue {

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

@@ -1,6 +1,10 @@
 import { QueueName, TQueueJobTypes } from "../queues";
 import { Queue } from "bullmq";
-import { createNewRedisInstance, redisQueueRetryOptions, getQueuePrefix } from "./redis";
+import {
+  createNewRedisInstance,
+  redisQueueRetryOptions,
+  getQueuePrefix,
+} from "./redis";
 import { logger } from "../logger";
 
 export class TraceUpsertQueue {

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

@@ -0,0 +1,45 @@
+import { QueueName, TQueueJobTypes } from "../queues";
+import { Queue } from "bullmq";
+import { createNewRedisInstance, redisQueueRetryOptions } from "./redis";
+import { logger } from "../logger";
+
+export class WebhookQueue {
+  private static instance: Queue<
+    TQueueJobTypes[QueueName.WebhookQueue]
+  > | null = null;
+
+  public static getInstance(): Queue<
+    TQueueJobTypes[QueueName.WebhookQueue]
+  > | null {
+    if (WebhookQueue.instance) return WebhookQueue.instance;
+
+    const newRedis = createNewRedisInstance({
+      enableOfflineQueue: false,
+      ...redisQueueRetryOptions,
+    });
+
+    WebhookQueue.instance = newRedis
+      ? new Queue<TQueueJobTypes[QueueName.WebhookQueue]>(
+          QueueName.WebhookQueue,
+          {
+            connection: newRedis,
+            defaultJobOptions: {
+              removeOnComplete: true,
+              removeOnFail: 100_000,
+              attempts: 5,
+              backoff: {
+                type: "exponential",
+                delay: 5000,
+              },
+            },
+          },
+        )
+      : null;
+
+    WebhookQueue.instance?.on("error", (err) => {
+      logger.error("WebhookQueue error", err);
+    });
+
+    return WebhookQueue.instance;
+  }
+}

packages/shared/src/server/repositories/automation-repository.ts

@@ -0,0 +1,242 @@
+import {
+  Action,
+  ActionExecutionStatus,
+  JobConfigState,
+  prisma,
+  Trigger,
+} from "../../db";
+import {
+  TriggerEventSource,
+  WebhookActionConfigWithSecrets,
+  TriggerDomain,
+  TriggerEventAction,
+  ActionDomain,
+  AutomationDomain,
+  SafeWebhookActionConfig,
+} from "../../domain/automations";
+import { FilterState } from "../../types";
+
+export const getActionByIdWithSecrets = async ({
+  projectId,
+  actionId,
+}: {
+  projectId: string;
+  actionId: string;
+}) => {
+  const actionConfig = await prisma.action.findFirst({
+    where: {
+      id: actionId,
+      projectId,
+    },
+  });
+
+  if (!actionConfig) {
+    return null;
+  }
+
+  const config = actionConfig.config as WebhookActionConfigWithSecrets;
+  return {
+    ...actionConfig,
+    config: {
+      type: config.type,
+      url: config.url,
+      headers: config.headers,
+      apiVersion: config.apiVersion,
+      displaySecretKey: config.displaySecretKey,
+      secretKey: config.secretKey,
+    },
+  };
+};
+
+export const getActionById = async ({
+  projectId,
+  actionId,
+}: {
+  projectId: string;
+  actionId: string;
+}): Promise<ActionDomain | null> => {
+  const actionConfig = await prisma.action.findFirst({
+    where: {
+      id: actionId,
+      projectId,
+    },
+  });
+
+  if (!actionConfig) {
+    return null;
+  }
+
+  const actionDomain = convertActionToDomain(actionConfig);
+
+  return actionDomain;
+};
+
+export type TriggerDomainWithActions = TriggerDomain & { actionIds: string[] };
+
+export const getTriggerConfigurations = async ({
+  projectId,
+  eventSource,
+  status,
+}: {
+  projectId: string;
+  eventSource: TriggerEventSource;
+  status: JobConfigState;
+}): Promise<TriggerDomainWithActions[]> => {
+  const triggers = await prisma.trigger.findMany({
+    where: {
+      projectId,
+      eventSource,
+      status,
+    },
+    include: {
+      automations: {
+        include: {
+          action: true,
+        },
+      },
+    },
+  });
+
+  const triggerConfigurations = triggers.map((trigger) => ({
+    ...convertTriggerToDomain(trigger),
+    actionIds: trigger.automations.map((automation) => automation.action.id),
+  }));
+
+  return triggerConfigurations;
+};
+
+const convertTriggerToDomain = (trigger: Trigger): TriggerDomain => {
+  return {
+    ...trigger,
+    eventActions: (trigger.eventActions || []) as TriggerEventAction[],
+    filter: (trigger.filter || []) as FilterState,
+    eventSource: trigger.eventSource as TriggerEventSource,
+  };
+};
+
+const convertActionToDomain = (action: Action): ActionDomain => {
+  const config = action.config as WebhookActionConfigWithSecrets;
+  return {
+    ...action,
+    config: {
+      type: config.type,
+      url: config.url,
+      headers: config.headers,
+      apiVersion: config.apiVersion,
+      displaySecretKey: config.displaySecretKey,
+    } as SafeWebhookActionConfig,
+  };
+};
+
+export const getAutomationById = async ({
+  projectId,
+  automationId,
+}: {
+  projectId: string;
+  automationId: string;
+}): Promise<AutomationDomain | null> => {
+  const automation = await prisma.automation.findFirst({
+    where: {
+      id: automationId,
+      projectId,
+    },
+    include: {
+      action: true,
+      trigger: true,
+    },
+  });
+
+  if (!automation) {
+    return null;
+  }
+
+  return {
+    id: automation.id,
+    name: automation.name,
+    trigger: convertTriggerToDomain(automation.trigger),
+    action: convertActionToDomain(automation.action),
+  };
+};
+
+export const getAutomations = async ({
+  projectId,
+  triggerId,
+  actionId,
+}: {
+  projectId: string;
+  triggerId?: string;
+  actionId?: string;
+}): Promise<AutomationDomain[]> => {
+  const automations = await prisma.automation.findMany({
+    where: {
+      projectId,
+      ...(triggerId ? { triggerId } : {}),
+      ...(actionId ? { actionId } : {}),
+    },
+    include: {
+      action: true,
+      trigger: true,
+    },
+    orderBy: {
+      createdAt: "desc",
+    },
+  });
+
+  return automations.map((automation) => ({
+    id: automation.id,
+    name: automation.name,
+    trigger: convertTriggerToDomain(automation.trigger),
+    action: convertActionToDomain(automation.action),
+  }));
+};
+
+export const getConsecutiveAutomationFailures = async ({
+  automationId,
+  projectId,
+}: {
+  automationId: string;
+  projectId: string;
+}): Promise<number> => {
+  // First get the automation to extract triggerId and actionId
+  const automation = await prisma.automation.findFirst({
+    where: {
+      id: automationId,
+      projectId,
+    },
+  });
+
+  if (!automation) {
+    return 0;
+  }
+
+  const { triggerId, actionId } = automation;
+  const executions = await prisma.automationExecution.findMany({
+    where: {
+      triggerId,
+      actionId,
+      projectId,
+      status: {
+        in: [ActionExecutionStatus.ERROR, ActionExecutionStatus.COMPLETED],
+      },
+    },
+    orderBy: {
+      createdAt: "desc",
+    },
+    take: 20,
+    select: {
+      status: true,
+    },
+  });
+
+  let consecutiveFailures = 0;
+  for (const execution of executions) {
+    if (execution.status === ActionExecutionStatus.ERROR) {
+      consecutiveFailures++;
+    } else if (execution.status === ActionExecutionStatus.COMPLETED) {
+      break; // Stop counting when we hit a successful execution
+    }
+    // Skip PENDING/CANCELLED executions in the count
+  }
+
+  return consecutiveFailures;
+};

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

@@ -9,6 +9,7 @@ import { randomUUID } from "crypto";
 import { getClickhouseEntityType } from "../clickhouse/schemaUtils";
 import { NodeClickHouseClientConfigOptions } from "@clickhouse/client/dist/config";
 import { context, SpanKind, trace } from "@opentelemetry/api";
+import { backOff } from "exponential-backoff";
 import {
   StorageService,
   StorageServiceFactory,
@@ -203,6 +204,18 @@ export async function* queryClickhouseStream<T>(opts: {
   }
 }
 
+/**
+ * Determines if an error is retryable (socket hang up, connection reset, etc.)
+ */
+function isRetryableError(error: unknown): boolean {
+  if (!error || typeof error !== "object") return false;
+
+  const errorMessage = (error as Error).message?.toLowerCase() || "";
+
+  // Check for socket hang up and other network-related errors
+  return errorMessage.includes("socket hang up");
+}
+
 export async function queryClickhouse<T>(opts: {
   query: string;
   params?: Record<string, unknown> | undefined;
@@ -218,40 +231,78 @@ export async function queryClickhouse<T>(opts: {
       span.setAttribute("db.query.text", opts.query);
       span.setAttribute("db.operation.name", "SELECT");
 
-      const res = await clickhouseClient(opts.clickhouseConfigs).query({
-        query: opts.query,
-        format: "JSONEachRow",
-        query_params: opts.params,
-        clickhouse_settings: {
-          log_comment: JSON.stringify(opts.tags ?? {}),
-        },
-      });
-      // same logic as for prisma. we want to see queries in development
-      if (env.NODE_ENV === "development") {
-        logger.info(`clickhouse:query ${res.query_id} ${opts.query}`);
-      }
+      // Retry logic for socket hang up and other network errors
+      return await backOff(
+        async () => {
+          const res = await clickhouseClient(opts.clickhouseConfigs).query({
+            query: opts.query,
+            format: "JSONEachRow",
+            query_params: opts.params,
+            clickhouse_settings: {
+              log_comment: JSON.stringify(opts.tags ?? {}),
+            },
+          });
 
-      span.setAttribute("ch.queryId", res.query_id);
-
-      // add summary headers to the span. Helps to tune performance
-      const summaryHeader = res.response_headers["x-clickhouse-summary"];
-      if (summaryHeader) {
-        try {
-          const summary = Array.isArray(summaryHeader)
-            ? JSON.parse(summaryHeader[0])
-            : JSON.parse(summaryHeader);
-          for (const key in summary) {
-            span.setAttribute(`ch.${key}`, summary[key]);
+          // same logic as for prisma. we want to see queries in development
+          if (env.NODE_ENV === "development") {
+            logger.info(`clickhouse:query ${res.query_id} ${opts.query}`);
           }
-        } catch (error) {
-          logger.debug(
-            `Failed to parse clickhouse summary header ${summaryHeader}`,
-            error,
-          );
-        }
-      }
 
-      return await res.json<T>();
+          span.setAttribute("ch.queryId", res.query_id);
+
+          // add summary headers to the span. Helps to tune performance
+          const summaryHeader = res.response_headers["x-clickhouse-summary"];
+          if (summaryHeader) {
+            try {
+              const summary = Array.isArray(summaryHeader)
+                ? JSON.parse(summaryHeader[0])
+                : JSON.parse(summaryHeader);
+              for (const key in summary) {
+                span.setAttribute(`ch.${key}`, summary[key]);
+              }
+            } catch (error) {
+              logger.debug(
+                `Failed to parse clickhouse summary header ${summaryHeader}`,
+                error,
+              );
+            }
+          }
+
+          return await res.json<T>();
+        },
+        {
+          numOfAttempts: env.LANGFUSE_CLICKHOUSE_QUERY_MAX_ATTEMPTS,
+          retry: (error: Error, attemptNumber: number) => {
+            const shouldRetry = isRetryableError(error);
+            if (shouldRetry) {
+              logger.warn(
+                `ClickHouse query failed with retryable error (attempt ${attemptNumber}/${env.LANGFUSE_CLICKHOUSE_QUERY_MAX_ATTEMPTS}): ${error.message}`,
+                {
+                  error: error.message,
+                  attemptNumber,
+                  tags: opts.tags,
+                },
+              );
+              span.addEvent("clickhouse-query-retry", {
+                "retry.attempt": attemptNumber,
+                "retry.error": error.message,
+              });
+            } else {
+              logger.error(
+                `ClickHouse query failed with non-retryable error: ${error.message}`,
+                {
+                  error: error.message,
+                  tags: opts.tags,
+                },
+              );
+            }
+            return shouldRetry;
+          },
+          startingDelay: 100,
+          timeMultiple: 1,
+          maxDelay: 100,
+        },
+      );
     },
   );
 }

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

@@ -128,12 +128,12 @@ export const traceMtRecordInsertSchema = z.object({
   id: z.string(),
   start_time: z.number(),
   end_time: z.number().nullish(),
-  name: z.string(),
+  name: z.string().nullish(),
 
   // Metadata properties
   metadata: z.record(z.string(), z.string()),
-  user_id: z.string(),
-  session_id: z.string(),
+  user_id: z.string().nullish(),
+  session_id: z.string().nullish(),
   environment: z.string(),
   tags: z.array(z.string()),
   version: z.string().nullish(),
@@ -411,3 +411,128 @@ export const convertPostgresScoreToInsert = (
     is_deleted: 0,
   };
 };
+
+export const convertTraceToTraceMt = (
+  traceRecord: TraceRecordInsertType,
+): TraceMtRecordInsertType => {
+  return {
+    // Identifiers
+    project_id: traceRecord.project_id,
+    id: traceRecord.id,
+    start_time: traceRecord.timestamp,
+    end_time: null, // traces don't have end_time, will be null
+    name: traceRecord.name || null,
+
+    // Metadata properties
+    metadata: traceRecord.metadata,
+    user_id: traceRecord.user_id || null,
+    session_id: traceRecord.session_id || null,
+    environment: traceRecord.environment,
+    tags: traceRecord.tags,
+    version: traceRecord.version || null,
+    release: traceRecord.release || null,
+
+    // UI properties - nullable to prevent absent values being interpreted as overwrites
+    bookmarked: traceRecord.bookmarked ?? null,
+    public: traceRecord.public ?? null,
+
+    // Aggregations - empty for now, will be populated by aggregation processes
+    observation_ids: [],
+    score_ids: [],
+    cost_details: {},
+    usage_details: {},
+
+    // Input/Output
+    input: traceRecord.input || "",
+    output: traceRecord.output || "",
+
+    created_at: traceRecord.created_at,
+    updated_at: traceRecord.updated_at,
+    event_ts: traceRecord.event_ts,
+  };
+};
+
+export const convertObservationToTraceMt = (
+  observationRecord: ObservationRecordInsertType,
+): TraceMtRecordInsertType => {
+  return {
+    // Identifiers
+    project_id: observationRecord.project_id,
+    // Use trace_id as the id in traces_mt. Always set given the conditions around calling the function
+    id: observationRecord.trace_id || "",
+    start_time: observationRecord.start_time,
+    end_time: observationRecord.end_time || null,
+    name: null,
+
+    // Metadata properties
+    metadata: {},
+    user_id: null,
+    session_id: null,
+    environment: observationRecord.environment,
+    tags: [],
+    version: null,
+    release: null,
+
+    // UI properties - nullable to prevent absent values being interpreted as overwrites
+    bookmarked: null,
+    public: null,
+
+    // Aggregations - include this observation ID
+    observation_ids: [observationRecord.id],
+    score_ids: [],
+    // We can fill the cost details here, but we shouldn't trust them.
+    // Only used for verification to estimate how big the double-counting is.
+    // Actually, we don't as this will make backfills challenging.
+    cost_details: {}, // observationRecord.cost_details || {},
+    usage_details: {}, // observationRecord.usage_details || {},
+
+    // Input/Output
+    input: "",
+    output: "",
+
+    created_at: observationRecord.created_at,
+    updated_at: observationRecord.updated_at,
+    event_ts: observationRecord.event_ts,
+  };
+};
+
+export const convertScoreToTraceMt = (
+  scoreRecord: ScoreRecordInsertType,
+): TraceMtRecordInsertType => {
+  return {
+    // Identifiers
+    project_id: scoreRecord.project_id,
+    // Use trace_id as the id in traces_mt. Always set given the conditions around calling the function
+    id: scoreRecord.trace_id || "",
+    start_time: scoreRecord.timestamp,
+    end_time: null, // scores don't have end_time
+    name: null,
+
+    // Metadata properties
+    metadata: {},
+    user_id: null,
+    session_id: null,
+    environment: scoreRecord.environment,
+    tags: [], // scores don't have tags
+    version: null, // scores don't have version
+    release: null, // scores don't have release
+
+    // UI properties - nullable to prevent absent values being interpreted as overwrites
+    bookmarked: null,
+    public: null,
+
+    // Aggregations - include this score ID
+    observation_ids: [],
+    score_ids: [scoreRecord.id],
+    cost_details: {},
+    usage_details: {},
+
+    // Input/Output
+    input: "", // scores don't have input
+    output: "", // scores don't have output
+
+    created_at: scoreRecord.created_at,
+    updated_at: scoreRecord.updated_at,
+    event_ts: scoreRecord.event_ts,
+  };
+};

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

@@ -12,3 +12,4 @@ export * from "./trace-sessions";
 export * from "./scores-utils";
 export * from "./blobStorageLog";
 export * from "./environments";
+export * from "./automation-repository";

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

@@ -28,6 +28,41 @@ import {
 import { env } from "../../env";
 import { ClickHouseClientConfigOptions } from "@clickhouse/client";
 import { recordDistribution } from "../instrumentation";
+import { measureAndReturn } from "../clickhouse/measureAndReturn";
+
+// eslint-disable-next-line no-unused-vars
+enum TracesAMTs {
+  Traces7dAMT = "traces_7d_amt", // eslint-disable-line no-unused-vars
+  Traces30dAMT = "traces_30d_amt", // eslint-disable-line no-unused-vars
+  TracesAllAMT = "traces_all_amt", // eslint-disable-line no-unused-vars
+}
+
+/**
+ * Returns which AMT table to use given the timestamp.
+ * For <= 6 days, we use traces_7d_amt,
+ * for <= 29 days, we use traces_30d_amt,
+ * for all other cases we use traces_all_amt.
+ *
+ * @param fromTimestamp
+ */
+export const getTimeframesTracesAMT = (
+  fromTimestamp: Date | undefined,
+): TracesAMTs => {
+  if (!fromTimestamp) {
+    return TracesAMTs.TracesAllAMT;
+  }
+
+  const now = new Date();
+  const diffInDays = Math.floor(
+    (now.getTime() - fromTimestamp.getTime()) / (1000 * 60 * 60 * 24),
+  );
+  if (diffInDays <= 6) {
+    return TracesAMTs.Traces7dAMT;
+  } else if (diffInDays <= 29) {
+    return TracesAMTs.Traces30dAMT;
+  }
+  return TracesAMTs.TracesAllAMT;
+};
 
 /**
  * Checks if trace exists in clickhouse.
@@ -82,66 +117,104 @@ export const checkTraceExists = async ({
   const tracesFilterRes = tracesFilter.apply();
   const observationFilterRes = observationFilter?.apply();
 
-  const query = `
+  const observations_cte = `
     WITH observations_agg AS (
-        SELECT
-            multiIf(
-              arrayExists(x -> x = 'ERROR', groupArray(level)), 'ERROR',
-              arrayExists(x -> x = 'WARNING', groupArray(level)), 'WARNING',
-              arrayExists(x -> x = 'DEFAULT', groupArray(level)), 'DEFAULT',
-              'DEBUG'
-            ) AS aggregated_level,
-            countIf(level = 'ERROR') as error_count,
-            countIf(level = 'WARNING') as warning_count,
-            countIf(level = 'DEFAULT') as default_count,
-            countIf(level = 'DEBUG') as debug_count,
-            trace_id,
-            project_id
-        FROM observations o FINAL 
-        WHERE o.project_id = {projectId: String}
+      SELECT
+        multiIf(
+          arrayExists(x -> x = 'ERROR', groupArray(level)), 'ERROR',
+          arrayExists(x -> x = 'WARNING', groupArray(level)), 'WARNING',
+          arrayExists(x -> x = 'DEFAULT', groupArray(level)), 'DEFAULT',
+          'DEBUG'
+        ) AS aggregated_level,
+        countIf(level = 'ERROR') as error_count,
+        countIf(level = 'WARNING') as warning_count,
+        countIf(level = 'DEFAULT') as default_count,
+        countIf(level = 'DEBUG') as debug_count,
+        trace_id,
+        project_id
+      FROM observations o FINAL
+      WHERE o.project_id = {projectId: String}
         ${timeStampFilter ? `AND o.start_time >= {traceTimestamp: DateTime64(3)} - ${OBSERVATIONS_TO_TRACE_INTERVAL}` : ""}
         AND o.start_time >= {timestamp: DateTime64(3)} - ${OBSERVATIONS_TO_TRACE_INTERVAL}
-        GROUP BY trace_id, project_id
-      )
-    SELECT 
-      t.id as id, 
-      t.project_id as project_id
-    FROM traces t FINAL
-    ${observationFilterRes ? `INNER JOIN observations_agg o ON t.id = o.trace_id AND t.project_id = o.project_id` : ""}
-    WHERE ${tracesFilterRes.query}
-    AND t.project_id = {projectId: String}
-    AND timestamp >= {timestamp: DateTime64(3)} - ${TRACE_TO_OBSERVATIONS_INTERVAL}
-    ${maxTimeStamp ? `AND timestamp <= {maxTimeStamp: DateTime64(3)}` : ""}
-    ${!maxTimeStamp ? `AND timestamp <= {timestamp: DateTime64(3)} + INTERVAL 2 DAY` : ""}
-    ${exactTimestamp ? `AND timestamp = {exactTimestamp: DateTime64(3)}` : ""}
-    GROUP BY t.id, t.project_id
+      GROUP BY trace_id, project_id
+    )
   `;
 
-  const rows = await queryClickhouse<{ id: string; project_id: string }>({
-    query,
-    params: {
-      projectId,
-      ...tracesFilterRes.params,
-      ...(observationFilterRes ? observationFilterRes.params : {}),
-      ...(timestamp
-        ? { timestamp: convertDateToClickhouseDateTime(timestamp) }
-        : {}),
-      ...(maxTimeStamp
-        ? { maxTimeStamp: convertDateToClickhouseDateTime(maxTimeStamp) }
-        : {}),
-      ...(exactTimestamp
-        ? { exactTimestamp: convertDateToClickhouseDateTime(exactTimestamp) }
-        : {}),
+  return measureAndReturn({
+    operationName: "checkTraceExists",
+    projectId,
+    input: {
+      params: {
+        projectId,
+        ...tracesFilterRes.params,
+        ...(observationFilterRes ? observationFilterRes.params : {}),
+        ...(timestamp
+          ? { timestamp: convertDateToClickhouseDateTime(timestamp) }
+          : {}),
+        ...(maxTimeStamp
+          ? { maxTimeStamp: convertDateToClickhouseDateTime(maxTimeStamp) }
+          : {}),
+        ...(exactTimestamp
+          ? { exactTimestamp: convertDateToClickhouseDateTime(exactTimestamp) }
+          : {}),
+      },
+      tags: {
+        feature: "tracing",
+        type: "trace",
+        kind: "exists",
+        projectId,
+      },
+      timestamp: timestamp ?? exactTimestamp,
     },
-    tags: {
-      feature: "tracing",
-      type: "trace",
-      kind: "exists",
-      projectId,
+    existingExecution: async (input) => {
+      const query = `
+        ${observations_cte}
+        SELECT
+          t.id as id,
+          t.project_id as project_id
+        FROM traces t FINAL
+        ${observationFilterRes ? `INNER JOIN observations_agg o ON t.id = o.trace_id AND t.project_id = o.project_id` : ""}
+        WHERE ${tracesFilterRes.query}
+        AND t.project_id = {projectId: String}
+        AND timestamp >= {timestamp: DateTime64(3)} - ${TRACE_TO_OBSERVATIONS_INTERVAL}
+        ${maxTimeStamp ? `AND timestamp <= {maxTimeStamp: DateTime64(3)}` : ""}
+        ${!maxTimeStamp ? `AND timestamp <= {timestamp: DateTime64(3)} + INTERVAL 2 DAY` : ""}
+        ${exactTimestamp ? `AND timestamp = {exactTimestamp: DateTime64(3)}` : ""}
+        GROUP BY t.id, t.project_id
+      `;
+
+      const rows = await queryClickhouse<{ id: string; project_id: string }>({
+        query,
+        params: input.params,
+        tags: input.tags,
+      });
+
+      return rows.length > 0;
+    },
+    newExecution: async (input) => {
+      const traceAmt = getTimeframesTracesAMT(input.timestamp);
+      const query = `
+        ${observations_cte}
+        SELECT
+          t.id as id,
+          t.project_id as project_id,
+          -- Add a timestamp alias to ensure we can filter on it
+          t.start_time as timestamp
+        FROM ${traceAmt} t
+        ${observationFilterRes ? `INNER JOIN observations_agg o ON t.id = o.trace_id AND t.project_id = o.project_id` : ""}
+        WHERE ${tracesFilterRes.query}
+        AND t.project_id = {projectId: String}
+      `;
+
+      const rows = await queryClickhouse<{ id: string; project_id: string }>({
+        query,
+        params: input.params,
+        tags: input.tags,
+      });
+
+      return rows.length > 0;
     },
   });
-
-  return rows.length > 0;
 };
 
 /**
@@ -171,28 +244,77 @@ export const getTracesByIds = async (
   timestamp?: Date,
   clickhouseConfigs?: ClickHouseClientConfigOptions | undefined,
 ) => {
-  const query = `
-      SELECT * 
-      FROM traces
-      WHERE id IN ({traceIds: Array(String)})
-      AND project_id = {projectId: String}
-      ${timestamp ? `AND timestamp >= {timestamp: DateTime64(3)}` : ""} 
-      ORDER BY event_ts DESC
-      LIMIT 1 by id, project_id;`;
-  const records = await queryClickhouse<TraceRecordReadType>({
-    query,
-    params: {
-      traceIds,
-      projectId,
-      timestamp: timestamp ? convertDateToClickhouseDateTime(timestamp) : null,
+  const records = await measureAndReturn({
+    operationName: "getTracesByIds",
+    projectId,
+    input: {
+      params: {
+        traceIds,
+        projectId,
+        timestamp: timestamp
+          ? convertDateToClickhouseDateTime(timestamp)
+          : null,
+      },
+      tags: {
+        feature: "tracing",
+        type: "trace",
+        kind: "byId",
+        projectId,
+      },
+      clickhouseConfigs,
     },
-    tags: {
-      feature: "tracing",
-      type: "trace",
-      kind: "byId",
-      projectId,
+    existingExecution: (input) => {
+      const query = `
+        SELECT *
+        FROM traces
+        WHERE id IN ({traceIds: Array(String)})
+        AND project_id = {projectId: String}
+        ${timestamp ? `AND timestamp >= {timestamp: DateTime64(3)}` : ""}
+        ORDER BY event_ts DESC
+        LIMIT 1 by id, project_id;
+      `;
+      return queryClickhouse<TraceRecordReadType>({
+        query,
+        params: input.params,
+        tags: input.tags,
+        clickhouseConfigs: input.clickhouseConfigs,
+      });
+    },
+    newExecution: (input) => {
+      const query = `
+        SELECT
+          id,
+          name as name,
+          user_id as user_id,
+          metadata as metadata,
+          release as release,
+          version as version,
+          project_id,
+          environment,
+          finalizeAggregation(public) as public,
+          finalizeAggregation(bookmarked) as bookmarked,
+          tags,
+          finalizeAggregation(input) as input,
+          finalizeAggregation(output) as output,
+          session_id as session_id,
+          0 as is_deleted,
+          start_time as timestamp,
+          created_at,
+          updated_at,
+          updated_at as event_ts
+        FROM traces_all_amt
+        WHERE id IN ({traceIds: Array(String)})
+        AND project_id = {projectId: String}
+        LIMIT 1 BY project_id, id
+      `;
+
+      return queryClickhouse<TraceRecordReadType>({
+        query,
+        params: input.params,
+        tags: input.tags,
+        clickhouseConfigs: input.clickhouseConfigs,
+      });
     },
-    clickhouseConfigs,
   });
 
   return records.map(convertClickhouseToDomain);
@@ -204,11 +326,11 @@ export const getTracesBySessionId = async (
   timestamp?: Date,
 ) => {
   const query = `
-      SELECT * 
+      SELECT *
       FROM traces
       WHERE session_id IN ({sessionIds: Array(String)})
       AND project_id = {projectId: String}
-      ${timestamp ? `AND timestamp >= {timestamp: DateTime64(3)}` : ""} 
+      ${timestamp ? `AND timestamp >= {timestamp: DateTime64(3)}` : ""}
       ORDER BY event_ts DESC
       LIMIT 1 by id, project_id;`;
   const records = await queryClickhouse<TraceRecordReadType>({
@@ -239,27 +361,55 @@ export const getTracesBySessionId = async (
 };
 
 export const hasAnyTrace = async (projectId: string) => {
-  const query = `
-    SELECT 1
-    FROM traces
-    WHERE project_id = {projectId: String}
-    LIMIT 1
-  `;
-
-  const rows = await queryClickhouse<{ 1: number }>({
-    query,
-    params: {
+  return measureAndReturn({
+    operationName: "hasAnyTrace",
+    projectId,
+    input: {
       projectId,
+      tags: {
+        feature: "tracing",
+        type: "trace",
+        kind: "hasAny",
+        projectId,
+      },
     },
-    tags: {
-      feature: "tracing",
-      type: "trace",
-      kind: "hasAny",
-      projectId,
+    existingExecution: async (input) => {
+      const query = `
+        SELECT 1
+        FROM traces
+        WHERE project_id = {projectId: String}
+        LIMIT 1
+      `;
+
+      const rows = await queryClickhouse<{ 1: number }>({
+        query,
+        params: {
+          projectId: input.projectId,
+        },
+        tags: input.tags,
+      });
+
+      return rows.length > 0;
+    },
+    newExecution: async (input) => {
+      const query = `
+        SELECT 1
+        FROM traces_all_amt
+        WHERE project_id = {projectId: String}
+        LIMIT 1
+      `;
+
+      const rows = await queryClickhouse<{ 1: number }>({
+        query,
+        params: {
+          projectId: input.projectId,
+        },
+        tags: input.tags,
+      });
+
+      return rows.length > 0;
     },
   });
-
-  return rows.length > 0;
 };
 
 export const getTraceCountsByProjectInCreationInterval = async ({
@@ -270,7 +420,7 @@ export const getTraceCountsByProjectInCreationInterval = async ({
   end: Date;
 }) => {
   const query = `
-    SELECT 
+    SELECT
       project_id,
       count(*) as count
     FROM traces
@@ -306,7 +456,7 @@ export const getTraceCountOfProjectsSinceCreationDate = async ({
   start: Date;
 }) => {
   const query = `
-    SELECT 
+    SELECT
       count(*) as count
     FROM traces
     WHERE project_id IN ({projectIds: Array(String)})
@@ -347,34 +497,78 @@ export const getTraceById = async ({
   timestamp?: Date;
   fromTimestamp?: Date;
 }) => {
-  const query = `
-    SELECT * 
-    FROM traces
-    WHERE id = {traceId: String} 
-    AND project_id = {projectId: String}
-    ${timestamp ? `AND toDate(timestamp) = toDate({timestamp: DateTime64(3)})` : ""} 
-    ${fromTimestamp ? `AND timestamp >= {fromTimestamp: DateTime64(3)}` : ""} 
-    ORDER BY event_ts DESC 
-    LIMIT 1
-  `;
-
-  const records = await queryClickhouse<TraceRecordReadType>({
-    query,
-    params: {
-      traceId,
-      projectId,
-      ...(timestamp
-        ? { timestamp: convertDateToClickhouseDateTime(timestamp) }
-        : {}),
-      ...(fromTimestamp
-        ? { fromTimestamp: convertDateToClickhouseDateTime(fromTimestamp) }
-        : {}),
+  const records = await measureAndReturn({
+    operationName: "getTraceById",
+    projectId,
+    input: {
+      params: {
+        traceId,
+        projectId,
+        ...(timestamp
+          ? { timestamp: convertDateToClickhouseDateTime(timestamp) }
+          : {}),
+        ...(fromTimestamp
+          ? { fromTimestamp: convertDateToClickhouseDateTime(fromTimestamp) }
+          : {}),
+      },
+      tags: {
+        feature: "tracing",
+        type: "trace",
+        kind: "byId",
+        projectId,
+      },
     },
-    tags: {
-      feature: "tracing",
-      type: "trace",
-      kind: "byId",
-      projectId,
+    existingExecution: (input) => {
+      const query = `
+        SELECT *
+        FROM traces
+        WHERE id = {traceId: String}
+        AND project_id = {projectId: String}
+        ${timestamp ? `AND toDate(timestamp) = toDate({timestamp: DateTime64(3)})` : ""}
+        ${fromTimestamp ? `AND timestamp >= {fromTimestamp: DateTime64(3)}` : ""}
+        ORDER BY event_ts DESC
+        LIMIT 1
+      `;
+
+      return queryClickhouse<TraceRecordReadType>({
+        query,
+        params: input.params,
+        tags: input.tags,
+      });
+    },
+    newExecution: (input) => {
+      const query = `
+        SELECT
+          id,
+          name as name,
+          user_id as user_id,
+          metadata as metadata,
+          release as release,
+          version as version,
+          project_id,
+          environment,
+          finalizeAggregation(public) as public,
+          finalizeAggregation(bookmarked) as bookmarked,
+          tags,
+          finalizeAggregation(input) as input,
+          finalizeAggregation(output) as output,
+          session_id as session_id,
+          0 as is_deleted,
+          start_time as timestamp,
+          created_at,
+          updated_at,
+          updated_at as event_ts
+        FROM traces_all_amt
+        WHERE id = {traceId: String}
+        AND project_id = {projectId: String}
+        LIMIT 1
+      `;
+
+      return queryClickhouse<TraceRecordReadType>({
+        query,
+        params: input.params,
+        tags: input.tags,
+      });
     },
   });
 
@@ -409,7 +603,7 @@ export const getTracesGroupedByName = async (
   // We mainly use queries like this to retrieve filter options.
   // Therefore, we can skip final as some inaccuracy in count is acceptable.
   const query = `
-      select 
+      select
         name as name,
         count(*) as count
       from traces t FINAL
@@ -466,7 +660,7 @@ export const getTracesGroupedByUsers = async (
   // We mainly use queries like this to retrieve filter options.
   // Therefore, we can skip final as some inaccuracy in count is acceptable.
   const query = `
-      select 
+      select
         user_id as user,
         count(*) as count
       from traces t
@@ -948,7 +1142,7 @@ export const getTracesForPostHog = async function* (
       GROUP BY o.project_id, o.trace_id
     )
 
-    SELECT 
+    SELECT
       t.id as id,
       t.timestamp as timestamp,
       t.name as name,

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

@@ -1,6 +1,6 @@
 import z from "zod/v4";
 import { prisma } from "../../../db";
-import { LangfuseNotFoundError } from "../../../errors";
+import { LangfuseNotFoundError, QUEUE_ERROR_MESSAGES } from "../../../errors";
 import { LLMApiKeySchema, ZodModelConfig } from "../../llm/types";
 
 type ValidConfig = {
@@ -156,7 +156,7 @@ export class DefaultEvalModelService {
     if (!selectedModel) {
       return {
         valid: false,
-        error: `No default model or custom model found for project ${projectId}.`,
+        error: `${QUEUE_ERROR_MESSAGES.NO_DEFAULT_MODEL_ERROR} ${projectId}.`,
       };
     }
 
@@ -173,7 +173,7 @@ export class DefaultEvalModelService {
     if (!parsedKey.success) {
       return {
         valid: false,
-        error: `API key for provider "${selectedModel.provider}" not found in project ${projectId}.`,
+        error: `${QUEUE_ERROR_MESSAGES.API_KEY_ERROR} "${selectedModel.provider}" not found in project ${projectId}.`,
       };
     }
 

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

@@ -13,8 +13,11 @@ export class InMemoryFilterService {
   static evaluateFilter<T>(
     data: T,
     filter: FilterState,
-    fieldMapper: (data: T, column: string) => unknown,
+    fieldMapper: (data: T, column: string) => unknown, // eslint-disable-line no-unused-vars
   ): boolean {
+    logger.debug(
+      `Evaluating filter ${JSON.stringify(filter)} for data ${JSON.stringify(data)}`,
+    );
     try {
       // If no filters, data matches
       if (!filter || filter.length === 0) {
@@ -45,7 +48,7 @@ export class InMemoryFilterService {
   private static evaluateFilterCondition<T>(
     data: T,
     condition: FilterCondition,
-    fieldMapper: (data: T, column: string) => unknown,
+    fieldMapper: (data: T, column: string) => unknown, // eslint-disable-line no-unused-vars
   ): boolean {
     const { column, type, operator } = condition;
 

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

@@ -13,8 +13,8 @@ export type PromptParams = {
 );
 
 export enum PromptServiceMetrics {
-  PromptCacheHit = "prompt_cache_hit",
-  PromptCacheMiss = "prompt_cache_miss",
+  PromptCacheHit = "prompt_cache_hit", // eslint-disable-line no-unused-vars
+  PromptCacheMiss = "prompt_cache_miss", // eslint-disable-line no-unused-vars
 }
 
 export type PartialPrompt = Pick<

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

@@ -28,20 +28,21 @@ type UploadFile = {
 };
 
 export interface StorageService {
-  uploadFile(params: UploadFile): Promise<{ signedUrl: string }>;
+  uploadFile(params: UploadFile): Promise<{ signedUrl: string }>; // eslint-disable-line no-unused-vars
 
-  uploadJson(path: string, body: Record<string, unknown>[]): Promise<void>;
+  uploadJson(path: string, body: Record<string, unknown>[]): Promise<void>; // eslint-disable-line no-unused-vars
 
-  download(path: string): Promise<string>;
+  download(path: string): Promise<string>; // eslint-disable-line no-unused-vars
 
-  listFiles(prefix: string): Promise<{ file: string; createdAt: Date }[]>;
+  listFiles(prefix: string): Promise<{ file: string; createdAt: Date }[]>; // eslint-disable-line no-unused-vars
 
   getSignedUrl(
-    fileName: string,
-    ttlSeconds: number,
-    asAttachment?: boolean,
+    fileName: string, // eslint-disable-line no-unused-vars
+    ttlSeconds: number, // eslint-disable-line no-unused-vars
+    asAttachment?: boolean, // eslint-disable-line no-unused-vars
   ): Promise<string>;
 
+  // eslint-disable-next-line no-unused-vars
   getSignedUploadUrl(params: {
     path: string;
     ttlSeconds: number;
@@ -50,7 +51,7 @@ export interface StorageService {
     contentLength: number;
   }): Promise<string>;
 
-  deleteFiles(paths: string[]): Promise<void>;
+  deleteFiles(paths: string[]): Promise<void>; // eslint-disable-line no-unused-vars
 }
 
 export class StorageServiceFactory {

packages/shared/src/server/services/email/passwordReset/sendResetPasswordVerificationRequest.tsx

@@ -66,10 +66,11 @@ const ResetPasswordTemplate = ({ token }: ResetPasswordTemplateProps) => {
 export async function sendResetPasswordVerificationRequest(
   params: SendVerificationRequestParams,
 ) {
-  const { identifier, token, provider } = params as SendVerificationRequestParams & { token: string };
+  const { identifier, token, provider } =
+    params as SendVerificationRequestParams & { token: string };
   const transport = createTransport(provider.server);
   const htmlTemplate = await render(<ResetPasswordTemplate token={token} />);
-        
+
   const result = await transport.sendMail({
     to: identifier,
     from: provider.from,

packages/shared/src/server/test-utils/clickhouse-helpers.ts

@@ -3,9 +3,20 @@ import {
   TraceRecordInsertType,
   ObservationRecordInsertType,
   ScoreRecordInsertType,
+  convertTraceToTraceMt,
 } from "../repositories/definitions";
+import { env } from "../../env";
 
 export const createTracesCh = async (trace: TraceRecordInsertType[]) => {
+  if (
+    env.LANGFUSE_EXPERIMENT_COMPARE_READ_FROM_AGGREGATING_MERGE_TREES === "true"
+  ) {
+    await clickhouseClient().insert({
+      table: "traces_mt",
+      format: "JSONEachRow",
+      values: trace.map(convertTraceToTraceMt),
+    });
+  }
   return await clickhouseClient().insert({
     table: "traces",
     format: "JSONEachRow",

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

@@ -24,12 +24,13 @@ export class DatabaseReadStream<EntityType> extends Readable {
 
   constructor(
     // the delegate function takes care of querying the database in a paginated manner
+    // eslint-disable-next-line no-unused-vars
     private queryDelegate: (
-      pageSize: number,
-      offset: number
+      pageSize: number, // eslint-disable-line no-unused-vars
+      offset: number, // eslint-disable-line no-unused-vars
     ) => Promise<Array<EntityType>>,
-    private pageSize: number,
-    private maxRecords?: number
+    private pageSize: number, // eslint-disable-line no-unused-vars
+    private maxRecords?: number, // eslint-disable-line no-unused-vars
   ) {
     super({ objectMode: true }); // Set object mode to true to allow pushing objects to the stream rather than strings or buffers
 

packages/shared/src/server/utils/transforms/stringify.ts

@@ -1,5 +1,5 @@
 export const stringify = (data: any): string => {
   return JSON.stringify(data, (key, value) =>
-    typeof value === "bigint" ? Number.parseInt(value.toString()) : value
+    typeof value === "bigint" ? Number.parseInt(value.toString()) : value,
   );
 };

packages/shared/src/server/utils/transforms/transformStreamToCsv.ts

@@ -11,7 +11,7 @@ export function transformStreamToCsv(): Transform {
     objectMode: true,
     transform(
       row: Record<string, any>,
-      encoding: BufferEncoding,
+      encoding: BufferEncoding, // eslint-disable-line
       callback: TransformCallback,
     ): void {
       if (isFirstChunk) {