chore: release v3.6.2

* fix: fix trace exports

* push

* fix: fix trace exports

.env.prod.example

@@ -245,6 +245,7 @@ OTEL_SERVICE_NAME="langfuse"
 # CLICKHOUSE_URL=
 # CLICKHOUSE_USER=
 # CLICKHOUSE_PASSWORD=
+# CLICKHOUSE_DB=
 
 # Ingestion
 # LANGFUSE_INGESTION_QUEUE_DELAY_MS=

docker-compose.yml

@@ -38,6 +38,8 @@ services:
       LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT: ${LANGFUSE_S3_MEDIA_UPLOAD_ENDPOINT:-http://minio:9000}
       LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE: ${LANGFUSE_S3_MEDIA_UPLOAD_FORCE_PATH_STYLE:-true}
       LANGFUSE_S3_MEDIA_UPLOAD_PREFIX: ${LANGFUSE_S3_MEDIA_UPLOAD_PREFIX:-media/}
+      LANGFUSE_INGESTION_QUEUE_DELAY_MS: ${LANGFUSE_INGESTION_QUEUE_DELAY_MS:-}
+      LANGFUSE_INGESTION_CLICKHOUSE_WRITE_INTERVAL_MS: ${LANGFUSE_INGESTION_CLICKHOUSE_WRITE_INTERVAL_MS:-}
       REDIS_HOST: ${REDIS_HOST:-redis}
       REDIS_PORT: ${REDIS_PORT:-6379}
       REDIS_AUTH: ${REDIS_AUTH:-myredissecret}

ee/package.json

@@ -47,7 +47,8 @@
   },
   "pnpm": {
     "overrides": {
-      "jsonpath-plus": "10.0.7"
+      "jsonpath-plus": "10.0.7",
+      "nanoid": "^3.3.8"
     }
   }
 }

package.json

@@ -1,6 +1,6 @@
 {
   "name": "langfuse",
-  "version": "3.5.3",
+  "version": "3.6.2",
   "author": "engineering@langfuse.com",
   "license": "MIT",
   "private": true,
@@ -84,7 +84,8 @@
   "packageManager": "pnpm@9.5.0",
   "pnpm": {
     "overrides": {
-      "jsonpath-plus": "10.0.7"
+      "jsonpath-plus": "10.0.7",
+      "nanoid": "^3.3.8"
     }
   }
 }

packages/shared/clickhouse/scripts/down.sh

@@ -18,21 +18,31 @@ then
     exit 1
 fi
 
+# Ensure CLICKHOUSE_DB is set
+if [ -z "${CLICKHOUSE_DB}" ]; then
+    export CLICKHOUSE_DB="default"
+fi
+
+# Ensure CLICKHOUSE_CLUSTER_NAME is set
+if [ -z "${CLICKHOUSE_CLUSTER_NAME}" ]; then
+    export CLICKHOUSE_CLUSTER_NAME="default"
+fi
+
 # Construct the database URL
 if [ "$CLICKHOUSE_CLUSTER_ENABLED" == "false" ] ; then
   if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
   else
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-migrations-table-engine=MergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-migrations-table-engine=MergeTree"
   fi
 
   # Execute the up command
   migrate -source file://clickhouse/migrations/unclustered -database "$DATABASE_URL" down
 else
   if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-cluster-name=default&x-migrations-table-engine=ReplicatedMergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-cluster-name=${CLICKHOUSE_CLUSTER_NAME}&x-migrations-table-engine=ReplicatedMergeTree"
   else
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-cluster-name=default&x-migrations-table-engine=ReplicatedMergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-cluster-name=${CLICKHOUSE_CLUSTER_NAME}&x-migrations-table-engine=ReplicatedMergeTree"
   fi
 
   # Execute the up command

packages/shared/clickhouse/scripts/drop.sh

@@ -12,11 +12,16 @@ then
     exit 1
 fi
 
+# Ensure CLICKHOUSE_DB is set
+if [ -z "${CLICKHOUSE_DB}" ]; then
+    export CLICKHOUSE_DB="default"
+fi
+
 # Construct the database URL
 if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
-    DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
+    DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
 else
-    DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-migrations-table-engine=MergeTree"
+    DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-migrations-table-engine=MergeTree"
 fi
 # Execute the drop command
 migrate -source file://clickhouse/migrations -database "$DATABASE_URL" drop

packages/shared/clickhouse/scripts/up.sh

@@ -18,21 +18,31 @@ then
     exit 1
 fi
 
+# Ensure CLICKHOUSE_DB is set
+if [ -z "${CLICKHOUSE_DB}" ]; then
+    export CLICKHOUSE_DB="default"
+fi
+
+# Ensure CLICKHOUSE_CLUSTER_NAME is set
+if [ -z "${CLICKHOUSE_CLUSTER_NAME}" ]; then
+    export CLICKHOUSE_CLUSTER_NAME="default"
+fi
+
 # Construct the database URL
 if [ "$CLICKHOUSE_CLUSTER_ENABLED" == "false" ] ; then
   if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-migrations-table-engine=MergeTree"
   else
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-migrations-table-engine=MergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-migrations-table-engine=MergeTree"
   fi
 
   # Execute the up command
   migrate -source file://clickhouse/migrations/unclustered -database "$DATABASE_URL" up
 else
 if [ "$CLICKHOUSE_MIGRATION_SSL" = true ] ; then
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&secure=true&skip_verify=true&x-cluster-name=default&x-migrations-table-engine=ReplicatedMergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&secure=true&skip_verify=true&x-cluster-name=${CLICKHOUSE_CLUSTER_NAME}&x-migrations-table-engine=ReplicatedMergeTree"
   else
-      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=default&x-multi-statement=true&x-cluster-name=default&x-migrations-table-engine=ReplicatedMergeTree"
+      DATABASE_URL="${CLICKHOUSE_MIGRATION_URL}?username=${CLICKHOUSE_USER}&password=${CLICKHOUSE_PASSWORD}&database=${CLICKHOUSE_DB}&x-multi-statement=true&x-cluster-name=${CLICKHOUSE_CLUSTER_NAME}&x-migrations-table-engine=ReplicatedMergeTree"
   fi
 
   # Execute the up command

packages/shared/package.json

@@ -121,7 +121,8 @@
   },
   "pnpm": {
     "overrides": {
-      "jsonpath-plus": "10.0.7"
+      "jsonpath-plus": "10.0.7",
+      "nanoid": "^3.3.8"
     }
   }
 }

packages/shared/prisma/generated/types.ts

@@ -479,6 +479,13 @@ export type Prompt = {
     tags: Generated<string[]>;
     labels: Generated<string[]>;
 };
