chore: release v3.98.2

* feat(traces): add more observation types

* observations table default filter for all types

* fix test running instructions

* add seeder

* add basic tests

* add to evaluator object selector

* add test to verify we default to span-create for all types

* allow generation like attributes on all types

* check if something is generation LIKE not strictly a GENERATION

* fix test

* ensure that GENERATION-like type's fields can also be null

* simplify create- logic

CLAUDE.md

@@ -83,7 +83,8 @@ Depending on the file location (sync, async)
 `web` related tests must go into the `web/src/__tests__/` folder.
 ```sh
 pnpm test-sync --testPathPattern="$FILE_LOCATION_PATTERN" --testNamePattern="$TEST_NAME_PATTERN"
-pnpm test-async --testPathPattern="$FILE_LOCATION_PATTERN" --testNamePattern="$TEST_NAME_PATTERN"
+# For tests in the async folder:
+pnpm test -- --testPathPattern="$FILE_LOCATION_PATTERN" --testNamePattern="$TEST_NAME_PATTERN"
 ```
 
 ### Testing in the Worker Package

CONTRIBUTING.md

@@ -241,7 +241,7 @@ We're using Jest with in the `web` package. Therefore, if you want to provide an
 There are three types of unit tests:
 
 - `test-sync`
-- `test-async`
+- `test` (for async folder tests)
 - `test-client`
 
 To run a specific test, for example the test: `"should handle special characters in prompt names"` in `prompts.v2.servertest.ts`, run:
@@ -249,6 +249,8 @@ To run a specific test, for example the test: `"should handle special characters
 ```sh
 cd web  # or with --filter=web
 pnpm test-sync --testPathPattern="prompts\.v2\.servertest" --testNamePattern="should handle special characters in prompt names"
+# for async folder tests:
+pnpm test -- --testPathPattern="observations-api" --testNamePattern="should fetch all observations"
 ```
 
 To run all tests:

REVIEW.md

@@ -0,0 +1,14 @@
+# Code Review Instructions
+
+## Database Migrations
+
+### ClickHouse
+
+- ClickHouse migrations in the `packages/shared/clickhouse/migrations/clustered` directory should include `ON CLUSTER default` and should use `Replicated` merge tree table types.
+  - E.g. `ReplacingMergeTree` is likely an error while `ReplicatedReplacingMergeTree` would be correct in most cases.
+- ClickHouse migrations in the `packages/shared/clickhouse/migrations/unclustered` directory must not include `ON CLUSTER` statements and must not use `Replicated` merge tree table types.
+- Migrations in `packages/shared/clickhouse/migrations/clustered` should match their counterparts in `packages/shared/clickhouse/migrations/unclustered` aside from the restrictions listed above.
+
+### Postgres
+
+- Most `schema.prisma` changes should produce a change in `packages/shared/prisma/migrations`.

fern/apis/server/definition/annotation-queues.yml

@@ -22,6 +22,13 @@ service:
             docs: limit of items per page
       response: PaginatedAnnotationQueues
 
+    createQueue:
+      docs: Create an annotation queue
+      method: POST
+      path: /annotation-queues
+      request: CreateAnnotationQueueRequest
+      response: AnnotationQueue
+
     getQueue:
       docs: Get an annotation queue by ID
       method: GET
@@ -168,6 +175,12 @@ types:
       data: list<AnnotationQueueItem>
       meta: pagination.MetaResponse
 
+  CreateAnnotationQueueRequest:
+    properties:
+      name: string
+      description: optional<string>
+      scoreConfigIds: list<string>
+
   CreateAnnotationQueueItemRequest:
     properties:
       objectId: string

fern/apis/server/definition/ingestion.yml

@@ -145,6 +145,13 @@ types:
       - SPAN
       - GENERATION
       - EVENT
+      - AGENT
+      - TOOL
+      - CHAIN
+      - RETRIEVER
+      - EVALUATOR
+      - EMBEDDING
+      - GUARDRAIL
 
   IngestionUsage:
     discriminated: false

package.json

@@ -1,6 +1,6 @@
 {
   "name": "langfuse",
-  "version": "3.97.2",
+  "version": "3.98.2",
   "author": "engineering@langfuse.com",
   "license": "MIT",
   "private": true,

packages/shared/clickhouse/migrations/clustered/0024_dataset_run_items.down.sql

@@ -0,0 +1 @@
+DROP TABLE dataset_run_items_rmt ON CLUSTER default;

packages/shared/clickhouse/migrations/clustered/0024_dataset_run_items.up.sql

@@ -0,0 +1,36 @@
+CREATE TABLE dataset_run_items_rmt ON CLUSTER default (
+    -- primary identifiers
+    `id` String,
+    `project_id` String,
+    `dataset_run_id` String,
+    `dataset_item_id` String,
+    `dataset_id` String,
+    `trace_id` String,
+    `observation_id` Nullable(String),
+
+    -- error field
+    `error` Nullable(String),
+
+     -- timestamps
+    `created_at` DateTime64(3) DEFAULT now(),
+    `updated_at` DateTime64(3) DEFAULT now(),
+
+    -- denormalized immutable dataset run fields
+    `dataset_run_name` String,
+    `dataset_run_description` Nullable(String),
+    `dataset_run_metadata` Map(LowCardinality(String), String),
+    `dataset_run_created_at` DateTime64(3),
+
+    -- denormalized dataset item fields (mutable, but snapshots are relevant)
+    `dataset_item_input` Nullable(String) CODEC(ZSTD(3)), -- json
+    `dataset_item_expected_output` Nullable(String) CODEC(ZSTD(3)), -- json
+    `dataset_item_metadata` Map(LowCardinality(String), String),
+
+    -- clickhouse engine fields
+    `event_ts` DateTime64(3),
+    `is_deleted` UInt8,
+
+    -- For dataset item lookups
+    INDEX idx_dataset_item dataset_item_id TYPE bloom_filter(0.001) GRANULARITY 1,
+) ENGINE = ReplicatedReplacingMergeTree(event_ts, is_deleted)
+ORDER BY (project_id, dataset_id, dataset_run_id, id);

packages/shared/clickhouse/migrations/unclustered/0024_dataset_run_items.down.sql

@@ -0,0 +1 @@
+DROP TABLE dataset_run_items_rmt;

packages/shared/clickhouse/migrations/unclustered/0024_dataset_run_items.up.sql

@@ -0,0 +1,36 @@
+CREATE TABLE dataset_run_items_rmt (
+    -- primary identifiers
+    `id` String,
+    `project_id` String,
+    `dataset_run_id` String,
+    `dataset_item_id` String,
+    `dataset_id` String,
+    `trace_id` String,
+    `observation_id` Nullable(String),
+
+    -- error field
+    `error` Nullable(String),
+
+     -- timestamps
+    `created_at` DateTime64(3) DEFAULT now(),
+    `updated_at` DateTime64(3) DEFAULT now(),
+
+    -- denormalized immutable dataset run fields
+    `dataset_run_name` String,
+    `dataset_run_description` Nullable(String),
+    `dataset_run_metadata` Map(LowCardinality(String), String),
+    `dataset_run_created_at` DateTime64(3),
+
+    -- denormalized dataset item fields (mutable, but snapshots are relevant)
+    `dataset_item_input` Nullable(String) CODEC(ZSTD(3)), -- json
+    `dataset_item_expected_output` Nullable(String) CODEC(ZSTD(3)), -- json
+    `dataset_item_metadata` Map(LowCardinality(String), String),
+
+    -- clickhouse engine fields
+    `event_ts` DateTime64(3),
+    `is_deleted` UInt8,
+
+    -- For dataset item lookups
+    INDEX idx_dataset_item dataset_item_id TYPE bloom_filter(0.001) GRANULARITY 1,
+) ENGINE = ReplacingMergeTree(event_ts, is_deleted)
+ORDER BY (project_id, dataset_id, dataset_run_id, id);

packages/shared/prisma/generated/types.ts

@@ -22,6 +22,13 @@ export const LegacyPrismaObservationType = {
   SPAN: "SPAN",
   EVENT: "EVENT",
   GENERATION: "GENERATION",
+  AGENT: "AGENT",
+  TOOL: "TOOL",
+  CHAIN: "CHAIN",
+  RETRIEVER: "RETRIEVER",
+  EVALUATOR: "EVALUATOR",
+  EMBEDDING: "EMBEDDING",
+  GUARDRAIL: "GUARDRAIL",
 } as const;
 export type LegacyPrismaObservationType =
   (typeof LegacyPrismaObservationType)[keyof typeof LegacyPrismaObservationType];

packages/shared/prisma/migrations/20250814090100_remove_dataset_run_items_pg_to_ch_background_migration/migration.sql

@@ -0,0 +1 @@
+DELETE FROM background_migrations WHERE id = '8d47f91b-3e5c-4a26-9f85-c12d6e4b9a3d';

packages/shared/prisma/migrations/20250814100100_add_dataset_run_items_rmt_pg_to_ch_background_migration/migration.sql

@@ -0,0 +1,2 @@
+INSERT INTO background_migrations (id, name, script, args)
+VALUES ('9f32e84c-7b1d-4f59-a803-d67ae5c9b2e8', '20250814_1001_migrate_dataset_run_items_rmt_pg_to_ch', 'migrateDatasetRunItemsFromPostgresToClickhouseRmt', '{}');

packages/shared/prisma/schema.prisma

@@ -368,7 +368,6 @@ model LegacyPrismaObservation {
   version              String?
   createdAt            DateTime                     @default(now()) @map("created_at")
   updatedAt            DateTime                     @default(now()) @updatedAt @map("updated_at")
-  // GENERATION ONLY
   model                String? // user-provided model attribute
   internalModel        String?                      @map("internal_model") // matched model.name that is matched at ingestion time, to be deprecated
   internalModelId      String?                      @map("internal_model_id") // matched model.id that is matched at ingestion time
@@ -409,6 +408,13 @@ enum LegacyPrismaObservationType {
   SPAN
   EVENT
   GENERATION
+  AGENT
+  TOOL
+  CHAIN
+  RETRIEVER
+  EVALUATOR
+  EMBEDDING
+  GUARDRAIL
 
   @@map("ObservationType")
 }

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

@@ -95,3 +95,57 @@ export const REALISTIC_METADATA_EXAMPLES = [
   },
   { file_type: "JSON", validation: "passed", schema_version: "v1.2" },
 ];
+
+export const REALISTIC_AGENT_NAMES = [
+  "AI-Coordinator",
+  "TaskManager",
+  "WorkflowOrchestrator",
+  "PlanningAgent",
+  "MultiStepAgent",
+  "SmartCoordinator",
+];
+
+export const REALISTIC_TOOL_NAMES = [
+  "WebSearchTool",
+  "CalculatorTool",
+  "WeatherForecastTool",
+  "EmailSenderTool",
+];
+
+export const REALISTIC_CHAIN_NAMES = [
+  "DataTransformationChain",
+  "ProcessingPipeline",
+  "TransformationChain",
+  "MultiStepProcess",
+];
+
+export const REALISTIC_RETRIEVER_NAMES = [
+  "DocumentRetriever",
+  "SemanticSearch",
+  "InformationRetriever",
+  "VectorSearch",
+  "MemoryRetriever",
+];
+
+export const REALISTIC_EVALUATOR_NAMES = [
+  "QualityEvaluator",
+  "RelevanceScorer",
+  "AccuracyChecker",
+  "ResponseEvaluator",
+  "ContentScorer",
+];
+
+export const REALISTIC_EMBEDDING_NAMES = [
+  "TextEmbedding",
+  "DocumentEncoder",
+  "SemanticEncoder",
+  "VectorEmbedding",
+];
+
+export const REALISTIC_GUARDRAIL_NAMES = [
+  "SafetyChecker",
+  "ContentModerator",
+  "ToxicityFilter",
+  "JailbreakDetector",
+  "PolicyEnforcer",
+];

packages/shared/scripts/seeder/utils/data-generators.ts

@@ -4,6 +4,13 @@ import {
   REALISTIC_SPAN_NAMES,
   REALISTIC_GENERATION_NAMES,
   REALISTIC_MODELS,
+  REALISTIC_AGENT_NAMES,
+  REALISTIC_TOOL_NAMES,
+  REALISTIC_CHAIN_NAMES,
+  REALISTIC_RETRIEVER_NAMES,
+  REALISTIC_EVALUATOR_NAMES,
+  REALISTIC_EMBEDDING_NAMES,
+  REALISTIC_GUARDRAIL_NAMES,
 } from "./clickhouse-seed-constants";
 import {
   generateDatasetItemId,
@@ -365,8 +372,56 @@ export class DataGenerator {
     const observations: ObservationRecordInsertType[] = [];
 
     traces.forEach((trace, traceIndex) => {
+      if (this.randomBoolean(0.1)) {
+        const { observations: workflowObservations } =
+          this.generateComprehensiveAIWorkflowTrace(trace.id, trace.project_id);
+        observations.push(...workflowObservations);
+        return;
+      }
+
       for (let i = 0; i < observationsPerTrace; i++) {
-        const obsType = this.randomElement(["GENERATION", "SPAN", "EVENT"]);
+        const obsType = this.randomBoolean(0.8) // More "traditional" types, are more common in app
+          ? this.randomElement(["GENERATION", "SPAN", "EVENT"])
+          : this.randomElement([
+              "AGENT",
+              "TOOL",
+              "CHAIN",
+              "RETRIEVER",
+              "EVALUATOR",
+              "EMBEDDING",
+              "GUARDRAIL",
+            ]);
+
+        let observationName: string;
+        switch (obsType) {
+          case "AGENT":
+            observationName = this.randomElement(REALISTIC_AGENT_NAMES);
+            break;
+          case "TOOL":
+            observationName = this.randomElement(REALISTIC_TOOL_NAMES);
+            break;
+          case "CHAIN":
+            observationName = this.randomElement(REALISTIC_CHAIN_NAMES);
+            break;
+          case "RETRIEVER":
+            observationName = this.randomElement(REALISTIC_RETRIEVER_NAMES);
+            break;
+          case "EVALUATOR":
+            observationName = this.randomElement(REALISTIC_EVALUATOR_NAMES);
+            break;
+          case "EMBEDDING":
+            observationName = this.randomElement(REALISTIC_EMBEDDING_NAMES);
+            break;
+          case "GUARDRAIL":
+            observationName = this.randomElement(REALISTIC_GUARDRAIL_NAMES);
+            break;
+          case "GENERATION":
+            observationName = this.randomElement(REALISTIC_GENERATION_NAMES);
+            break;
+          default:
+            observationName = this.randomElement(REALISTIC_SPAN_NAMES);
+            break;
+        }
 
         const observation: ObservationRecordInsertType = createObservation({
           id: `obs-synthetic-${traceIndex}-${i}`,
@@ -375,28 +430,28 @@ export class DataGenerator {
           parent_observation_id:
             i > 0 ? `obs-synthetic-${traceIndex}-${i - 1}` : undefined,
           type: obsType as any,
-          name:
-            obsType === "GENERATION"
-              ? this.randomElement(REALISTIC_GENERATION_NAMES)
-              : this.randomElement(REALISTIC_SPAN_NAMES),
+          name: observationName,
           input:
-            obsType === "GENERATION"
+            obsType === "GENERATION" || obsType === "EMBEDDING"
               ? this.generateObservationInput()
               : undefined,
           output:
-            obsType === "GENERATION"
+            obsType === "GENERATION" ||
+            obsType === "RETRIEVER" ||
+            obsType === "EVALUATOR" ||
+            obsType === "GUARDRAIL"
               ? this.generateObservationOutput()
               : undefined,
           provided_model_name:
-            obsType === "GENERATION"
+            obsType === "GENERATION" || obsType === "EMBEDDING"
               ? this.randomElement(REALISTIC_MODELS)
               : undefined,
           model_parameters:
-            obsType === "GENERATION"
+            obsType === "GENERATION" || obsType === "EMBEDDING"
               ? JSON.stringify({ temperature: 0.7 })
               : undefined,
           usage_details:
-            obsType === "GENERATION"
+            obsType === "GENERATION" || obsType === "EMBEDDING"
               ? {
                   input: this.randomInt(20, 200),
                   output: this.randomInt(10, 100),
@@ -404,7 +459,7 @@ export class DataGenerator {
                 }
               : undefined,
           provided_usage_details:
-            obsType === "GENERATION"
+            obsType === "GENERATION" || obsType === "EMBEDDING"
               ? {
                   input: this.randomInt(20, 200),
                   output: this.randomInt(10, 100),
@@ -412,7 +467,7 @@ export class DataGenerator {
                 }
               : undefined,
           cost_details:
-            obsType === "GENERATION"
+            obsType === "GENERATION" || obsType === "EMBEDDING"
               ? {
                   input: this.randomInt(1, 10) / 100000,
                   output: this.randomInt(1, 20) / 100000,
@@ -420,7 +475,7 @@ export class DataGenerator {
                 }
               : undefined,
           provided_cost_details:
-            obsType === "GENERATION"
+            obsType === "GENERATION" || obsType === "EMBEDDING"
               ? {
                   input: this.randomInt(1, 10) / 100000,
                   output: this.randomInt(1, 20) / 100000,
@@ -500,6 +555,252 @@ export class DataGenerator {
     return scores;
   }
 
+  /**
+   * Creates a workflow trace with all possible observation types.
+   */
+  generateComprehensiveAIWorkflowTrace(
+    traceId: string,
+    projectId: string,
+  ): {
+    trace: TraceRecordInsertType;
+    observations: ObservationRecordInsertType[];
+  } {
+    // Create the main trace
+    const trace = createTrace({
+      id: traceId,
+      project_id: projectId,
+      name: "AI-Agent-Workflow",
+      input:
+        "Analyze and summarize the latest research papers on quantum computing",
+      output:
+        "Here is a comprehensive summary of quantum computing research trends with key insights and recommendations.",
+      user_id: this.randomBoolean(0.3)
+        ? `user_${this.randomInt(1, 1000)}`
+        : null,
+      session_id: this.randomBoolean(0.3)
+        ? `session_${this.randomInt(1, 100)}`
+        : undefined,
+      environment: "default",
+      metadata: { workflowType: "comprehensive-ai", purpose: "demonstration" },
+      tags: ["ai-agent", "multi-step", "comprehensive"],
+      public: true,
+      bookmarked: this.randomBoolean(0.2),
+    });
+
+    const observations: ObservationRecordInsertType[] = [];
+    const baseTime = Date.now();
+
+    // 1. AGENT - Main coordinator
+    observations.push(
+      createObservation({
+        id: `${traceId}-agent`,
+        trace_id: trace.id,
+        project_id: projectId,
+        type: "AGENT",
+        name: this.randomElement(REALISTIC_AGENT_NAMES),
+        input: "Plan and coordinate the research analysis workflow",
+        output:
+          "Workflow planned: retrieve documents → create embeddings → analyze → evaluate → check safety",
+        start_time: baseTime,
+        end_time: baseTime + 500,
+        level: "DEFAULT",
+        environment: trace.environment,
+        metadata: { role: "coordinator", step: "1" },
+      }),
+    );
+
+    // 2. RETRIEVER - Document retrieval
+    observations.push(
+      createObservation({
+        id: `${traceId}-retriever`,
+        trace_id: trace.id,
+        project_id: projectId,
+        parent_observation_id: `${traceId}-agent`,
+        type: "RETRIEVER",
+        name: this.randomElement(REALISTIC_RETRIEVER_NAMES),
+        input: "query: quantum computing research papers 2024",
+        output: "Retrieved 15 relevant research papers from arXiv and IEEE",
+        start_time: baseTime + 500,
+        end_time: baseTime + 2000,
+        level: "DEFAULT",
+        environment: trace.environment,
+        metadata: { documentsFound: "15", sources: "arXiv,IEEE" },
+      }),
+    );
+
+    // 3. EMBEDDING - Create document embeddings
+    observations.push(
+      createObservation({
+        id: `${traceId}-embedding`,
+        trace_id: trace.id,
+        project_id: projectId,
+        parent_observation_id: `${traceId}-retriever`,
+        type: "EMBEDDING",
+        name: this.randomElement(REALISTIC_EMBEDDING_NAMES),
+        input: "15 research paper abstracts and titles",
+        output: "Generated 1536-dimensional embeddings for semantic similarity",
+        start_time: baseTime + 2000,
+        end_time: baseTime + 3500,
+        level: "DEFAULT",
+        environment: trace.environment,
+        metadata: {
+          embeddingModel: "text-embedding-ada-002",
+          dimensions: "1536",
+        },
+        usage_details: {
+          input: this.randomInt(2000, 4000),
+          total: this.randomInt(2000, 4000),
+        },
+        provided_usage_details: {
+          input: this.randomInt(2000, 4000),
+          total: this.randomInt(2000, 4000),
+        },
+        cost_details: {
+          input: this.randomInt(5, 15) / 100000,
+          total: this.randomInt(5, 15) / 100000,
+        },
+        provided_cost_details: {
+          input: this.randomInt(5, 15) / 100000,
+          total: this.randomInt(5, 15) / 100000,
+        },
+      }),
+    );
+
+    // 4. CHAIN - Processing pipeline
+    observations.push(
+      createObservation({
+        id: `${traceId}-chain`,
+        trace_id: trace.id,
+        project_id: projectId,
+        parent_observation_id: `${traceId}-embedding`,
+        type: "CHAIN",
+        name: this.randomElement(REALISTIC_CHAIN_NAMES),
+        input: "Research papers with embeddings",
+        output:
+          "Processed and analyzed 15 papers through multi-step analysis chain",
+        start_time: baseTime + 3500,
+        end_time: baseTime + 8000,
+        level: "DEFAULT",
+        environment: trace.environment,
+        metadata: { steps: "4", processed: "15" },
+      }),
+    );
+
+    // 5. TOOL - External API call for additional context
+    observations.push(
+      createObservation({
+        id: `${traceId}-tool`,
+        trace_id: trace.id,
+        project_id: projectId,
+        parent_observation_id: `${traceId}-chain`,
+        type: "TOOL",
+        name: this.randomElement(REALISTIC_TOOL_NAMES),
+        input: "Search for quantum computing market trends",
+        output:
+          "Market data: $1.2B industry, 25% YoY growth, key players identified",
+        start_time: baseTime + 8000,
+        end_time: baseTime + 10000,
+        level: "DEFAULT",
+        environment: trace.environment,
+        metadata: { toolType: "api-call", endpoint: "market-research" },
+      }),
+    );
+
+    // 6. GENERATION - Final summary generation
+    observations.push(
+      createObservation({
+        id: `${traceId}-generation`,
+        trace_id: trace.id,
+        project_id: projectId,
+        parent_observation_id: `${traceId}-tool`,
+        type: "GENERATION",
+        name: this.randomElement(REALISTIC_GENERATION_NAMES),
+        input:
+          "Synthesize research analysis and market data into comprehensive summary",
+        output:
+          "Generated comprehensive 2000-word analysis of quantum computing research trends",
+        provided_model_name: this.randomElement(REALISTIC_MODELS),
+        model_parameters: JSON.stringify({
+          temperature: 0.3,
+          max_tokens: 2000,
+        }),
+        start_time: baseTime + 10000,
+        end_time: baseTime + 15000,
+        level: "DEFAULT",
+        environment: trace.environment,
+        usage_details: {
+          input: this.randomInt(1500, 2500),
+          output: this.randomInt(1800, 2200),
+          total: this.randomInt(3300, 4700),
+        },
+        provided_usage_details: {
+          input: this.randomInt(1500, 2500),
+          output: this.randomInt(1800, 2200),
+          total: this.randomInt(3300, 4700),
+        },
+        cost_details: {
+          input: this.randomInt(15, 25) / 100000,
+          output: this.randomInt(35, 45) / 100000,
+          total: this.randomInt(50, 70) / 100000,
+        },
+        provided_cost_details: {
+          input: this.randomInt(15, 25) / 100000,
+          output: this.randomInt(35, 45) / 100000,
+          total: this.randomInt(50, 70) / 100000,
+        },
+      }),
+    );
+
+    // 7. EVALUATOR - Quality evaluation
+    observations.push(
+      createObservation({
+        id: `${traceId}-evaluator`,
+        trace_id: trace.id,
+        project_id: projectId,
+        parent_observation_id: `${traceId}-generation`,
+        type: "EVALUATOR",
+        name: this.randomElement(REALISTIC_EVALUATOR_NAMES),
+        input: "Evaluate summary quality, accuracy, and completeness",
+        output: "Quality score: 8.7/10, High accuracy, Comprehensive coverage",
+        start_time: baseTime + 15000,
+        end_time: baseTime + 16500,
+        level: "DEFAULT",
+        environment: trace.environment,
+        metadata: {
+          qualityScore: "8.7",
+          accuracy: "high",
+          completeness: "comprehensive",
+        },
+      }),
+    );
+
+    // 8. GUARDRAIL - Safety and compliance check
+    observations.push(
+      createObservation({
+        id: `${traceId}-guardrail`,
+        trace_id: trace.id,
+        project_id: projectId,
+        parent_observation_id: `${traceId}-evaluator`,
+        type: "GUARDRAIL",
+        name: this.randomElement(REALISTIC_GUARDRAIL_NAMES),
+        input: "Check content for safety, bias, and compliance issues",
+        output:
+          "✓ Content approved: No safety issues, Low bias detected, Compliant",
+        start_time: baseTime + 16500,
+        end_time: baseTime + 17000,
+        level: "DEFAULT",
+        environment: trace.environment,
+        metadata: {
+          safetyCheck: "passed",
+          biasLevel: "low",
+          compliance: "approved",
+        },
+      }),
+    );
+
+    return { trace, observations };
+  }
+
   /**
    * Creates evaluation traces for testing evaluator configurations.
    * Use for: Evaluation testing, score validation, evaluator development.

packages/shared/src/domain/observations.ts

@@ -6,8 +6,27 @@ export const ObservationType = {
   SPAN: "SPAN",
   EVENT: "EVENT",
   GENERATION: "GENERATION",
+  AGENT: "AGENT",
+  TOOL: "TOOL",
+  CHAIN: "CHAIN",
+  RETRIEVER: "RETRIEVER",
+  EVALUATOR: "EVALUATOR",
+  EMBEDDING: "EMBEDDING",
+  GUARDRAIL: "GUARDRAIL",
 } as const;
-export const ObservationTypeDomain = z.enum(["SPAN", "EVENT", "GENERATION"]);
+
+export const ObservationTypeDomain = z.enum([
+  "SPAN",
+  "EVENT",
+  "GENERATION",
+  "AGENT",
+  "TOOL",
+  "CHAIN",
+  "RETRIEVER",
+  "EVALUATOR",
+  "EMBEDDING",
+  "GUARDRAIL",
+]);
 export type ObservationType = z.infer<typeof ObservationTypeDomain>;
 
 export const ObservationLevel = {
@@ -74,3 +93,29 @@ export const ObservationSchema = z.object({
 });
 
 export type Observation = z.infer<typeof ObservationSchema>;
+
+/**
+ * Returns true if an observation type is generation-like, meaning it could include LLM calls
+ * and potentially has similar input/output fields.
+ */
+export const GenerationLikeObservationTypes = [
+  ObservationType.GENERATION,
+  ObservationType.AGENT,
+  ObservationType.TOOL,
+  ObservationType.CHAIN,
+  ObservationType.RETRIEVER,
+  ObservationType.EVALUATOR,
+  ObservationType.EMBEDDING,
+  ObservationType.GUARDRAIL,
+] as const;
+
+export const isGenerationLike = (observationType: ObservationType): boolean => {
+  return GenerationLikeObservationTypes.includes(observationType as any);
+};
+
+/**
+ * Returns all generation-like observation types for use in filters and queries.
+ */
+export const getGenerationLikeTypes = (): ObservationType[] => {
+  return [...GenerationLikeObservationTypes];
+};

packages/shared/src/env.ts

@@ -158,6 +158,12 @@ const EnvSchema = z.object({
   LANGFUSE_EXPERIMENT_INSERT_INTO_AGGREGATING_MERGE_TREES: z
     .enum(["true", "false"])
     .default("false"),
+  LANGFUSE_EXPERIMENT_WHITELISTED_AMT_TABLES: z
+    .string()
+    .optional()
+    .transform((s) =>
+      s ? s.split(",").map((s) => s.toLowerCase().trim()) : [],
+    ),
   LANGFUSE_INGESTION_PROCESSING_SAMPLED_PROJECTS: z
     .string()
     .optional()

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

@@ -5,6 +5,13 @@ export const langfuseObjects = [
   "span",
   "generation",
   "event",
+  "agent",
+  "tool",
+  "chain",
+  "retriever",
+  "evaluator",
+  "embedding",
+  "guardrail",
   "dataset_item",
 ] as const;
 
@@ -80,6 +87,41 @@ export const availableTraceEvalVariables = [
     display: "Event",
     availableColumns: observationCols,
   },
+  {
+    id: "agent",
+    display: "Agent",
+    availableColumns: observationCols,
+  },
+  {
+    id: "tool",
+    display: "Tool",
+    availableColumns: observationCols,
+  },
+  {
+    id: "chain",
+    display: "Chain",
+    availableColumns: observationCols,
+  },
+  {
+    id: "retriever",
+    display: "Retriever",
+    availableColumns: observationCols,
+  },
+  {
+    id: "evaluator",
+    display: "Evaluator",
+    availableColumns: observationCols,
+  },
+  {
+    id: "embedding",
+    display: "Embedding",
+    availableColumns: observationCols,
+  },
+  {
+    id: "guardrail",
+    display: "Guardrail",
+    availableColumns: observationCols,
+  },
 ];
 
 export const availableDatasetEvalVariables = [

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

@@ -2,7 +2,7 @@ export const ClickhouseTableNames = {
   traces: "traces",
   observations: "observations",
   scores: "scores",
-  dataset_run_items: "dataset_run_items",
+  dataset_run_items_rmt: "dataset_run_items_rmt",
 
   // Virtual tables for dashboards
   // TODO: Check if we can do this more elegantly

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

@@ -27,6 +27,13 @@ export const getClickhouseEntityType = (
     case eventTypes.SPAN_UPDATE:
     case eventTypes.GENERATION_CREATE:
     case eventTypes.GENERATION_UPDATE:
+    case eventTypes.AGENT_CREATE:
+    case eventTypes.TOOL_CREATE:
+    case eventTypes.CHAIN_CREATE:
+    case eventTypes.RETRIEVER_CREATE:
+    case eventTypes.EVALUATOR_CREATE:
+    case eventTypes.EMBEDDING_CREATE:
+    case eventTypes.GUARDRAIL_CREATE:
       return "observation";
     case eventTypes.SCORE_CREATE:
       return "score";

packages/shared/src/server/data-deletion/ingestionFileDeletion.ts

@@ -88,7 +88,7 @@ async function removeIngestionEventsFromS3AndDeleteClickhouseRefs(p: {
   );
   await softDeleteInClickhouse(blobStorageRefs);
   logger.info(
-    `Deleted batch ${batch} of size ${blobStorageRefs.length} for ${projectId} of deleting s3 refs`,
+    `Deleted last batch ${batch} of size ${blobStorageRefs.length} for ${projectId} of deleting s3 refs`,
   );
 }
 

packages/shared/src/server/dataset-run-items/datasetExecution.ts

@@ -88,6 +88,7 @@ export async function executeWithDatasetRunItemsStrategy<TInput, TOutput>({
         return await postgresExecution(input);
       }
     } else {
+      // Read from PostgreSQL
       return await postgresExecution(input);
     }
   }

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

@@ -223,6 +223,13 @@ export const eventTypes = {
   SPAN_UPDATE: "span-update",
   GENERATION_CREATE: "generation-create",
   GENERATION_UPDATE: "generation-update",
+  AGENT_CREATE: "agent-create",
+  TOOL_CREATE: "tool-create",
+  CHAIN_CREATE: "chain-create",
+  RETRIEVER_CREATE: "retriever-create",
+  EVALUATOR_CREATE: "evaluator-create",
+  EMBEDDING_CREATE: "embedding-create",
+  GUARDRAIL_CREATE: "guardrail-create",
   SDK_LOG: "sdk-log",
   DATASET_RUN_ITEM_CREATE: "dataset-run-item-create",
   // LEGACY, only required for backwards compatibility
@@ -564,6 +571,41 @@ const createAllIngestionSchemas = ({
     body: UpdateGenerationBody,
   });
 
+  const agentCreateEvent = base.extend({
+    type: z.literal(eventTypes.AGENT_CREATE),
+    body: CreateGenerationBody,
+  });
+
+  const toolCreateEvent = base.extend({
+    type: z.literal(eventTypes.TOOL_CREATE),
+    body: CreateGenerationBody,
+  });
+
+  const chainCreateEvent = base.extend({
+    type: z.literal(eventTypes.CHAIN_CREATE),
+    body: CreateGenerationBody,
+  });
+
+  const retrieverCreateEvent = base.extend({
+    type: z.literal(eventTypes.RETRIEVER_CREATE),
+    body: CreateGenerationBody,
+  });
+
+  const evaluatorCreateEvent = base.extend({
+    type: z.literal(eventTypes.EVALUATOR_CREATE),
+    body: CreateGenerationBody,
+  });
+
+  const embeddingCreateEvent = base.extend({
+    type: z.literal(eventTypes.EMBEDDING_CREATE),
+    body: CreateGenerationBody,
+  });
+
+  const guardrailCreateEvent = base.extend({
+    type: z.literal(eventTypes.GUARDRAIL_CREATE),
+    body: CreateGenerationBody,
+  });
+
   const scoreEvent = base.extend({
     type: z.literal(eventTypes.SCORE_CREATE),
     body: ScoreBody,
@@ -603,6 +645,13 @@ const createAllIngestionSchemas = ({
     spanUpdateEvent,
     generationCreateEvent,
     generationUpdateEvent,
+    agentCreateEvent,
+    toolCreateEvent,
+    chainCreateEvent,
+    retrieverCreateEvent,
+    evaluatorCreateEvent,
+    embeddingCreateEvent,
+    guardrailCreateEvent,
     sdkLogEvent,
     datasetRunItemCreateEvent,
     // LEGACY, only required for backwards compatibility
@@ -629,6 +678,13 @@ const createAllIngestionSchemas = ({
     spanUpdateEvent,
     generationCreateEvent,
     generationUpdateEvent,
+    agentCreateEvent,
+    toolCreateEvent,
+    chainCreateEvent,
+    retrieverCreateEvent,
+    evaluatorCreateEvent,
+    embeddingCreateEvent,
+    guardrailCreateEvent,
     scoreEvent,
     datasetRunItemCreateEvent,
     sdkLogEvent,
@@ -666,6 +722,13 @@ export const spanCreateEvent = publicSchemas.spanCreateEvent;
 export const spanUpdateEvent = publicSchemas.spanUpdateEvent;
 export const generationCreateEvent = publicSchemas.generationCreateEvent;
 export const generationUpdateEvent = publicSchemas.generationUpdateEvent;
+export const agentCreateEvent = publicSchemas.agentCreateEvent;
+export const toolCreateEvent = publicSchemas.toolCreateEvent;
+export const chainCreateEvent = publicSchemas.chainCreateEvent;
+export const retrieverCreateEvent = publicSchemas.retrieverCreateEvent;
+export const evaluatorCreateEvent = publicSchemas.evaluatorCreateEvent;
+export const embeddingCreateEvent = publicSchemas.embeddingCreateEvent;
+export const guardrailCreateEvent = publicSchemas.guardrailCreateEvent;
 export const scoreEvent = publicSchemas.scoreEvent;
 export const sdkLogEvent = publicSchemas.sdkLogEvent;
 export const datasetRunItemCreateEvent =
@@ -707,4 +770,11 @@ export type ObservationEvent =
   | z.infer<typeof spanCreateEvent>
   | z.infer<typeof spanUpdateEvent>
   | z.infer<typeof generationCreateEvent>
-  | z.infer<typeof generationUpdateEvent>;
+  | z.infer<typeof generationUpdateEvent>
+  | z.infer<typeof agentCreateEvent>
+  | z.infer<typeof toolCreateEvent>
+  | z.infer<typeof chainCreateEvent>
+  | z.infer<typeof retrieverCreateEvent>
+  | z.infer<typeof evaluatorCreateEvent>
+  | z.infer<typeof embeddingCreateEvent>
+  | z.infer<typeof guardrailCreateEvent>;

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

@@ -242,6 +242,7 @@ export async function fetchLLMCompletion(
         timeout: 1000 * 60 * 2, // 2 minutes timeout
         ...(proxyAgent && { httpAgent: proxyAgent }),
       },
+      invocationKwargs: modelParams.providerOptions,
     });
   } else if (modelParams.adapter === LLMAdapter.OpenAI) {
     chatModel = new ChatOpenAI({
@@ -258,6 +259,7 @@ export async function fetchLLMCompletion(
         defaultHeaders: extraHeaders,
         ...(proxyAgent && { httpAgent: proxyAgent }),
       },
+      modelKwargs: modelParams.providerOptions,
       timeout: 1000 * 60 * 2, // 2 minutes timeout
     });
   } else if (modelParams.adapter === LLMAdapter.Azure) {
@@ -276,6 +278,7 @@ export async function fetchLLMCompletion(
         defaultHeaders: extraHeaders,
         ...(proxyAgent && { httpAgent: proxyAgent }),
       },
+      modelKwargs: modelParams.providerOptions,
     });
   } else if (modelParams.adapter === LLMAdapter.Bedrock) {
     const { region } = BedrockConfigSchema.parse(config);
@@ -295,6 +298,7 @@ export async function fetchLLMCompletion(
       callbacks: finalCallbacks,
       maxRetries,
       timeout: 1000 * 60 * 2, // 2 minutes timeout
+      additionalModelRequestFields: modelParams.providerOptions as any,
     });
   } else if (modelParams.adapter === LLMAdapter.VertexAI) {
     const credentials = GCPServiceAccountKeySchema.parse(JSON.parse(apiKey));
@@ -351,52 +355,6 @@ 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
-  - top_p
-  - max_tokens at all, one has to use max_completion_tokens instead
-  - temperature different than 1
-
-  Reference: https://platform.openai.com/docs/guides/reasoning/beta-limitations
-  */
-    if (
-      modelParams.model.startsWith("o1-") ||
-      modelParams.model.startsWith("o3-")
-    ) {
-      const filteredMessages = finalMessages.filter((message) => {
-        return (
-          modelParams.model.startsWith("o3-") || message._getType() !== "system"
-        );
-      });
-
-      return {
-        completion: await new ChatOpenAI({
-          openAIApiKey: apiKey,
-          modelName: modelParams.model,
-          temperature: 1,
-          maxTokens: undefined,
-          topP: undefined,
-          callbacks,
-          maxRetries,
-          modelKwargs: {
-            max_completion_tokens: modelParams.max_tokens,
-          },
-          configuration: {
-            baseURL,
-            ...(proxyAgent && { httpAgent: proxyAgent }),
-          },
-          timeout: 1000 * 60 * 2, // 2 minutes timeout
-        })
-          .pipe(new StringOutputParser())
-          .invoke(filteredMessages, runConfig),
-        processTracedEvents,
-      };
-    }
-
     if (tools && tools.length > 0) {
       const langchainTools = tools.map((tool) => ({
         type: "function",

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

@@ -22,7 +22,7 @@ export const testModelCall = async ({
   apiKey: z.infer<typeof LLMApiKeySchema>;
   prompt?: string;
   modelConfig?: ModelConfig | null;
-}): Promise<void> => {
+}) => {
   (
     await fetchLLMCompletion({
       streaming: false,

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

@@ -6,6 +6,7 @@ import {
 } from "../../interfaces/customLLMProviderConfigSchemas";
 import { TokenCountDelegate } from "../ingestion/processEventBatch";
 import { AuthHeaderValidVerificationResult } from "../auth/types";
+import { JSONObjectSchema } from "../../utils/zod";
 
 /* eslint-disable no-unused-vars */
 // disable lint as this is exported and used in web/worker
@@ -271,6 +272,7 @@ export const ZodModelConfig = z.object({
   max_tokens: z.coerce.number().optional(),
   temperature: z.coerce.number().optional(),
   top_p: z.coerce.number().optional(),
+  providerOptions: JSONObjectSchema.optional(),
 });
 
 // Experiment config

packages/shared/src/server/repositories/dataset-run-items.ts

@@ -78,13 +78,13 @@ const getProjectDatasetIdDefaultFilter = (
   return {
     datasetRunItemsFilter: new FilterList([
       new StringFilter({
-        clickhouseTable: "dataset_run_items",
+        clickhouseTable: "dataset_run_items_rmt",
         field: "project_id",
         operator: "=",
         value: projectId,
       }),
       new StringFilter({
-        clickhouseTable: "dataset_run_items",
+        clickhouseTable: "dataset_run_items_rmt",
         field: "dataset_id",
         operator: "=",
         value: datasetId,
@@ -132,13 +132,13 @@ const getDatasetRunsTableInternal = async <T>(
       WHERE o.project_id = {projectId: String}
         AND o.start_time >= (
           SELECT min(dri.dataset_run_created_at) - INTERVAL 1 DAY 
-          FROM dataset_run_items dri 
+          FROM dataset_run_items_rmt dri 
           WHERE dri.project_id = {projectId: String} 
             AND dri.dataset_id = {datasetId: String}
         )
         AND o.start_time <= (
           SELECT max(dri.dataset_run_created_at) + INTERVAL 1 DAY 
-          FROM dataset_run_items dri 
+          FROM dataset_run_items_rmt dri 
           WHERE dri.project_id = {projectId: String} 
             AND dri.dataset_id = {datasetId: String}
         )
@@ -150,7 +150,7 @@ const getDatasetRunsTableInternal = async <T>(
         dateDiff('millisecond', min(of.start_time), max(of.end_time)) as latency_ms,
         sum(of.total_cost) as total_cost
       FROM observations_filtered of
-      JOIN dataset_run_items dri ON dri.trace_id = of.trace_id 
+      JOIN dataset_run_items_rmt dri ON dri.trace_id = of.trace_id 
         AND dri.project_id = of.project_id
         AND dri.observation_id IS NULL  -- Only for trace-level dataset run items
       WHERE dri.dataset_id = {datasetId: String}
@@ -163,7 +163,7 @@ const getDatasetRunsTableInternal = async <T>(
         dri.trace_id,
         of.total_cost,
         dateDiff('millisecond', of.start_time, of.end_time) as latency_ms
-      FROM dataset_run_items dri
+      FROM dataset_run_items_rmt dri
       JOIN observations_filtered of ON dri.observation_id = of.id
         AND dri.project_id = of.project_id
         AND dri.trace_id = of.trace_id
@@ -190,7 +190,7 @@ const getDatasetRunsTableInternal = async <T>(
         THEN od.total_cost
         ELSE COALESCE(ta.total_cost, 0)
       END) as avg_total_cost
-    FROM dataset_run_items dri 
+    FROM dataset_run_items_rmt dri 
     LEFT JOIN traces_aggregated ta
       ON dri.trace_id = ta.trace_id
       AND dri.project_id = ta.project_id
@@ -317,7 +317,7 @@ const getDatasetRunItemsTableInternal = async <T>(
   const query = `
     SELECT
       ${selectString}