+export type QueueBackUp = {
+    id: string;
+    project_id: string | null;
+    queue_name: string;
+    content: unknown;
+    created_at: Generated<Timestamp>;
+};
 export type Score = {
     id: string;
     timestamp: Generated<Timestamp>;
@@ -628,6 +635,7 @@ export type DB = {
     project_memberships: ProjectMembership;
     projects: Project;
     prompts: Prompt;
+    queue_backups: QueueBackUp;
     score_configs: ScoreConfig;
     scores: Score;
     Session: Session;

packages/shared/prisma/migrations/20250108220721_add_queue_backup_table/migration.sql

@@ -0,0 +1,13 @@
+-- CreateTable
+CREATE TABLE "queue_backups" (
+    "id" TEXT NOT NULL,
+    "project_id" TEXT,
+    "queue_name" TEXT NOT NULL,
+    "content" JSONB NOT NULL,
+    "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "queue_backups_pkey" PRIMARY KEY ("id")
+);
+
+-- AddForeignKey
+ALTER TABLE "traces" ADD CONSTRAINT "traces_session_id_project_id_fkey" FOREIGN KEY ("session_id", "project_id") REFERENCES "trace_sessions"("id", "project_id") ON DELETE RESTRICT ON UPDATE CASCADE;

packages/shared/prisma/migrations/20250109083346_drop_trace_tracesession_fk/migration.sql

@@ -0,0 +1,2 @@
+-- DropForeignKey
+ALTER TABLE "traces" DROP CONSTRAINT "traces_session_id_project_id_fkey";

packages/shared/prisma/schema.prisma

@@ -272,7 +272,6 @@ model TraceSession {
   project    Project  @relation(fields: [projectId], references: [id], onDelete: Cascade)
   bookmarked Boolean  @default(false)
   public     Boolean  @default(false)
-  traces     Trace[]
 
   @@id([id, projectId])
   @@index([projectId])
@@ -284,25 +283,24 @@ model TraceSession {
 // Update TraceView below when making changes to this model!
 
 model Trace {
-  id         String        @id @default(cuid())
-  externalId String?       @map("external_id")
-  timestamp  DateTime      @default(now())
+  id         String   @id @default(cuid())
+  externalId String?  @map("external_id")
+  timestamp  DateTime @default(now())
   name       String?
-  userId     String?       @map("user_id")
+  userId     String?  @map("user_id")
   metadata   Json?
   release    String?
   version    String?
-  projectId  String        @map("project_id")
-  project    Project       @relation(fields: [projectId], references: [id], onDelete: Cascade)
-  public     Boolean       @default(false)
-  bookmarked Boolean       @default(false)
-  tags       String[]      @default([])
+  projectId  String   @map("project_id")
+  project    Project  @relation(fields: [projectId], references: [id], onDelete: Cascade)
+  public     Boolean  @default(false)
+  bookmarked Boolean  @default(false)
+  tags       String[] @default([])
   input      Json?
   output     Json?
-  sessionId  String?       @map("session_id")
-  session    TraceSession? @relation(fields: [sessionId, projectId], references: [id, projectId])
-  createdAt  DateTime      @default(now()) @map("created_at")
-  updatedAt  DateTime      @default(now()) @updatedAt @map("updated_at")
+  sessionId  String?  @map("session_id")
+  createdAt  DateTime @default(now()) @map("created_at")
+  updatedAt  DateTime @default(now()) @updatedAt @map("updated_at")
 
   @@index([projectId, timestamp])
   @@index([sessionId])
@@ -1017,3 +1015,14 @@ model ObservationMedia {
   @@index([projectId, observationId])
   @@map("observation_media")
 }
+
+model QueueBackUp {
+  id        String  @id @default(cuid())
+  projectId String? @map("project_id")
+  queueName String  @map("queue_name")
+  content   Json
+
+  createdAt DateTime @default(now()) @map("created_at")
+
+  @@map("queue_backups")
+}

packages/shared/src/env.ts

@@ -29,6 +29,8 @@ const EnvSchema = z.object({
   LANGFUSE_CACHE_PROMPT_ENABLED: z.enum(["true", "false"]).default("false"),
   LANGFUSE_CACHE_PROMPT_TTL_SECONDS: z.coerce.number().default(60 * 60),
   CLICKHOUSE_URL: z.string().url(),
+  CLICKHOUSE_CLUSTER_NAME: z.string().default("default"),
+  CLICKHOUSE_DB: z.string().default("default"),
   CLICKHOUSE_USER: z.string(),
   CLICKHOUSE_PASSWORD: z.string(),
   LANGFUSE_SDK_CI_SYNC_PROCESSING_ENABLED: z

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

@@ -10,7 +10,7 @@ export const clickhouseClient = (opts?: NodeClickHouseClientConfigOptions) =>
     url: env.CLICKHOUSE_URL,
     username: env.CLICKHOUSE_USER,
     password: env.CLICKHOUSE_PASSWORD,
-    database: "default",
+    database: env.CLICKHOUSE_DB,
     clickhouse_settings: {
       async_insert: 1,
       wait_for_async_insert: 1, // if disabled, we won't get errors from clickhouse

packages/shared/src/server/index.ts

@@ -45,6 +45,7 @@ export * from "./logger";
 export * from "./queries";
 export * from "./repositories";
 export * from "./redis/evalExecutionQueue";
+export * from "./services/sessions-ui-table-service";
 
 // test utils
 export * from "./test-utils";

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

@@ -9,7 +9,6 @@ import Decimal from "decimal.js";
 import { parseClickhouseUTCDateTimeFormat } from "./clickhouse";
 import { ObservationRecordReadType } from "./definitions";
 import { parseJsonPrioritised } from "../../utils/json";
-import { jsonSchema } from "../../utils/zod";
 
 export const convertObservationToView = (
   record: ObservationRecordReadType,
@@ -62,15 +61,22 @@ export const convertObservation = (
       ? parseClickhouseUTCDateTimeFormat(record.end_time)
       : null,
     name: record.name ?? null,
-    metadata: record.metadata,
+    metadata:
+      record.metadata &&
+      Object.fromEntries(
+        Object.entries(record.metadata ?? {}).map(([key, val]) => [
+          key,
+          val && parseJsonPrioritised(val),
+        ]),
+      ),
     level: record.level as ObservationLevel,
     statusMessage: record.status_message ?? null,
     version: record.version ?? null,
     input: (record.input
-      ? jsonSchema.parse(parseJsonPrioritised(record.input))
+      ? parseJsonPrioritised(record.input)
       : null) as Prisma.JsonValue | null,
     output: (record.output
-      ? jsonSchema.parse(parseJsonPrioritised(record.output))
+      ? parseJsonPrioritised(record.output)
       : null) as Prisma.JsonValue | null,
     modelParameters: record.model_parameters
       ? JSON.parse(record.model_parameters)

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

@@ -135,13 +135,16 @@ export const upsertScore = async (score: Partial<ScoreRecordReadType>) => {
   });
 };
 
-export const getScoresForTraces = async (
-  projectId: string,
-  traceIds: string[],
-  timestamp?: Date,
-  limit?: number,
-  offset?: number,
-) => {
+export type GetScoresForTracesProps = {
+  projectId: string;
+  traceIds: string[];
+  timestamp?: Date;
+  limit?: number;
+  offset?: number;
+};
+
+export const getScoresForTraces = async (props: GetScoresForTracesProps) => {
+  const { projectId, traceIds, timestamp, limit, offset } = props;
   const query = `
       select 
         *

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

@@ -16,10 +16,7 @@ import {
 } from "../queries/clickhouse-sql/clickhouse-filter";
 import { TraceRecordReadType } from "./definitions";
 import { tracesTableUiColumnDefinitions } from "../../tableDefinitions/mapTracesTable";
-import { OrderByState } from "../../interfaces/orderBy";
-import { orderByToClickhouseSql } from "../queries/clickhouse-sql/orderby-factory";
 import { UiColumnMapping } from "../../tableDefinitions";
-import { sessionCols } from "../../tableDefinitions/mapSessionTable";
 import { convertDateToClickhouseDateTime } from "../clickhouse/client";
 import { convertClickhouseToDomain } from "./traces_converters";
 import { clickhouseSearchCondition } from "../queries/clickhouse-sql/search";
@@ -383,196 +380,6 @@ export const getTracesGroupedByTags = async (props: GroupedTracesQueryProp) => {
   return rows;
 };
 
-export type SessionDataReturnType = {
-  session_id: string;
-  max_timestamp: string;
-  min_timestamp: string;
-  trace_ids: string[];
-  user_ids: string[];
-  trace_count: number;
-  trace_tags: string[];
-  total_observations: number;
-  duration: number;
-  session_usage_details: Record<string, number>;
-  session_cost_details: Record<string, number>;
-  session_input_cost: string;
-  session_output_cost: string;
-  session_total_cost: string;
-  session_input_usage: string;
-  session_output_usage: string;
-  session_total_usage: string;
-};
-
-export const getSessionsTableCount = async (props: {
-  projectId: string;
-  filter: FilterState;
-  orderBy?: OrderByState;
-  limit?: number;
-  page?: number;
-}) => {
-  const rows = await getSessionsTableGeneric<{ count: string }>({
-    select: `
-      count(session_id) as count
-    `,
-    projectId: props.projectId,
-    filter: props.filter,
-    orderBy: props.orderBy,
-    limit: props.limit,
-    page: props.page,
-  });
-
-  return rows.length > 0 ? Number(rows[0].count) : 0;
-};
-
-export const getSessionsTable = async (props: {
-  projectId: string;
-  filter: FilterState;
-  orderBy?: OrderByState;
-  limit?: number;
-  page?: number;
-}) => {
-  const rows = await getSessionsTableGeneric<SessionDataReturnType>({
-    select: `
-    session_id, 
-    max_timestamp, 
-    min_timestamp, 
-    trace_ids, 
-    user_ids, 
-    trace_count, 
-    trace_tags,
-    total_observations,
-    duration,
-    session_usage_details,
-    session_cost_details,
-    session_input_cost,
-    session_output_cost,
-    session_total_cost,
-    session_input_usage,
-    session_output_usage,
-    session_total_usage
-    `,
-    projectId: props.projectId,
-    filter: props.filter,
-    orderBy: props.orderBy,
-    limit: props.limit,
-    page: props.page,
-  });
-
-  return rows;
-};
-
-export type FetchSessionsTableProps = {
-  select: string;
-  projectId: string;
-  filter: FilterState;
-  searchQuery?: string;
-  orderBy?: OrderByState;
-  limit?: number;
-  page?: number;
-};
-
-const getSessionsTableGeneric = async <T>(props: FetchSessionsTableProps) => {
-  const { select, projectId, filter, orderBy, limit, page } = props;
-
-  const { tracesFilter, scoresFilter, observationsFilter } =
-    getProjectIdDefaultFilter(projectId, { tracesPrefix: "s" });
-
-  tracesFilter.push(...createFilterFromFilterState(filter, sessionCols));
-
-  const tracesFilterRes = tracesFilter.apply();
-  const scoresAvgFilterRes = scoresFilter.apply();
-  const observationsStatsRes = observationsFilter.apply();
-
-  const traceTimestampFilter: DateTimeFilter | undefined = tracesFilter.find(
-    (f) =>
-      f.field === "min_timestamp" &&
-      (f.operator === ">=" || f.operator === ">"),
-  ) as DateTimeFilter | undefined;
-
-  const singleTraceFilter = traceTimestampFilter
-    ? new FilterList([
-        new DateTimeFilter({
-          clickhouseTable: "traces",
-          field: "timestamp",
-          operator: traceTimestampFilter.operator,
-          value: traceTimestampFilter.value,
-        }),
-      ]).apply()
-    : undefined;
-
-  const query = `
-      WITH observations_agg AS (
-        SELECT o.trace_id,
-              count(*) as obs_count,
-              min(o.start_time) as min_start_time,
-              max(o.end_time) as max_end_time,
-              sumMap(usage_details) as sum_usage_details,
-              sumMap(cost_details) as sum_cost_details,
-              anyLast(project_id) as project_id
-        FROM observations o FINAL
-        WHERE o.project_id = {projectId: String}
-        ${traceTimestampFilter ? `AND o.start_time >= {observationsStartTime: DateTime64(3)} - ${TRACE_TO_OBSERVATIONS_INTERVAL}` : ""}
-        GROUP BY o.trace_id
-    ),
-    session_data AS (
-        SELECT
-            t.session_id,
-            anyLast(t.project_id) as project_id,
-            max(t.timestamp) as max_timestamp,
-            min(t.timestamp) as min_timestamp,
-            groupArray(t.id) AS trace_ids,
-            groupUniqArray(t.user_id) AS user_ids,
-            count(*) as trace_count,
-            groupUniqArrayArray(t.tags) as trace_tags,
-            -- Aggregate observations data at session level
-            sum(o.obs_count) as total_observations,
-            date_diff('millisecond', min(min_start_time), max(max_end_time)) as duration,
-            sumMap(o.sum_usage_details) as session_usage_details,
-            sumMap(o.sum_cost_details) as session_cost_details,
-            arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'input') > 0, sumMap(o.sum_cost_details)))) as session_input_cost,
-            arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'output') > 0, sumMap(o.sum_cost_details)))) as session_output_cost,
-            sumMap(o.sum_cost_details)['total'] as session_total_cost,          
-            arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'input') > 0, sumMap(o.sum_usage_details)))) as session_input_usage,
-            arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'output') > 0, sumMap(o.sum_usage_details)))) as session_output_usage,
-            sumMap(o.sum_usage_details)['total'] as session_total_usage
-        FROM traces t FINAL
-        LEFT JOIN observations_agg o
-        ON t.id = o.trace_id AND t.project_id = o.project_id
-        WHERE t.session_id IS NOT NULL
-            AND t.project_id = {projectId: String}
-            ${singleTraceFilter?.query ? ` AND ${singleTraceFilter.query}` : ""}
-        GROUP BY t.session_id
-    )
-    SELECT ${select}
-    FROM session_data s
-    WHERE ${tracesFilterRes.query ? tracesFilterRes.query : ""}
-    ${orderByToClickhouseSql(orderBy ?? null, sessionCols)}
-    ${limit !== undefined && page !== undefined ? `LIMIT {limit: Int32} OFFSET {offset: Int32}` : ""}
-    `;
-
-  const obsStartTimeValue = traceTimestampFilter
-    ? convertDateToClickhouseDateTime(traceTimestampFilter.value)
-    : null;
-
-  const res = await queryClickhouse<T>({
-    query: query,
-    params: {
-      projectId,
-      limit: limit,
-      offset: limit && page ? limit * page : 0,
-      ...tracesFilterRes.params,
-      ...observationsStatsRes.params,
-      ...scoresAvgFilterRes.params,
-      ...singleTraceFilter?.params,
-      ...(obsStartTimeValue
-        ? { observationsStartTime: obsStartTimeValue }
-        : {}),
-    },
-  });
-
-  return res;
-};
-
 export const getTracesIdentifierForSession = async (
   projectId: string,
   sessionId: string,

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

@@ -3,7 +3,6 @@ import { parseClickhouseUTCDateTimeFormat } from "./clickhouse";
 import { TraceRecordReadType } from "./definitions";
 import { convertDateToClickhouseDateTime } from "../clickhouse/client";
 import { parseJsonPrioritised } from "../../utils/json";
-import { jsonSchema } from "../../utils/zod";
 
 export const convertTraceDomainToClickhouse = (
   trace: Trace,
@@ -46,12 +45,19 @@ export const convertClickhouseToDomain = (
     sessionId: record.session_id ?? null,
     public: record.public,
     input: (record.input
-      ? jsonSchema.parse(parseJsonPrioritised(record.input))
+      ? parseJsonPrioritised(record.input)
       : null) as Prisma.JsonValue | null,
     output: (record.output
-      ? jsonSchema.parse(parseJsonPrioritised(record.output))
+      ? parseJsonPrioritised(record.output)
       : null) as Prisma.JsonValue | null,
-    metadata: record.metadata,
+    metadata:
+      record.metadata &&
+      Object.fromEntries(
+        Object.entries(record.metadata ?? {}).map(([key, val]) => [
+          key,
+          val && parseJsonPrioritised(val),
+        ]),
+      ),
     createdAt: parseClickhouseUTCDateTimeFormat(record.created_at),
     updatedAt: parseClickhouseUTCDateTimeFormat(record.updated_at),
     externalId: null,

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

@@ -30,7 +30,7 @@ export interface StorageService {
 
   download(path: string): Promise<string>;
 
-  listFiles(prefix: string): Promise<string[]>;
+  listFiles(prefix: string): Promise<{ file: string; createdAt: Date }[]>;
 
   getSignedUrl(
     fileName: string,
@@ -201,7 +201,9 @@ class AzureBlobStorageService implements StorageService {
     }
   }
 
-  public async listFiles(prefix: string): Promise<string[]> {
+  public async listFiles(
+    prefix: string,
+  ): Promise<{ file: string; createdAt: Date }[]> {
     try {
       await this.createContainerIfNotExists();
 
@@ -209,7 +211,10 @@ class AzureBlobStorageService implements StorageService {
       const files = [];
       for await (const blob of result) {
         if (blob.name.startsWith(prefix)) {
-          files.push(blob.name);
+          files.push({
+            file: blob.name,
+            createdAt: blob?.properties?.createdOn ?? new Date(),
+          });
         }
       }
       return files;
@@ -364,7 +369,9 @@ class S3StorageService implements StorageService {
     }
   }
 
-  public async listFiles(prefix: string): Promise<string[]> {
+  public async listFiles(
+    prefix: string,
+  ): Promise<{ file: string; createdAt: Date }[]> {
     const listCommand = new ListObjectsV2Command({
       Bucket: this.bucketName,
       Prefix: prefix,
@@ -373,7 +380,11 @@ class S3StorageService implements StorageService {
     try {
       const response = await this.client.send(listCommand);
       return (
-        response.Contents?.flatMap((file) => (file.Key ? [file.Key] : [])) ?? []
+        response.Contents?.flatMap((file) =>
+          file.Key
+            ? [{ file: file.Key, createdAt: file.LastModified ?? new Date() }]
+            : [],
+        ) ?? []
       );
     } catch (err) {
       logger.error(`Failed to list files from S3 ${prefix}`, err);

packages/shared/src/server/services/sessions-ui-table-service.ts

@@ -0,0 +1,307 @@
+import { OrderByState } from "../../interfaces/orderBy";
+import { sessionCols } from "../../tableDefinitions/mapSessionTable";
+import { FilterState } from "../../types";
+import { convertDateToClickhouseDateTime } from "../clickhouse/client";
+import { DateTimeFilter, FilterList, orderByToClickhouseSql } from "../queries";
+import {
+  getProjectIdDefaultFilter,
+  createFilterFromFilterState,
+} from "../queries/clickhouse-sql/factory";
+import {
+  TRACE_TO_OBSERVATIONS_INTERVAL,
+  queryClickhouse,
+} from "../repositories";
+
+export type SessionDataReturnType = {
+  session_id: string;
+  max_timestamp: string;
+  min_timestamp: string;
+  trace_ids: string[];
+  user_ids: string[];
+  trace_count: number;
+  trace_tags: string[];
+};
+
+export type SessionWithMetricsReturnType = SessionDataReturnType & {
+  total_observations: number;
+  duration: number;
+  session_usage_details: Record<string, number>;
+  session_cost_details: Record<string, number>;
+  session_input_cost: string;
+  session_output_cost: string;
+  session_total_cost: string;
+  session_input_usage: string;
+  session_output_usage: string;
+  session_total_usage: string;
+};
+
+export const getSessionsTableCount = async (props: {
+  projectId: string;
+  filter: FilterState;
+  orderBy?: OrderByState;
+  limit?: number;
+  page?: number;
+}) => {
+  const rows = await getSessionsTableGeneric<{ count: string }>({
+    select: "count",
+    projectId: props.projectId,
+    filter: props.filter,
+    orderBy: props.orderBy,
+    limit: props.limit,
+    page: props.page,
+  });
+
+  return rows.length > 0 ? Number(rows[0].count) : 0;
+};
+
+export const getSessionsTable = async (props: {
+  projectId: string;
+  filter: FilterState;
+  orderBy?: OrderByState;
+  limit?: number;
+  page?: number;
+}) => {
+  const rows = await getSessionsTableGeneric<SessionDataReturnType>({
+    select: "rows",
+    projectId: props.projectId,
+    filter: props.filter,
+    orderBy: props.orderBy,
+    limit: props.limit,
+    page: props.page,
+  });
+
+  return rows.map((row) => ({
+    ...row,
+    trace_count: Number(row.trace_count),
+  }));
+};
+
+export const getSessionsWithMetrics = async (props: {
+  projectId: string;
+  filter: FilterState;
+  orderBy?: OrderByState;
+  limit?: number;
+  page?: number;
+}) => {
+  const rows = await getSessionsTableGeneric<SessionWithMetricsReturnType>({
+    select: "metrics",
+    projectId: props.projectId,
+    filter: props.filter,
+    orderBy: props.orderBy,
+    limit: props.limit,
+    page: props.page,
+  });
+
+  return rows.map((row) => ({
+    ...row,
+    trace_count: Number(row.trace_count),
+    total_observations: Number(row.total_observations),
+  }));
+};
+
+export type FetchSessionsTableProps = {
+  select: "count" | "rows" | "metrics";
+  projectId: string;
+  filter: FilterState;
+  searchQuery?: string;
+  orderBy?: OrderByState;
+  limit?: number;
+  page?: number;
+};
+
+const getSessionsTableGeneric = async <T>(props: FetchSessionsTableProps) => {
+  const { select, projectId, filter, orderBy, limit, page } = props;
+
+  let sqlSelect: string;
+  switch (select) {
+    case "count":
+      sqlSelect = "count(session_id) as count";
+      break;
+    case "rows":
+      sqlSelect = `
+          session_id, 
+          max_timestamp, 
+          min_timestamp, 
+          trace_ids, 
+          user_ids, 
+          trace_count, 
+          trace_tags`;
+      break;
+    case "metrics":
+      sqlSelect = `
+        session_id, 
+        max_timestamp, 
+        min_timestamp, 
+        trace_ids, 
+        user_ids, 
+        trace_count, 
+        trace_tags,
+        total_observations,
+        duration,
+        session_usage_details,
+        session_cost_details,
+        session_input_cost,
+        session_output_cost,
+        session_total_cost,
+        session_input_usage,
+        session_output_usage,
+        session_total_usage`;
+      break;
+    default:
+      const exhaustiveCheckDefault: never = select;
+      throw new Error(`Unknown select type: ${select}`);
+  }
+
+  const { tracesFilter } = getProjectIdDefaultFilter(projectId, {
+    tracesPrefix: "s",
+  });
+
+  tracesFilter.push(...createFilterFromFilterState(filter, sessionCols));
+
+  const tracesFilterRes = tracesFilter.apply();
+
+  const traceTimestampFilter: DateTimeFilter | undefined = tracesFilter.find(
+    (f) =>
+      f.field === "min_timestamp" &&
+      (f.operator === ">=" || f.operator === ">"),
+  ) as DateTimeFilter | undefined;
+
+  const filters = [];
+  if (traceTimestampFilter) {
+    filters.push(
+      new DateTimeFilter({
+        clickhouseTable: "traces",
+        field: "timestamp",
+        operator: traceTimestampFilter.operator,
+        value: traceTimestampFilter.value,
+      }),
+    );
+  }
+
+  const additionalSingleTraceFilter = tracesFilter.find(
+    (f) => f.field === "bookmarked" || f.field === "session_id",
+  );
+
+  if (additionalSingleTraceFilter) {
+    filters.push(additionalSingleTraceFilter);
+  }
+
+  const singleTraceFilter =
+    filters.length > 0 ? new FilterList(filters).apply() : undefined;
+
+  const hasMetricsFilter = tracesFilter.find((f) =>
+    [
+      "session_total_cost",
+      "session_input_cost",
+      "session_output_cost",
+      "duration",
+      "session_total_usage",
+      "session_output_usage",
+      "session_input_usage",
+    ].includes(f.field),
+  );
+
+  const selectMetrics = select === "metrics" || hasMetricsFilter;
+
+  // We use deduplicated traces and observations CTEs instead of final to be able to use Skip indices in Clickhouse.
+  const query = `
+    WITH deduplicated_traces AS (
+      SELECT * EXCEPT input, output, metadata
+      FROM traces t
+      WHERE t.session_id IS NOT NULL 
+        AND t.project_id = {projectId: String}
+        ${singleTraceFilter?.query ? ` AND ${singleTraceFilter.query}` : ""}
+        ORDER BY event_ts DESC
+        LIMIT 1 BY id, project_id
+    ),
+    deduplicated_observations AS (
+        SELECT * 
+        FROM observations o
+        WHERE o.project_id = {projectId: String}
+        ${traceTimestampFilter ? `AND o.start_time >= {observationsStartTime: DateTime64(3)} - ${TRACE_TO_OBSERVATIONS_INTERVAL}` : ""}
+        AND o.trace_id IN (
+          SELECT id
+          FROM deduplicated_traces
+        )
+        ORDER BY event_ts DESC
+        LIMIT 1 BY id, project_id
+    ),
+    observations_agg AS (
+      SELECT o.trace_id,
+            count(*) as obs_count,
+            min(o.start_time) as min_start_time,
+            max(o.end_time) as max_end_time,
+            sumMap(usage_details) as sum_usage_details,
+            sumMap(cost_details) as sum_cost_details,
+            anyLast(project_id) as project_id
+      FROM deduplicated_observations o
+      WHERE o.project_id = {projectId: String}
+      ${traceTimestampFilter ? `AND o.start_time >= {observationsStartTime: DateTime64(3)} - ${TRACE_TO_OBSERVATIONS_INTERVAL}` : ""}
+      GROUP BY o.trace_id
+    ),
+    session_data AS (
+        SELECT
+            t.session_id,
+            anyLast(t.project_id) as project_id,
+            max(t.timestamp) as max_timestamp,
+            min(t.timestamp) as min_timestamp,
+            groupArray(t.id) AS trace_ids,
+            groupUniqArray(t.user_id) AS user_ids,
+            count(*) as trace_count,
+            groupUniqArrayArray(t.tags) as trace_tags
+            -- Aggregate observations data at session level
+            ${
+              selectMetrics
+                ? `
+            ,
+            sum(o.obs_count) as total_observations,
+            date_diff('millisecond', min(min_start_time), max(max_end_time)) as duration,
+            sumMap(o.sum_usage_details) as session_usage_details,
+            sumMap(o.sum_cost_details) as session_cost_details,
+            arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'input') > 0, sumMap(o.sum_cost_details)))) as session_input_cost,
+            arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'output') > 0, sumMap(o.sum_cost_details)))) as session_output_cost,
+            sumMap(o.sum_cost_details)['total'] as session_total_cost,          
+            arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'input') > 0, sumMap(o.sum_usage_details)))) as session_input_usage,
+            arraySum(mapValues(mapFilter(x -> positionCaseInsensitive(x.1, 'output') > 0, sumMap(o.sum_usage_details)))) as session_output_usage,
+            sumMap(o.sum_usage_details)['total'] as session_total_usage`
+                : ""
+            }
+        FROM deduplicated_traces t
+        ${
+          selectMetrics
+            ? `LEFT JOIN observations_agg o
+        ON t.id = o.trace_id AND t.project_id = o.project_id`
+            : ""
+        }
+        WHERE t.session_id IS NOT NULL
+            AND t.project_id = {projectId: String}
+            ${singleTraceFilter?.query ? ` AND ${singleTraceFilter.query}` : ""}
+        GROUP BY t.session_id
+    )
+    SELECT ${sqlSelect}
+    FROM session_data s
+    WHERE ${tracesFilterRes.query ? tracesFilterRes.query : ""}
+    ${orderByToClickhouseSql(orderBy ?? null, sessionCols)}
+    ${limit !== undefined && page !== undefined ? `LIMIT {limit: Int32} OFFSET {offset: Int32}` : ""}
+    `;
+
+  const obsStartTimeValue = traceTimestampFilter
+    ? convertDateToClickhouseDateTime(traceTimestampFilter.value)
+    : null;
+
+  const res = await queryClickhouse<T>({
+    query: query,
+    params: {
+      projectId,
+      limit: limit,
+      offset: limit && page ? limit * page : 0,
+      ...tracesFilterRes.params,
+      ...singleTraceFilter?.params,
+      ...(obsStartTimeValue
+        ? { observationsStartTime: obsStartTimeValue }
+        : {}),
+    },
+  });
+
+  return res;
+};

packages/shared/src/utils/json.ts

@@ -55,25 +55,9 @@ export const parseJsonPrioritised = (
 ): z.infer<typeof jsonSchema> | string | undefined => {
   try {
     const parsedJson = JSON.parse(json);
-
-    if (Object.keys(parsedJson).length === 0) {
-      return parsedJson;
-    }
-
-    const parsedArray = z.array(jsonSchemaNullable).safeParse(parsedJson);
-    if (parsedArray.success) {
-      return parsedArray.data;
-    }
-
-    const parsedObject = z.record(jsonSchemaNullable).safeParse(parsedJson);
-    if (parsedObject.success) {
-      return parsedObject.data;
-    }
-
     return jsonSchema.parse(parsedJson);
   } catch (error) {
     const parsed = jsonSchema.safeParse(json);
-
     return parsed.success ? parsed.data : json;
   }
 };

packages/shared/src/utils/zod.ts

@@ -23,18 +23,18 @@ type Json = Root | { [key: string]: JsonNested } | JsonNested[];
 // Here, you define the schema recursively
 export const jsonSchemaNullable: z.ZodType<JsonNested> = z.lazy(() =>
   z.union([
-    nestedLiteralSchema,
     z.array(jsonSchemaNullable),
     z.record(jsonSchemaNullable),
+    nestedLiteralSchema,
   ]),
 );
 
 // Root schema that does not allow nulls at the root level
 export const jsonSchema: z.ZodType<Json> = z.lazy(() =>
   z.union([
-    rootLiteralSchema,
     z.array(jsonSchemaNullable),
     z.record(jsonSchemaNullable),
+    rootLiteralSchema,
   ]),
 );
 

pnpm-lock.yaml

@@ -6,6 +6,7 @@ settings:
 
 overrides:
   jsonpath-plus: 10.0.7
+  nanoid: ^3.3.8
 
 importers:
 
@@ -742,7 +743,7 @@ importers:
         version: 7.12.0(eslint@8.57.0)(typescript@5.4.5)
       autoprefixer:
         specifier: ^10.4.19
-        version: 10.4.19(postcss@8.4.47)
+        version: 10.4.19(postcss@8.4.49)
       dotenv-cli:
         specifier: ^7.4.2
         version: 7.4.2
@@ -761,9 +762,6 @@ importers:
       node-mocks-http:
         specifier: ^1.14.1
         version: 1.14.1
-      postcss:
-        specifier: ^8.4.47
-        version: 8.4.47
       prettier:
         specifier: ^3.3.3
         version: 3.3.3
@@ -8976,13 +8974,8 @@ packages:
   mz@2.7.0:
     resolution: {integrity: sha512-z81GNO7nnYMEhrGh9LeymoE4+Yr0Wn5McHIZMK5cfQCl+NDX08sCZgUc9/6MHni9IWuFLm1Z3HTCXu2z9fN62Q==}
 
-  nanoid@3.3.6:
-    resolution: {integrity: sha512-BGcqMMJuToF7i1rt+2PWSNVnWIkGCU78jBG3RxO/bZlnZPK2Cmi2QaffxGO/2RvWi9sL+FAiRiXMgsyxQ1DIDA==}
-    engines: {node: ^10 || ^12 || ^13.7 || ^14 || >=15.0.1}
-    hasBin: true
-
-  nanoid@3.3.7:
-    resolution: {integrity: sha512-eSRppjcPIatRIMC1U6UngP8XFcz8MQWGQdt1MTBQ7NaAmvXDfvNxbvWV3x2y6CdEUciCSsDHDQZbhYaB8QEo2g==}
+  nanoid@3.3.8:
+    resolution: {integrity: sha512-WNLf5Sd8oZxOm+TzppcYk8gVOgP+l58xNy58D0nbUnOxOWRWvlcCV4kUF7ltmI6PsrLl/BgKEyS4mqsGChFN0w==}
     engines: {node: ^10 || ^12 || ^13.7 || ^14 || >=15.0.1}
     hasBin: true
 
@@ -9518,9 +9511,6 @@ packages:
   picocolors@1.0.0:
     resolution: {integrity: sha512-1fygroTLlHu66zi26VoTDv8yRgm0Fccecssto+MhsZ0D/DGW2sm8E8AjW7NU5VVTRt5GxbeZ5qBuJr+HyLYkjQ==}
 
-  picocolors@1.1.0:
-    resolution: {integrity: sha512-TQ92mBOW0l3LeMeyLV6mzy/kWr8lkd/hp3mTg7wYK7zJhuBStmGMBG0BdeDZS/dZx1IukaX6Bk11zcln25o1Aw==}
-
   picocolors@1.1.1:
     resolution: {integrity: sha512-xceH2snhtb5M9liqDsmEw56le376mTZkEX/jEb/RxNFyegNul7eNslCXP9FDj/Lcu0X8KEyMceP2ntpaHrDEVA==}
 
@@ -9612,10 +9602,6 @@ packages:
     resolution: {integrity: sha512-PS08Iboia9mts/2ygV3eLpY5ghnUcfLV/EXTOW1E2qYxJKGGBUtNjN76FYHnMs36RmARn41bC0AZmn+rR0OVpQ==}
     engines: {node: ^10 || ^12 || >=14}
 
-  postcss@8.4.47:
-    resolution: {integrity: sha512-56rxCq7G/XfB4EkXq9Egn5GCqugWvDFjafDOThIdMBsI15iqPqR5r15TfSr1YPYeEI19YeaXMCbY6u88Y76GLQ==}
-    engines: {node: ^10 || ^12 || >=14}
-
   postcss@8.4.49:
     resolution: {integrity: sha512-OCVPnIObs4N29kxTjzLfUryOkvZEq+pf8jTF0lg8E7uETuWHA+v7j3c/xJmiqpX450191LlmZfUKkXxkTry7nA==}
     engines: {node: ^10 || ^12 || >=14}
@@ -11575,7 +11561,7 @@ snapshots:
     dependencies:
       '@ai-sdk/provider': 0.0.24
       eventsource-parser: 1.1.2
-      nanoid: 3.3.6
+      nanoid: 3.3.8
       secure-json-parse: 2.7.0
     optionalDependencies:
       zod: 3.23.8
@@ -17720,7 +17706,7 @@ snapshots:
       eventsource-parser: 1.1.2
       json-schema: 0.4.0
       jsondiffpatch: 0.6.0
-      nanoid: 3.3.6
+      nanoid: 3.3.8
       secure-json-parse: 2.7.0
       zod-to-json-schema: 3.23.2(zod@3.23.8)
     optionalDependencies:
@@ -17933,14 +17919,14 @@ snapshots:
 
   asynckit@0.4.0: {}
 
-  autoprefixer@10.4.19(postcss@8.4.47):
+  autoprefixer@10.4.19(postcss@8.4.49):
     dependencies:
       browserslist: 4.23.0
       caniuse-lite: 1.0.30001599
       fraction.js: 4.3.7
       normalize-range: 0.1.2
       picocolors: 1.0.0
-      postcss: 8.4.47
+      postcss: 8.4.49
       postcss-value-parser: 4.2.0
 
   available-typed-arrays@1.0.7:
@@ -22282,9 +22268,7 @@ snapshots:
       object-assign: 4.1.1
       thenify-all: 1.6.0
 
-  nanoid@3.3.6: {}
-
-  nanoid@3.3.7: {}
+  nanoid@3.3.8: {}
 
   natural-compare@1.4.0: {}
 
@@ -22851,8 +22835,6 @@ snapshots:
 
   picocolors@1.0.0: {}
 
-  picocolors@1.1.0: {}
-
   picocolors@1.1.1: {}
 
   picomatch@2.3.1: {}
@@ -22926,19 +22908,13 @@ snapshots:
 
   postcss@8.4.31:
     dependencies:
-      nanoid: 3.3.7
+      nanoid: 3.3.8
       picocolors: 1.1.1
       source-map-js: 1.2.1
 
-  postcss@8.4.47:
-    dependencies:
-      nanoid: 3.3.7
-      picocolors: 1.1.0
-      source-map-js: 1.2.1
-
   postcss@8.4.49:
     dependencies:
-      nanoid: 3.3.7
+      nanoid: 3.3.8
       picocolors: 1.1.1
       source-map-js: 1.2.1
 

web/Dockerfile

@@ -95,6 +95,9 @@ ARG BUILDPLATFORM
 
 WORKDIR /app
 
+ARG NEXT_PUBLIC_BUILD_ID
+ENV BUILD_ID=$NEXT_PUBLIC_BUILD_ID
+
 ENV NODE_ENV production
 # Uncomment the following line in case you want to disable telemetry during runtime.
 ENV NEXT_TELEMETRY_DISABLED 1

web/package.json

@@ -1,6 +1,6 @@
 {
   "name": "web",
-  "version": "3.5.3",
+  "version": "3.6.2",
   "private": true,
   "license": "MIT",
   "engines": {
@@ -175,7 +175,6 @@
     "jest": "^29.7.0",
     "jest-environment-jsdom": "^29.7.0",
     "node-mocks-http": "^1.14.1",
-    "postcss": "^8.4.47",
     "prettier": "^3.3.3",
     "prettier-plugin-tailwindcss": "^0.6.6",
     "tailwindcss": "^3.4.17",
@@ -193,7 +192,8 @@
   },
   "pnpm": {
     "overrides": {
-      "jsonpath-plus": "10.0.7"
+      "jsonpath-plus": "10.0.7",
+      "nanoid": "^3.3.8"
     }
   }
 }

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

@@ -204,6 +204,7 @@ describe("/api/public/observations API Endpoint", () => {
       trace_id: traceId,
       name: "generation-name",
       start_time: new Date("2021-01-01T00:00:00.000Z").getTime(),
+      event_ts: new Date("2021-01-01T00:00:00.000Z").getTime(),
       end_time: new Date("2021-01-01T00:00:00.000Z").getTime(),
       project_id: "7a88fb47-b4e2-43b8-a06c-a5ce950dc53a",
       type: "GENERATION",
@@ -214,6 +215,7 @@ describe("/api/public/observations API Endpoint", () => {
       trace_id: traceId,
       name: "generation-name",
       start_time: new Date("2021-02-01T00:00:00.000Z").getTime(),
+      event_ts: new Date("2021-02-01T00:00:00.000Z").getTime(),
       end_time: new Date("2021-02-01T00:00:00.000Z").getTime(),
       project_id: "7a88fb47-b4e2-43b8-a06c-a5ce950dc53a",
       type: "SPAN",
@@ -223,6 +225,7 @@ describe("/api/public/observations API Endpoint", () => {
       trace_id: traceId,
       name: "generation-name",
       start_time: new Date("2021-03-01T00:00:00.000Z").getTime(),
+      event_ts: new Date("2021-03-01T00:00:00.000Z").getTime(),
       project_id: "7a88fb47-b4e2-43b8-a06c-a5ce950dc53a",
       type: "EVENT",
     });
@@ -231,6 +234,7 @@ describe("/api/public/observations API Endpoint", () => {
       trace_id: traceId,
       name: "generation-name",
       start_time: new Date("2021-04-01T00:00:00.000Z").getTime(),
+      event_ts: new Date("2021-04-01T00:00:00.000Z").getTime(),
       end_time: new Date("2021-04-01T00:00:00.000Z").getTime(),
       project_id: "7a88fb47-b4e2-43b8-a06c-a5ce950dc53a",
       type: "GENERATION",

web/src/__tests__/async/sessions-ui-table.servertest.ts

@@ -0,0 +1,192 @@
+import { v4 } from "uuid";
+import { prisma } from "@langfuse/shared/src/db";
+import {
+  createObservation,
+  createObservationsCh,
+  createOrgProjectAndApiKey,
+  createTracesCh,
+  getSessionsWithMetrics,
+} from "@langfuse/shared/src/server";
+import { createTrace, getSessionsTable } from "@langfuse/shared/src/server";
+
+describe("trpc.sessions", () => {
+  describe("GET sessions.all", () => {
+    it("should GET all session", async () => {
+      const { projectId } = await createOrgProjectAndApiKey();
+      const sessionId = v4();
+
+      await prisma.traceSession.create({
+        data: {
+          id: sessionId,
+          projectId: projectId,
+        },
+      });
+
+      const traces = [
+        createTrace({ session_id: sessionId, project_id: projectId }),
+        createTrace({ session_id: sessionId, project_id: projectId }),
+      ];
+
+      await createTracesCh(traces);
+
+      const uiSessions = await getSessionsTable({
+        projectId: projectId,
+        filter: [],
+        orderBy: null,
+        limit: 10000,
+        page: 0,
+      });
+
+      expect(uiSessions.length).toBe(1);
+      expect(uiSessions[0].session_id).toBe(sessionId);
+      expect(uiSessions[0].trace_count).toBe(2);
+      expect(uiSessions[0].trace_tags).toEqual(["doe", "john"]);
+    });
+  });
+
+  it("should GET all session filtered by trace attribute only", async () => {
+    const { projectId } = await createOrgProjectAndApiKey();
+    const sessionId = v4();
+
+    await prisma.traceSession.create({
+      data: {
+        id: sessionId,
+        projectId: projectId,
+      },
+    });
+
+    const traces = [
+      createTrace({
+        session_id: sessionId,
+        project_id: projectId,
+        user_id: "user1",
+      }),
+      createTrace({
+        session_id: sessionId,
+        project_id: projectId,
+        user_id: undefined,
+      }),
+    ];
+
+    await createTracesCh(traces);
+
+    const uiSessions = await getSessionsTable({
+      projectId: projectId,
+      filter: [
+        {
+          column: "userIds",
+          type: "stringOptions",
+          operator: "any of",
+          value: ["user1"],
+        },
+      ],
+      orderBy: null,
+      limit: 10000,
+      page: 0,
+    });
+
+    expect(uiSessions.length).toBe(1);
+    expect(uiSessions[0].session_id).toBe(sessionId);
+    expect(uiSessions[0].trace_count).toBe(2);
+    expect(uiSessions[0].trace_tags).toEqual(["doe", "john"]);
+    expect(uiSessions[0].user_ids).toEqual(["user1"]);
+  });
+
+  it("should GET metrics for a list of sessions", async () => {
+    const { projectId } = await createOrgProjectAndApiKey();
+    const sessionId1 = v4();
+    const sessionId2 = v4();
+
+    await prisma.traceSession.createMany({
+      data: [
+        {
+          id: sessionId1,
+          projectId: projectId,
+        },
+        {
+          id: sessionId2,
+          projectId: projectId,
+        },
+      ],
+    });
+
+    const traces = [
+      createTrace({
+        session_id: sessionId1,
+        project_id: projectId,
+        user_id: "user1",
+      }),
+      createTrace({
+        session_id: sessionId1,
+        project_id: projectId,
+        user_id: "user2",
+      }),
+      createTrace({
+        session_id: sessionId2,
+        project_id: projectId,
+        user_id: "user3",
+      }),
+    ];
+
+    const observations = traces.flatMap((trace) => [
+      createObservation({
+        trace_id: trace.id,
+        project_id: projectId,
+      }),
+      createObservation({
+        trace_id: trace.id,
+        project_id: projectId,
+      }),
+    ]);
+
+    await createObservationsCh(observations);
+
+    await createTracesCh(traces);
+
+    const sessions = await getSessionsWithMetrics({
+      projectId: projectId,
+      filter: [
+        {
+          column: "id",
+          type: "stringOptions",
+          operator: "any of",
+          value: [sessionId1, sessionId2],
+        },
+      ],
+    });
+
+    expect(sessions.length).toBe(2);
+
+    // Session 1 checks
+    const session1 = sessions.find((s) => s.session_id === sessionId1);
+    expect(session1).toBeDefined();
+    expect(session1?.trace_count).toBe(2);
+    expect(session1?.user_ids).toEqual(
+      expect.arrayContaining(["user1", "user2"]),
+    );
+    expect(session1?.trace_tags).toEqual(["doe", "john"]);
+    expect(session1?.total_observations).toEqual(4);
+
+    expect(Number(session1?.session_input_cost)).toBeGreaterThan(0);
+    expect(Number(session1?.session_output_cost)).toBeGreaterThan(0);
+    expect(Number(session1?.session_total_cost)).toBeGreaterThan(0);
+    expect(Number(session1?.session_input_usage)).toBeGreaterThan(0);
+    expect(Number(session1?.session_output_usage)).toBeGreaterThan(0);
+    expect(Number(session1?.session_total_usage)).toBeGreaterThan(0);
+
+    // Session 2 checks
+    const session2 = sessions.find((s) => s.session_id === sessionId2);
+    expect(session2).toBeDefined();
+    expect(session2?.trace_count).toBe(1);
+    expect(session2?.user_ids).toEqual(["user3"]);
+    expect(session2?.trace_tags).toEqual(["doe", "john"]);
+    expect(session2?.total_observations).toEqual(2);
+
+    expect(Number(session2?.session_input_cost)).toBeGreaterThan(0);
+    expect(Number(session2?.session_output_cost)).toBeGreaterThan(0);
+    expect(Number(session2?.session_total_cost)).toBeGreaterThan(0);
+    expect(Number(session2?.session_input_usage)).toBeGreaterThan(0);
+    expect(Number(session2?.session_output_usage)).toBeGreaterThan(0);
+    expect(Number(session2?.session_total_usage)).toBeGreaterThan(0);
+  });
+});

web/src/__tests__/zod.servertest.ts

@@ -1,4 +1,4 @@
-import { paginationZod } from "@langfuse/shared";
+import { paginationZod, parseJsonPrioritised } from "@langfuse/shared";
 import { ZodError } from "zod";
 
 // Create test cases
@@ -24,3 +24,31 @@ describe("Pagination Zod Schema", () => {
     expect(() => paginationZod.limit.parse("abc")).toThrowError(ZodError);
   });
 });
+
+describe("parseJsonPrioritised", () => {
+  it.each([
+    ["test", "test"], // Raw string
+    ['{"hello": "world"}', { hello: "world" }], // Simple object
+    ["[1, 2, 3]", [1, 2, 3]], // Array
+    ['{"nested": {"key": "value"}}', { nested: { key: "value" } }], // Nested object
+    ["[]", []], // Empty array
+    ["{}", {}], // Empty object
+    ['"simple string"', "simple string"], // Quoted string
+    ["42", 42], // Number
+    ["true", true], // Boolean true
+    ["false", false], // Boolean false
+    ["null", "null"], // Null
+    ["", ""], // Empty string
+    ["invalid{json}", "invalid{json}"], // Invalid JSON string
+    ['[null, 123, "abc"]', [null, 123, "abc"]], // Mixed array
+    [
+      '{"array": [1, 2], "nested": {"key": "value"}}',
+      { array: [1, 2], nested: { key: "value" } },
+    ], // Complex object
+  ])(
+    "should parse input correctly  (%s, %s)",
+    (input: string, expectedOutput: any) => {
+      expect(parseJsonPrioritised(input)).toEqual(expectedOutput);
+    },
+  );
+});

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

@@ -34,6 +34,7 @@ export type ModelTableRow = {
   prices?: Record<string, number>;
   tokenizerId?: string;
   config?: Prisma.JsonValue;
+  lastUsed?: Date | null;
   serverResponse: GetModelResult;
 };
 
@@ -51,6 +52,7 @@ const modelConfigDescriptions = {
     "Some tokenizers require additional configuration (e.g. openai tiktoken). See docs for details.",
   maintainer:
     "Maintainer of the model. Langfuse managed models can be cloned, user managed models can be edited and deleted. To supersede a Langfuse managed model, set the custom model name to the Langfuse model name.",
+  lastUsed: "Start time of the latest generation using this model",
 } as const;
 
 export default function ModelTable({ projectId }: { projectId: string }) {
@@ -198,6 +200,20 @@ export default function ModelTable({ projectId }: { projectId: string }) {
         ) : null;
       },
     },
+    {
+      accessorKey: "lastUsed",
+      id: "lastUsed",
+      header: "Last used",
+      headerTooltip: {
+        description: modelConfigDescriptions.lastUsed,
+      },
+      enableHiding: true,
+      size: 120,
+      cell: ({ row }) => {
+        const value: Date | null | undefined = row.getValue("lastUsed");
+        return value?.toLocaleString() ?? "";
+      },
+    },
     {
       accessorKey: "actions",
       header: "Actions",
@@ -246,6 +262,7 @@ export default function ModelTable({ projectId }: { projectId: string }) {
       prices: model.prices,
       tokenizerId: model.tokenizerId ?? undefined,
       config: model.tokenizerConfig,
+      lastUsed: model.lastUsed,
       serverResponse: model,
     };
   };

web/src/constants/VERSION.ts

@@ -1 +1 @@
-export const VERSION = "v3.5.3";
+export const VERSION = "v3.6.2";

web/src/env.mjs

@@ -9,6 +9,7 @@ export const env = createEnv({
   server: {
     DATABASE_URL: z.string().url(),
     NODE_ENV: z.enum(["development", "test", "production"]),
+    BUILD_ID: z.string().optional(),
     NEXTAUTH_SECRET:
       process.env.NODE_ENV === "production"
         ? z.string().min(1)
@@ -150,9 +151,11 @@ export const env = createEnv({
 
     // clickhouse
     CLICKHOUSE_URL: z.string().url(),
+    CLICKHOUSE_CLUSTER_NAME: z.string().default("default"),
+    CLICKHOUSE_DB: z.string().default("default"),
     CLICKHOUSE_USER: z.string(),
     CLICKHOUSE_PASSWORD: z.string(),
-    CLICKHOUSE_CLUSTER_ENABLED: z.enum(["true", "false"]).default("false"),
+    CLICKHOUSE_CLUSTER_ENABLED: z.enum(["true", "false"]).default("true"),
 
     // EE ui customization
     LANGFUSE_UI_API_HOST: z.string().optional(),
@@ -301,6 +304,7 @@ export const env = createEnv({
     NEXT_PUBLIC_DEMO_ORG_ID: process.env.NEXT_PUBLIC_DEMO_ORG_ID,
     DATABASE_URL: process.env.DATABASE_URL,
     NODE_ENV: process.env.NODE_ENV,
+    BUILD_ID: process.env.BUILD_ID,
     NEXT_PUBLIC_BUILD_ID: process.env.NEXT_PUBLIC_BUILD_ID,
     NEXTAUTH_SECRET: process.env.NEXTAUTH_SECRET,
     NEXTAUTH_COOKIE_DOMAIN: process.env.NEXTAUTH_COOKIE_DOMAIN,
@@ -378,12 +382,10 @@ export const env = createEnv({
     AUTH_CUSTOM_ISSUER: process.env.AUTH_CUSTOM_ISSUER,
     AUTH_CUSTOM_NAME: process.env.AUTH_CUSTOM_NAME,
     AUTH_CUSTOM_SCOPE: process.env.AUTH_CUSTOM_SCOPE,
-    AUTH_CUSTOM_CLIENT_AUTH_METHOD:
-      process.env.AUTH_CUSTOM_CLIENT_AUTH_METHOD,
+    AUTH_CUSTOM_CLIENT_AUTH_METHOD: process.env.AUTH_CUSTOM_CLIENT_AUTH_METHOD,
     AUTH_CUSTOM_ALLOW_ACCOUNT_LINKING:
       process.env.AUTH_CUSTOM_ALLOW_ACCOUNT_LINKING,
-    AUTH_IGNORE_ACCOUNT_FIELDS:
-      process.env.AUTH_IGNORE_ACCOUNT_FIELDS,
+    AUTH_IGNORE_ACCOUNT_FIELDS: process.env.AUTH_IGNORE_ACCOUNT_FIELDS,
     AUTH_DOMAINS_WITH_SSO_ENFORCEMENT:
       process.env.AUTH_DOMAINS_WITH_SSO_ENFORCEMENT,
     AUTH_DISABLE_USERNAME_PASSWORD: process.env.AUTH_DISABLE_USERNAME_PASSWORD,
@@ -445,6 +447,8 @@ export const env = createEnv({
     NEXT_PUBLIC_CRISP_WEBSITE_ID: process.env.NEXT_PUBLIC_CRISP_WEBSITE_ID,
     // clickhouse
     CLICKHOUSE_URL: process.env.CLICKHOUSE_URL,
+    CLICKHOUSE_CLUSTER_NAME: process.env.CLICKHOUSE_CLUSTER_NAME,
+    CLICKHOUSE_DB: process.env.CLICKHOUSE_DB,
     CLICKHOUSE_USER: process.env.CLICKHOUSE_USER,
     CLICKHOUSE_PASSWORD: process.env.CLICKHOUSE_PASSWORD,
     CLICKHOUSE_CLUSTER_ENABLED: process.env.CLICKHOUSE_CLUSTER_ENABLED,

web/src/features/datasets/components/DatasetCompareRunsTable.tsx

@@ -26,7 +26,7 @@ import { DatasetCompareRunPeekView } from "@/src/features/datasets/components/Da
 import { useClickhouse } from "@/src/components/layouts/ClickhouseAdminToggle";
 import { getQueryKey } from "@trpc/react-query";
 import { useQueryClient } from "@tanstack/react-query";
-
+import _ from "lodash";
 export type RunMetrics = {
   id: string;
   scores: ScoreAggregate;
@@ -182,7 +182,7 @@ export function DatasetCompareRunsTable(props: {
           ({ datasetItemId, trace, observation, scores }) => {
             if (!itemsAcc[datasetItemId]) itemsAcc[datasetItemId] = {};
 
-            itemsAcc[datasetItemId][runId] = {
+            _.set(itemsAcc[datasetItemId], runId, {
               id: runId,
               traceId: trace?.id ?? "",
               observationId: observation?.id ?? undefined,
@@ -196,7 +196,7 @@ export function DatasetCompareRunsTable(props: {
                     : usdFormatter(trace?.totalCost)) ?? undefined,
               },
               scores,
-            };
+            });
           },
         );
 

web/src/features/datasets/server/service.ts

@@ -176,7 +176,7 @@ export const createTempTableInClickhouse = async (
   clickhouseSession: string,
 ) => {
   const query = `