-    FROM dataset_run_items dri 
+    FROM dataset_run_items_rmt dri 
     WHERE ${appliedFilter.query}
     ${orderByClause}
     ${opts.select === "rows" ? "LIMIT 1 BY dri.project_id, dri.dataset_id, dri.dataset_run_id, dri.dataset_item_id" : ""}
@@ -371,7 +371,7 @@ export const deleteDatasetRunItemsByProjectId = async ({
   projectId: string;
 }) => {
   const query = `
-      DELETE FROM dataset_run_items
+      DELETE FROM dataset_run_items_rmt
       WHERE project_id = {projectId: String};
     `;
   await commandClickhouse({
@@ -399,7 +399,7 @@ export const deleteDatasetRunItemsByDatasetId = async ({
   datasetId: string;
 }) => {
   const query = `
-  DELETE FROM dataset_run_items
+  DELETE FROM dataset_run_items_rmt
   WHERE project_id = {projectId: String}
   AND dataset_id = {datasetId: String}
 `;
@@ -432,7 +432,7 @@ export const deleteDatasetRunItemsByDatasetRunIds = async ({
   datasetId: string;
 }) => {
   const query = `
-    DELETE FROM dataset_run_items
+    DELETE FROM dataset_run_items_rmt
     WHERE project_id = {projectId: String}
     AND dataset_id = {datasetId: String}
     AND dataset_run_id IN ({datasetRunIds: Array(String)})

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

@@ -297,7 +297,7 @@ export const getTraceScoresForDatasetRuns = async (
       s.* EXCEPT (metadata),
       length(mapKeys(s.metadata)) > 0 AS has_metadata,
       dri.dataset_run_id as run_id
-    FROM dataset_run_items dri 
+    FROM dataset_run_items_rmt dri 
     JOIN scores s FINAL ON dri.trace_id = s.trace_id 
       AND dri.project_id = s.project_id
     WHERE dri.project_id = {projectId: String}

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

@@ -47,12 +47,16 @@ enum TracesAMTs {
  * for <= 29 days, we use traces_30d_amt,
  * for all other cases we use traces_all_amt.
  *
+ * If LANGFUSE_EXPERIMENT_WHITELISTED_AMT_TABLES is set, we only return timeframes
+ * that are whitelisted or fallback to the traces_all_amt.
+ *
  * @param fromTimestamp
  */
 export const getTimeframesTracesAMT = (
   fromTimestamp: Date | undefined,
 ): TracesAMTs => {
   if (!fromTimestamp) {
+    // The TracesAllAMT must always be returned if there is no timestamp.
     return TracesAMTs.TracesAllAMT;
   }
 
@@ -60,12 +64,21 @@ export const getTimeframesTracesAMT = (
   const diffInDays = Math.floor(
     (now.getTime() - fromTimestamp.getTime()) / (1000 * 60 * 60 * 24),
   );
+
+  let selectedTable: TracesAMTs;
   if (diffInDays <= 6) {
-    return TracesAMTs.Traces7dAMT;
+    selectedTable = TracesAMTs.Traces7dAMT;
   } else if (diffInDays <= 29) {
-    return TracesAMTs.Traces30dAMT;
+    selectedTable = TracesAMTs.Traces30dAMT;
+  } else {
+    selectedTable = TracesAMTs.TracesAllAMT;
   }
-  return TracesAMTs.TracesAllAMT;
+
+  // Check if the selected table is whitelisted, fallback to TracesAllAMT if not
+  return env.LANGFUSE_EXPERIMENT_WHITELISTED_AMT_TABLES.length === 0 ||
+    env.LANGFUSE_EXPERIMENT_WHITELISTED_AMT_TABLES.includes(selectedTable)
+    ? selectedTable
+    : TracesAMTs.TracesAllAMT;
 };
 
 /**

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

@@ -45,7 +45,7 @@ export const createDatasetRunItemsCh = async (
   datasetRunItems: DatasetRunItemRecordInsertType[],
 ) => {
   return await clickhouseClient().insert({
-    table: "dataset_run_items",
+    table: "dataset_run_items_rmt",
     format: "JSONEachRow",
     values: datasetRunItems,
   });

packages/shared/src/server/test-utils/org-factory.ts

@@ -15,6 +15,7 @@ export function createBasicAuthHeader(
 
 export type CreateOrgProjectAndApiKeyOptions = {
   projectId?: string;
+  plan?: "Team" | "Hobby" | "Core" | "Pro" | "Enterprise";
 };
 export const createOrgProjectAndApiKey = async (
   props?: CreateOrgProjectAndApiKeyOptions,
@@ -25,7 +26,7 @@ export const createOrgProjectAndApiKey = async (
       id: v4(),
       name: v4(),
       cloudConfig: CloudConfigSchema.parse({
-        plan: "Team",
+        plan: props?.plan ?? "Team",
       }),
     },
   });

packages/shared/src/tableDefinitions/mapDatasetRunItemsTable.ts

@@ -4,25 +4,25 @@ export const datasetRunItemsTableUiColumnDefinitions: UiColumnMappings = [
   {
     uiTableName: "Dataset Run ID",
     uiTableId: "datasetRunId",
-    clickhouseTableName: "dataset_run_items",
+    clickhouseTableName: "dataset_run_items_rmt",
     clickhouseSelect: 'dri."dataset_run_id"',
   },
   {
     uiTableName: "Created At",
     uiTableId: "createdAt",
-    clickhouseTableName: "dataset_run_items",
+    clickhouseTableName: "dataset_run_items_rmt",
     clickhouseSelect: 'dri."created_at"',
   },
   {
     uiTableName: "Event Timestamp",
     uiTableId: "eventTs",
-    clickhouseTableName: "dataset_run_items",
+    clickhouseTableName: "dataset_run_items_rmt",
     clickhouseSelect: 'dri."event_ts"',
   },
   {
     uiTableName: "Dataset Item ID",
     uiTableId: "datasetItemId",
-    clickhouseTableName: "dataset_run_items",
+    clickhouseTableName: "dataset_run_items_rmt",
     clickhouseSelect: 'dri."dataset_item_id"',
   },
 ];

packages/shared/src/utils/zod.ts

@@ -1,3 +1,4 @@
+import { InputJsonValue } from "@prisma/client/runtime/library";
 import { z } from "zod/v4";
 
 // to be used for Prisma JSON type
@@ -111,3 +112,26 @@ export const validateZodSchema = <T extends z.ZodTypeAny>(
 ): z.infer<T> => {
   return schema.parse(object);
 };
+
+// JSON Schema validation
+export const JSONPrimitiveValueSchema = z.union([
+  z.string(),
+  z.number().finite(),
+  z.boolean(),
+]);
+
+export const JSONValueSchema: z.ZodType<InputJsonValue> = z.lazy(() =>
+  z.union([
+    JSONPrimitiveValueSchema,
+    z.array(JSONValueSchema),
+    z.record(z.string(), JSONValueSchema),
+  ]),
+);
+
+export const JSONObjectSchema = z.record(z.string(), JSONValueSchema);
+export const JSONArraySchema = z.array(JSONValueSchema);
+
+export type JSONPrimitiveValue = z.infer<typeof JSONPrimitiveValueSchema>;
+export type JSONValue = z.infer<typeof JSONValueSchema>;
+export type JSONObject = z.infer<typeof JSONObjectSchema>;
+export type JSONArray = z.infer<typeof JSONArraySchema>;

web/package.json

@@ -1,6 +1,6 @@
 {
   "name": "web",
-  "version": "3.97.2",
+  "version": "3.98.2",
   "private": true,
   "license": "MIT",
   "engines": {

web/public/generated/api/openapi.yml

@@ -79,6 +79,52 @@ paths:
               schema: {}
       security:
         - BasicAuth: []
+    post:
+      description: Create an annotation queue
+      operationId: annotationQueues_createQueue
+      tags:
+        - AnnotationQueues
+      parameters: []
+      responses:
+        '200':
+          description: ''
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/AnnotationQueue'
+        '400':
+          description: ''
+          content:
+            application/json:
+              schema: {}
+        '401':
+          description: ''
+          content:
+            application/json:
+              schema: {}
+        '403':
+          description: ''
+          content:
+            application/json:
+              schema: {}
+        '404':
+          description: ''
+          content:
+            application/json:
+              schema: {}
+        '405':
+          description: ''
+          content:
+            application/json:
+              schema: {}
+      security:
+        - BasicAuth: []
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/CreateAnnotationQueueRequest'
   /api/public/annotation-queues/{queueId}:
     get:
       description: Get an annotation queue by ID
@@ -3254,7 +3300,7 @@ paths:
       parameters:
         - name: filter
           in: query
-          description: Filter expression (e.g. userName eq 'value')
+          description: Filter expression (e.g. userName eq "value")
           required: false
           schema:
             type: string
@@ -4459,6 +4505,22 @@ components:
       required:
         - data
         - meta
+    CreateAnnotationQueueRequest:
+      title: CreateAnnotationQueueRequest
+      type: object
+      properties:
+        name:
+          type: string
+        description:
+          type: string
+          nullable: true
+        scoreConfigIds:
+          type: array
+          items:
+            type: string
+      required:
+        - name
+        - scoreConfigIds
     CreateAnnotationQueueItemRequest:
       title: CreateAnnotationQueueItemRequest
       type: object
@@ -5908,6 +5970,13 @@ components:
         - SPAN
         - GENERATION
         - EVENT
+        - AGENT
+        - TOOL
+        - CHAIN
+        - RETRIEVER
+        - EVALUATOR
+        - EMBEDDING
+        - GUARDRAIL
     IngestionUsage:
       title: IngestionUsage
       oneOf:

web/public/generated/postman/collection.json

@@ -78,6 +78,39 @@
           },
           "response": []
         },
+        {
+          "_type": "endpoint",
+          "name": "Create Queue",
+          "request": {
+            "description": "Create an annotation queue",
+            "url": {
+              "raw": "{{baseUrl}}/api/public/annotation-queues",
+              "host": [
+                "{{baseUrl}}"
+              ],
+              "path": [
+                "api",
+                "public",
+                "annotation-queues"
+              ],
+              "query": [],
+              "variable": []
+            },
+            "header": [],
+            "method": "POST",
+            "auth": null,
+            "body": {
+              "mode": "raw",
+              "raw": "{\n    \"name\": \"example\",\n    \"description\": \"example\",\n    \"scoreConfigIds\": [\n        \"example\"\n    ]\n}",
+              "options": {
+                "raw": {
+                  "language": "json"
+                }
+              }
+            }
+          },
+          "response": []
+        },
         {
           "_type": "endpoint",
           "name": "Get Queue",

web/src/__tests__/annotation-queues-api.servertest.ts

@@ -14,6 +14,7 @@ import {
   CreateAnnotationQueueItemResponse,
   UpdateAnnotationQueueItemResponse,
   DeleteAnnotationQueueItemResponse,
+  CreateAnnotationQueueResponse,
 } from "@/src/features/public-api/types/annotation-queues";
 import {
   AnnotationQueueObjectType,
@@ -173,6 +174,120 @@ describe("Annotation Queues API Endpoints", () => {
     });
   });
 
+  describe("POST /annotation-queues", () => {
+    it("should create a new annotation queue", async () => {
+      const scoreConfig = await prisma.scoreConfig.create({
+        data: {
+          name: "Test Score Config",
+          description: "Test Score Config Description",
+          projectId,
+          dataType: "NUMERIC",
+        },
+      });
+
+      const response = await makeZodVerifiedAPICall(
+        CreateAnnotationQueueResponse,
+        "POST",
+        "/api/public/annotation-queues",
+        {
+          name: "Test Queue",
+          description: "Test Queue Description",
+          scoreConfigIds: [scoreConfig.id],
+        },
+        auth,
+      );
+
+      expect(response.status).toBe(200);
+      expect(response.body.id).toBeDefined();
+      expect(response.body.name).toBe("Test Queue");
+      expect(response.body.description).toBe("Test Queue Description");
+      expect(response.body.scoreConfigIds).toEqual([scoreConfig.id]);
+    });
+
+    it("should return 400 if the queue name already exists", async () => {
+      const response = await makeAPICall(
+        "POST",
+        "/api/public/annotation-queues",
+        {
+          name: "Test Queue",
+          description: "Test Queue Description",
+          scoreConfigIds: [],
+        },
+        auth,
+      );
+
+      expect(response.status).toBe(400);
+    });
+
+    it("should return 400 if no score config IDs are provided", async () => {
+      const response = await makeAPICall(
+        "POST",
+        "/api/public/annotation-queues",
+        {
+          name: "No configs queue",
+          description: "Test Queue Description",
+          scoreConfigIds: [],
+        },
+        auth,
+      );
+
+      expect(response.status).toBe(400);
+    });
+
+    it("should return 400 if the score config IDs are invalid", async () => {
+      const response = await makeAPICall(
+        "POST",
+        "/api/public/annotation-queues",
+        {
+          name: "Invalid configs queue",
+          description: "Test Queue Description",
+          scoreConfigIds: ["invalid-score-config-id"],
+        },
+        auth,
+      );
+
+      expect(response.status).toBe(400);
+    });
+
+    it("should return 405 if the user is on the Hobby plan and has reached the maximum number of annotation queues", async () => {
+      const { auth: hobbyPlanAuth, projectId: hobbyProjectId } =
+        await createOrgProjectAndApiKey({
+          plan: "Hobby",
+        });
+
+      const config = await prisma.scoreConfig.create({
+        data: {
+          name: "Test Score Config",
+          description: "Test Score Config Description",
+          projectId: hobbyProjectId,
+          dataType: "NUMERIC",
+        },
+      });
+
+      await prisma.annotationQueue.create({
+        data: {
+          name: "First queue",
+          description: "First queue description",
+          scoreConfigIds: [config.id],
+          projectId: hobbyProjectId,
+        },
+      });
+
+      const response = await makeAPICall(
+        "POST",
+        "/api/public/annotation-queues",
+        {
+          name: "Hobby plan queue",
+          description: "Test Queue Description",
+          scoreConfigIds: [config.id],
+        },
+        hobbyPlanAuth,
+      );
+
+      expect(response.status).toBe(405);
+    });
+  });
+
   describe("GET /annotation-queues/:queueId", () => {
     it("should get a specific annotation queue", async () => {
       const response = await makeZodVerifiedAPICall(

web/src/__tests__/async/ingestion-api.servertest.ts

@@ -205,6 +205,162 @@ describe("/api/public/ingestion API Endpoint", () => {
         },
       },
     ],
+    [
+      "agent",
+      "AGENT",
+      {
+        id: randomUUID(),
+        type: "agent-create",
+        timestamp: new Date().toISOString(),
+        body: {
+          id: randomUUID(),
+          traceId: randomUUID(),
+          startTime: new Date().toISOString(),
+          endTime: new Date(Date.now() + 1000).toISOString(),
+          name: "AI Agent",
+          input: "Process user request",
+          output: "Request processed successfully",
+          model: "claude-3-haiku",
+          modelParameters: { temperature: 0.7, max_tokens: 1000 },
+          usage: {
+            input: 150,
+            output: 75,
+            total: 225,
+            unit: "TOKENS",
+            inputCost: 0.0015,
+            outputCost: 0.003,
+            totalCost: 0.0045,
+          },
+          usageDetails: { input: 150, output: 75, total: 225 },
+          costDetails: { input: 0.0015, output: 0.003, total: 0.0045 },
+        },
+      },
+    ],
+    [
+      "tool",
+      "TOOL",
+      {
+        id: randomUUID(),
+        type: "tool-create",
+        timestamp: new Date().toISOString(),
+        body: {
+          id: randomUUID(),
+          traceId: randomUUID(),
+          startTime: new Date().toISOString(),
+          endTime: new Date(Date.now() + 2000).toISOString(),
+          name: "Web Search Tool",
+          input: "Search for current weather",
+          output: "Weather data retrieved",
+          model: "gpt-4o-mini",
+          usage: {
+            input: 50,
+            output: 25,
+            total: 75,
+            unit: "TOKENS",
+            inputCost: 0.0001,
+            outputCost: 0.0002,
+            totalCost: 0.0003,
+          },
+        },
+      },
+    ],
+    [
+      "chain",
+      "CHAIN",
+      {
+        id: randomUUID(),
+        type: "chain-create",
+        timestamp: new Date().toISOString(),
+        body: {
+          id: randomUUID(),
+          traceId: randomUUID(),
+          startTime: new Date().toISOString(),
+          endTime: new Date(Date.now() + 3000).toISOString(),
+          name: "Processing Chain",
+          input: "Multi-step task",
+          output: "All steps completed",
+          model: "gpt-4",
+          usageDetails: { input: 800, output: 400, total: 1200 },
+          costDetails: { input: 0.024, output: 0.048, total: 0.072 },
+        },
+      },
+    ],
+    [
+      "retriever",
+      "RETRIEVER",
+      {
+        id: randomUUID(),
+        type: "retriever-create",
+        timestamp: new Date().toISOString(),
+        body: {
+          id: randomUUID(),
+          traceId: randomUUID(),
+          startTime: new Date().toISOString(),
+          endTime: new Date(Date.now() + 1500).toISOString(),
+          name: "Document Retriever",
+          input: "Query document database",
+          output: "Retrieved 5 relevant documents",
+        },
+      },
+    ],
+    [
+      "evaluator",
+      "EVALUATOR",
+      {
+        id: randomUUID(),
+        type: "evaluator-create",
+        timestamp: new Date().toISOString(),
+        body: {
+          id: randomUUID(),
+          traceId: randomUUID(),
+          startTime: new Date().toISOString(),
+          endTime: new Date(Date.now() + 800).toISOString(),
+          name: "Quality Evaluator",
+          input: "Evaluate response quality",
+          output: "Quality score: 0.85",
+        },
+      },
+    ],
+    [
+      "embedding",
+      "EMBEDDING",
+      {
+        id: randomUUID(),
+        type: "embedding-create",
+        timestamp: new Date().toISOString(),
+        body: {
+          id: randomUUID(),
+          traceId: randomUUID(),
+          startTime: new Date().toISOString(),
+          endTime: new Date(Date.now() + 500).toISOString(),
+          name: "Text Embedding",
+          input: "Text to embed",
+          output: "Embedding vector generated",
+          model: "text-embedding-ada-002",
+          usage: {
+            input: 20,
+            output: 0,
+            total: 20,
+            unit: "TOKENS",
+            totalCost: 0.00004,
+          },
+        },
+      },
+    ],
+    [
+      "guardrail",
+      "GUARDRAIL",
+      {
+        id: randomUUID(),
+        type: "guardrail-create",
+        timestamp: new Date().toISOString(),
+        body: {
+          id: randomUUID(),
+          traceId: randomUUID(),
+          startTime: new Date().toISOString(),
+        },
+      },
+    ],
   ])(
     "should create observations via the ingestion API (%s)",
     async (_name: string, type: string, entity: any) => {

web/src/__tests__/async/observation-api.servertest.ts

@@ -45,6 +45,41 @@ describe("/api/public/observations API Endpoint", () => {
         output: observation.output,
       });
     });
+
+    it.each([
+      ["AGENT", "agent-observation"],
+      ["TOOL", "tool-observation"],
+      ["CHAIN", "chain-observation"],
+      ["RETRIEVER", "retriever-observation"],
+      ["EVALUATOR", "evaluator-observation"],
+      ["EMBEDDING", "embedding-observation"],
+      ["GUARDRAIL", "guardrail-observation"],
+    ])("should GET observation with type %s", async (type, name) => {
+      const observationId = uuidv4();
+      const traceId = uuidv4();
+
+      const observation = createObservationObject({
+        id: observationId,
+        project_id: projectId,
+        trace_id: traceId,
+        type: type,
+        name: name,
+      });
+
+      await createObservationsInClickhouse([observation]);
+
+      const getEventRes = await makeZodVerifiedAPICall(
+        GetObservationV1Response,
+        "GET",
+        "/api/public/observations/" + observationId,
+      );
+      expect(getEventRes.body).toMatchObject({
+        id: observationId,
+        traceId: traceId,
+        type: type,
+        name: name,
+      });
+    });
   });
 
   describe("GET /api/public/observations", () => {

web/src/__tests__/async/observations-api.servertest.ts

@@ -66,6 +66,98 @@ describe("/api/public/observations API Endpoint", () => {
           input: "User action recorded",
           metadata: { eventType: "click", target: "submit-button" },
         }),
+        createObservation({
+          id: randomUUID(),
+          trace_id: traceId,
+          project_id: projectId,
+          name: "agent-observation",
+          type: "AGENT",
+          level: "DEFAULT",
+          start_time: timestamp.getTime() + 2000,
+          end_time: timestamp.getTime() + 2500,
+          provided_model_name: "claude-3-haiku",
+          provided_usage_details: { input: 100, output: 50, total: 150 },
+          provided_cost_details: { input: 0.001, output: 0.002, total: 0.003 },
+        }),
+        createObservation({
+          id: randomUUID(),
+          trace_id: traceId,
+          project_id: projectId,
+          name: "tool-observation",
+          type: "TOOL",
+          level: "DEFAULT",
+          start_time: timestamp.getTime() + 2500,
+          end_time: timestamp.getTime() + 3000,
+          input: "Search web for information",
+          output: "Found relevant results",
+          provided_model_name: "gpt-4o-mini",
+          provided_usage_details: { input: 200, output: 100, total: 300 },
+          provided_cost_details: { input: 0.002, output: 0.004, total: 0.006 },
+        }),
+        createObservation({
+          id: randomUUID(),
+          trace_id: traceId,
+          project_id: projectId,
+          name: "chain-observation",
+          type: "CHAIN",
+          level: "DEFAULT",
+          start_time: timestamp.getTime() + 3000,
+          end_time: timestamp.getTime() + 3500,
+          input: "Process multi-step workflow",
+          output: "Workflow completed",
+          provided_model_name: "gpt-4",
+          provided_usage_details: { input: 500, output: 300, total: 800 },
+          provided_cost_details: { input: 0.015, output: 0.03, total: 0.045 },
+        }),
+        createObservation({
+          id: randomUUID(),
+          trace_id: traceId,
+          project_id: projectId,
+          name: "retriever-observation",
+          type: "RETRIEVER",
+          level: "DEFAULT",
+          start_time: timestamp.getTime() + 3500,
+          end_time: timestamp.getTime() + 4000,
+          input: "Query document database",
+          output: "Retrieved 5 relevant documents",
+        }),
+        createObservation({
+          id: randomUUID(),
+          trace_id: traceId,
+          project_id: projectId,
+          name: "evaluator-observation",
+          type: "EVALUATOR",
+          level: "DEFAULT",
+          start_time: timestamp.getTime() + 4000,
+          input: null,
+          output: null,
+          end_time: null,
+        }),
+        createObservation({
+          id: randomUUID(),
+          trace_id: traceId,
+          project_id: projectId,
+          name: "embedding-observation",
+          type: "EMBEDDING",
+          level: "DEFAULT",
+          start_time: timestamp.getTime() + 4500,
+          end_time: timestamp.getTime() + 4750,
+          input: "Text to embed",
+          output: "Vector embedding generated",
+          provided_model_name: "text-embedding-ada-002",
+          provided_usage_details: { input: 10, output: 0, total: 10 },
+          provided_cost_details: { input: 0.0001, output: 0, total: 0.0001 },
+        }),
+        createObservation({
+          id: randomUUID(),
+          trace_id: traceId,
+          project_id: projectId,
+          name: "guardrail-observation",
+          type: "GUARDRAIL",
+          level: "DEFAULT",
+          start_time: timestamp.getTime() + 5000,
+          provided_cost_details: { input: 0.0001, output: 0, total: 0.0001 },
+        }),
       ];
 
       await createTracesCh([createdTrace]);
@@ -80,14 +172,14 @@ describe("/api/public/observations API Endpoint", () => {
       expect(response.status).toBe(200);
       expect(response.body.data).toBeDefined();
       expect(response.body.meta).toBeDefined();
-      expect(response.body.meta.totalItems).toBeGreaterThanOrEqual(3);
-      expect(response.body.data.length).toBeGreaterThanOrEqual(3);
+      expect(response.body.meta.totalItems).toBeGreaterThanOrEqual(10);
+      expect(response.body.data.length).toBeGreaterThanOrEqual(10);
 
       // Find our created observations in the response
       const createdObservations = response.body.data.filter(
         (obs) => obs.traceId === traceId,
       );
-      expect(createdObservations.length).toBe(3);
+      expect(createdObservations.length).toBe(10);
 
       // Verify data structure and content
       const generationObs = createdObservations.find(
@@ -118,6 +210,75 @@ describe("/api/public/observations API Endpoint", () => {
         eventType: "click",
         target: "submit-button",
       });
+
+      // Verify new observation types exist and have correct type
+      const agentObs = createdObservations.find((obs) => obs.type === "AGENT");
+      expect(agentObs).toBeDefined();
+      expect(agentObs?.name).toBe("agent-observation");
+
+      const toolObs = createdObservations.find((obs) => obs.type === "TOOL");
+      expect(toolObs).toBeDefined();
+      expect(toolObs?.name).toBe("tool-observation");
+
+      const chainObs = createdObservations.find((obs) => obs.type === "CHAIN");
+      expect(chainObs).toBeDefined();
+      expect(chainObs?.name).toBe("chain-observation");
+
+      const retrieverObs = createdObservations.find(
+        (obs) => obs.type === "RETRIEVER",
+      );
+      expect(retrieverObs).toBeDefined();
+      expect(retrieverObs?.name).toBe("retriever-observation");
+
+      const evaluatorObs = createdObservations.find(
+        (obs) => obs.type === "EVALUATOR",
+      );
+      expect(evaluatorObs).toBeDefined();
+      expect(evaluatorObs?.name).toBe("evaluator-observation");
+      // Test that input, output, and endTime can be null (optional fields)
+      expect(evaluatorObs?.input).toBeNull();
+      expect(evaluatorObs?.output).toBeNull();
+      expect(evaluatorObs?.endTime).toBeNull();
+
+      const embeddingObs = createdObservations.find(
+        (obs) => obs.type === "EMBEDDING",
+      );
+      expect(embeddingObs).toBeDefined();
+      expect(embeddingObs?.name).toBe("embedding-observation");
+
+      const guardrailObs = createdObservations.find(
+        (obs) => obs.type === "GUARDRAIL",
+      );
+      expect(guardrailObs).toBeDefined();
+      expect(guardrailObs?.name).toBe("guardrail-observation");
+
+      // Verify new observation types support model and cost attributes
+      // The key verification is that new observation types now have model and cost fields populated
+      // (even if with default values from the factory, proving the schema changes work)
+      expect(agentObs?.model).toBe("claude-3-haiku");
+      expect(agentObs?.input).toBe("Hello World");
+      expect(agentObs?.output).toBe("Hello John");
+      // Verify that model/cost fields are present (core functionality test)
+      expect(agentObs?.usageDetails).toBeDefined();
+      expect(agentObs?.costDetails).toBeDefined();
+
+      expect(toolObs?.model).toBe("gpt-4o-mini");
+      expect(toolObs?.input).toBe("Search web for information");
+      expect(toolObs?.output).toBe("Found relevant results");
+      expect(toolObs?.usageDetails).toBeDefined();
+      expect(toolObs?.costDetails).toBeDefined();
+
+      expect(chainObs?.model).toBe("gpt-4");
+      expect(chainObs?.input).toBe("Process multi-step workflow");
+      expect(chainObs?.output).toBe("Workflow completed");
+      expect(chainObs?.usageDetails).toBeDefined();
+      expect(chainObs?.costDetails).toBeDefined();
+
+      expect(embeddingObs?.model).toBe("text-embedding-ada-002");
+      expect(embeddingObs?.input).toBe("Text to embed");
+      expect(embeddingObs?.output).toBe("Vector embedding generated");
+      expect(embeddingObs?.usageDetails).toBeDefined();
+      expect(embeddingObs?.costDetails).toBeDefined();
     }, 20_000);
 
     it("should filter observations by level parameter", async () => {

web/src/__tests__/server/api/otel/otelMapping.servertest.ts

@@ -3304,5 +3304,48 @@ describe("OTel Resource Span Mapping", () => {
       expect(traceEvent).toBeDefined();
       expect(traceEvent.body.sessionId).toBe("session-id-123");
     });
+
+    it("should default to span-create for unknown observation type", async () => {
+      const otelSpans = [
+        {
+          resource: { attributes: [] },
+          scopeSpans: [
+            {
+              scope: { name: "test-scope" },
+              spans: [
+                {
+                  traceId: {
+                    data: [
+                      1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16,
+                    ],
+                  },
+                  spanId: { data: [1, 2, 3, 4, 5, 6, 7, 8] },
+                  name: "test-span",
+                  startTimeUnixNano: 1000000000,
+                  endTimeUnixNano: 2000000000,
+                  attributes: [
+                    {
+                      key: "langfuse.observation.type",
+                      value: { stringValue: "invalid_type" },
+                    },
+                  ],
+                  status: {},
+                },
+              ],
+            },
+          ],
+        },
+      ];
+
+      const events = await convertOtelSpanToIngestionEvent(
+        otelSpans[0],
+        new Set(),
+        publicKey,
+      );
+      const spanEvents = events.filter((e) => e.type === "span-create");
+
+      expect(spanEvents.length).toBe(1);
+      expect(spanEvents[0].body.name).toBe("test-span");
+    });
   });
 });

web/src/components/ItemBadge.tsx

@@ -14,6 +14,11 @@ import {
   TestTubeDiagonal,
   Clock,
   Bot,
+  Wrench,
+  Link,
+  Search,
+  Layers3,
+  ShieldCheck,
 } from "lucide-react";
 import { cva } from "class-variance-authority";
 import { type ObservationType } from "@langfuse/shared";
@@ -38,6 +43,12 @@ const iconMap = {
   GENERATION: Fan,
   EVENT: CircleDot,
   SPAN: MoveHorizontal,
+  AGENT: Bot,
+  TOOL: Wrench,
+  CHAIN: Link,
+  RETRIEVER: Search,
+  EMBEDDING: Layers3,
+  GUARDRAIL: ShieldCheck,
   SESSION: Clock,
   USER: User,
   QUEUE_ITEM: ClipboardPen,
@@ -57,6 +68,12 @@ const iconVariants = cva(cn("h-4 w-4"), {
       GENERATION: "text-muted-magenta",
       EVENT: "text-muted-green",
       SPAN: "text-muted-blue",
+      AGENT: "text-purple-600",
+      TOOL: "text-orange-600",
+      CHAIN: "text-indigo-600",
+      RETRIEVER: "text-teal-600",
+      EMBEDDING: "text-amber-600",
+      GUARDRAIL: "text-red-600",
       SESSION: "text-primary-accent",
       USER: "text-primary-accent",
       QUEUE_ITEM: "text-primary-accent",

web/src/components/ModelParameters/index.tsx

@@ -15,19 +15,27 @@ import { CreateLLMApiKeyDialog } from "@/src/features/public-api/components/Crea
 import useProjectIdFromURL from "@/src/hooks/useProjectIdFromURL";
 import { cn } from "@/src/utils/tailwind";
 import {
-  type LLMAdapter,
+  type JSONObject,
+  JSONObjectSchema,
+  LLMAdapter,
   type supportedModels,
   type UIModelParams,
 } from "@langfuse/shared";
-import { Settings2 } from "lucide-react";
+import { InfoIcon, Settings2 } from "lucide-react";
 import {
   Popover,
   PopoverContent,
   PopoverTrigger,
 } from "@/src/components/ui/popover";
+import {
+  Tooltip,
+  TooltipContent,
+  TooltipTrigger,
+} from "@/src/components/ui/tooltip";
 
 import { LLMApiKeyComponent } from "./LLMApiKeyComponent";
 import { FormDescription } from "@/src/components/ui/form";
+import { CodeMirrorEditor } from "../editor";
 
 export type ModelParamsContext = {
   modelParams: UIModelParams;
@@ -99,6 +107,11 @@ export const ModelParameters: React.FC<ModelParamsContext> = ({
     );
   }
 
+  const isProviderOptionsSupported = ![
+    LLMAdapter.GoogleAIStudio,
+    LLMAdapter.VertexAI,
+  ].includes(modelParams.adapter.value);
+
   // Settings button component for reuse
   const SettingsButton = (
     <Popover open={modelSettingsOpen} onOpenChange={setModelSettingsOpen}>
@@ -166,6 +179,15 @@ export const ModelParameters: React.FC<ModelParamsContext> = ({
             tooltip="An alternative to sampling with temperature, called nucleus sampling, where the model considers the results of the tokens with top_p probability mass. So 0.1 means only the tokens comprising the top 10% probability mass are considered. We generally recommend altering this or temperature but not both."
             updateModelParam={updateModelParamValue}
           />
+          {isProviderOptionsSupported ? (
+            <ProviderOptionsInput
+              value={modelParams.providerOptions.value}
+              formDisabled={formDisabled}
+              enabled={modelParams.providerOptions.enabled}
+              setModelParamEnabled={setModelParamEnabled}
+              updateModelParam={updateModelParamValue}
+            />
+          ) : null}
           <LLMApiKeyComponent {...{ projectId, modelParams }} />
         </div>
       </PopoverContent>
@@ -273,21 +295,6 @@ export const ModelParameters: React.FC<ModelParamsContext> = ({
             layout="vertical"
           />
         </div>
-
-        {modelParams.model.value?.startsWith("o1-") ? (
-          <p className="mt-1 text-xs text-dark-yellow">
-            For {modelParams.model.value}, the system message and the
-            temperature, max_tokens and top_p setting are not supported while it
-            is in beta.{" "}
-            <a
-              href="https://platform.openai.com/docs/guides/reasoning/beta-limitations"
-              target="_blank"
-              rel="noreferrer noopener"
-            >
-              More info ↗
-            </a>
-          </p>
-        ) : null}
       </div>
     </div>
   );
@@ -484,3 +491,92 @@ const ModelParamsSlider = ({
     </div>
   );
 };
+
+type ProviderOptionsInputProps = {
+  value: JSONObject | undefined;
+  updateModelParam: ModelParamsContext["updateModelParamValue"];
+  setModelParamEnabled: ModelParamsContext["setModelParamEnabled"];
+  enabled: boolean;
+  formDisabled: boolean;
+};
+const ProviderOptionsInput = ({
+  value,
+  updateModelParam,
+  setModelParamEnabled,
+  enabled,
+  formDisabled,
+}: ProviderOptionsInputProps) => {
+  const [inputValue, setInputValue] = useState<string>(
+    value ? JSON.stringify(value, null, 2) : "{}",
+  );
+  const [error, setError] = useState<string | null>(null);
+
+  return (
+    <div
+      className="space-y-3"
+      title="Additional options to pass to the invocation. Please check your provider's API reference for supported values."
+    >
+      <div className="flex flex-row">
+        <div className="flex-1 flex-row space-x-1">
+          <span
+            className={cn(
+              "text-xs font-semibold",
+              (!enabled || formDisabled) && "text-muted-foreground",
+            )}
+          >
+            Additional options
+          </span>
+          <Tooltip>
+            <TooltipTrigger>
+              <InfoIcon className="size-3 text-muted-foreground" />
+            </TooltipTrigger>
+            <TooltipContent className="max-w-[200px] p-2">
+              Additional options to pass to the invocation. Please check your
+              provider&apos;s API reference for supported values.
+            </TooltipContent>
+          </Tooltip>
+        </div>
+        <div className="flex flex-row space-x-3">
+          {setModelParamEnabled ? (
+            <Switch
+              title={`Control sending the additional options parameter`}
+              disabled={formDisabled}
+              checked={enabled}
+              onCheckedChange={(checked) => {
+                setModelParamEnabled("providerOptions", checked);
+              }}
+            />
+          ) : null}
+        </div>
+      </div>
+
+      {enabled && (
+        <div>
+          <CodeMirrorEditor
+            value={inputValue}
+            onChange={(value) => {
+              setInputValue(value);
+
+              try {
+                const parsed = JSONObjectSchema.parse(JSON.parse(value));
+                updateModelParam("providerOptions", parsed);
+                setError(null);
+              } catch {
+                setError("Invalid JSON Object");
+              }
+            }}
+            editable={enabled && !formDisabled}
+            mode="json"
+            minHeight="none"
+            lineNumbers={false}
+          />
+          {error && (
+            <span className="pt-6">
+              <p className="text-[12px] text-red-500">{error}</p>
+            </span>
+          )}
+        </div>
+      )}
+    </div>
+  );
+};

web/src/components/table/use-cases/observations.tsx

@@ -148,7 +148,16 @@ export default function ObservationsTable({
             column: "type",
             type: "stringOptions",
             operator: "any of",
-            value: ["GENERATION"],
+            value: [
+              "GENERATION",
+              "AGENT",
+              "TOOL",
+              "CHAIN",
+              "RETRIEVER",
+              "EVALUATOR",
+              "EMBEDDING",
+              "GUARDRAIL",
+            ],
           },
         ]
       : [],

web/src/components/trace/ObservationPreview.tsx

@@ -1,5 +1,9 @@
 import { PrettyJsonView } from "@/src/components/ui/PrettyJsonView";
-import { AnnotationQueueObjectType, type APIScoreV2 } from "@langfuse/shared";
+import {
+  AnnotationQueueObjectType,
+  type APIScoreV2,
+  isGenerationLike,
+} from "@langfuse/shared";
 import { Badge } from "@/src/components/ui/badge";
 import { type ObservationReturnType } from "@/src/server/api/routers/traces";
 import { api } from "@/src/utils/api";
@@ -177,14 +181,15 @@ export const ObservationPreview = ({
                     objectType={AnnotationQueueObjectType.OBSERVATION}
                   />
                 </div>
-                {observationWithInputAndOutput.data?.type === "GENERATION" && (
-                  <JumpToPlaygroundButton
-                    source="generation"
-                    generation={observationWithInputAndOutput.data}
-                    analyticsEventName="trace_detail:test_in_playground_button_click"
-                    className={cn(isTimeline ? "!hidden" : "")}
-                  />
-                )}
+                {observationWithInputAndOutput.data &&
+                  isGenerationLike(observationWithInputAndOutput.data.type) && (
+                    <JumpToPlaygroundButton
+                      source="generation"
+                      generation={observationWithInputAndOutput.data}
+                      analyticsEventName="trace_detail:test_in_playground_button_click"
+                      className={cn(isTimeline ? "!hidden" : "")}
+                    />
+                  )}
                 <CommentDrawerButton
                   projectId={preloadedObservation.projectId}
                   objectId={preloadedObservation.id}
@@ -259,7 +264,7 @@ export const ObservationPreview = ({
                       projectId={preloadedObservation.projectId}
                     />
                   ) : undefined}
-                  {preloadedObservation.type === "GENERATION" && (
+                  {isGenerationLike(preloadedObservation.type) && (
                     <BreakdownTooltip
                       details={preloadedObservation.usageDetails}
                       isCost={false}

web/src/components/trace/TracePreview.tsx

@@ -3,6 +3,7 @@ import {
   type APIScoreV2,
   type TraceDomain,
   AnnotationQueueObjectType,
+  isGenerationLike,
 } from "@langfuse/shared";
 import { AggUsageBadge } from "@/src/components/token-usage-badge";
 import { Badge } from "@/src/components/ui/badge";
@@ -102,7 +103,7 @@ export const TracePreview = ({
   const usageDetails = useMemo(
     () =>
       observations
-        .filter((o) => o.type === "GENERATION")
+        .filter((o) => isGenerationLike(o.type))
         .map((o) => o.usageDetails),
     [observations],
   );
@@ -201,7 +202,7 @@ export const TracePreview = ({
                   {totalCost && (
                     <BreakdownTooltip
                       details={observations
-                        .filter((o) => o.type === "GENERATION")
+                        .filter((o) => isGenerationLike(o.type))
                         .map((o) => o.costDetails)}
                       isCost
                     >

web/src/constants/VERSION.ts

@@ -1 +1 @@
-export const VERSION = "v3.97.2";
+export const VERSION = "v3.98.2";

web/src/features/dashboard/components/LatencyChart.tsx

@@ -1,5 +1,5 @@
 import { api } from "@/src/utils/api";
-import { type FilterState } from "@langfuse/shared";
+import { type FilterState, getGenerationLikeTypes } from "@langfuse/shared";
 import {
   extractTimeSeriesData,
   fillMissingValuesAndTransform,
@@ -69,9 +69,9 @@ export const GenerationLatencyChart = ({
       ...mapLegacyUiTableFilterToView("observations", globalFilterState),
       {
         column: "type",
-        operator: "=",
-        value: "GENERATION",
-        type: "string",
+        operator: "any of",
+        value: getGenerationLikeTypes(),
+        type: "stringOptions",
       },
       {
         column: "providedModelName",

web/src/features/dashboard/components/LatencyTables.tsx

@@ -1,7 +1,7 @@
 import { RightAlignedCell } from "@/src/features/dashboard/components/RightAlignedCell";
 import { DashboardCard } from "@/src/features/dashboard/components/cards/DashboardCard";
 import { DashboardTable } from "@/src/features/dashboard/components/cards/DashboardTable";
-import { type FilterState } from "@langfuse/shared";
+import { type FilterState, getGenerationLikeTypes } from "@langfuse/shared";
 import { api } from "@/src/utils/api";
 
 import { formatIntervalSeconds } from "@/src/utils/dates";
@@ -38,9 +38,9 @@ export const LatencyTables = ({
       ...mapLegacyUiTableFilterToView("observations", globalFilterState),
       {
         column: "type",
-        operator: "=",
-        value: "GENERATION",
-        type: "string",
+        operator: "any of",
+        value: getGenerationLikeTypes(),
+        type: "stringOptions",
       },
     ],
     timeDimension: null,

web/src/features/dashboard/components/ModelCostTable.tsx

@@ -3,7 +3,7 @@ import { RightAlignedCell } from "@/src/features/dashboard/components/RightAlign
 import { LeftAlignedCell } from "@/src/features/dashboard/components/LeftAlignedCell";
 import { DashboardCard } from "@/src/features/dashboard/components/cards/DashboardCard";
 import { DashboardTable } from "@/src/features/dashboard/components/cards/DashboardTable";
-import { type FilterState } from "@langfuse/shared";
+import { type FilterState, getGenerationLikeTypes } from "@langfuse/shared";
 import { api } from "@/src/utils/api";
 import { compactNumberFormatter } from "@/src/utils/numbers";
 import { TotalMetric } from "./TotalMetric";
@@ -40,9 +40,9 @@ export const ModelCostTable = ({
       ...mapLegacyUiTableFilterToView("observations", globalFilterState),
       {
         column: "type",
-        operator: "=",
-        value: "GENERATION",
-        type: "string",
+        operator: "any of",
+        value: getGenerationLikeTypes(),
+        type: "stringOptions",
       },
     ],
     timeDimension: null,

web/src/features/dashboard/components/ModelUsageChart.tsx

@@ -15,7 +15,7 @@ import {
   dashboardDateRangeAggregationSettings,
 } from "@/src/utils/date-range-utils";
 import { compactNumberFormatter } from "@/src/utils/numbers";
-import { type FilterState } from "@langfuse/shared";
+import { type FilterState, getGenerationLikeTypes } from "@langfuse/shared";
 import {
   ModelSelectorPopover,
   useModelSelection,
@@ -70,9 +70,9 @@ export const ModelUsageChart = ({
       ...mapLegacyUiTableFilterToView("observations", userAndEnvFilterState),
       {
         column: "type",
-        operator: "=",
-        value: "GENERATION",
-        type: "string",
+        operator: "any of",
+        value: getGenerationLikeTypes(),
+        type: "stringOptions",
       },
       {
         column: "providedModelName",
@@ -115,7 +115,12 @@ export const ModelUsageChart = ({
       ],
       filter: [
         ...globalFilterState,
-        { type: "string", column: "type", operator: "=", value: "GENERATION" },
+        {
+          type: "stringOptions",
+          column: "type",
+          operator: "any of",
+          value: getGenerationLikeTypes(),
+        },
         {
           type: "stringOptions",
           column: "model",
@@ -160,7 +165,12 @@ export const ModelUsageChart = ({
       ],
       filter: [
         ...globalFilterState,
-        { type: "string", column: "type", operator: "=", value: "GENERATION" },
+        {
+          type: "stringOptions",
+          column: "type",
+          operator: "any of",
+          value: getGenerationLikeTypes(),
+        },
         {
           type: "stringOptions",
           column: "model",

web/src/features/dashboard/components/UserChart.tsx

@@ -1,5 +1,5 @@
 import { api } from "@/src/utils/api";
-import { type FilterState } from "@langfuse/shared";
+import { type FilterState, getGenerationLikeTypes } from "@langfuse/shared";
 import { DashboardCard } from "@/src/features/dashboard/components/cards/DashboardCard";
 import { compactNumberFormatter } from "@/src/utils/numbers";
 import { TabComponent } from "@/src/features/dashboard/components/TabsComponent";
@@ -46,9 +46,9 @@ export const UserChart = ({
       ...mapLegacyUiTableFilterToView("observations", globalFilterState),
       {
         column: "type",
-        operator: "=",
-        value: "GENERATION",
-        type: "string",
+        operator: "any of",
+        value: getGenerationLikeTypes(),
+        type: "stringOptions",
       },
     ],
     timeDimension: null,

web/src/features/dashboard/components/hooks.ts

@@ -1,5 +1,5 @@
 import { type TimeSeriesChartDataPoint } from "@/src/features/dashboard/components/BaseTimeSeriesChart";
-import { type FilterState } from "@langfuse/shared";
+import { type FilterState, getGenerationLikeTypes } from "@langfuse/shared";
 import { type DatabaseRow } from "@/src/server/api/services/sqlInterface";
 import { api } from "@/src/utils/api";
 import { mapLegacyUiTableFilterToView } from "@/src/features/query";
@@ -21,9 +21,9 @@ export const getAllModels = (
           ...mapLegacyUiTableFilterToView("observations", globalFilterState),
           {
             column: "type",
-            operator: "=",
-            value: "GENERATION",
-            type: "string",
+            operator: "any of",
+            value: getGenerationLikeTypes(),
+            type: "stringOptions",
           },
         ],
         timeDimension: null,

web/src/features/datasets/server/dataset-router.ts

@@ -410,7 +410,7 @@ export const datasetRouter = createTRPCRouter({
           // Get all runs from PostgreSQL and merge with ClickHouse metrics to maintain consistent count
           const [runsWithMetrics, totalRuns, allRunsBasicInfo] =
             await Promise.all([
-              // Get runs that have metrics (only runs with dataset_run_items)
+              // Get runs that have metrics (only runs with dataset_run_items_rmt)
               getDatasetRunsTableMetricsCh({
                 projectId: queryInput.projectId,
                 datasetId: queryInput.datasetId,
@@ -420,14 +420,14 @@ export const datasetRouter = createTRPCRouter({
                     ? queryInput.page * queryInput.limit
                     : undefined,
               }),
-              // Count all runs (including those without dataset_run_items)
+              // Count all runs (including those without dataset_run_items_rmt)
               ctx.prisma.datasetRuns.count({
                 where: {
                   datasetId: queryInput.datasetId,
                   projectId: queryInput.projectId,
                 },
               }),
-              // Get basic info for all runs to ensure we return all runs, even those without dataset_run_items
+              // Get basic info for all runs to ensure we return all runs, even those without dataset_run_items_rmt
               ctx.prisma.datasetRuns.findMany({
                 where: {
                   datasetId: queryInput.datasetId,
@@ -460,7 +460,7 @@ export const datasetRouter = createTRPCRouter({
             runsWithMetrics.map((run) => [run.id, run]),
           );
 
-          // Only fetch scores for runs that have metrics (runs without dataset_run_items won't have trace scores)
+          // Only fetch scores for runs that have metrics (runs without dataset_run_items_rmt won't have trace scores)
           const runsWithMetricsIds = runsWithMetrics.map((run) => run.id);
           const [traceScores, runScores] = await Promise.all([
             runsWithMetricsIds.length > 0
@@ -483,7 +483,7 @@ export const datasetRouter = createTRPCRouter({
 
             return {
               ...run,
-              // Use ClickHouse metrics if available, otherwise use defaults for runs without dataset_run_items
+              // Use ClickHouse metrics if available, otherwise use defaults for runs without dataset_run_items_rmt
               countRunItems: metrics?.countRunItems ?? 0,
               avgTotalCost: metrics?.avgTotalCost ?? null,
               avgLatency: metrics?.avgLatency ?? null,

web/src/features/evals/components/eval-template-detail.tsx

@@ -33,9 +33,6 @@ export const EvalTemplateDetail = () => {
   const templateId = router.query.id as string;
 
   const [isEditing, setIsEditing] = useState(false);
-  const [selectedTemplate, setSelectedTemplate] = useState<EvalTemplate | null>(
-    null,
-  );
 
   // get the current template by id
   const template = api.evals.templateById.useQuery({
@@ -58,15 +55,7 @@ export const EvalTemplateDetail = () => {
     },
   );
 
-  // Set the selected template when data is loaded
-  React.useEffect(() => {
-    if (template.data && !selectedTemplate) {
-      setSelectedTemplate(template.data);
-    }
-  }, [template.data, selectedTemplate]);
-
   const handleTemplateSelect = (newTemplate: EvalTemplate) => {
-    setSelectedTemplate(newTemplate);
     // Update URL without full page reload
     router.push(
       `/project/${projectId}/evals/templates/${newTemplate.id}`,
@@ -75,13 +64,10 @@ export const EvalTemplateDetail = () => {
     );
   };
 
-  // Get the appropriate template to display
-  const displayTemplate = selectedTemplate || template.data;
-
   return (
     <Page
       headerProps={{
-        title: `${displayTemplate?.name || ""}`,
+        title: `${template.data?.name ?? ""}`,
         itemType: "EVALUATOR",
         breadcrumb: [
           {
@@ -95,7 +81,7 @@ export const EvalTemplateDetail = () => {
               projectId={projectId}
               isEditing={isEditing}
               setIsEditing={setIsEditing}
-              isCustom={!!displayTemplate?.projectId}
+              isCustom={!!template.data?.projectId}
             />
 
             {/* TODO: moved to LFE-4573 */}
@@ -114,14 +100,14 @@ export const EvalTemplateDetail = () => {
         ),
       }}
     >
-      {allTemplates.isLoading || !allTemplates.data || !displayTemplate ? (
+      {allTemplates.isLoading || !allTemplates.data || !template.data ? (
         <div className="p-3">Loading...</div>
       ) : isEditing ? (
         <div className="overflow-y-auto p-3 pt-1">
           <EvalTemplateForm
             useDialog={false}
             projectId={projectId}
-            existingEvalTemplate={displayTemplate}
+            existingEvalTemplate={template.data}
             isEditing={isEditing}
             setIsEditing={setIsEditing}
           />
@@ -132,7 +118,7 @@ export const EvalTemplateDetail = () => {
             <EvalTemplateForm
               useDialog={false}
               projectId={projectId}
-              existingEvalTemplate={displayTemplate}
+              existingEvalTemplate={template.data}
               isEditing={isEditing}
               setIsEditing={setIsEditing}
             />
@@ -150,7 +136,7 @@ export const EvalTemplateDetail = () => {
                     <div
                       key={template.id}
                       className={`flex cursor-pointer flex-col rounded-md px-2 py-1.5 hover:bg-accent ${
-                        template.id === displayTemplate.id ? "bg-accent" : ""
+                        template.id === templateId ? "bg-accent" : ""
                       }`}
                       onClick={() => handleTemplateSelect(template)}
                     >

web/src/features/evals/components/evaluator-selector.tsx

@@ -58,6 +58,12 @@ export function EvaluatorSelector({
     },
   );
 
+  // Ensure per-name arrays are sorted by createdAt ascending so last is latest
+  const sortByCreatedAt = (arr: EvalTemplate[]) =>
+    arr.sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime());
+  Object.values(groupedTemplates.custom).forEach(sortByCreatedAt);
+  Object.values(groupedTemplates.langfuse).forEach(sortByCreatedAt);
+
   // Filter templates based on search
   const filteredTemplates = {
     langfuse: Object.entries(groupedTemplates.langfuse)

web/src/features/evals/pages/default-evaluation-model.tsx

@@ -15,6 +15,7 @@ import { DeleteEvaluationModelButton } from "@/src/components/deleteButton";
 import { ManageDefaultEvalModel } from "@/src/features/evals/components/manage-default-eval-model";
 import { useState } from "react";
 import { DialogContent, DialogTrigger } from "@/src/components/ui/dialog";
+import { getFinalModelParams } from "@/src/utils/getFinalModelParams";
 import { Dialog } from "@/src/components/ui/dialog";
 import { Pencil } from "lucide-react";
 import {
@@ -80,11 +81,7 @@ export default function DefaultEvaluationModelPage() {
       provider: modelParams.provider.value,
       adapter: modelParams.adapter.value,
       model: modelParams.model.value,
-      modelParams: {
-        max_tokens: modelParams.max_tokens.value,
-        temperature: modelParams.temperature.value,
-        top_p: modelParams.top_p.value,
-      },
+      modelParams: getFinalModelParams(modelParams),
     });
   };
 

web/src/features/llm-api-key/server/router.ts

@@ -289,6 +289,7 @@ export const llmApiKeyRouter = createTRPCRouter({
               customModels: true,
               withDefaultModels: true,
               extraHeaderKeys: true,
+              config: true,
             },
             where: {
               projectId: input.projectId,

web/src/features/otel/server/OtelIngestionProcessor.ts

@@ -1,6 +1,10 @@
 import { randomUUID } from "crypto";
 
-import { ForbiddenError, ObservationLevel } from "@langfuse/shared";
+import {
+  ForbiddenError,
+  ObservationLevel,
+  ObservationTypeDomain,
+} from "@langfuse/shared";
 import {
   type TraceEventType,
   type IngestionEventType,
@@ -634,23 +638,30 @@ export class OtelIngestionProcessor {
       }),
     };
 
+    const observationType = attributes[
+      LangfuseOtelSpanAttributes.OBSERVATION_TYPE
+    ] as string;
     const isGeneration =
-      attributes[LangfuseOtelSpanAttributes.OBSERVATION_TYPE] ===
-        "generation" ||
+      observationType === "generation" ||
       Boolean(observation.model) ||
       ("openinference.span.kind" in attributes &&
         attributes["openinference.span.kind"] === "LLM");
 
-    const isEvent =
-      attributes[LangfuseOtelSpanAttributes.OBSERVATION_TYPE] === "event";
+    const isKnownObservationType =
+      observationType &&
+      ObservationTypeDomain.safeParse(observationType.toUpperCase()).success;
+
+    const getIngestionEventType = (): string => {
+      if (isGeneration) return "generation-create";
+      if (isKnownObservationType) {
+        return `${observationType.toLowerCase()}-create`;
+      }
+      return "span-create";
+    };
 
     return {
       id: randomUUID(),
-      type: isGeneration
-        ? "generation-create"
-        : isEvent
-          ? "event-create"
-          : "span-create",
+      type: getIngestionEventType(),
       timestamp: new Date().toISOString(),
       body: observation,
     } as unknown as IngestionEventType;

web/src/features/playground/page/components/JumpToPlaygroundButton.tsx

@@ -38,6 +38,7 @@ import {
   type PlaceholderMessage,
   isPlaceholder,
   PromptType,
+  isGenerationLike,
 } from "@langfuse/shared";
 import {
   LANGGRAPH_NODE_TAG,
@@ -478,7 +479,7 @@ const parseGeneration = (
   },
   modelToProviderMap: Record<string, string>,
 ): PlaygroundCache => {
-  if (generation.type !== "GENERATION") return null;
+  if (!isGenerationLike(generation.type)) return null;
 
   const isLangGraph = isLangGraphTrace(generation);
   const modelParams = parseModelParams(generation, modelToProviderMap);
@@ -611,7 +612,7 @@ function parseModelParams(
           if (!modelParams) return;
 
           modelParams[key as keyof typeof parsedParams.data] = {
-            value,
+            value: value as any,
             enabled: true,
           };
         });

web/src/features/playground/page/hooks/useModelParams.ts

@@ -213,6 +213,7 @@ function getDefaultAdapterParams(
         maxTemperature: { value: 2, enabled: false },
         max_tokens: { value: 4096, enabled: false },
         top_p: { value: 1, enabled: false },
+        providerOptions: { value: {}, enabled: false },
       };
 
     case LLMAdapter.Azure:
@@ -225,6 +226,7 @@ function getDefaultAdapterParams(
         maxTemperature: { value: 2, enabled: false },
         max_tokens: { value: 4096, enabled: false },
         top_p: { value: 1, enabled: false },
+        providerOptions: { value: {}, enabled: false },
       };
 
     // Docs: https://docs.anthropic.com/claude/reference/messages_post
@@ -238,6 +240,7 @@ function getDefaultAdapterParams(
         maxTemperature: { value: 1, enabled: false },
         max_tokens: { value: 4096, enabled: false },
         top_p: { value: 1, enabled: false },
+        providerOptions: { value: {}, enabled: false },
       };
 
     case LLMAdapter.Bedrock:
@@ -250,6 +253,7 @@ function getDefaultAdapterParams(
         maxTemperature: { value: 1, enabled: false },
         max_tokens: { value: 4096, enabled: false },
         top_p: { value: 1, enabled: false },
+        providerOptions: { value: {}, enabled: false },
       };
 
     case LLMAdapter.VertexAI:
@@ -262,6 +266,7 @@ function getDefaultAdapterParams(
         maxTemperature: { value: 2, enabled: false },
         max_tokens: { value: 4096, enabled: false },
         top_p: { value: 1, enabled: false },
+        providerOptions: { value: {}, enabled: false },
       };
 
     case LLMAdapter.GoogleAIStudio:
@@ -274,6 +279,7 @@ function getDefaultAdapterParams(
         maxTemperature: { value: 2, enabled: false },
         max_tokens: { value: 4096, enabled: false },
         top_p: { value: 1, enabled: false },
+        providerOptions: { value: {}, enabled: false },
       };
   }
 }

web/src/features/playground/server/validateChatCompletionBody.ts

@@ -4,6 +4,7 @@ import {
   LLMJSONSchema,
   LLMToolDefinitionSchema,
   ChatMessageSchema,
+  JSONObjectSchema,
 } from "@langfuse/shared";
 
 const ModelParamsSchema = z.object({
@@ -13,6 +14,7 @@ const ModelParamsSchema = z.object({
   temperature: z.number().optional(),
   max_tokens: z.number().optional(),
   top_p: z.number().optional(),
+  providerOptions: JSONObjectSchema.optional(),
 });
 
 export const ChatCompletionBodySchema = z.object({

web/src/features/public-api/server/dataset-runs.ts

@@ -14,6 +14,19 @@ const isUniqueConstraintError = (error: any): boolean => {
   );
 };
 
+/**
+ * Create or fetch a dataset run with optimistic concurrency handling.
+ *
+ * Behavior:
+ * - First tries to find an existing run by (projectId, datasetId, name).
+ * - If not found, attempts to create it.
+ * - If creation fails due to a unique constraint (likely created concurrently),
+ *   fetches and returns the existing run.
+ * - If all steps fail, throws an error.
+ *
+ * Rationale: The public API can receive many POST requests almost simultaneously,
+ * which is not concurrency-safe without this guard.
+ */
 export const createOrFetchDatasetRun = async ({
   projectId,
   datasetId,
@@ -28,7 +41,21 @@ export const createOrFetchDatasetRun = async ({
   metadata?: Json | null;
 }) => {
   try {
-    // Attempt optimistic creation
+    // Attempt to fetch existing run
+    const existingRun = await prisma.datasetRuns.findUnique({
+      where: {
+        datasetId_projectId_name: {
+          datasetId,
+          projectId,
+          name: name,
+        },
+      },
+    });
+    if (existingRun) {
+      return existingRun;
+    }
+
+    // Attempt creation
     const datasetRun = await prisma.datasetRuns.create({
       data: {
         id: v4(),

web/src/features/public-api/server/traces.ts

@@ -84,10 +84,9 @@ export const generateTracesForPublicApi = async ({
       SELECT
         trace_id,
         project_id,
-        sum(total_cost) as total_cost,
-        date_diff('millisecond', least(min(start_time), min(end_time)), greatest(max(start_time), max(end_time))) as latency_milliseconds,
-        groupArray(id) as observation_ids
-      FROM observations FINAL
+         ${includeMetrics ? "sum(total_cost) as total_cost, date_diff('millisecond', least(min(start_time), min(end_time)), greatest(max(start_time), max(end_time))) as latency_milliseconds, " : ""}
+        groupUniqArray(id) as observation_ids
+      FROM observations ${includeMetrics ? "FINAL" : ""}
       WHERE project_id = {projectId: String}
       ${timeFilter ? `AND start_time >= {cteTimeFilter: DateTime64(3)} - ${TRACE_TO_OBSERVATIONS_INTERVAL}` : ""}
       ${environmentFilter.length() > 0 ? `AND ${appliedEnvironmentFilter.query}` : ""}

web/src/features/public-api/types/annotation-queues.ts

@@ -57,6 +57,17 @@ export const GetAnnotationQueuesResponse = z
   })
   .strict();
 
+// POST /annotation-queues
+export const CreateAnnotationQueueBody = z
+  .object({
+    name: z.string(),
+    description: z.string().nullable(),
+    scoreConfigIds: z.array(z.string()).min(1),
+  })
+  .strict();
+
+export const CreateAnnotationQueueResponse = AnnotationQueueSchema;
+
 // GET /annotation-queues/:queueId
 export const GetAnnotationQueueByIdQuery = z
   .object({

web/src/features/public-api/types/observations.ts

@@ -16,7 +16,18 @@ import { z } from "zod/v4";
  * Objects
  */
 
-const ObservationType = z.enum(["GENERATION", "SPAN", "EVENT"]);
+const ObservationType = z.enum([
+  "GENERATION",
+  "SPAN",
+  "EVENT",
+  "AGENT",
+  "TOOL",
+  "CHAIN",
+  "RETRIEVER",
+  "EVALUATOR",
+  "EMBEDDING",
+  "GUARDRAIL",
+]);
 
 export const APIObservation = z
   .object({
@@ -38,7 +49,6 @@ export const APIObservation = z
     level: z.enum(["DEBUG", "DEFAULT", "WARNING", "ERROR"]),
     statusMessage: z.string().nullable(),
 
-    // GENERATION only
     model: z.string().nullable(),
     modelParameters: z.any(),
     completionStartTime: z.coerce.date().nullable(),

web/src/features/tag/components/TagInput.tsx

@@ -1,50 +1,79 @@
 import React from "react";
 import { cn } from "@/src/utils/tailwind";
+import { X } from "lucide-react";
 import { Button } from "@/src/components/ui/button";
 import { Command as CommandPrimitive } from "cmdk";
+import { usePostHogClientCapture } from "@/src/features/posthog-analytics/usePostHogClientCapture";
 
 type TagInputProps = React.ComponentPropsWithoutRef<
   typeof CommandPrimitive.Input
 > & {
   selectedTags: string[];
   setSelectedTags?: (tags: string[]) => void;
+  allowTagRemoval?: boolean;
 };
 
 export const TagInput = React.forwardRef<
   React.ElementRef<typeof CommandPrimitive.Input>,
   TagInputProps
->(({ className, selectedTags, ...props }, ref) => {
-  return (
-    <div
-      className="flex flex-wrap items-center overflow-auto rounded-lg border px-2"
-      cmdk-input-wrapper=""
-    >
-      {selectedTags.length > 0 && (
-        <div className="flex flex-wrap items-center gap-x-2 gap-y-1 pt-2">
-          {selectedTags.map((tag: string) => (
-            <Button
-              key={tag}
-              variant="tertiary"
-              size="icon-sm"
-              disabled
-              className="cursor-default"
-            >
-              {tag}
-            </Button>
-          ))}
-        </div>
-      )}
-      <CommandPrimitive.Input
-        ref={ref}
-        className={cn(
-          "placeholder:muted-foreground flex h-8 w-full rounded-md border-transparent bg-transparent px-1 text-sm outline-none focus:border-0 focus:border-none focus:border-transparent focus:ring-0 disabled:cursor-not-allowed disabled:opacity-50",
-          className,
+>(
+  (
+    {
+      className,
+      selectedTags,
+      setSelectedTags,
+      allowTagRemoval = false,
+      ...props
+    },
+    ref,
+  ) => {
+    const capture = usePostHogClientCapture();
+
+    const removeTag = (tagToRemove: string) => {
+      if (setSelectedTags && allowTagRemoval) {
+        setSelectedTags(selectedTags.filter((t) => t !== tagToRemove));
+        capture("tag:remove_tag", {
+          name: tagToRemove,
+        });
+      }
+    };
+
+    return (
+      <div
+        className="flex flex-wrap items-center overflow-auto rounded-lg border px-2"
+        cmdk-input-wrapper=""
+      >
+        {selectedTags.length > 0 && (
+          <div className="flex flex-wrap items-center gap-x-2 gap-y-1 pt-2">
+            {selectedTags.map((tag: string) => (
+              <Button
+                key={tag}
+                variant="tertiary"
+                size="icon-sm"
+                disabled={!allowTagRemoval}
+                className={
+                  allowTagRemoval ? "cursor-pointer" : "cursor-default"
+                }
+                onClick={allowTagRemoval ? () => removeTag(tag) : undefined}
+              >
+                {tag}
+                {allowTagRemoval && <X className="ml-1 h-3 w-3" />}
+              </Button>
+            ))}
+          </div>
         )}
-        autoFocus
-        {...props}
-      />
-    </div>
-  );
-});
+        <CommandPrimitive.Input
+          ref={ref}
+          className={cn(
+            "placeholder:muted-foreground flex h-8 w-full rounded-md border-transparent bg-transparent px-1 text-sm outline-none focus:border-0 focus:border-none focus:border-transparent focus:ring-0 disabled:cursor-not-allowed disabled:opacity-50",
+            className,
+          )}
+          autoFocus
+          {...props}
+        />
+      </div>
+    );
+  },
+);
 
 TagInput.displayName = CommandPrimitive.Input.displayName;

web/src/features/tag/components/TagManager.tsx

@@ -22,6 +22,7 @@ type TagManagerProps = {
   mutateTags: (value: string[]) => void;
   className?: string;
   isTableCell?: boolean;
+  allowTagRemoval?: boolean;
 };
 
 const TagManager = ({
@@ -33,6 +34,7 @@ const TagManager = ({
   mutateTags,
   className,
   isTableCell = false,
+  allowTagRemoval = true,
 }: TagManagerProps) => {
   const {
     selectedTags,
@@ -114,6 +116,7 @@ const TagManager = ({
             onValueChange={setInputValue}
             selectedTags={selectedTags}
             setSelectedTags={setSelectedTags}
+            allowTagRemoval={allowTagRemoval}
           />
           <CommandList>
             <CommandGroup

web/src/features/tag/components/TagPromptDetailsPopover.tsx

@@ -2,7 +2,7 @@ import React, { useState } from "react";
 import { api } from "@/src/utils/api";
 import { useHasProjectAccess } from "@/src/features/rbac/utils/checkProjectAccess";
 import { type RouterOutput } from "@/src/utils/types";
-import TagManager from "@/src/features/tag/components/TagMananger";
+import TagManager from "@/src/features/tag/components/TagManager";
 import { trpcErrorToast } from "@/src/utils/trpcErrorToast";
 
 type TagPromptDetailsPopoverProps = {

web/src/features/tag/components/TagPromptPopover.tsx

@@ -2,7 +2,7 @@ import React, { useState } from "react";
 import { api } from "@/src/utils/api";
 import { useHasProjectAccess } from "@/src/features/rbac/utils/checkProjectAccess";
 import { type RouterOutput, type RouterInput } from "@/src/utils/types";
-import TagManager from "@/src/features/tag/components/TagMananger";
+import TagManager from "@/src/features/tag/components/TagManager";
 import { trpcErrorToast } from "@/src/utils/trpcErrorToast";
 
 type TagPromptPopverProps = {

web/src/features/tag/components/TagTraceDetailsPopover.tsx

@@ -2,7 +2,7 @@ import React, { useState } from "react";
 import { api } from "@/src/utils/api";
 import { useHasProjectAccess } from "@/src/features/rbac/utils/checkProjectAccess";
 import { type RouterOutput } from "@/src/utils/types";
-import TagManager from "@/src/features/tag/components/TagMananger";
+import TagManager from "@/src/features/tag/components/TagManager";
 import { trpcErrorToast } from "@/src/utils/trpcErrorToast";
 
 type TagTraceDetailsPopoverProps = {
@@ -85,6 +85,7 @@ export function TagTraceDetailsPopover({
       isLoading={isLoading}
       mutateTags={mutateTags}
       className={className}
+      allowTagRemoval={false}
     />
   );
 }

web/src/features/tag/components/TagTracePopver.tsx

@@ -2,7 +2,7 @@ import React, { useState } from "react";
 import { api } from "@/src/utils/api";
 import { useHasProjectAccess } from "@/src/features/rbac/utils/checkProjectAccess";
 import { type RouterOutput, type RouterInput } from "@/src/utils/types";
-import TagManager from "@/src/features/tag/components/TagMananger";
+import TagManager from "@/src/features/tag/components/TagManager";
 import { trpcErrorToast } from "@/src/utils/trpcErrorToast";
 
 type TagTracePopoverProps = {
@@ -75,6 +75,7 @@ export function TagTracePopover({
       mutateTags={mutateTags}
       className={className}
       isTableCell
+      allowTagRemoval={false}
     />
   );
 }

web/src/pages/api/admin/bullmq/index.ts

@@ -56,13 +56,6 @@ export default async function handler(
       return;
     }
 
-    const body = ManageBullBody.safeParse(req.body);
-
-    if (!body.success) {
-      res.status(400).json({ error: body.error });
-      return;
-    }
-
     if (req.method === "GET") {
       const queues: string[] = Object.values(QueueName);
       queues.push(...IngestionQueue.getShardNames());
@@ -94,6 +87,13 @@ export default async function handler(
       return res.status(200).json(queueCounts);
     }
 
+    const body = ManageBullBody.safeParse(req.body);
+
+    if (!body.success) {
+      res.status(400).json({ error: body.error });
+      return;
+    }
+
     if (req.method === "POST" && body.data.action === "remove") {
       logger.info(
         `Removing jobs for queues ${body.data.queueNames.join(", ")}`,

web/src/pages/api/public/annotation-queues/index.ts

@@ -2,12 +2,14 @@ import { prisma } from "@langfuse/shared/src/db";
 import { withMiddlewares } from "@/src/features/public-api/server/withMiddlewares";
 import { createAuthedProjectAPIRoute } from "@/src/features/public-api/server/createAuthedProjectAPIRoute";
 import {
+  CreateAnnotationQueueBody,
+  CreateAnnotationQueueResponse,
   GetAnnotationQueuesQuery,
   GetAnnotationQueuesResponse,
 } from "@/src/features/public-api/types/annotation-queues";
+import { InvalidRequestError, MethodNotAllowedError } from "@langfuse/shared";
 
 export default withMiddlewares({
-  // NOTE: Post API requires entitlement check
   GET: createAuthedProjectAPIRoute({
     name: "Get annotation queues",
     querySchema: GetAnnotationQueuesQuery,
@@ -54,4 +56,72 @@ export default withMiddlewares({
       };
     },
   }),
+
+  POST: createAuthedProjectAPIRoute({
+    name: "Create annotation queue",
+    bodySchema: CreateAnnotationQueueBody,
+    responseSchema: CreateAnnotationQueueResponse,
+    fn: async ({ body, auth }) => {
+      // entitlement check
+      if (auth.scope.plan === "cloud:hobby") {
+        if (
+          (await prisma.annotationQueue.count({
+            where: {
+              projectId: auth.scope.projectId,
+            },
+          })) >= 1
+        ) {
+          throw new MethodNotAllowedError(
+            "Maximum number of annotation queues reached on Hobby plan.",
+          );
+        }
+      }
+
+      const existingQueue = await prisma.annotationQueue.findFirst({
+        where: {
+          projectId: auth.scope.projectId,
+          name: body.name,
+        },
+      });
+
+      if (existingQueue) {
+        throw new InvalidRequestError("A queue with this name already exists.");
+      }
+
+      // verify the score configs exist
+      const scoreConfigs = await prisma.scoreConfig.findMany({
+        where: {
+          id: { in: body.scoreConfigIds },
+          projectId: auth.scope.projectId,
+        },
+        select: {
+          id: true,
+        },
+      });
+      const scoreConfigIdSet = new Set(scoreConfigs.map((config) => config.id));
+      if (body.scoreConfigIds.some((id) => !scoreConfigIdSet.has(id))) {
+        throw new InvalidRequestError(
+          "At least one of the score config IDs cannot be found for the given project.",
+        );
+      }
+
+      const queue = await prisma.annotationQueue.create({
+        data: {
+          projectId: auth.scope.projectId,
+          name: body.name,
+          description: body.description,
+          scoreConfigIds: body.scoreConfigIds,
+        },
+      });
+
+      return {
+        id: queue.id,
+        name: queue.name,
+        description: queue.description,
+        scoreConfigIds: queue.scoreConfigIds,
+        createdAt: queue.createdAt,
+        updatedAt: queue.updatedAt,
+      };
+    },
+  }),
 });

web/src/server/api/routers/generations/filterOptionsQuery.ts

@@ -160,7 +160,18 @@ export const filterOptionsQuery = protectedProjectProcedure
         .map((i) => ({
           value: i.tag as string,
         })),
-      type: ["GENERATION", "SPAN", "EVENT"].map((i) => ({
+      type: [
+        "GENERATION",
+        "SPAN",
+        "EVENT",
+        "AGENT",
+        "TOOL",
+        "CHAIN",
+        "RETRIEVER",
+        "EVALUATOR",
+        "EMBEDDING",
+        "GUARDRAIL",
+      ].map((i) => ({
         value: i,
       })),
     };

web/src/utils/types.ts

@@ -34,6 +34,34 @@ export type Generation = Observation & {
   };
 };
 
+export type Agent = Observation & {
+  type: "AGENT";
+};
+
+export type Tool = Observation & {
+  type: "TOOL";
+};
+
+export type Chain = Observation & {
+  type: "CHAIN";
+};
+
+export type Retriever = Observation & {
+  type: "RETRIEVER";
+};
+
+export type Evaluator = Observation & {
+  type: "EVALUATOR";
+};
+
+export type Embedding = Observation & {
+  type: "EMBEDDING";
+};
+
+export type Guardrail = Observation & {
+  type: "GUARDRAIL";
+};
+
 export type RouterInput = inferRouterInputs<AppRouter>;
 export type RouterOutput = inferRouterOutputs<AppRouter>;
 

worker/package.json

@@ -1,6 +1,6 @@
 {
   "name": "worker",
-  "version": "3.97.2",
+  "version": "3.98.2",
   "description": "",
   "license": "MIT",
   "private": true,

worker/src/backgroundMigrations/migrateDatasetRunItemsFromPostgresToClickhouse.ts

@@ -1,26 +1,18 @@
 import { IBackgroundMigration } from "./IBackgroundMigration";
-import {
-  clickhouseClient,
-  convertPostgresDatasetRunItemToInsert,
-  logger,
-} from "@langfuse/shared/src/server";
-import { parseArgs } from "node:util";
-import { prisma, Prisma } from "@langfuse/shared/src/db";
+import { logger } from "@langfuse/shared/src/server";
 import { env } from "../env";
 
 // This is hard-coded in our migrations and uniquely identifies the row in background_migrations table
-const backgroundMigrationId = "8d47f91b-3e5c-4a26-9f85-c12d6e4b9a3d";
+// In this case it is not used as we will skip this migration and run the RMT migration instead
+// const backgroundMigrationId = "8d47f91b-3e5c-4a26-9f85-c12d6e4b9a3d";
 
 export default class MigrateDatasetRunItemsFromPostgresToClickhouse
   implements IBackgroundMigration
 {
-  private isAborted = false;
-  private isFinished = false;
-
-  async validate(
-    args: Record<string, unknown>,
-    attempts = 5,
-  ): Promise<{ valid: boolean; invalidReason: string | undefined }> {
+  async validate(): Promise<{
+    valid: boolean;
+    invalidReason: string | undefined;
+  }> {
     // Check if Clickhouse credentials are configured
     if (
       !env.CLICKHOUSE_URL ||
@@ -34,157 +26,13 @@ export default class MigrateDatasetRunItemsFromPostgresToClickhouse
       };
     }
 
-    // Check if ClickHouse dataset_run_items table exists
-    const tables = await clickhouseClient().query({
-      query: "SHOW TABLES",
-    });
-    const tableNames = (await tables.json()).data as { name: string }[];
-    if (!tableNames.some((r) => r.name === "dataset_run_items")) {
-      // Retry if the table does not exist as this may mean migrations are still pending
-      if (attempts > 0) {
-        logger.info(
-          `ClickHouse dataset_run_items table does not exist. Retrying in 10s...`,
-        );
-        return new Promise((resolve) => {
-          setTimeout(() => resolve(this.validate(args, attempts - 1)), 10_000);
-        });
-      }
-
-      // If all retries are exhausted, return as invalid
-      return {
-        valid: false,
-        invalidReason: "ClickHouse dataset_run_items table does not exist",
-      };
-    }
-
+    // Return true as we will skip this migration and run the RMT migration instead
     return { valid: true, invalidReason: undefined };
   }
 
-  async run(args: Record<string, unknown>): Promise<void> {
-    const start = Date.now();
+  async run(): Promise<void> {
     logger.info(
-      `Migrating dataset_run_items from postgres to clickhouse with ${JSON.stringify(args)}`,
-    );
-
-    // @ts-ignore
-    const initialMigrationState: { state: { maxDate: string | undefined } } =
-      await prisma.backgroundMigration.findUniqueOrThrow({
-        where: { id: backgroundMigrationId },
-        select: { state: true },
-      });
-
-    const maxRowsToProcess = Number(args.maxRowsToProcess ?? Infinity);
-    const batchSize = Number(args.batchSize ?? 1000);
-    const maxDate = initialMigrationState.state?.maxDate
-      ? new Date(initialMigrationState.state.maxDate)
-      : new Date((args.maxDate as string) ?? new Date());
-
-    await prisma.backgroundMigration.update({
-      where: { id: backgroundMigrationId },
-      data: { state: { maxDate } },
-    });
-
-    let processedRows = 0;
-    while (
-      !this.isAborted &&
-      !this.isFinished &&
-      processedRows < maxRowsToProcess
-    ) {
-      const fetchStart = Date.now();
-
-      // @ts-ignore
-      const migrationState: { state: { maxDate: string } } =
-        await prisma.backgroundMigration.findUniqueOrThrow({
-          where: { id: backgroundMigrationId },
-          select: { state: true },
-        });
-
-      const datasetRunItems = await prisma.$queryRaw<
-        Array<Record<string, any>>
-      >(Prisma.sql`
-        SELECT 
-          dri.id as id,
-          dri.project_id as project_id,
-          dri.dataset_run_id as dataset_run_id,
-          dri.dataset_item_id as dataset_item_id, 
-          dri.trace_id as trace_id,
-          dri.observation_id as observation_id,
-          dri.created_at as created_at,
-          dri.updated_at as updated_at,
-          
-          -- Denormalized dataset run fields
-          dr.name as dataset_run_name,
-          dr.description as dataset_run_description,
-          dr.metadata as dataset_run_metadata,
-          dr.created_at as dataset_run_created_at,
-          
-          -- Denormalized dataset item fields  
-          di.input as dataset_item_input,
-          di.expected_output as dataset_item_expected_output,
-          di.metadata as dataset_item_metadata,
-          
-          -- Dataset ID
-          d.id as dataset_id
-
-        FROM dataset_run_items dri
-        JOIN dataset_runs dr ON dri.dataset_run_id = dr.id
-        JOIN dataset_items di ON dri.dataset_item_id = di.id  
-        JOIN datasets d ON di.dataset_id = d.id
-        WHERE dri.created_at <= ${new Date(migrationState.state.maxDate)}
-        ORDER BY dri.created_at DESC
-        LIMIT ${batchSize};
-      `);
-      if (datasetRunItems.length === 0) {
-        logger.info("No more dataset_run_items to migrate. Exiting...");
-        break;
-      }
-
-      logger.info(
-        `Got ${datasetRunItems.length} records from Postgres in ${Date.now() - fetchStart}ms`,
-      );
-
-      const insertStart = Date.now();
-      await clickhouseClient().insert({
-        table: "dataset_run_items",
-        values: datasetRunItems.map(convertPostgresDatasetRunItemToInsert),
-        format: "JSONEachRow",
-      });
-
-      logger.info(
-        `Inserted ${datasetRunItems.length} dataset_run_items into Clickhouse in ${Date.now() - insertStart}ms`,
-      );
-
-      await prisma.backgroundMigration.update({
-        where: { id: backgroundMigrationId },
-        data: {
-          state: {
-            maxDate: new Date(
-              datasetRunItems[datasetRunItems.length - 1].created_at,
-            ),
-          },
-        },
-      });
-
-      if (datasetRunItems.length < batchSize) {
-        logger.info("No more dataset_run_items to migrate. Exiting...");
-        this.isFinished = true;
-      }
-
-      processedRows += datasetRunItems.length;
-      logger.info(
-        `Processed batch in ${Date.now() - fetchStart}ms. Oldest record in batch: ${new Date(datasetRunItems[datasetRunItems.length - 1].created_at).toISOString()}`,
-      );
-    }
-
-    if (this.isAborted) {
-      logger.info(
-        `Migration of traces from Postgres to Clickhouse aborted after processing ${processedRows} rows. Skipping cleanup.`,
-      );
-      return;
-    }
-
-    logger.info(
-      `Finished migration of traces from Postgres to Clickhouse in ${Date.now() - start}ms`,
+      `Migration of dataset run items from postgres to clickhouse skipped as we will run the RMT migration instead`,
     );
   }
 
@@ -192,26 +40,13 @@ export default class MigrateDatasetRunItemsFromPostgresToClickhouse
     logger.info(
       `Aborting migration of dataset run items from Postgres to clickhouse`,
     );
-    this.isAborted = true;
   }
 }
 
 async function main() {
-  const args = parseArgs({
-    options: {
-      batchSize: { type: "string", short: "b", default: "1000" },
-      maxRowsToProcess: { type: "string", short: "r", default: "Infinity" },
-      maxDate: {
-        type: "string",
-        short: "d",
-        default: new Date().toISOString(),
-      },
-    },
-  });
-
   const migration = new MigrateDatasetRunItemsFromPostgresToClickhouse();
-  await migration.validate(args.values);
-  await migration.run(args.values);
+  await migration.validate();
+  await migration.run();
 }
 
 // If the script is being executed directly (not imported), run the main function

worker/src/backgroundMigrations/migrateDatasetRunItemsFromPostgresToClickhouseRmt.ts

@@ -0,0 +1,227 @@
+import { IBackgroundMigration } from "./IBackgroundMigration";
+import {
+  clickhouseClient,
+  convertPostgresDatasetRunItemToInsert,
+  logger,
+} from "@langfuse/shared/src/server";
+import { parseArgs } from "node:util";
+import { prisma, Prisma } from "@langfuse/shared/src/db";
+import { env } from "../env";
+
+// This is hard-coded in our migrations and uniquely identifies the row in background_migrations table
+const backgroundMigrationId = "9f32e84c-7b1d-4f59-a803-d67ae5c9b2e8";
+
+export default class MigrateDatasetRunItemsFromPostgresToClickhouseRmt
+  implements IBackgroundMigration
+{
+  private isAborted = false;
+  private isFinished = false;
+
+  async validate(
+    args: Record<string, unknown>,
+    attempts = 5,
+  ): Promise<{ valid: boolean; invalidReason: string | undefined }> {
+    // Check if Clickhouse credentials are configured
+    if (
+      !env.CLICKHOUSE_URL ||
+      !env.CLICKHOUSE_USER ||
+      !env.CLICKHOUSE_PASSWORD
+    ) {
+      return {
+        valid: false,
+        invalidReason:
+          "Clickhouse credentials must be configured to perform migration",
+      };
+    }
+
+    // Check if ClickHouse dataset_run_items_rmt table exists
+    const tables = await clickhouseClient().query({
+      query: "SHOW TABLES",
+    });
+    const tableNames = (await tables.json()).data as { name: string }[];
+    if (!tableNames.some((r) => r.name === "dataset_run_items_rmt")) {
+      // Retry if the table does not exist as this may mean migrations are still pending
+      if (attempts > 0) {
+        logger.info(
+          `ClickHouse dataset_run_items_rmt table does not exist. Retrying in 10s...`,
+        );
+        return new Promise((resolve) => {
+          setTimeout(() => resolve(this.validate(args, attempts - 1)), 10_000);
+        });
+      }
+
+      // If all retries are exhausted, return as invalid
+      return {
+        valid: false,
+        invalidReason: "ClickHouse dataset_run_items_rmt table does not exist",
+      };
+    }
+
+    return { valid: true, invalidReason: undefined };
+  }
+
+  async run(args: Record<string, unknown>): Promise<void> {
+    const start = Date.now();
+    logger.info(
+      `Migrating dataset run items from postgres to clickhouse with ${JSON.stringify(args)}`,
+    );
+
+    // @ts-ignore
+    const initialMigrationState: { state: { maxDate: string | undefined } } =
+      await prisma.backgroundMigration.findUniqueOrThrow({
+        where: { id: backgroundMigrationId },
+        select: { state: true },
+      });
+
+    const maxRowsToProcess = Number(args.maxRowsToProcess ?? Infinity);
+    const batchSize = Number(args.batchSize ?? 1000);
+    const maxDate = initialMigrationState.state?.maxDate
+      ? new Date(initialMigrationState.state.maxDate)
+      : new Date((args.maxDate as string) ?? new Date());
+
+    await prisma.backgroundMigration.update({
+      where: { id: backgroundMigrationId },
+      data: { state: { maxDate } },
+    });
+
+    let processedRows = 0;
+    while (
+      !this.isAborted &&
+      !this.isFinished &&
+      processedRows < maxRowsToProcess
+    ) {
+      const fetchStart = Date.now();
+
+      // @ts-ignore
+      const migrationState: { state: { maxDate: string } } =
+        await prisma.backgroundMigration.findUniqueOrThrow({
+          where: { id: backgroundMigrationId },
+          select: { state: true },
+        });
+
+      const datasetRunItems = await prisma.$queryRaw<
+        Array<Record<string, any>>
+      >(Prisma.sql`
+        SELECT 
+          dri.id as id,
+          dri.project_id as project_id,
+          dri.dataset_run_id as dataset_run_id,
+          dri.dataset_item_id as dataset_item_id, 
+          dri.trace_id as trace_id,
+          dri.observation_id as observation_id,
+          dri.created_at as created_at,
+          dri.updated_at as updated_at,
+          
+          -- Denormalized dataset run fields
+          dr.name as dataset_run_name,
+          dr.description as dataset_run_description,
+          dr.metadata as dataset_run_metadata,
+          dr.created_at as dataset_run_created_at,
+          
+          -- Denormalized dataset item fields  
+          di.input as dataset_item_input,
+          di.expected_output as dataset_item_expected_output,
+          di.metadata as dataset_item_metadata,
+          
+          -- Dataset ID
+          d.id as dataset_id
+
+        FROM dataset_run_items dri
+        JOIN dataset_runs dr ON dri.dataset_run_id = dr.id
+        JOIN dataset_items di ON dri.dataset_item_id = di.id  
+        JOIN datasets d ON di.dataset_id = d.id
+        WHERE dri.created_at <= ${new Date(migrationState.state.maxDate)}
+        ORDER BY dri.created_at DESC
+        LIMIT ${batchSize};
+      `);
+      if (datasetRunItems.length === 0) {
+        logger.info("No more dataset run items to migrate. Exiting...");
+        break;
+      }
+
+      logger.info(
+        `Got ${datasetRunItems.length} records from Postgres in ${Date.now() - fetchStart}ms`,
+      );
+
+      const insertStart = Date.now();
+      await clickhouseClient().insert({
+        table: "dataset_run_items_rmt",
+        values: datasetRunItems.map(convertPostgresDatasetRunItemToInsert),
+        format: "JSONEachRow",
+      });
+
+      logger.info(
+        `Inserted ${datasetRunItems.length} dataset run items into Clickhouse in ${Date.now() - insertStart}ms`,
+      );
+
+      await prisma.backgroundMigration.update({
+        where: { id: backgroundMigrationId },
+        data: {
+          state: {
+            maxDate: new Date(
+              datasetRunItems[datasetRunItems.length - 1].created_at,
+            ),
+          },
+        },
+      });
+
+      if (datasetRunItems.length < batchSize) {
+        logger.info("No more dataset run items to migrate. Exiting...");
+        this.isFinished = true;
+      }
+
+      processedRows += datasetRunItems.length;
+      logger.info(
+        `Processed batch in ${Date.now() - fetchStart}ms. Oldest record in batch: ${new Date(datasetRunItems[datasetRunItems.length - 1].created_at).toISOString()}`,
+      );
+    }
+
+    if (this.isAborted) {
+      logger.info(
+        `Migration of dataset run items from Postgres to Clickhouse aborted after processing ${processedRows} rows. Skipping cleanup.`,
+      );
+      return;
+    }
+
+    logger.info(
+      `Finished migration of dataset run items from Postgres to Clickhouse in ${Date.now() - start}ms`,
+    );
+  }
+
+  async abort(): Promise<void> {
+    logger.info(
+      `Aborting migration of dataset run items from Postgres to clickhouse`,
+    );
+    this.isAborted = true;
+  }
+}
+
+async function main() {
+  const args = parseArgs({
+    options: {
+      batchSize: { type: "string", short: "b", default: "1000" },
+      maxRowsToProcess: { type: "string", short: "r", default: "Infinity" },
+      maxDate: {
+        type: "string",
+        short: "d",
+        default: new Date().toISOString(),
+      },
+    },
+  });
+
+  const migration = new MigrateDatasetRunItemsFromPostgresToClickhouseRmt();
+  await migration.validate(args.values);
+  await migration.run(args.values);
+}
+
+// If the script is being executed directly (not imported), run the main function
+if (require.main === module) {
+  main()
+    .then(() => {
+      process.exit(0);
+    })
+    .catch((error) => {
+      logger.error(`Migration execution failed: ${error}`, error);
+      process.exit(1); // Exit with an error code
+    });
+}

worker/src/constants/VERSION.ts

@@ -1 +1 @@
-export const VERSION = "v3.97.2";
+export const VERSION = "v3.98.2";

worker/src/ee/dataRetention/handleDataRetentionProcessingJob.ts

@@ -56,22 +56,21 @@ export const handleDataRetentionProcessingJob = async (job: Job) => {
     );
   }
 
-  await removeIngestionEventsFromS3AndDeleteClickhouseRefsForProject(
-    projectId,
-    cutoffDate,
-  );
-
   // Delete ClickHouse (TTL / Delete Queries)
   logger.info(
-    `[Data Retention] Deleting ClickHouse data older than ${retention} days for project ${projectId}`,
+    `[Data Retention] Deleting ClickHouse and S3 data older than ${retention} days for project ${projectId}`,
   );
   await Promise.all([
+    removeIngestionEventsFromS3AndDeleteClickhouseRefsForProject(
+      projectId,
+      cutoffDate,
+    ),
     deleteTracesOlderThanDays(projectId, cutoffDate),
     deleteObservationsOlderThanDays(projectId, cutoffDate),
     deleteScoresOlderThanDays(projectId, cutoffDate),
   ]);
   logger.info(
-    `[Data Retention] Deleted ClickHouse data older than ${retention} days for project ${projectId}`,
+    `[Data Retention] Deleted ClickHouse and S3 data older than ${retention} days for project ${projectId}`,
   );
 
   // Set S3 Lifecycle for deletion (Future)

worker/src/env.ts

@@ -262,6 +262,9 @@ const EnvSchema = z.object({
     .positive()
     .default(2),
   LANGFUSE_DELETE_BATCH_SIZE: z.coerce.number().positive().default(2000),
+  LANGFUSE_EXPERIMENT_DATASET_RUN_ITEMS_TRACE_SOURCE_CH: z
+    .enum(["true", "false"])
+    .default("false"),
 });
 
 export const env: z.infer<typeof EnvSchema> =

worker/src/features/evaluation/evalService.ts

@@ -27,6 +27,7 @@ import {
   getObservationForTraceIdByName,
   InMemoryFilterService,
   recordIncrement,
+  getCurrentSpan,
 } from "@langfuse/shared/src/server";
 import {
   mapTraceFilterColumn,
@@ -159,6 +160,11 @@ export const createEvalJobs = async ({
   jobTimestamp: Date;
   enforcedJobTimeScope?: JobTimeScope;
 }) => {
+  const span = getCurrentSpan();
+  if (span) {
+    span.setAttribute("messaging.bullmq.job.input.projectId", event.projectId);
+  }
+
   // Fetch all configs for a given project. Those may be dataset or trace configs.
   let configsQuery = kyselyPrisma.$kysely
     .selectFrom("job_configurations")
@@ -471,6 +477,11 @@ export const evaluate = async ({
 }: {
   event: z.infer<typeof EvalExecutionEvent>;
 }) => {
+  const span = getCurrentSpan();
+  if (span) {
+    span.setAttribute("messaging.bullmq.job.input.projectId", event.projectId);
+  }
+
   logger.debug(
     `Evaluating job ${event.jobExecutionId} for project ${event.projectId}`,
   );

worker/src/features/experiments/experimentServiceClickhouse.ts

@@ -35,7 +35,7 @@ async function getExistingRunItemDatasetItemIds(
 ): Promise<Set<string>> {
   const query = `
   SELECT dataset_item_id as id
-  FROM dataset_run_items
+  FROM dataset_run_items_rmt
   WHERE project_id = {projectId: String}
   AND dataset_id = {datasetId: String}
   AND dataset_run_id = {runId: String}

worker/src/features/experiments/utils.ts

@@ -29,6 +29,7 @@ import {
 import { kyselyPrisma, prisma } from "@langfuse/shared/src/db";
 import z from "zod/v4";
 import { createHash } from "crypto";
+import { env } from "../../env";
 
 export enum TraceExecutionSource {
   // eslint-disable-next-line no-unused-vars
@@ -52,7 +53,9 @@ export enum TraceExecutionSource {
  *
  */
 export const shouldCreateTrace = (source: TraceExecutionSource) => {
-  // TODO: use env variable instead
+  if (env.LANGFUSE_EXPERIMENT_DATASET_RUN_ITEMS_TRACE_SOURCE_CH === "true") {
+    return source === TraceExecutionSource.CLICKHOUSE;
+  }
   return source === TraceExecutionSource.POSTGRES;
 };
 

worker/src/services/ClickhouseWriter/index.ts

@@ -404,7 +404,7 @@ export enum TableName {
   Scores = "scores", // eslint-disable-line no-unused-vars
   Observations = "observations", // eslint-disable-line no-unused-vars
   BlobStorageFileLog = "blob_storage_file_log", // eslint-disable-line no-unused-vars
-  DatasetRunItems = "dataset_run_items", // eslint-disable-line no-unused-vars
+  DatasetRunItems = "dataset_run_items_rmt", // eslint-disable-line no-unused-vars
 }
 
 type RecordInsertType<T extends TableName> = T extends TableName.Scores

worker/src/services/IngestionService/index.ts

@@ -450,8 +450,8 @@ export class IngestionService {
         await this.prisma.$executeRaw`
           INSERT INTO trace_sessions (id, project_id, environment, created_at, updated_at)
           VALUES (${traceRecordWithSession.session_id}, ${projectId}, ${traceRecordWithSession.environment}, NOW(), NOW())
-          ON CONFLICT (id, project_id) 
-          DO UPDATE SET 
+          ON CONFLICT (id, project_id)
+          DO UPDATE SET
             environment = EXCLUDED.environment,
             updated_at = NOW()
           WHERE trace_sessions.environment IS DISTINCT FROM EXCLUDED.environment
@@ -1118,7 +1118,17 @@ export class IngestionService {
 
   private getObservationType(
     observation: ObservationEvent,
-  ): "EVENT" | "SPAN" | "GENERATION" {
+  ):
+    | "EVENT"
+    | "SPAN"
+    | "GENERATION"
+    | "AGENT"
+    | "TOOL"
+    | "CHAIN"
+    | "RETRIEVER"
+    | "EVALUATOR"
+    | "GUARDRAIL"
+    | "EMBEDDING" {
     switch (observation.type) {
       case eventTypes.OBSERVATION_CREATE:
       case eventTypes.OBSERVATION_UPDATE:
@@ -1131,6 +1141,20 @@ export class IngestionService {
       case eventTypes.GENERATION_CREATE:
       case eventTypes.GENERATION_UPDATE:
         return "GENERATION" as const;
+      case eventTypes.AGENT_CREATE:
+        return "AGENT" as const;
+      case eventTypes.TOOL_CREATE:
+        return "TOOL" as const;
+      case eventTypes.CHAIN_CREATE:
+        return "CHAIN" as const;
+      case eventTypes.RETRIEVER_CREATE:
+        return "RETRIEVER" as const;
+      case eventTypes.EVALUATOR_CREATE:
+        return "EVALUATOR" as const;
+      case eventTypes.EMBEDDING_CREATE:
+        return "EMBEDDING" as const;
+      case eventTypes.GUARDRAIL_CREATE:
+        return "GUARDRAIL" as const;
     }
   }