-      CREATE TABLE IF NOT EXISTS ${tableName} ${env.CLICKHOUSE_CLUSTER_ENABLED === "true" ? "ON CLUSTER default" : ""}
+      CREATE TABLE IF NOT EXISTS ${tableName} ${env.CLICKHOUSE_CLUSTER_ENABLED === "true" ? "ON CLUSTER " + env.CLICKHOUSE_CLUSTER_NAME : ""}
       (
           project_id String,    
           run_id String,  
@@ -531,10 +531,10 @@ export const getRunItemsByRunIdOrItemId = async (
 ) => {
   const [traceScores, observationAggregates, traceAggregate] =
     await Promise.all([
-      getScoresForTraces(
+      getScoresForTraces({
         projectId,
-        runItems.map((ri) => ri.traceId),
-      ),
+        traceIds: runItems.map((ri) => ri.traceId),
+      }),
       getLatencyAndTotalCostForObservations(
         projectId,
         runItems

web/src/features/models/validation.ts

@@ -17,11 +17,13 @@ export const GetModelResultSchema = z.object({
   projectId: z.string().nullable(),
   modelName: z.string(),
   matchPattern: z.string(),
-  tokenizerConfig: z
-    .record(z.union([z.string(), z.coerce.number()]))
-    .nullable(),
+  tokenizerConfig: z.union([
+    z.record(z.union([z.string(), z.coerce.number()])).nullable(),
+    z.string(),
+  ]),
   tokenizerId: TokenizerSchema,
   prices: PriceMapSchema,
+  lastUsed: z.date().nullish(),
 });
 
 export type GetModelResult = z.infer<typeof GetModelResultSchema>;
@@ -75,3 +77,10 @@ export enum PriceUnit {
   Per1KUnits = "per 1K units",
   Per1MUnits = "per 1M units",
 }
+
+export const ModelLastUsedQueryResult = z.array(
+  z.object({
+    modelId: z.string(),
+    lastUsed: z.coerce.date(),
+  }),
+);

web/src/features/projects/server/projectsRouter.ts

@@ -132,18 +132,18 @@ export const projectsRouter = createTRPCRouter({
         action: "delete",
       });
 
-      // Delete API keys from DB first
+      // API keys need to be deleted from cache. Otherwise, they will still be valid.
+      await new ApiAuthService(ctx.prisma, redis).invalidateProjectApiKeys(
+        input.projectId,
+      );
+
+      // Delete API keys from DB
       await ctx.prisma.apiKey.deleteMany({
         where: {
           projectId: input.projectId,
         },
       });
 
-      // API keys need to be deleted from cache. Otherwise, they will still be valid.
-      await new ApiAuthService(ctx.prisma, redis).invalidateProjectApiKeys(
-        input.projectId,
-      );
-
       await ctx.prisma.project.update({
         where: {
           id: input.projectId,

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

@@ -61,7 +61,7 @@ export const generateObservationsForPublicApi = async (props: QueryType) => {
       WHERE o.project_id = {projectId: String}
       ${traceFilter ? `AND t.project_id = {projectId: String}` : ""}
       AND ${appliedFilter.query}
-      ORDER BY start_time desc
+      ORDER BY event_ts desc
       LIMIT 1 by id, project_id
       ${props.limit !== undefined && props.page !== undefined ? `LIMIT {limit: Int32} OFFSET {offset: Int32}` : ""}
       `;

web/src/features/telemetry/index.ts

@@ -10,11 +10,11 @@ import {
 } from "@langfuse/shared/src/server";
 import { env } from "@/src/env.mjs";
 
-// Interval between jobs in milliseconds
-const JOB_INTERVAL_MINUTES = Prisma.raw("60");
+// Interval between jobs in minutes
+const JOB_INTERVAL_MINUTES = Prisma.raw("24 * 60"); // 24 hours
 
 // Timeout for job in minutes, if job is not finished in this time, it will be retried
-const JOB_TIMEOUT_MINUTES = Prisma.raw("10");
+const JOB_TIMEOUT_MINUTES = Prisma.raw("10"); // 10 minutes
 
 export async function telemetry() {
   try {

web/src/observability.config.ts

@@ -26,6 +26,7 @@ if (!process.env.VERCEL && process.env.NEXT_PUBLIC_LANGFUSE_CLOUD_REGION) {
   const sdk = new NodeSDK({
     resource: new Resource({
       "service.name": env.OTEL_SERVICE_NAME,
+      "service.version": env.BUILD_ID,
     }),
     traceExporter: new OTLPTraceExporter({
       url: `${env.OTEL_EXPORTER_OTLP_ENDPOINT}/v1/traces`,
@@ -34,6 +35,14 @@ if (!process.env.VERCEL && process.env.NEXT_PUBLIC_LANGFUSE_CLOUD_REGION) {
       new IORedisInstrumentation(),
       new HttpInstrumentation({
         requireParentforOutgoingSpans: true,
+        ignoreIncomingRequestHook: (req) => {
+          // Ignore health checks
+          return [
+            "/api/public/health",
+            "/api/public/ready",
+            "/api/health",
+          ].some((path) => req.url?.includes(path));
+        },
         ignoreOutgoingRequestHook: (req) => {
           return req.host === "127.0.0.1";
         },

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

@@ -2,11 +2,22 @@ import { type NextApiRequest, type NextApiResponse } from "next";
 import { z } from "zod";
 import { logger, QueueName, getQueue } from "@langfuse/shared/src/server";
 import { env } from "@/src/env.mjs";
+import { prisma } from "@langfuse/shared/src/db";
 
 /* 
 This API route is used by Langfuse Cloud to retry failed bullmq jobs.
 */
 
+const BullStatus = z.enum([
+  "completed",
+  "failed",
+  "active",
+  "delayed",
+  "prioritized",
+  "paused",
+  "wait",
+]);
+
 const ManageBullBody = z.discriminatedUnion("action", [
   z.object({
     action: z.literal("retry"),
@@ -15,15 +26,18 @@ const ManageBullBody = z.discriminatedUnion("action", [
   z.object({
     action: z.literal("remove"),
     queueNames: z.array(z.string()),
-    bullStatus: z.enum([
-      "completed",
-      "failed",
-      "active",
-      "delayed",
-      "prioritized",
-      "paused",
-      "wait",
-    ]),
+    bullStatus: BullStatus,
+  }),
+  z.object({
+    action: z.literal("backup"),
+    queueName: z.string(),
+    bullStatus: BullStatus,
+    numberOfEvents: z.number(),
+  }),
+  z.object({
+    action: z.literal("restore"),
+    queueName: z.string(),
+    numberOfEvents: z.number(),
   }),
 ]);
 
@@ -162,6 +176,21 @@ export default async function handler(
       return res.status(200).json({ message: "Retried all jobs" });
     }
 
+    if (req.method === "POST" && body.data.action === "backup") {
+      await backUpEvents(
+        body.data.queueName as QueueName,
+        body.data.numberOfEvents,
+        body.data.bullStatus,
+      );
+    }
+
+    if (req.method === "POST" && body.data.action === "restore") {
+      await restoreEvents(
+        body.data.queueName as QueueName,
+        body.data.numberOfEvents,
+      );
+    }
+
     // return not implemented error
     res.status(404).json({ error: "Action does not exist" });
   } catch (e) {
@@ -169,3 +198,69 @@ export default async function handler(
     res.status(500).json({ error: e });
   }
 }
+
+const backUpEvents = async (
+  queueName: QueueName,
+  numberOfEvents: number,
+  bullStatus: z.infer<typeof BullStatus>,
+) => {
+  const queue = getQueue(queueName);
+  let processedEvents = 0;
+  const batchSize = 1000;
+
+  while (processedEvents < numberOfEvents) {
+    const remainingEvents = numberOfEvents - processedEvents;
+    const currentBatchSize = Math.min(batchSize, remainingEvents);
+
+    const events = await queue?.getJobs(
+      [bullStatus],
+      0,
+      currentBatchSize,
+      true,
+    );
+
+    if (!events || events.length === 0) {
+      break;
+    }
+
+    await prisma.queueBackUp.createMany({
+      data: events.map((event) => ({
+        queueName,
+        content: event,
+        projectId: event.data.projectId ?? undefined,
+        createdAt: new Date(),
+      })),
+    });
+
+    // remove events from the queue but might throw in case if the job is already processing
+    await Promise.all(
+      events.map(async (event) => {
+        try {
+          await event.remove();
+        } catch (error) {
+          logger.error(`Failed to remove event ${event.id}:`, error);
+        }
+      }),
+    );
+
+    processedEvents += events.length;
+  }
+};
+
+const restoreEvents = async (queueName: QueueName, numberOfEvents: number) => {
+  const queue = getQueue(queueName);
+
+  const queueBackUp = await prisma.queueBackUp.findMany({
+    where: {
+      queueName,
+    },
+    orderBy: {
+      createdAt: "desc",
+    },
+    take: numberOfEvents,
+  });
+
+  await queue?.addBulk(
+    queueBackUp.map((event) => ({ name: queueName, data: event.content })),
+  );
+};

web/src/pages/api/public/media/[mediaId].ts

@@ -73,6 +73,7 @@ export default withMiddlewares({
     }),
     bodySchema: PatchMediaBodySchema,
     responseSchema: z.void(),
+    rateLimitResource: "ingestion",
     fn: async ({ query, body, auth }) => {
       if (auth.scope.accessLevel !== "all") throw new ForbiddenError();
 

web/src/pages/api/public/media/index.ts

@@ -26,6 +26,7 @@ export default withMiddlewares({
     bodySchema: GetMediaUploadUrlQuerySchema,
     responseSchema: GetMediaUploadUrlResponseSchema,
     successStatusCode: 201,
+    rateLimitResource: "ingestion",
     fn: async ({ body, auth }) => {
       if (auth.scope.accessLevel !== "all") throw new ForbiddenError();
 

web/src/pages/api/public/sessions/[sessionId].ts

@@ -22,28 +22,9 @@ export default withMiddlewares({
         operation: "scores.countAll",
         user: null,
         pgExecution: async () => {
-          const session = await prisma.traceSession.findUnique({
-            where: {
-              id_projectId: {
-                id: sessionId,
-                projectId: auth.scope.projectId,
-              },
-            },
-            select: {
-              id: true,
-              createdAt: true,
-              projectId: true,
-              traces: true,
-            },
-          });
-
-          if (!session) {
-            throw new LangfuseNotFoundError(
-              "Session not found within authorized project",
-            );
-          }
-
-          return session;
+          throw new LangfuseNotFoundError(
+            "Session not found within authorized project",
+          );
         },
         clickhouseExecution: async () => {
           const session = await prisma.traceSession.findUnique({

web/src/pages/api/public/traces/[traceId].ts

@@ -91,11 +91,11 @@ export default withMiddlewares({
               trace?.timestamp,
               true,
             ),
-            getScoresForTraces(
-              auth.scope.projectId,
-              [traceId],
-              trace?.timestamp,
-            ),
+            getScoresForTraces({
+              projectId: auth.scope.projectId,
+              traceIds: [traceId],
+              timestamp: trace?.timestamp,
+            }),
           ]);
 
           const uniqueModels: string[] = Array.from(

web/src/server/api/routers/comments.ts

@@ -11,7 +11,10 @@ import { auditLog } from "@/src/features/audit-logs/auditLog";
 import { TRPCError } from "@trpc/server";
 import { validateCommentReferenceObject } from "@/src/features/comments/validateCommentReferenceObject";
 import { measureAndReturnApi } from "@/src/server/utils/checkClickhouseAccess";
-import { getTracesIdentifierForSession } from "@langfuse/shared/src/server";
+import {
+  getTracesIdentifierForSession,
+  logger,
+} from "@langfuse/shared/src/server";
 
 export const commentsRouter = createTRPCRouter({
   create: protectedProjectProcedure
@@ -56,7 +59,7 @@ export const commentsRouter = createTRPCRouter({
 
         return comment;
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call comments.create", error);
         if (error instanceof TRPCError) {
           throw error;
         }
@@ -115,7 +118,7 @@ export const commentsRouter = createTRPCRouter({
           before: comment,
         });
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call comments.delete", error);
         if (error instanceof TRPCError) {
           throw error;
         }
@@ -172,7 +175,7 @@ export const commentsRouter = createTRPCRouter({
 
         return comments;
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call comments.getByObjectId", error);
         if (error instanceof TRPCError) {
           throw error;
         }
@@ -207,7 +210,7 @@ export const commentsRouter = createTRPCRouter({
         });
         return new Map([[input.objectId, commentCount]]);
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call comments.getCountByObjectId", error);
         if (error instanceof TRPCError) {
           throw error;
         }
@@ -251,7 +254,7 @@ export const commentsRouter = createTRPCRouter({
           ]),
         );
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call comments.getCountByObjectType", error);
         if (error instanceof TRPCError) {
           throw error;
         }
@@ -276,44 +279,10 @@ export const commentsRouter = createTRPCRouter({
           operation: "comments.getTraceCommentCountsBySessionId",
           user: ctx.session.user ?? undefined,
           pgExecution: async () => {
-            const session = await ctx.prisma.traceSession.findFirst({
-              where: {
-                id: input.sessionId,
-                projectId: input.projectId,
-              },
-              include: {
-                traces: {
-                  select: {
-                    id: true,
-                  },
-                },
-              },
+            throw new TRPCError({
+              code: "NOT_FOUND",
+              message: "Session not found in project",
             });
-            if (!session) {
-              throw new TRPCError({
-                code: "NOT_FOUND",
-                message: "Session not found in project",
-              });
-            }
-
-            const allTraceCommentCounts = await ctx.prisma.$queryRaw<
-              Array<{ objectId: string; count: bigint }>
-            >`
-            SELECT object_id as "objectId", COUNT(*) as count
-            FROM comments
-            WHERE project_id = ${input.projectId}
-            AND object_type = 'TRACE'
-            GROUP BY object_id
-          `;
-
-            const traceIds = new Set(
-              session.traces.map((t: { id: string }) => t.id),
-            );
-            return new Map(
-              allTraceCommentCounts
-                .filter((c) => traceIds.has(c.objectId))
-                .map(({ objectId, count }) => [objectId, Number(count)]),
-            );
           },
           clickhouseExecution: async () => {
             const clickhouseTraces = await getTracesIdentifierForSession(
@@ -339,8 +308,11 @@ export const commentsRouter = createTRPCRouter({
             );
           },
         });
-      } catch (e) {
-        console.error(e);
+      } catch (error) {
+        logger.error(
+          "Failed to call comments.getTraceCommentCountBySessionId",
+          error,
+        );
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
           message: "Unable to get trace comment counts by session id",

web/src/server/api/routers/models.ts

@@ -5,6 +5,7 @@ import { auditLog } from "@/src/features/audit-logs/auditLog";
 import { isValidPostgresRegex } from "@/src/features/models/server/isValidPostgresRegex";
 import {
   GetModelResultSchema,
+  ModelLastUsedQueryResult,
   UpsertModelSchema,
 } from "@/src/features/models/validation";
 import { throwIfNoProjectAccess } from "@/src/features/rbac/utils/checkProjectAccess";
@@ -13,6 +14,7 @@ import {
   protectedProjectProcedure,
 } from "@/src/server/api/trpc";
 import { ModelUsageUnit, paginationZod } from "@langfuse/shared";
+import { queryClickhouse } from "@langfuse/shared/src/server";
 import { TRPCError } from "@trpc/server";
 
 const ModelAllOptions = z.object({
@@ -20,6 +22,18 @@ const ModelAllOptions = z.object({
   ...paginationZod,
 });
 
+const paginateArray = <T>(params: {
+  limit: number;
+  page: number;
+  data: Array<T>;
+}): Array<T> => {
+  const { data, limit, page } = params;
+  const startIndex = limit * page;
+  const endIndex = startIndex + limit;
+
+  return data.slice(startIndex, endIndex);
+};
+
 export const modelRouter = createTRPCRouter({
   getById: protectedProjectProcedure
     .input(z.object({ projectId: z.string(), modelId: z.string() }))
@@ -68,6 +82,8 @@ export const modelRouter = createTRPCRouter({
   getAll: protectedProjectProcedure
     .input(ModelAllOptions)
     .query(async ({ input, ctx }) => {
+      const { projectId, page, limit } = input;
+
       const [allModelsQueryResult, totalCountQuery] = await Promise.all([
         // All models
         ctx.prisma.$queryRaw`
@@ -93,12 +109,11 @@ export const modelRouter = createTRPCRouter({
             models m
           WHERE
             project_id IS NULL
-            OR project_id = ${input.projectId}
+            OR project_id = ${projectId}
           ORDER BY
             project_id,
             model_name,
             m.created_at DESC NULLS LAST 
-          LIMIT ${input.limit} OFFSET ${input.page * input.limit};
           `,
 
         // Total count
@@ -110,7 +125,7 @@ export const modelRouter = createTRPCRouter({
           SELECT COUNT(DISTINCT (project_id, model_name))
           FROM models
           WHERE project_id IS NULL 
-          OR project_id = '${input.projectId}';
+          OR project_id = ${projectId};
         `,
       ]);
 
@@ -119,8 +134,57 @@ export const modelRouter = createTRPCRouter({
         .parse(allModelsQueryResult);
       const totalCount = z.coerce.number().parse(totalCountQuery[0].count);
 
+      // Check whether model was used in last month
+      const lastUsedQuery = `
+            SELECT 
+                internal_model_id as modelId,
+                MAX(start_time) as lastUsed
+            FROM 
+                observations
+            WHERE
+                project_id = {projectId: String}
+                AND internal_model_id IS NOT NULL
+            GROUP BY
+                internal_model_id
+      `;
+
+      const lastUsedQueryResult = ModelLastUsedQueryResult.safeParse(
+        await queryClickhouse({
+          query: lastUsedQuery,
+          params: { projectId },
+        }),
+      );
+
+      if (lastUsedQueryResult.success) {
+        const lastUsedMap = new Map(
+          lastUsedQueryResult.data.map((res) => [res.modelId, res.lastUsed]),
+        );
+
+        const sortedModels = allModels
+          .map((m) => ({ ...m, lastUsed: lastUsedMap.get(m.id) }))
+          .sort((a, b) => {
+            const lastUsedA = lastUsedMap.get(a.id);
+            const lastUsedB = lastUsedMap.get(b.id);
+
+            if (lastUsedA && lastUsedB) {
+              return lastUsedB.getTime() - lastUsedA.getTime();
+            } else if (lastUsedA) {
+              return -1;
+            } else if (lastUsedB) {
+              return 1;
+            }
+
+            return 0;
+          });
+
+        return {
+          models: paginateArray({ data: sortedModels, page, limit }),
+          totalCount,
+        };
+      }
+
       return {
-        models: allModels,
+        models: paginateArray({ data: allModels, page, limit }),
         totalCount,
       };
     }),

web/src/server/api/routers/sessions.ts

@@ -32,6 +32,8 @@ import {
   getCostForTraces,
   getTracesGroupedByUsers,
   getPublicSessionsFilter,
+  logger,
+  getSessionsWithMetrics,
 } from "@langfuse/shared/src/server";
 
 const SessionFilterOptions = z.object({
@@ -108,14 +110,7 @@ export const sessionRouter = createTRPCRouter({
               sessions: sessions.map((s) => ({
                 id: s.session_id,
                 userIds: s.user_ids,
-                countTraces: s.trace_ids.length,
-                sessionDuration: Number(s.duration) / 1000,
-                inputCost: new Decimal(s.session_input_cost),
-                outputCost: new Decimal(s.session_output_cost),
-                totalCost: new Decimal(s.session_total_cost),
-                promptTokens: Number(s.session_input_usage),
-                completionTokens: Number(s.session_output_usage),
-                totalTokens: Number(s.session_total_usage),
+                countTraces: s.trace_count,
                 traceTags: s.trace_tags,
                 createdAt: new Date(s.min_timestamp),
                 bookmarked:
@@ -129,7 +124,7 @@ export const sessionRouter = createTRPCRouter({
           },
         });
       } catch (e) {
-        console.error(e);
+        logger.error("Unable to call sessions.all", e);
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
           message: "unable to get sessions",
@@ -196,7 +191,7 @@ export const sessionRouter = createTRPCRouter({
           },
         });
       } catch (e) {
-        console.error(e);
+        logger.error("Error in sessions.countAll", e);
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
           message: "unable to get session count",
@@ -273,11 +268,59 @@ export const sessionRouter = createTRPCRouter({
             }));
           },
           clickhouseExecution: async () => {
-            return []; // initial endpoint returns all data from CH.
+            const finalFilter = await getPublicSessionsFilter(input.projectId, [
+              {
+                column: "id",
+                type: "stringOptions",
+                operator: "any of",
+                value: input.sessionIds,
+              },
+            ]);
+            const sessions = await getSessionsWithMetrics({
+              projectId: input.projectId,
+              filter: finalFilter,
+            });
+
+            const prismaSessionInfo = await ctx.prisma.traceSession.findMany({
+              where: {
+                id: {
+                  in: sessions.map((s) => s.session_id),
+                },
+                projectId: input.projectId,
+              },
+              select: {
+                id: true,
+                bookmarked: true,
+                public: true,
+              },
+            });
+
+            return sessions.map((s) => ({
+              id: s.session_id,
+              userIds: s.user_ids,
+              countTraces: s.trace_count,
+              traceTags: s.trace_tags,
+              createdAt: new Date(s.min_timestamp),
+              bookmarked:
+                prismaSessionInfo.find((p) => p.id === s.session_id)
+                  ?.bookmarked ?? false,
+              public:
+                prismaSessionInfo.find((p) => p.id === s.session_id)?.public ??
+                false,
+              trace_count: Number(s.trace_count),
+              total_observations: Number(s.total_observations),
+              sessionDuration: Number(s.duration) / 1000,
+              inputCost: new Decimal(s.session_input_cost),
+              outputCost: new Decimal(s.session_output_cost),
+              totalCost: new Decimal(s.session_total_cost),
+              promptTokens: Number(s.session_input_usage),
+              completionTokens: Number(s.session_output_usage),
+              totalTokens: Number(s.session_total_usage),
+            }));
           },
         });
       } catch (e) {
-        console.error(e);
+        logger.error("Error in sessions.metrics", e);
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
           message: "unable to get session metrics",
@@ -391,7 +434,7 @@ export const sessionRouter = createTRPCRouter({
           },
         });
       } catch (e) {
-        console.error(e);
+        logger.error("Unable to get sessions.filterOptions", e);
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
           message: "unable to get session filter options",
@@ -413,75 +456,10 @@ export const sessionRouter = createTRPCRouter({
           operation: "sessions.byId",
           user: ctx.session.user ?? undefined,
           pgExecution: async () => {
-            const session = await ctx.prisma.traceSession.findFirst({
-              where: {
-                id: input.sessionId,
-                projectId: input.projectId,
-              },
-              include: {
-                traces: {
-                  orderBy: {
-                    timestamp: "asc",
-                  },
-                  select: {
-                    id: true,
-                    userId: true,
-                    name: true,
-                    timestamp: true,
-                  },
-                },
-              },
+            throw new TRPCError({
+              code: "NOT_FOUND",
+              message: "Session not found in project",
             });
-            if (!session) {
-              throw new TRPCError({
-                code: "NOT_FOUND",
-                message: "Session not found in project",
-              });
-            }
-
-            const totalCostQuery = Prisma.sql`
-              SELECT
-                SUM(COALESCE(o."calculated_total_cost", 0)) AS "totalCost"
-              FROM observations_view o
-              JOIN traces t ON t.id = o.trace_id
-              WHERE
-                t."session_id" = ${input.sessionId}
-                AND t."project_id" = ${input.projectId}
-            `;
-
-            const [scores, costData] = await Promise.all([
-              ctx.prisma.score.findMany({
-                where: {
-                  traceId: {
-                    in: session.traces.map((t) => t.id),
-                  },
-                  projectId: input.projectId,
-                },
-              }),
-              // costData
-              ctx.prisma.$queryRaw<Array<{ totalCost: number }>>(
-                totalCostQuery,
-              ),
-            ]);
-
-            const validatedScores = filterAndValidateDbScoreList(
-              scores,
-              traceException,
-            );
-
-            return {
-              ...session,
-              traces: session.traces.map((t) => ({
-                ...t,
-                scores: validatedScores.filter((s) => s.traceId === t.id),
-              })),
-              totalCost: costData[0].totalCost ?? 0,
-              users: [
-                ...new Set(
-                  session.traces.map((t) => t.userId).filter((t) => t !== null),
-                ),
-              ],
-            };
           },
           clickhouseExecution: async () => {
             const postgresSession = await ctx.prisma.traceSession.findFirst({
@@ -514,7 +492,10 @@ export const sessionRouter = createTRPCRouter({
             const [scores, costs] = await Promise.all([
               Promise.all(
                 chunks.map((chunk) =>
-                  getScoresForTraces(input.projectId, chunk),
+                  getScoresForTraces({
+                    projectId: input.projectId,
+                    traceIds: chunk,
+                  }),
                 ),
               ).then((results) => results.flat()),
               Promise.all(
@@ -549,7 +530,7 @@ export const sessionRouter = createTRPCRouter({
           },
         });
       } catch (e) {
-        console.error(e);
+        logger.error("Unable to get sessions.byId", e);
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
           message: "unable to get session",
@@ -593,7 +574,7 @@ export const sessionRouter = createTRPCRouter({
         });
         return session;
       } catch (error) {
-        console.error(error);
+        logger.error("Unable to call sessions.bookmark", error);
         if (
           error instanceof Prisma.PrismaClientKnownRequestError &&
           error.code === "P2025" // Record to update not found
@@ -643,7 +624,7 @@ export const sessionRouter = createTRPCRouter({
           },
         });
       } catch (e) {
-        console.error(e);
+        logger.error("Unable to call sessions.publish", e);
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
           message: "unable to publish session",

web/src/server/api/routers/traces.ts

@@ -291,13 +291,12 @@ export const traceRouter = createTRPCRouter({
             ],
           });
 
-          const scores = await getScoresForTraces(
-            ctx.session.projectId,
-            res.map((r) => r.id),
-            undefined,
-            1000,
-            0,
-          );
+          const scores = await getScoresForTraces({
+            projectId: ctx.session.projectId,
+            traceIds: res.map((r) => r.id),
+            limit: 1000,
+            offset: 0,
+          });
 
           const validatedScores = filterAndValidateDbScoreList(
             scores,
@@ -576,13 +575,11 @@ export const traceRouter = createTRPCRouter({
               input.projectId,
               input.timestamp ?? undefined,
             ),
-            getScoresForTraces(
-              input.projectId,
-              [input.traceId],
-              input.timestamp ?? undefined,
-              undefined,
-              undefined,
-            ),
+            getScoresForTraces({
+              projectId: input.projectId,
+              traceIds: [input.traceId],
+              timestamp: input.timestamp ?? undefined,
+            }),
           ]);
 
           if (!trace) {
@@ -769,7 +766,7 @@ export const traceRouter = createTRPCRouter({
 
         return trace;
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call traces.bookmark", error);
         if (
           error instanceof Prisma.PrismaClientKnownRequestError &&
           error.code === "P2025" // Record to update not found
@@ -839,7 +836,7 @@ export const traceRouter = createTRPCRouter({
           return clickhouseTrace;
         }
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call traces.publish", error);
         if (
           error instanceof Prisma.PrismaClientKnownRequestError &&
           error.code === "P2025" // Record to update not found
@@ -907,7 +904,7 @@ export const traceRouter = createTRPCRouter({
           await upsertTrace(convertTraceDomainToClickhouse(clickhouseTrace));
         }
       } catch (error) {
-        console.error(error);
+        logger.error("Failed to call traces.updateTags", error);
         throw new TRPCError({
           code: "INTERNAL_SERVER_ERROR",
         });

worker/Dockerfile

@@ -54,6 +54,9 @@ RUN apk add --no-cache dumb-init
 
 WORKDIR /app
 
+ARG NEXT_PUBLIC_BUILD_ID
+ENV BUILD_ID=$NEXT_PUBLIC_BUILD_ID
+
 ENV NODE_ENV production
 ENV DOCKER_BUILD 0
 

worker/package.json

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

worker/src/__tests__/batch-export.test.ts

@@ -380,7 +380,17 @@ describe("batch export test suite", () => {
       value: 123,
     });
 
-    await createScoresCh([score]);
+    const qualitativeScore = createScore({
+      project_id: projectId,
+      trace_id: traces[0].id,
+      observation_id: generations[0].id,
+      name: "qualitative_test",
+      value: undefined,
+      string_value: "This is some qualitative text",
+      data_type: "CATEGORICAL",
+    });
+
+    await createScoresCh([score, qualitativeScore]);
     await createObservationsCh(generations);
 
     const stream = await getDatabaseReadStream({
@@ -405,10 +415,13 @@ describe("batch export test suite", () => {
           id: traces[0].id,
           latency: 1,
           test: [score.value],
+          qualitative_test: ["This is some qualitative text"],
         }),
         expect.objectContaining({
           id: traces[1].id,
           latency: 2.123,
+          test: null,
+          qualitative_test: null,
         }),
       ]),
     );

worker/src/constants/VERSION.ts

@@ -1 +1 @@
-export const VERSION = "v3.5.3";
+export const VERSION = "v3.6.2";

worker/src/constants/default-model-prices.json

@@ -8,6 +8,7 @@
     "prices": {
       "input": 2.5e-6,
       "input_cached_tokens": 1.25e-6,
+      "input_cache_read": 1.25e-6,
       "output": 10e-6
     },
     "tokenizer_config": {
@@ -844,7 +845,9 @@
       "output": 15e-6,
       "output_tokens": 15e-6,
       "cache_creation_input_tokens": 3.75e-6,
-      "cache_read_input_tokens": 0.3e-6
+      "input_cache_creation": 3.75e-6,
+      "cache_read_input_tokens": 0.3e-6,
+      "input_cache_read": 0.3e-6
     },
     "tokenizer_config": null,
     "tokenizer_id": "claude"
@@ -1010,7 +1013,9 @@
       "output": 15e-6,
       "output_tokens": 15e-6,
       "cache_creation_input_tokens": 3.75e-6,
-      "cache_read_input_tokens": 0.3e-6
+      "input_cache_creation": 3.75e-6,
+      "cache_read_input_tokens": 0.3e-6,
+      "input_cache_read": 0.3e-6
     },
     "tokenizer_config": null,
     "tokenizer_id": "claude"
@@ -1024,7 +1029,8 @@
     "prices": {
       "input": 0.15e-6,
       "output": 0.6e-6,
-      "input_cached_tokens": 0.075e-6
+      "input_cached_tokens": 0.075e-6,
+      "input_cache_read": 0.075e-6
     },
     "tokenizer_config": {
       "tokensPerName": 1,
@@ -1042,6 +1048,7 @@
     "prices": {
       "input": 0.15e-6,
       "input_cached_tokens": 0.075e-6,
+      "input_cache_read": 0.075e-6,
       "output": 0.6e-6
     },
     "tokenizer_config": {
@@ -1060,6 +1067,7 @@
     "prices": {
       "input": 2.5e-6,
       "input_cached_tokens": 1.25e-6,
+      "input_cache_read": 1.25e-6,
       "output": 10e-6
     },
     "tokenizer_config": {
@@ -1078,8 +1086,10 @@
     "prices": {
       "input": 15e-6,
       "input_cached_tokens": 7.5e-6,
+      "input_cache_read": 7.5e-6,
       "output": 60e-6,
-      "output_reasoning_tokens": 60e-6
+      "output_reasoning_tokens": 60e-6,
+      "output_reasoning": 60e-6
     },
     "tokenizer_config": null,
     "tokenizer_id": null
@@ -1093,8 +1103,10 @@
     "prices": {
       "input": 15e-6,
       "input_cached_tokens": 7.5e-6,
+      "input_cache_read": 7.5e-6,
       "output": 60e-6,
-      "output_reasoning_tokens": 60e-6
+      "output_reasoning_tokens": 60e-6,
+      "output_reasoning": 60e-6
     },
     "tokenizer_config": null,
     "tokenizer_id": null
@@ -1108,8 +1120,10 @@
     "prices": {
       "input": 3e-6,
       "input_cached_tokens": 1.5e-6,
+      "input_cache_read": 1.5e-6,
       "output": 12e-6,
-      "output_reasoning_tokens": 12e-6
+      "output_reasoning_tokens": 12e-6,
+      "output_reasoning": 12e-6
     },
     "tokenizer_config": null,
     "tokenizer_id": null
@@ -1123,8 +1137,10 @@
     "prices": {
       "input": 3e-6,
       "input_cached_tokens": 1.5e-6,
+      "input_cache_read": 1.5e-6,
       "output": 12e-6,
-      "output_reasoning_tokens": 12e-6
+      "output_reasoning_tokens": 12e-6,
+      "output_reasoning": 12e-6
     },
     "tokenizer_config": null,
     "tokenizer_id": null
@@ -1141,7 +1157,9 @@
       "output": 15e-6,
       "output_tokens": 15e-6,
       "cache_creation_input_tokens": 3.75e-6,
-      "cache_read_input_tokens": 0.3e-6
+      "input_cache_creation": 3.75e-6,
+      "cache_read_input_tokens": 0.3e-6,
+      "input_cache_read": 0.3e-6
     },
     "tokenizer_config": null,
     "tokenizer_id": "claude"
@@ -1158,7 +1176,9 @@
       "output": 15e-6,
       "output_tokens": 15e-6,
       "cache_creation_input_tokens": 3.75e-6,
-      "cache_read_input_tokens": 0.3e-6
+      "input_cache_creation": 3.75e-6,
+      "cache_read_input_tokens": 0.3e-6,
+      "input_cache_read": 0.3e-6
     },
     "tokenizer_config": null,
     "tokenizer_id": "claude"
@@ -1175,7 +1195,9 @@
       "output": 4e-6,
       "output_tokens": 4e-6,
       "cache_creation_input_tokens": 1e-6,
-      "cache_read_input_tokens": 0.08e-6
+      "input_cache_creation": 1e-6,
+      "cache_read_input_tokens": 0.08e-6,
+      "input_cache_read": 0.08e-6
     },
     "tokenizer_config": null,
     "tokenizer_id": "claude"
@@ -1192,7 +1214,9 @@
       "output": 4e-6,
       "output_tokens": 4e-6,
       "cache_creation_input_tokens": 1e-6,
-      "cache_read_input_tokens": 0.08e-6
+      "input_cache_creation": 1e-6,
+      "cache_read_input_tokens": 0.08e-6,
+      "input_cache_read": 0.08e-6
     },
     "tokenizer_config": null,
     "tokenizer_id": "claude"
@@ -1223,6 +1247,7 @@
     "prices": {
       "input": 2.5e-6,
       "input_cached_tokens": 1.25e-6,
+      "input_cache_read": 1.25e-6,
       "output": 10e-6
     },
     "tokenizer_config": {
@@ -1242,7 +1267,9 @@
       "input_text_tokens": 2.5e-6,
       "output_text_tokens": 10e-6,
       "input_audio_tokens": 100e-6,
-      "output_audio_tokens": 200e-6
+      "input_audio": 100e-6,
+      "output_audio_tokens": 200e-6,
+      "output_audio": 200e-6
     },
     "tokenizer_config": null,
     "tokenizer_id": null
@@ -1257,7 +1284,9 @@
       "input_text_tokens": 2.5e-6,
       "output_text_tokens": 10e-6,
       "input_audio_tokens": 100e-6,
-      "output_audio_tokens": 200e-6
+      "input_audio": 100e-6,
+      "output_audio_tokens": 200e-6,
+      "output_audio": 200e-6
     },
     "tokenizer_config": null,
     "tokenizer_id": null
@@ -1273,8 +1302,10 @@
       "input_cached_text_tokens": 2.5e-6,
       "output_text_tokens": 20e-6,
       "input_audio_tokens": 100e-6,
+      "input_audio": 100e-6,
       "input_cached_audio_tokens": 20e-6,
-      "output_audio_tokens": 200e-6
+      "output_audio_tokens": 200e-6,
+      "output_audio": 200e-6
     },
     "tokenizer_config": null,
     "tokenizer_id": null
@@ -1290,8 +1321,10 @@
       "input_cached_text_tokens": 2.5e-6,
       "output_text_tokens": 20e-6,
       "input_audio_tokens": 100e-6,
+      "input_audio": 100e-6,
       "input_cached_audio_tokens": 20e-6,
-      "output_audio_tokens": 200e-6
+      "output_audio_tokens": 200e-6,
+      "output_audio": 200e-6
     },
     "tokenizer_config": null,
     "tokenizer_id": null

worker/src/env.ts

@@ -2,6 +2,7 @@ import { removeEmptyEnvVariables } from "@langfuse/shared";
 import { z } from "zod";
 
 const EnvSchema = z.object({
+  BUILD_ID: z.string().optional(),
   NODE_ENV: z
     .enum(["development", "test", "production"])
     .default("development"),
@@ -82,6 +83,8 @@ const EnvSchema = z.object({
 
   CLICKHOUSE_URL: z.string().url(),
   CLICKHOUSE_USER: z.string(),
+  CLICKHOUSE_CLUSTER_NAME: z.string().default("default"),
+  CLICKHOUSE_DB: z.string().default("default"),
   CLICKHOUSE_PASSWORD: z.string(),
 
   LANGFUSE_LEGACY_INGESTION_WORKER_CONCURRENCY: z.coerce
@@ -107,6 +110,11 @@ const EnvSchema = z.object({
   // TODO: Remove for go-live
   LANGFUSE_RETURN_FROM_CLICKHOUSE: z.enum(["true", "false"]).default("true"),
 
+  // Skip the read from ClickHouse within the Ingestion pipeline for the given
+  // project ids. Applicable for projects that were created after the S3 write
+  // was activated and which don't rely on historic updates.
+  LANGFUSE_SKIP_INGESTION_CLICKHOUSE_READ_PROJECT_IDS: z.string().default(""),
+
   // Otel
   OTEL_EXPORTER_OTLP_ENDPOINT: z.string().default("http://localhost:4318"),
   OTEL_SERVICE_NAME: z.string().default("worker"),

worker/src/features/batchExport/handleBatchExportJob.ts

@@ -24,7 +24,6 @@ import {
   parseTraceAllFilters,
   FullObservationsWithScores,
   getPublicSessionsFilter,
-  getSessionsTable,
   getScoresForObservations,
   getObservationsTableWithModelData,
   getDistinctScoreNames,
@@ -33,6 +32,7 @@ import {
   getScoresForTraces,
   logger,
   getTracesByIds,
+  getSessionsWithMetrics,
 } from "@langfuse/shared/src/server";
 import { env } from "../../env";
 import { BatchExportSessionsRow, BatchExportTracesRow } from "./types";
@@ -182,7 +182,7 @@ export const getDatabaseReadStream = async ({
               projectId,
               finalFilter ?? [],
             );
-            const sessions = await getSessionsTable({
+            const sessions = await getSessionsWithMetrics({
               projectId: projectId,
               filter: sessionsFilter,
               orderBy: orderBy,
@@ -382,10 +382,11 @@ export const getDatabaseReadStream = async ({
               ),
             ]);
 
-            const scores = await getScoresForTraces(
+            const scores = await getScoresForTraces({
               projectId,
-              traces.map((t) => t.id),
-            );
+              traceIds: traces.map((t) => t.id),
+            });
+
             chunk = traces.map((t) => {
               const metric = metrics.find((m) => m.id === t.id);
               const filteredScores = scores.filter((s) => s.traceId === t.id);
@@ -408,7 +409,15 @@ export const getDatabaseReadStream = async ({
                   completionTokens: metric?.completionTokens,
                   totalTokens: metric?.totalTokens,
                 },
+                inputCost: metric?.calculatedInputCost,
+                outputCost: metric?.calculatedOutputCost,
+                totalCost: metric?.calculatedTotalCost,
+                level: metric?.level,
+                observationCount: Number(metric?.observationCount),
                 scores: outputScores,
+                inputTokens: metric?.promptTokens,
+                outputTokens: metric?.completionTokens,
+                totalTokens: metric?.totalTokens,
               };
             });
           } else {
@@ -604,7 +613,8 @@ function prepareScoresForOutput(
     (acc, score) => {
       // If this score name already exists in acc, use its existing type
       const existingValues = acc[score.name];
-      const newValue = score.value ?? score.stringValue;
+      const newValue =
+        score.dataType === "NUMERIC" ? score.value : score.stringValue;
       if (!newValue) return acc;
 
       if (!existingValues) {

worker/src/instrumentation.ts

@@ -25,6 +25,7 @@ dd.init({
 const sdk = new NodeSDK({
   resource: new Resource({
     "service.name": env.OTEL_SERVICE_NAME,
+    "service.version": env.BUILD_ID,
   }),
   traceExporter: new OTLPTraceExporter({
     url: `${env.OTEL_EXPORTER_OTLP_ENDPOINT}/v1/traces`,
@@ -33,6 +34,12 @@ const sdk = new NodeSDK({
     new IORedisInstrumentation(),
     new HttpInstrumentation({
       requireParentforOutgoingSpans: true,
+      ignoreIncomingRequestHook: (req) => {
+        // Ignore health checks
+        return ["/api/public/health", "/api/public/ready", "/api/health"].some(
+          (path) => req.url?.includes(path),
+        );
+      },
       ignoreOutgoingRequestHook: (req) => {
         return req.host === "127.0.0.1";
       },

worker/src/queues/evalQueue.ts

@@ -1,5 +1,5 @@
 import { Job } from "bullmq";
-import { BaseError } from "@langfuse/shared";
+import { ApiError, BaseError } from "@langfuse/shared";
 import { kyselyPrisma } from "@langfuse/shared/src/db";
 import { sql } from "kysely";
 import {
@@ -7,8 +7,11 @@ import {
   TQueueJobTypes,
   logger,
   traceException,
+  EvalExecutionQueue,
+  QueueJobs,
 } from "@langfuse/shared/src/server";
 import { createEvalJobs, evaluate } from "../ee/evaluation/evalService";
+import { randomUUID } from "crypto";
 
 export const evalJobTraceCreatorQueueProcessor = async (
   job: Job<TQueueJobTypes[QueueName.TraceUpsert]>,
@@ -50,6 +53,51 @@ export const evalJobExecutorQueueProcessor = async (
     await evaluate({ event: job.data.payload });
     return true;
   } catch (e) {
+    // If the job fails with a 429, we want to retry it unless it's older than 24h.
+    if (e instanceof ApiError && e.httpCode === 429) {
+      try {
+        // Check if the job execution is older than 24h
+        const jobExecution = await kyselyPrisma.$kysely
+          .selectFrom("job_executions")
+          .select("created_at")
+          .where("id", "=", job.data.payload.jobExecutionId)
+          .where("project_id", "=", job.data.payload.projectId)
+          .executeTakeFirstOrThrow();
+        if (
+          // Do nothing if job execution is older than 24h
+          jobExecution.created_at < new Date(Date.now() - 24 * 60 * 60 * 1000)
+        ) {
+          logger.info(
+            `Job ${job.data.payload.jobExecutionId} is rate limited for more than 24h. Stop retrying.`,
+          );
+        } else {
+          // Add the job into the queue with a random delay between 1 and 10min and return
+          const delay = Math.floor(Math.random() * 9 + 1) * 60 * 1000;
+          logger.info(
+            `Job ${job.data.payload.jobExecutionId} is rate limited. Retrying in ${delay}ms.`,
+          );
+          await EvalExecutionQueue.getInstance()?.add(
+            QueueName.EvaluationExecution,
+            {
+              name: QueueJobs.EvaluationExecution,
+              id: randomUUID(),
+              timestamp: new Date(),
+              payload: job.data.payload,
+            },
+            {
+              delay,
+            },
+          );
+          return;
+        }
+      } catch (innerErr) {
+        logger.error(
+          `Failed to handle 429 retry for ${job.data.payload.jobExecutionId}. Continuing regular processing.`,
+          innerErr,
+        );
+      }
+    }
+
     const displayError =
       e instanceof BaseError ? e.message : "An internal error occurred";
 
@@ -64,22 +112,22 @@ export const evalJobExecutorQueueProcessor = async (
 
     // do not log expected errors (api failures + missing api keys not provided by the user)
     if (
-      !(e instanceof BaseError && e.message.includes("API key for provider")) &&
-      !(
-        e instanceof BaseError &&
+      (e instanceof BaseError && e.message.includes("API key for provider")) || // api key not provided
+      (e instanceof ApiError && e.httpCode >= 400 && e.httpCode < 500) || // do not error and retry on 4xx errors. They are visible to the user in the UI but do not alert us.
+      (e instanceof ApiError && e.message.includes("TypeError")) || // Zod parsing the response failed. User should update prompt to consistently return expected output structure.
+      (e instanceof BaseError &&
         e.message.includes(
           "Please ensure the mapped data exists and consider extending the job delay.",
-        )
-      )
+        )) // Trace not found.
     ) {
-      traceException(e);
-      logger.error(
-        `Failed Evaluation_Execution job for id ${job.data.payload.jobExecutionId}`,
-        e,
-      );
-      throw e;
+      return;
     }
 
-    return;
+    traceException(e);
+    logger.error(
+      `Failed Evaluation_Execution job for id ${job.data.payload.jobExecutionId}`,
+      e,
+    );
+    throw e;
   }
 };

worker/src/queues/ingestionQueue.ts

@@ -97,10 +97,15 @@ export const ingestionQueueProcessorBuilder = (
         `${env.LANGFUSE_S3_EVENT_UPLOAD_PREFIX}${job.data.payload.authCheck.scope.projectId}/${getClickhouseEntityType(job.data.payload.data.type)}/${job.data.payload.data.eventBodyId}/`,
       );
 
+      const firstS3WriteTime =
+        eventFiles
+          .map((fileRef) => fileRef.createdAt)
+          .sort()
+          .shift() ?? new Date();
       const events: IngestionEventType[] = (
         await Promise.all(
-          eventFiles.map(async (key) => {
-            const file = await s3Client.download(key);
+          eventFiles.map(async (fileRef) => {
+            const file = await s3Client.download(fileRef.file);
             const parsedFile = JSON.parse(file);
             return Array.isArray(parsedFile) ? parsedFile : [parsedFile];
           }),
@@ -126,6 +131,7 @@ export const ingestionQueueProcessorBuilder = (
         getClickhouseEntityType(events[0].type),
         job.data.payload.authCheck.scope.projectId,
         job.data.payload.data.eventBodyId,
+        firstS3WriteTime,
         events,
       );
     } catch (e) {

worker/src/services/IngestionService/index.ts

@@ -39,6 +39,7 @@ import { tokenCount } from "../../features/tokenisation/usage";
 import { ClickhouseWriter, TableName } from "../ClickhouseWriter";
 import { convertJsonSchemaToRecord, overwriteObject } from "./utils";
 import { randomUUID } from "crypto";
+import { env } from "../../env";
 
 type InsertRecord =
   | TraceRecordInsertType
@@ -83,6 +84,7 @@ export class IngestionService {
     eventType: IngestionEntityTypes,
     projectId: string,
     eventBodyId: string,
+    createdAtTimestamp: Date,
     events: IngestionEventType[],
   ): Promise<void> {
     logger.debug(
@@ -94,18 +96,21 @@ export class IngestionService {
         return await this.processTraceEventList({
           projectId,
           entityId: eventBodyId,
+          createdAtTimestamp,
           traceEventList: events as TraceEventType[],
         });
       case "observation":
         return await this.processObservationEventList({
           projectId,
           entityId: eventBodyId,
+          createdAtTimestamp,
           observationEventList: events as ObservationEvent[],
         });
       case "score": {
         return await this.processScoreEventList({
           projectId,
           entityId: eventBodyId,
+          createdAtTimestamp,
           scoreEventList: events as ScoreEventType[],
         });
       }
@@ -115,9 +120,10 @@ export class IngestionService {
   private async processScoreEventList(params: {
     projectId: string;
     entityId: string;
+    createdAtTimestamp: Date;
     scoreEventList: ScoreEventType[];
   }) {
-    const { projectId, entityId, scoreEventList } = params;
+    const { projectId, entityId, createdAtTimestamp, scoreEventList } = params;
     if (scoreEventList.length === 0) return;
 
     const timeSortedEvents =
@@ -185,6 +191,8 @@ export class IngestionService {
         clickhouseScoreRecord,
         scoreRecords,
       });
+    finalScoreRecord.created_at =
+      clickhouseScoreRecord?.created_at ?? createdAtTimestamp.getTime();
 
     this.clickHouseWriter.addToQueue(TableName.Scores, finalScoreRecord);
   }
@@ -192,9 +200,10 @@ export class IngestionService {
   private async processTraceEventList(params: {
     projectId: string;
     entityId: string;
+    createdAtTimestamp: Date;
     traceEventList: TraceEventType[];
   }) {
-    const { projectId, entityId, traceEventList } = params;
+    const { projectId, entityId, createdAtTimestamp, traceEventList } = params;
     if (traceEventList.length === 0) return;
 
     const timeSortedEvents =
@@ -238,6 +247,8 @@ export class IngestionService {
       clickhouseTraceRecord,
       traceRecords,
     });
+    finalTraceRecord.created_at =
+      clickhouseTraceRecord?.created_at ?? createdAtTimestamp.getTime();
 
     // If the trace has a sessionId, we upsert the corresponding session into Postgres.
     if (finalTraceRecord.session_id) {
@@ -291,9 +302,11 @@ export class IngestionService {
   private async processObservationEventList(params: {
     projectId: string;
     entityId: string;
+    createdAtTimestamp: Date;
     observationEventList: ObservationEvent[];
   }) {
-    const { projectId, entityId, observationEventList } = params;
+    const { projectId, entityId, createdAtTimestamp, observationEventList } =
+      params;
     if (observationEventList.length === 0) return;
 
     const timeSortedEvents =
@@ -345,6 +358,8 @@ export class IngestionService {
       observationRecords,
       clickhouseObservationRecord,
     });
+    finalObservationRecord.created_at =
+      clickhouseObservationRecord?.created_at ?? createdAtTimestamp.getTime();
 
     // Backward compat: create wrapper trace for SDK < 2.0.0 events that do not have a traceId
     if (!finalObservationRecord.trace_id) {
@@ -742,6 +757,15 @@ export class IngestionService {
       params: Record<string, unknown>;
     };
   }) {
+    if (
+      env.LANGFUSE_SKIP_INGESTION_CLICKHOUSE_READ_PROJECT_IDS &&
+      env.LANGFUSE_SKIP_INGESTION_CLICKHOUSE_READ_PROJECT_IDS.split(
+        ",",
+      ).includes(params.projectId)
+    ) {
+      return null;
+    }
+
     const recordParser = {
       traces: traceRecordReadSchema,
       scores: scoreRecordReadSchema,

worker/src/services/IngestionService/tests/IngestionService.integration.test.ts

@@ -73,6 +73,7 @@ describe("Ingestion end-to-end tests", () => {
     await ingestionService.processTraceEventList({
       projectId,
       entityId: traceId,
+      createdAtTimestamp: new Date(timestamp),
       traceEventList: eventList,
     });
     await clickhouseWriter.flushAll(true);
@@ -442,21 +443,25 @@ describe("Ingestion end-to-end tests", () => {
         ingestionService.processTraceEventList({
           projectId,
           entityId: traceId,
+          createdAtTimestamp: new Date(),
           traceEventList,
         }),
         ingestionService.processObservationEventList({
           projectId,
           entityId: spanId,
+          createdAtTimestamp: new Date(),
           observationEventList: spanEventList,
         }),
         ingestionService.processObservationEventList({
           projectId,
           entityId: generationId,
+          createdAtTimestamp: new Date(),
           observationEventList: generationEventList,
         }),
         ingestionService.processScoreEventList({
           projectId,
           entityId: scoreId,
+          createdAtTimestamp: new Date(),
           scoreEventList,
         }),
       ]);
@@ -778,11 +783,13 @@ describe("Ingestion end-to-end tests", () => {
         ingestionService.processTraceEventList({
           projectId,
           entityId: traceId,
+          createdAtTimestamp: new Date(),
           traceEventList,
         }),
         ingestionService.processObservationEventList({
           projectId,
           entityId: generationId,
+          createdAtTimestamp: new Date(),
           observationEventList: generationEventList,
         }),
       ]);
@@ -918,26 +925,31 @@ describe("Ingestion end-to-end tests", () => {
       ingestionService.processTraceEventList({
         projectId,
         entityId: traceId,
+        createdAtTimestamp: new Date(),
         traceEventList,
       }),
       ingestionService.processObservationEventList({
         projectId,
         entityId: spanId,
+        createdAtTimestamp: new Date(),
         observationEventList: spanEventList,
       }),
       ingestionService.processObservationEventList({
         projectId,
         entityId: generationId,
+        createdAtTimestamp: new Date(),
         observationEventList: generationEventList,
       }),
       ingestionService.processObservationEventList({
         projectId,
         entityId: eventId,
+        createdAtTimestamp: new Date(),
         observationEventList: eventEventList,
       }),
       ingestionService.processScoreEventList({
         projectId,
         entityId: scoreId,
+        createdAtTimestamp: new Date(),
         scoreEventList,
       }),
     ]);
@@ -1009,6 +1021,7 @@ describe("Ingestion end-to-end tests", () => {
     await ingestionService.processTraceEventList({
       projectId,
       entityId: traceId,
+      createdAtTimestamp: new Date(),
       traceEventList: traceEventList1,
     });
 
@@ -1032,6 +1045,7 @@ describe("Ingestion end-to-end tests", () => {
     await ingestionService.processTraceEventList({
       projectId,
       entityId: traceId,
+      createdAtTimestamp: new Date(),
       traceEventList: traceEventList2,
     });
 
@@ -1088,6 +1102,7 @@ describe("Ingestion end-to-end tests", () => {
     await ingestionService.processTraceEventList({
       projectId,
       entityId: traceId,
+      createdAtTimestamp: new Date(),
       traceEventList,
     });
 
@@ -1135,6 +1150,7 @@ describe("Ingestion end-to-end tests", () => {
     await ingestionService.processTraceEventList({
       projectId,
       entityId: traceId,
+      createdAtTimestamp: new Date(),
       traceEventList,
     });
 
@@ -1190,6 +1206,7 @@ describe("Ingestion end-to-end tests", () => {
     await ingestionService.processScoreEventList({
       projectId,
       entityId: scoreId,
+      createdAtTimestamp: new Date(),
       scoreEventList,
     });
 
@@ -1312,6 +1329,7 @@ describe("Ingestion end-to-end tests", () => {
     await ingestionService.processObservationEventList({
       projectId,
       entityId: observationId,
+      createdAtTimestamp: new Date(),
       observationEventList,
     });
 
@@ -1451,6 +1469,7 @@ describe("Ingestion end-to-end tests", () => {
     await ingestionService.processObservationEventList({
       projectId,
       entityId: observationId,
+      createdAtTimestamp: new Date(),
       observationEventList,
     });
 
@@ -1504,6 +1523,7 @@ describe("Ingestion end-to-end tests", () => {
     await ingestionService.processObservationEventList({
       projectId,
       entityId: observationId,
+      createdAtTimestamp: new Date(),
       observationEventList: observationEventList1,
     });
     await clickhouseWriter.flushAll(true);
@@ -1525,6 +1545,7 @@ describe("Ingestion end-to-end tests", () => {
     await ingestionService.processObservationEventList({
       projectId,
       entityId: observationId,
+      createdAtTimestamp: new Date(),
       observationEventList: observationEventList2,
     });
     await clickhouseWriter.flushAll(true);
@@ -1574,6 +1595,7 @@ describe("Ingestion end-to-end tests", () => {
     await ingestionService.processObservationEventList({
       projectId,
       entityId: generationId,
+      createdAtTimestamp: new Date(),
       observationEventList: generationEventList,
     });
 
@@ -1635,11 +1657,13 @@ describe("Ingestion end-to-end tests", () => {
       ingestionService.processTraceEventList({
         projectId,
         entityId: traceId,
+        createdAtTimestamp: new Date(),
         traceEventList,
       }),
       ingestionService.processObservationEventList({
         projectId,
         entityId: generationId,
+        createdAtTimestamp: new Date(),
         observationEventList: generationEventList1,
       }),
     ]);
@@ -1680,6 +1704,7 @@ describe("Ingestion end-to-end tests", () => {
     await ingestionService.processObservationEventList({
       projectId,
       entityId: generationId,
+      createdAtTimestamp: new Date(),
       observationEventList: generationEventList2,
     });
 
@@ -1771,11 +1796,13 @@ describe("Ingestion end-to-end tests", () => {
       ingestionService.processTraceEventList({
         projectId,
         entityId: traceId,
+        createdAtTimestamp: new Date(),
         traceEventList,
       }),
       ingestionService.processObservationEventList({
         projectId,
         entityId: generationId,
+        createdAtTimestamp: new Date(),
         observationEventList: generationEventList,
       }),
     ]);
@@ -1871,11 +1898,13 @@ describe("Ingestion end-to-end tests", () => {
       ingestionService.processTraceEventList({
         projectId,
         entityId: traceId,
+        createdAtTimestamp: new Date(),
         traceEventList,
       }),
       ingestionService.processObservationEventList({
         projectId,
         entityId: generationId,
+        createdAtTimestamp: new Date(),
         observationEventList: generationEventList,
       }),
     ]);
@@ -1939,6 +1968,7 @@ describe("Ingestion end-to-end tests", () => {
     await ingestionService.processTraceEventList({
       projectId,
       entityId: traceId,
+      createdAtTimestamp: new Date(),
       traceEventList,
     });
 
@@ -2052,11 +2082,13 @@ describe("Ingestion end-to-end tests", () => {
         ingestionService.processTraceEventList({
           projectId,
           entityId: traceId,
+          createdAtTimestamp: new Date(),
           traceEventList,
         }),
         ingestionService.processObservationEventList({
           projectId,
           entityId: generationId,
+          createdAtTimestamp: new Date(),
           observationEventList: generationEventList,
         }),
       ]);

worker/src/services/IngestionService/tests/calculateTokenCost.unit.test.ts

@@ -427,6 +427,7 @@ describe("Token Cost Calculation", () => {
     await (mockIngestionService as any).processObservationEventList({
       projectId,
       entityId: generationId,
+      createdAtTimestamp: new Date(),
       observationEventList: events,
     });
     expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -507,6 +508,7 @@ describe("Token Cost Calculation", () => {
     await (mockIngestionService as any).processObservationEventList({
       projectId,
       entityId: generationId,
+      createdAtTimestamp: new Date(),
       observationEventList: events,
     });
     expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -585,6 +587,7 @@ describe("Token Cost Calculation", () => {
     await (mockIngestionService as any).processObservationEventList({
       projectId,
       entityId: generationId,
+      createdAtTimestamp: new Date(),
       observationEventList: events,
     });
     expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -665,6 +668,7 @@ describe("Token Cost Calculation", () => {
     await (mockIngestionService as any).processObservationEventList({
       projectId,
       entityId: generationId,
+      createdAtTimestamp: new Date(),
       observationEventList: events,
     });
     expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -748,6 +752,7 @@ describe("Token Cost Calculation", () => {
     await (mockIngestionService as any).processObservationEventList({
       projectId,
       entityId: generationId,
+      createdAtTimestamp: new Date(),
       observationEventList: events,
     });
     expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -837,6 +842,7 @@ describe("Token Cost Calculation", () => {
     await (mockIngestionService as any).processObservationEventList({
       projectId,
       entityId: generationId,
+      createdAtTimestamp: new Date(),
       observationEventList: events,
     });
     expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -921,6 +927,7 @@ describe("Token Cost Calculation", () => {
     await (mockIngestionService as any).processObservationEventList({
       projectId,
       entityId: generationId,
+      createdAtTimestamp: new Date(),
       observationEventList: events,
     });
     expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -1002,6 +1009,7 @@ describe("Token Cost Calculation", () => {
     await (mockIngestionService as any).processObservationEventList({
       projectId,
       entityId: generationId,
+      createdAtTimestamp: new Date(),
       observationEventList: events,
     });
     expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -1088,6 +1096,7 @@ describe("Token Cost Calculation", () => {
     await (mockIngestionService as any).processObservationEventList({
       projectId,
       entityId: generationId,
+      createdAtTimestamp: new Date(),
       observationEventList: events,
     });
     expect(mockAddToClickhouseWriter).toHaveBeenCalled();
@@ -1159,6 +1168,7 @@ describe("Token Cost Calculation", () => {
     await (mockIngestionService as any).processObservationEventList({
       projectId,
       entityId: generationId,
+      createdAtTimestamp: new Date(),
       observationEventList: events,
     });
     expect(mockAddToClickhouseWriter).toHaveBeenCalled();