feat(langfuse): support media fanout gateway

2026-09-03 21:21:34 +00:00 · 2026-06-25 10:30:30 +02:00 · 2026-06-25 10:30:30 +02:00 · 50caff53d3
commit 50caff53d3
parent ac9154ed55
21 changed files with 2947 additions and 334 deletions
--- a/.env.example
+++ b/.env.example
@ -123,7 +123,7 @@ NODE_MAX_OLD_SPACE_SIZE=6144
 # LANGFUSE_BASE_URL=

 # Optional Langfuse fanout for tenant-scoped Langfuse projects.
-# The collector is opt-in: add docker-compose.langfuse-fanout.yml,
+# The fanout gateway is opt-in: add docker-compose.langfuse-fanout.yml,
 # deploy-compose.langfuse-fanout.yml, or enable helm langfuseFanout.
 # Tenant public/secret keys are read from LibreChat tenant app configuration.
 # Tenant Langfuse base URLs must be set in tenant app configuration and match
@ -132,22 +132,44 @@ NODE_MAX_OLD_SPACE_SIZE=6144
 # See otel/langfuse-fanout/README.md.
 # LANGFUSE_FANOUT_ENABLED=false
 # LANGFUSE_FANOUT_COLLECTOR_URL=http://langfuse-fanout-collector:4318
-# Emergency switch: unset/false defaults enabled; set true to keep central collector export but skip tenant trace/score export.
+# Emergency switch: unset/false defaults enabled; set true to keep central fanout export but skip tenant trace/score export.
 # LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED=false
 # Langfuse Cloud base URL options: https://cloud.langfuse.com (EU),
 # https://us.cloud.langfuse.com (US), https://jp.cloud.langfuse.com (JP).
-# Collector-only central trace export URL. LibreChat feedback scores use
+# Gateway-only central trace/media export URL. LibreChat feedback scores use
 # LANGFUSE_BASE_URL, so set both URLs to the same non-EU region when applicable.
 # LANGFUSE_FANOUT_CENTRAL_BASE_URL=https://cloud.langfuse.com
-# Collector-only Basic auth header for central trace export. LibreChat feedback
+# Gateway-only Basic auth header for central trace/media export. LibreChat feedback
 # scores use LANGFUSE_PUBLIC_KEY/LANGFUSE_SECRET_KEY instead.
 # LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=Basic <base64-public-colon-secret>
-# Compose's included collector config supports the three listed destination keys.
-# Add custom keys only when the collector routing table/exporters are updated too.
+# Compose's included gateway config supports the three listed destination keys.
+# Add custom keys only when the gateway is started with matching destination URLs.
 # LANGFUSE_FANOUT_TENANT_DESTINATIONS=eu=https://cloud.langfuse.com,us=https://us.cloud.langfuse.com,jp=https://jp.cloud.langfuse.com
+# Compose's collector config routes only these destination keys. The gateway
+# fails startup when LANGFUSE_FANOUT_TENANT_DESTINATIONS contains another key.
+# LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS=eu,us,jp
+# Gateway base URL used to build one-time media upload URLs. Compose sets this
+# to its private service URL; Helm derives an internal service URL unless set.
+# LANGFUSE_FANOUT_PUBLIC_URL=http://langfuse-fanout-collector:4318
+# Internal gateway-to-collector trace endpoint. Compose sets this automatically.
+# LANGFUSE_FANOUT_TRACE_COLLECTOR_URL=http://langfuse-fanout-otel:4319
+# Redis-backed one-time upload plans let multiple gateway pods handle Langfuse
+# media create/upload requests. Compose sets this to its private Redis service.
+# LANGFUSE_FANOUT_REDIS_URI=redis://langfuse-fanout-redis:6379
+# LANGFUSE_FANOUT_REDIS_USERNAME=
+# LANGFUSE_FANOUT_REDIS_PASSWORD=
+# LANGFUSE_FANOUT_REDIS_KEY_PREFIX=langfuse-fanout
+# Internal collector receiver bind address. Helm uses 127.0.0.1 because the
+# collector is a sidecar; Compose uses 0.0.0.0 on the private fanout network.
+# LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT=0.0.0.0:4319
+# Static Compose collector destination URLs. Helm derives these from values.
 # LANGFUSE_FANOUT_TENANT_EU_BASE_URL=https://cloud.langfuse.com
 # LANGFUSE_FANOUT_TENANT_US_BASE_URL=https://us.cloud.langfuse.com
 # LANGFUSE_FANOUT_TENANT_JP_BASE_URL=https://jp.cloud.langfuse.com
+# LANGFUSE_FANOUT_UPSTREAM_TIMEOUT=30s
+# Optional bearer token for scraping the fanout gateway /metrics endpoint.
+# If unset, /metrics returns 401. The gateway also accepts METRICS_SECRET when present.
+# LANGFUSE_FANOUT_METRICS_SECRET=
 # LANGFUSE_FANOUT_MEMORY_LIMIT_MIB=256
 # LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB=64
 # LANGFUSE_FANOUT_BATCH_TIMEOUT=1s
--- a/.github/workflows/eslint-ci.yml
+++ b/.github/workflows/eslint-ci.yml
@ -11,10 +11,6 @@ on:
      - 'api/**'
      - 'client/**'
      - 'packages/**'
-      - 'helm/librechat/templates/langfuse-fanout-*.yaml'
-      - 'helm/librechat/tests/langfuse_fanout_*.sh'
-      - 'helm/librechat/values.yaml'
-      - 'otel/langfuse-fanout/**'
      - '.github/workflows/eslint-ci.yml'

 jobs:
@ -130,21 +126,3 @@ jobs:
            echo "::error::Or rely on the lint-staged pre-commit hook (do not bypass with --no-verify)."
            exit 1
          fi
-
-      - name: Check Langfuse fanout collector config drift
-        run: |
-          BASE_SHA=$(jq --raw-output .pull_request.base.sha "$GITHUB_EVENT_PATH")
-          mapfile -d '' -t CHANGED_FILES < <(
-            git diff -z --name-only --diff-filter=ACMRTUXB "$BASE_SHA" HEAD |
-              grep -zE '^(otel/langfuse-fanout/|helm/librechat/(templates/langfuse-fanout-|tests/langfuse_fanout_|values\.yaml$))' || true
-          )
-
-          if [[ ${#CHANGED_FILES[@]} -eq 0 ]]; then
-            echo "No Langfuse fanout collector files changed. Skipping drift check."
-            exit 0
-          fi
-
-          echo "Files triggering drift check:"
-          printf '%s\n' "${CHANGED_FILES[@]}"
-
-          helm/librechat/tests/langfuse_fanout_config_drift_test.sh
--- a/deploy-compose.langfuse-fanout.yml
+++ b/deploy-compose.langfuse-fanout.yml
@ -12,6 +12,42 @@ services:
      - langfuse-fanout

  langfuse-fanout-collector:
+    build:
+      context: .
+      dockerfile: otel/langfuse-fanout/Dockerfile
+    image: librechat-langfuse-fanout:local
+    restart: always
+    depends_on:
+      - langfuse-fanout-otel
+      - langfuse-fanout-redis
+    env_file:
+      - .env
+    environment:
+      - LANGFUSE_FANOUT_CENTRAL_BASE_URL=${LANGFUSE_FANOUT_CENTRAL_BASE_URL:-https://cloud.langfuse.com}
+      - LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=${LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER:?Set LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER to central Langfuse Basic auth header}
+      - LANGFUSE_FANOUT_TENANT_DESTINATIONS=${LANGFUSE_FANOUT_TENANT_DESTINATIONS:-eu=https://cloud.langfuse.com,us=https://us.cloud.langfuse.com,jp=https://jp.cloud.langfuse.com}
+      - LANGFUSE_FANOUT_UPSTREAM_TIMEOUT=${LANGFUSE_FANOUT_UPSTREAM_TIMEOUT:-30s}
+      - LANGFUSE_FANOUT_PUBLIC_URL=${LANGFUSE_FANOUT_PUBLIC_URL:-http://langfuse-fanout-collector:4318}
+      - LANGFUSE_FANOUT_TRACE_COLLECTOR_URL=http://langfuse-fanout-otel:4319
+      - LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS=eu,us,jp
+      - LANGFUSE_FANOUT_REDIS_URI=${LANGFUSE_FANOUT_REDIS_URI:-redis://langfuse-fanout-redis:6379}
+      - LANGFUSE_FANOUT_REDIS_USERNAME=${LANGFUSE_FANOUT_REDIS_USERNAME:-}
+      - LANGFUSE_FANOUT_REDIS_PASSWORD=${LANGFUSE_FANOUT_REDIS_PASSWORD:-}
+      - LANGFUSE_FANOUT_REDIS_KEY_PREFIX=${LANGFUSE_FANOUT_REDIS_KEY_PREFIX:-langfuse-fanout}
+    expose:
+      - '4318'
+    networks:
+      - langfuse-fanout
+
+  langfuse-fanout-redis:
+    image: redis:7.4-alpine
+    restart: always
+    expose:
+      - '6379'
+    networks:
+      - langfuse-fanout
+
+  langfuse-fanout-otel:
    image: otel/opentelemetry-collector-contrib:0.143.0
    restart: always
    command: ['--config=/etc/otelcol/otelcol.yaml']
@ -20,6 +56,7 @@ services:
    environment:
      - LANGFUSE_FANOUT_CENTRAL_BASE_URL=${LANGFUSE_FANOUT_CENTRAL_BASE_URL:-https://cloud.langfuse.com}
      - LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=${LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER:?Set LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER to central Langfuse Basic auth header}
+      - LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT=0.0.0.0:4319
      - LANGFUSE_FANOUT_TENANT_EU_BASE_URL=${LANGFUSE_FANOUT_TENANT_EU_BASE_URL:-https://cloud.langfuse.com}
      - LANGFUSE_FANOUT_TENANT_US_BASE_URL=${LANGFUSE_FANOUT_TENANT_US_BASE_URL:-https://us.cloud.langfuse.com}
      - LANGFUSE_FANOUT_TENANT_JP_BASE_URL=${LANGFUSE_FANOUT_TENANT_JP_BASE_URL:-https://jp.cloud.langfuse.com}
@ -28,10 +65,10 @@ services:
      - LANGFUSE_FANOUT_BATCH_TIMEOUT=${LANGFUSE_FANOUT_BATCH_TIMEOUT:-1s}
      - LANGFUSE_FANOUT_BATCH_SEND_SIZE=${LANGFUSE_FANOUT_BATCH_SEND_SIZE:-128}
      - LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT=${LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT:-1000}
-    expose:
-      - '4318'
    volumes:
      - ./otel/langfuse-fanout/otelcol.yaml:/etc/otelcol/otelcol.yaml:ro
+    expose:
+      - '4319'
    networks:
      - langfuse-fanout

--- a/docker-compose.langfuse-fanout.yml
+++ b/docker-compose.langfuse-fanout.yml
@ -12,6 +12,42 @@ services:
      - langfuse-fanout

  langfuse-fanout-collector:
+    build:
+      context: .
+      dockerfile: otel/langfuse-fanout/Dockerfile
+    image: librechat-langfuse-fanout:local
+    restart: always
+    depends_on:
+      - langfuse-fanout-otel
+      - langfuse-fanout-redis
+    env_file:
+      - .env
+    environment:
+      - LANGFUSE_FANOUT_CENTRAL_BASE_URL=${LANGFUSE_FANOUT_CENTRAL_BASE_URL:-https://cloud.langfuse.com}
+      - LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=${LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER:?Set LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER to central Langfuse Basic auth header}
+      - LANGFUSE_FANOUT_TENANT_DESTINATIONS=${LANGFUSE_FANOUT_TENANT_DESTINATIONS:-eu=https://cloud.langfuse.com,us=https://us.cloud.langfuse.com,jp=https://jp.cloud.langfuse.com}
+      - LANGFUSE_FANOUT_UPSTREAM_TIMEOUT=${LANGFUSE_FANOUT_UPSTREAM_TIMEOUT:-30s}
+      - LANGFUSE_FANOUT_PUBLIC_URL=${LANGFUSE_FANOUT_PUBLIC_URL:-http://langfuse-fanout-collector:4318}
+      - LANGFUSE_FANOUT_TRACE_COLLECTOR_URL=http://langfuse-fanout-otel:4319
+      - LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS=eu,us,jp
+      - LANGFUSE_FANOUT_REDIS_URI=${LANGFUSE_FANOUT_REDIS_URI:-redis://langfuse-fanout-redis:6379}
+      - LANGFUSE_FANOUT_REDIS_USERNAME=${LANGFUSE_FANOUT_REDIS_USERNAME:-}
+      - LANGFUSE_FANOUT_REDIS_PASSWORD=${LANGFUSE_FANOUT_REDIS_PASSWORD:-}
+      - LANGFUSE_FANOUT_REDIS_KEY_PREFIX=${LANGFUSE_FANOUT_REDIS_KEY_PREFIX:-langfuse-fanout}
+    expose:
+      - '4318'
+    networks:
+      - langfuse-fanout
+
+  langfuse-fanout-redis:
+    image: redis:7.4-alpine
+    restart: always
+    expose:
+      - '6379'
+    networks:
+      - langfuse-fanout
+
+  langfuse-fanout-otel:
    image: otel/opentelemetry-collector-contrib:0.143.0
    restart: always
    command: ['--config=/etc/otelcol/otelcol.yaml']
@ -20,6 +56,7 @@ services:
    environment:
      - LANGFUSE_FANOUT_CENTRAL_BASE_URL=${LANGFUSE_FANOUT_CENTRAL_BASE_URL:-https://cloud.langfuse.com}
      - LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=${LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER:?Set LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER to central Langfuse Basic auth header}
+      - LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT=0.0.0.0:4319
      - LANGFUSE_FANOUT_TENANT_EU_BASE_URL=${LANGFUSE_FANOUT_TENANT_EU_BASE_URL:-https://cloud.langfuse.com}
      - LANGFUSE_FANOUT_TENANT_US_BASE_URL=${LANGFUSE_FANOUT_TENANT_US_BASE_URL:-https://us.cloud.langfuse.com}
      - LANGFUSE_FANOUT_TENANT_JP_BASE_URL=${LANGFUSE_FANOUT_TENANT_JP_BASE_URL:-https://jp.cloud.langfuse.com}
@ -28,10 +65,10 @@ services:
      - LANGFUSE_FANOUT_BATCH_TIMEOUT=${LANGFUSE_FANOUT_BATCH_TIMEOUT:-1s}
      - LANGFUSE_FANOUT_BATCH_SEND_SIZE=${LANGFUSE_FANOUT_BATCH_SEND_SIZE:-128}
      - LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT=${LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT:-1000}
-    expose:
-      - '4318'
    volumes:
      - ./otel/langfuse-fanout/otelcol.yaml:/etc/otelcol/otelcol.yaml:ro
+    expose:
+      - '4319'
    networks:
      - langfuse-fanout

--- a/helm/librechat/readme.md
+++ b/helm/librechat/readme.md
@ -56,24 +56,42 @@ https://<librechat-domain>/api/admin/oauth/openid/callback

 ## Langfuse Fanout

-The chart can optionally deploy an OpenTelemetry Collector that forwards
-tenant-scoped Langfuse traces to both a central Langfuse project and the tenant
-Langfuse project. It is disabled by default.
+The chart can optionally deploy a Langfuse fanout gateway with an internal
+OpenTelemetry Collector sidecar. The gateway handles Langfuse media fanout and
+proxies traces to the collector; the collector forwards tenant-scoped Langfuse
+traces to both a central Langfuse project and the tenant Langfuse project. It is
+disabled by default.

 When enabled, the chart also sets `LANGFUSE_FANOUT_ENABLED` and
 `LANGFUSE_FANOUT_COLLECTOR_URL` for the LibreChat app unless those values are
 already provided in `librechat.configEnv`.

 Set `librechat.configEnv.LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED=true` to keep
-central trace export flowing through the collector while disabling tenant trace
+central trace export flowing through the fanout gateway while disabling tenant trace
 and score export. When omitted, false, or blank, tenant export remains available
 if tenant keys and a known destination are configured.

 Langfuse tenant base URLs are selected from the startup-configured destination
-map rendered into LibreChat and the collector. Tenant API keys can still be added
+map rendered into LibreChat and the fanout gateway. Tenant API keys can still be added
 through tenant app configuration at runtime without restarting either component.
-Collector batching can be tuned with `langfuseFanout.batchTimeout` and
-`langfuseFanout.batchSendSize`; defaults are `1s` and `128`.
+The internal collector provides trace memory limiting, batching, tenant routing,
+and removal of LibreChat-only routing attributes before export.
+
+The fanout gateway stores one-time media upload plans in Redis so media create
+and byte-upload requests can land on different gateway replicas. Set
+`langfuseFanout.redis.uri` for an external Redis service, or enable the bundled
+Redis chart with `redis.enabled=true` and let the chart derive the internal URI.
+Scale the gateway manually with `langfuseFanout.replicaCount`; the chart does
+not create a fanout HPA.
+The internal collector receiver is bound to `127.0.0.1:4319` by default because
+only the gateway sidecar should send traces to it.
+
+The gateway exposes Prometheus metrics at `/metrics`. Configure
+`langfuseFanout.metrics.secret.name` and `.key` to pass a bearer token secret to
+the gateway; if omitted, `/metrics` returns 401. Use
+`langfuseFanout.service.annotations` for scrape annotations when your cluster
+uses annotation-based discovery. The gateway container also has configurable
+`/healthz` liveness and readiness probes under `langfuseFanout`.

 See [`otel/langfuse-fanout/README.md`](../../otel/langfuse-fanout/README.md)
 for the central Langfuse secret and values example.
--- a/helm/librechat/templates/_helpers.tpl
+++ b/helm/librechat/templates/_helpers.tpl
@ -86,6 +86,38 @@ already use the same lowercase key shape for collector routing to match.
 {{- end -}}
 {{- end }}

+{{/*
+Render the fanout destination list consumed by LibreChat and the fanout gateway.
+*/}}
+{{- define "librechat.langfuseFanout.tenantDestinationsEnv" -}}
+{{- $tenantDestinations := list -}}
+{{- range $name, $destination := .Values.langfuseFanout.tenant.destinations -}}
+{{- include "librechat.langfuseFanout.validateDestinationKey" $name -}}
+{{- $tenantDestinations = append $tenantDestinations (printf "%s=%s" $name $destination.baseUrl) -}}
+{{- end -}}
+{{- join "," $tenantDestinations -}}
+{{- end }}
+
+{{/*
+Render the fanout destination key list consumed by the gateway as a startup
+guard against media destinations the collector cannot route traces to.
+*/}}
+{{- define "librechat.langfuseFanout.tenantDestinationKeysEnv" -}}
+{{- $tenantDestinationKeys := list -}}
+{{- range $name, $_destination := .Values.langfuseFanout.tenant.destinations -}}
+{{- include "librechat.langfuseFanout.validateDestinationKey" $name -}}
+{{- $tenantDestinationKeys = append $tenantDestinationKeys $name -}}
+{{- end -}}
+{{- join "," $tenantDestinationKeys -}}
+{{- end }}
+
+{{/*
+Bundled Redis URI used when the Redis subchart is enabled.
+*/}}
+{{- define "librechat.bundledRedisURI" -}}
+{{- printf "redis://%s-master.%s.svc.cluster.local:6379" (include "common.names.fullname" .Subcharts.redis) (.Release.Namespace | lower) -}}
+{{- end }}
+
 {{/*
 RAG Selector labels
 */}}
--- a/helm/librechat/templates/configmap-env.yaml
+++ b/helm/librechat/templates/configmap-env.yaml
@ -19,7 +19,7 @@ data:
  USE_REDIS: "true"
  {{- end }}
  {{- if and (not (dig "configEnv" "REDIS_URI" "" .Values.librechat)) .Values.redis.enabled }}
-  REDIS_URI: redis://{{ include "common.names.fullname" .Subcharts.redis }}-master.{{ .Release.Namespace | lower }}.svc.cluster.local:6379
+  REDIS_URI: {{ include "librechat.bundledRedisURI" . }}
  {{- end }}
  {{- if and $adminPanelUrl (not $configAdminPanelUrl) }}
  ADMIN_PANEL_URL: {{ $adminPanelUrl | quote }}
--- a/helm/librechat/templates/langfuse-fanout-configmap.yaml
+++ b/helm/librechat/templates/langfuse-fanout-configmap.yaml
@ -18,7 +18,7 @@ data:
      otlp:
        protocols:
          http:
-            endpoint: 0.0.0.0:4318
+            endpoint: ${env:LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT}
            include_metadata: true
            traces_url_path: /api/public/otel/v1/traces

--- a/helm/librechat/templates/langfuse-fanout-deployment.yaml
+++ b/helm/librechat/templates/langfuse-fanout-deployment.yaml
@ -1,4 +1,15 @@
 {{- if .Values.langfuseFanout.enabled }}
+{{- $redisURI := .Values.langfuseFanout.redis.uri }}
+{{- if and (not $redisURI) .Values.redis.enabled }}
+{{- $redisURI = include "librechat.bundledRedisURI" . }}
+{{- end }}
+{{- if not $redisURI }}
+{{- fail "langfuseFanout.redis.uri is required when langfuseFanout.enabled=true unless redis.enabled=true" }}
+{{- end }}
+{{- $publicURL := .Values.langfuseFanout.publicUrl }}
+{{- if not $publicURL }}
+{{- $publicURL = printf "http://%s.%s.svc.cluster.local:%v" (include "librechat.langfuseFanout.fullname" .) (.Release.Namespace | lower) .Values.langfuseFanout.service.port }}
+{{- end }}
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@ -13,7 +24,7 @@ spec:
  template:
    metadata:
      annotations:
-        checksum/langfuse-fanout-config: {{ include (print $.Template.BasePath "/langfuse-fanout-configmap.yaml") . | sha256sum }}
+        checksum/langfuse-fanout-config: {{ toYaml .Values.langfuseFanout | sha256sum }}
        {{- with .Values.langfuseFanout.podAnnotations }}
        {{- toYaml . | nindent 8 }}
        {{- end }}
@ -24,10 +35,9 @@ spec:
        {{- end }}
    spec:
      containers:
-        - name: otelcol
+        - name: langfuse-fanout
          image: "{{ .Values.langfuseFanout.image.repository }}:{{ .Values.langfuseFanout.image.tag }}"
          imagePullPolicy: {{ .Values.langfuseFanout.image.pullPolicy }}
-          args: ["--config=/etc/otelcol/otelcol.yaml"]
          ports:
            - name: otlp-http
              containerPort: 4318
@ -40,6 +50,66 @@ spec:
                secretKeyRef:
                  name: {{ required "langfuseFanout.central.authHeaderSecret.name is required when langfuseFanout.enabled=true" .Values.langfuseFanout.central.authHeaderSecret.name | quote }}
                  key: {{ .Values.langfuseFanout.central.authHeaderSecret.key | quote }}
+            {{- /* Validate destination keys before rendering the env map. */}}
+            {{- range $name, $_ := .Values.langfuseFanout.tenant.destinations }}
+            {{- include "librechat.langfuseFanout.validateDestinationKey" $name }}
+            {{- end }}
+            - name: LANGFUSE_FANOUT_TENANT_DESTINATIONS
+              value: {{ include "librechat.langfuseFanout.tenantDestinationsEnv" . | quote }}
+            - name: LANGFUSE_FANOUT_UPSTREAM_TIMEOUT
+              value: {{ .Values.langfuseFanout.upstreamTimeout | quote }}
+            - name: LANGFUSE_FANOUT_TRACE_COLLECTOR_URL
+              value: {{ .Values.langfuseFanout.traceCollectorUrl | quote }}
+            - name: LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS
+              value: {{ include "librechat.langfuseFanout.tenantDestinationKeysEnv" . | quote }}
+            - name: LANGFUSE_FANOUT_REDIS_URI
+              value: {{ $redisURI | quote }}
+            {{- with .Values.langfuseFanout.redis.username }}
+            - name: LANGFUSE_FANOUT_REDIS_USERNAME
+              value: {{ . | quote }}
+            {{- end }}
+            {{- with .Values.langfuseFanout.redis.passwordSecret.name }}
+            - name: LANGFUSE_FANOUT_REDIS_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: {{ . | quote }}
+                  key: {{ $.Values.langfuseFanout.redis.passwordSecret.key | quote }}
+            {{- end }}
+            - name: LANGFUSE_FANOUT_REDIS_KEY_PREFIX
+              value: {{ .Values.langfuseFanout.redis.keyPrefix | quote }}
+            - name: LANGFUSE_FANOUT_PUBLIC_URL
+              value: {{ $publicURL | quote }}
+            {{- with .Values.langfuseFanout.metrics.secret.name }}
+            - name: LANGFUSE_FANOUT_METRICS_SECRET
+              valueFrom:
+                secretKeyRef:
+                  name: {{ . | quote }}
+                  key: {{ $.Values.langfuseFanout.metrics.secret.key | quote }}
+            {{- end }}
+          livenessProbe:
+            {{- toYaml .Values.langfuseFanout.livenessProbe | nindent 12 }}
+          readinessProbe:
+            {{- toYaml .Values.langfuseFanout.readinessProbe | nindent 12 }}
+          resources:
+            {{- toYaml .Values.langfuseFanout.resources | nindent 12 }}
+        - name: otelcol
+          image: "{{ .Values.langfuseFanout.otelCollector.image.repository }}:{{ .Values.langfuseFanout.otelCollector.image.tag }}"
+          imagePullPolicy: {{ .Values.langfuseFanout.otelCollector.image.pullPolicy }}
+          args: ["--config=/etc/otelcol/otelcol.yaml"]
+          ports:
+            - name: otlp-internal
+              containerPort: 4319
+              protocol: TCP
+          env:
+            - name: LANGFUSE_FANOUT_CENTRAL_BASE_URL
+              value: {{ .Values.langfuseFanout.central.baseUrl | quote }}
+            - name: LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER
+              valueFrom:
+                secretKeyRef:
+                  name: {{ required "langfuseFanout.central.authHeaderSecret.name is required when langfuseFanout.enabled=true" .Values.langfuseFanout.central.authHeaderSecret.name | quote }}
+                  key: {{ .Values.langfuseFanout.central.authHeaderSecret.key | quote }}
+            - name: LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT
+              value: {{ .Values.langfuseFanout.otelCollector.receiverEndpoint | quote }}
            {{- range $name, $destination := .Values.langfuseFanout.tenant.destinations }}
            {{- include "librechat.langfuseFanout.validateDestinationKey" $name }}
            - name: LANGFUSE_FANOUT_TENANT_{{ $name | upper | replace "-" "_" }}_BASE_URL
@ -61,7 +131,7 @@ spec:
              subPath: otelcol.yaml
              readOnly: true
          resources:
-            {{- toYaml .Values.langfuseFanout.resources | nindent 12 }}
+            {{- toYaml .Values.langfuseFanout.otelCollector.resources | nindent 12 }}
      volumes:
        - name: config
          configMap:
--- a/helm/librechat/templates/langfuse-fanout-service.yaml
+++ b/helm/librechat/templates/langfuse-fanout-service.yaml
@ -5,6 +5,12 @@ metadata:
  name: {{ include "librechat.langfuseFanout.fullname" . }}
  labels:
    {{- include "librechat.langfuseFanout.labels" . | nindent 4 }}
+  {{- with .Values.langfuseFanout.service.annotations }}
+  annotations:
+    {{- range $key, $value := . }}
+    {{ $key }}: {{ $value | quote }}
+    {{- end }}
+  {{- end }}
 spec:
  type: {{ .Values.langfuseFanout.service.type }}
  ports:
--- a/helm/librechat/tests/langfuse_fanout_config_drift_test.sh
+++ b/helm/librechat/tests/langfuse_fanout_config_drift_test.sh
@ -1,212 +0,0 @@
-#!/usr/bin/env bash
-# Regression test for Langfuse fanout collector config drift.
-#
-# Compose mounts otel/langfuse-fanout/otelcol.yaml directly. Helm renders the
-# same collector topology into a ConfigMap from templates/langfuse-fanout-configmap.yaml.
-# This test renders Helm with the default tenant destinations and compares the
-# semantic collector topology, not raw YAML formatting.
-
-set -euo pipefail
-
-SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
-CHART_DIR="$(cd "${SCRIPT_DIR}/.." && pwd)"
-REPO_ROOT="$(cd "${CHART_DIR}/../.." && pwd)"
-STATIC_CONFIG="${REPO_ROOT}/otel/langfuse-fanout/otelcol.yaml"
-RENDER_CHART_DIR="$(mktemp -d -t librechat-fanout-chart.XXXXXX)"
-RENDERED_FILE="$(mktemp -t librechat-fanout-config-render.XXXXXX)"
-STATIC_SUMMARY="$(mktemp -t librechat-fanout-static-summary.XXXXXX)"
-HELM_SUMMARY="$(mktemp -t librechat-fanout-helm-summary.XXXXXX)"
-trap 'rm -rf "${RENDER_CHART_DIR}"; rm -f "${RENDERED_FILE}" "${STATIC_SUMMARY}" "${HELM_SUMMARY}"' EXIT
-
-if ! command -v helm >/dev/null 2>&1; then
-  echo "FAIL: helm not on PATH" >&2
-  exit 1
-fi
-
-if ! command -v node >/dev/null 2>&1; then
-  echo "FAIL: node not on PATH" >&2
-  exit 1
-fi
-
-mkdir -p "${RENDER_CHART_DIR}/templates"
-awk '/^dependencies:/{ exit } { print }' "${CHART_DIR}/Chart.yaml" > "${RENDER_CHART_DIR}/Chart.yaml"
-cp "${CHART_DIR}/values.yaml" "${RENDER_CHART_DIR}/values.yaml"
-cp "${CHART_DIR}/templates/_helpers.tpl" "${RENDER_CHART_DIR}/templates/_helpers.tpl"
-cp "${CHART_DIR}/templates/langfuse-fanout-configmap.yaml" \
-  "${RENDER_CHART_DIR}/templates/langfuse-fanout-configmap.yaml"
-
-helm template librechat "${RENDER_CHART_DIR}" \
-  --set langfuseFanout.enabled=true \
-  --set langfuseFanout.central.authHeaderSecret.name=langfuse-central \
-  --show-only templates/langfuse-fanout-configmap.yaml \
-  > "${RENDERED_FILE}"
-
-NODE_PATH="${REPO_ROOT}/node_modules${NODE_PATH:+:${NODE_PATH}}" \
-STATIC_CONFIG="${STATIC_CONFIG}" \
-RENDERED_FILE="${RENDERED_FILE}" \
-STATIC_SUMMARY="${STATIC_SUMMARY}" \
-HELM_SUMMARY="${HELM_SUMMARY}" \
-node <<'NODE'
-const fs = require('fs');
-const yaml = require('js-yaml');
-
-const DEFAULT_DESTINATIONS = ['eu', 'jp', 'us'];
-
-function fail(message) {
-  console.error(`FAIL: ${message}`);
-  process.exit(1);
-}
-
-function requireObject(value, path) {
-  if (value == null || typeof value !== 'object' || Array.isArray(value)) {
-    fail(`${path} must be an object`);
-  }
-  return value;
-}
-
-function sorted(value) {
-  return [...(value ?? [])].sort();
-}
-
-function extractHelmCollectorConfig(renderedFile) {
-  const docs = yaml.loadAll(fs.readFileSync(renderedFile, 'utf8')).filter(Boolean);
-  const configMap = docs.find(
-    (doc) =>
-      doc.kind === 'ConfigMap' &&
-      doc.metadata?.name === 'librechat-librechat-langfuse-fanout-config',
-  );
-  if (!configMap) {
-    fail('missing rendered Langfuse fanout ConfigMap');
-  }
-  const body = configMap.data?.['otelcol.yaml'];
-  if (typeof body !== 'string' || body.trim() === '') {
-    fail('rendered ConfigMap is missing data.otelcol.yaml');
-  }
-  return yaml.load(body);
-}
-
-function extractRouteDestinations(config) {
-  const routes = config.connectors?.['routing/langfuse_tenant_destination']?.table;
-  if (!Array.isArray(routes)) {
-    fail('routing/langfuse_tenant_destination.table must be an array');
-  }
-
-  return Object.fromEntries(
-    routes.map((route) => {
-      const condition = route.condition;
-      const match = /attributes\["librechat\.langfuse\.destination"\]\s*==\s*"([^"]+)"/.exec(
-        condition,
-      );
-      if (!match) {
-        fail(`unsupported routing condition: ${condition}`);
-      }
-      return [
-        match[1],
-        {
-          context: route.context,
-          pipelines: sorted(route.pipelines),
-        },
-      ];
-    }).sort(([left], [right]) => left.localeCompare(right)),
-  );
-}
-
-function summarize(config) {
-  requireObject(config, 'collector config');
-  const processors = requireObject(config.processors, 'processors');
-  const exporters = requireObject(config.exporters, 'exporters');
-  const pipelines = requireObject(config.service?.pipelines, 'service.pipelines');
-
-  const destinations = Object.keys(extractRouteDestinations(config)).sort();
-  const tenantExporters = Object.fromEntries(
-    destinations.map((destination) => {
-      const exporter = exporters[`otlphttp/tenant_${destination}`];
-      return [
-        destination,
-        {
-          endpoint: exporter?.endpoint,
-          authenticator: exporter?.auth?.authenticator,
-          headers: exporter?.headers ?? {},
-        },
-      ];
-    }),
-  );
-  const tenantPipelines = Object.fromEntries(
-    destinations.map((destination) => {
-      const pipeline = pipelines[`traces/tenant_${destination}`];
-      return [
-        destination,
-        {
-          receivers: sorted(pipeline?.receivers),
-          processors: sorted(pipeline?.processors),
-          exporters: sorted(pipeline?.exporters),
-        },
-      ];
-    }),
-  );
-  const tenantBatches = Object.fromEntries(
-    destinations.map((destination) => {
-      const batch = processors[`batch/by_auth_${destination}`];
-      return [
-        destination,
-        {
-          timeout: batch?.timeout,
-          sendBatchSize: batch?.send_batch_size,
-          metadataKeys: sorted(batch?.metadata_keys),
-          metadataCardinalityLimit: batch?.metadata_cardinality_limit,
-        },
-      ];
-    }),
-  );
-
-  return {
-    destinations,
-    extensions: config.service?.extensions ?? [],
-    headersSetter: config.extensions?.['headers_setter/tenant_passthrough'] ?? {},
-    receiver: config.receivers?.otlp?.protocols?.http ?? {},
-    routes: extractRouteDestinations(config),
-    tenantExportFilter: processors['filter/tenant_export']?.traces?.span ?? [],
-    dropRoutingAttributes: processors['attributes/drop_librechat_routing']?.actions ?? [],
-    centralExporter: exporters['otlphttp/central'] ?? {},
-    centralBatch: {
-      timeout: processors['batch/central']?.timeout,
-      sendBatchSize: processors['batch/central']?.send_batch_size,
-    },
-    centralPipeline: pipelines['traces/central'] ?? {},
-    tenantRouterPipeline: pipelines['traces/tenant'] ?? {},
-    tenantExporters,
-    tenantPipelines,
-    tenantBatches,
-  };
-}
-
-const staticConfig = yaml.load(fs.readFileSync(process.env.STATIC_CONFIG, 'utf8'));
-const helmConfig = extractHelmCollectorConfig(process.env.RENDERED_FILE);
-const staticSummary = summarize(staticConfig);
-const helmSummary = summarize(helmConfig);
-
-if (JSON.stringify(staticSummary.destinations) !== JSON.stringify(DEFAULT_DESTINATIONS)) {
-  fail(
-    `static collector destinations changed from default set: ${JSON.stringify(
-      staticSummary.destinations,
-    )}`,
-  );
-}
-if (JSON.stringify(helmSummary.destinations) !== JSON.stringify(DEFAULT_DESTINATIONS)) {
-  fail(
-    `Helm default collector destinations changed from default set: ${JSON.stringify(
-      helmSummary.destinations,
-    )}`,
-  );
-}
-
-fs.writeFileSync(process.env.STATIC_SUMMARY, `${JSON.stringify(staticSummary, null, 2)}\n`);
-fs.writeFileSync(process.env.HELM_SUMMARY, `${JSON.stringify(helmSummary, null, 2)}\n`);
-NODE
-
-if ! diff -u "${STATIC_SUMMARY}" "${HELM_SUMMARY}"; then
-  echo "FAIL: static Compose collector config and Helm-rendered collector config drifted" >&2
-  exit 1
-fi
-
-echo "PASS: Langfuse fanout Compose and Helm collector configs match for default destinations"
--- a/helm/librechat/tests/langfuse_fanout_selector_test.sh
+++ b/helm/librechat/tests/langfuse_fanout_selector_test.sh
@ -28,23 +28,22 @@ cp "${CHART_DIR}/templates/langfuse-fanout-service.yaml" \
  "${RENDER_CHART_DIR}/templates/langfuse-fanout-service.yaml"
 cp "${CHART_DIR}/templates/langfuse-fanout-deployment.yaml" \
  "${RENDER_CHART_DIR}/templates/langfuse-fanout-deployment.yaml"
-cp "${CHART_DIR}/templates/langfuse-fanout-configmap.yaml" \
-  "${RENDER_CHART_DIR}/templates/langfuse-fanout-configmap.yaml"

 helm template librechat "${RENDER_CHART_DIR}" \
  --set langfuseFanout.enabled=true \
  --set langfuseFanout.central.authHeaderSecret.name=langfuse-central \
+  --set langfuseFanout.redis.uri=redis://langfuse-fanout-redis:6379 \
  --show-only templates/service.yaml \
  --show-only templates/langfuse-fanout-service.yaml \
  --show-only templates/langfuse-fanout-deployment.yaml \
-  --show-only templates/langfuse-fanout-configmap.yaml \
  > "${RENDERED_FILE}"

 if helm template librechat "${RENDER_CHART_DIR}" \
  --set langfuseFanout.enabled=true \
  --set langfuseFanout.central.authHeaderSecret.name=langfuse-central \
+  --set langfuseFanout.redis.uri=redis://langfuse-fanout-redis:6379 \
  --set langfuseFanout.tenant.destinations.EU.baseUrl=https://cloud.langfuse.com \
-  --show-only templates/langfuse-fanout-configmap.yaml \
+  --show-only templates/langfuse-fanout-deployment.yaml \
  > /dev/null 2> "${INVALID_RENDER_ERROR}"; then
  echo "FAIL: Helm accepted invalid uppercase Langfuse fanout destination key" >&2
  exit 1
@ -87,17 +86,25 @@ function isSubset(subset, labels) {
  return Object.entries(subset ?? {}).every(([key, value]) => labels?.[key] === value);
 }

+function envValue(env, name) {
+  return (env ?? []).find((entry) => entry.name === name)?.value;
+}
+
 const mainService = find('Service', 'librechat-librechat');
 const fanoutService = find('Service', 'librechat-librechat-langfuse-fanout');
 const fanoutDeployment = find('Deployment', 'librechat-librechat-langfuse-fanout');
-const fanoutConfigMap = find('ConfigMap', 'librechat-librechat-langfuse-fanout-config');
+const fanoutContainer = fanoutDeployment.spec?.template?.spec?.containers?.find(
+  (container) => container.name === 'langfuse-fanout',
+);
+if (!fanoutContainer) {
+  fail('missing langfuse-fanout container');
+}

 const mainSelector = mainService.spec?.selector ?? {};
 const fanoutSelector = fanoutService.spec?.selector ?? {};
 const fanoutMatchLabels = fanoutDeployment.spec?.selector?.matchLabels ?? {};
 const fanoutPodLabels = fanoutDeployment.spec?.template?.metadata?.labels ?? {};
 const fanoutMetadataLabels = fanoutDeployment.metadata?.labels ?? {};
-const fanoutConfigLabels = fanoutConfigMap.metadata?.labels ?? {};

 if (isSubset(mainSelector, fanoutPodLabels)) {
  fail('main Service selector matches fanout pod labels');
@ -114,8 +121,20 @@ if (mainSelector['app.kubernetes.io/name'] === fanoutSelector['app.kubernetes.io
 if (fanoutMetadataLabels['app.kubernetes.io/name'] !== fanoutSelector['app.kubernetes.io/name']) {
  fail('fanout Deployment metadata labels do not use fanout app name');
 }
-if (fanoutConfigLabels['app.kubernetes.io/name'] !== fanoutSelector['app.kubernetes.io/name']) {
-  fail('fanout ConfigMap labels do not use fanout app name');
+if (envValue(fanoutContainer.env, 'LANGFUSE_FANOUT_REDIS_URI') !== 'redis://langfuse-fanout-redis:6379') {
+  fail('fanout Deployment did not render configured Redis URI');
+}
+if (
+  envValue(fanoutContainer.env, 'LANGFUSE_FANOUT_PUBLIC_URL') !==
+  'http://librechat-librechat-langfuse-fanout.default.svc.cluster.local:4318'
+) {
+  fail('fanout Deployment did not render derived public URL');
+}
+if (fanoutContainer.livenessProbe?.httpGet?.path !== '/healthz') {
+  fail('fanout Deployment missing /healthz liveness probe');
+}
+if (fanoutContainer.readinessProbe?.httpGet?.path !== '/healthz') {
+  fail('fanout Deployment missing /healthz readiness probe');
 }

 console.log('PASS: Langfuse fanout selectors are isolated from the main LibreChat Service');
--- a/helm/librechat/values.yaml
+++ b/helm/librechat/values.yaml
@ -288,20 +288,44 @@ langfuseFanout:
  enabled: false
  replicaCount: 1
  image:
-    repository: otel/opentelemetry-collector-contrib
-    tag: "0.143.0"
+    repository: librechat-langfuse-fanout
+    tag: "latest"
    pullPolicy: IfNotPresent
+  otelCollector:
+    receiverEndpoint: 127.0.0.1:4319
+    image:
+      repository: otel/opentelemetry-collector-contrib
+      tag: "0.143.0"
+      pullPolicy: IfNotPresent
+    resources: {}
  service:
    type: ClusterIP
    port: 4318
+    annotations: {}
  central:
    baseUrl: https://cloud.langfuse.com
    authHeaderSecret:
      name: ""
      key: LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER
+  metrics:
+    secret:
+      # Optional bearer token secret for scraping the gateway's /metrics endpoint.
+      # When omitted, /metrics returns 401.
+      name: ""
+      key: METRICS_SECRET
+  redis:
+    # Redis stores short-lived Langfuse media upload plans so multiple gateway
+    # replicas can handle create/upload requests. If empty and redis.enabled is
+    # true, the chart derives the bundled Redis service URI.
+    uri: ""
+    username: ""
+    passwordSecret:
+      name: ""
+      key: REDIS_PASSWORD
+    keyPrefix: langfuse-fanout
  tenant:
    # Destination map keys are emitted by LibreChat as trace attributes and
-    # matched by the collector. Use lowercase keys matching ^[a-z][a-z0-9_-]*$.
+    # matched by the gateway. Use lowercase keys matching ^[a-z][a-z0-9_-]*$.
    destinations:
      eu:
        baseUrl: https://cloud.langfuse.com
@ -309,11 +333,24 @@ langfuseFanout:
        baseUrl: https://us.cloud.langfuse.com
      jp:
        baseUrl: https://jp.cloud.langfuse.com
+  upstreamTimeout: 30s
+  # Optional override for one-time media upload URLs returned by the gateway.
+  # When empty, the chart derives the internal fanout Service URL.
+  publicUrl: ""
+  traceCollectorUrl: http://127.0.0.1:4319
  memoryLimitMiB: 256
  memorySpikeLimitMiB: 64
  batchTimeout: 1s
  batchSendSize: 128
  metadataCardinalityLimit: 1000
+  livenessProbe:
+    httpGet:
+      path: /healthz
+      port: 4318
+  readinessProbe:
+    httpGet:
+      path: /healthz
+      port: 4318
  resources: {}
  podAnnotations: {}
  podLabels: {}
--- a/otel/langfuse-fanout/Dockerfile
+++ b/otel/langfuse-fanout/Dockerfile
@ -0,0 +1,13 @@
+FROM golang:1.23-alpine AS builder
+WORKDIR /src
+COPY otel/langfuse-fanout/go.mod otel/langfuse-fanout/go.sum* ./
+RUN go mod download
+COPY otel/langfuse-fanout/ ./
+RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" -o /out/langfuse-fanout ./cmd/langfuse-fanout
+
+FROM alpine:3.22
+RUN addgroup -S app && adduser -S app -G app
+COPY --from=builder /out/langfuse-fanout /usr/local/bin/langfuse-fanout
+USER app
+EXPOSE 4318
+ENTRYPOINT ["/usr/local/bin/langfuse-fanout"]
--- a/otel/langfuse-fanout/README.md
+++ b/otel/langfuse-fanout/README.md
@ -1,59 +1,106 @@
-# Langfuse Fanout Collector
+# Langfuse Fanout Gateway

 LibreChat can send tenant-scoped agent traces to a tenant Langfuse project and
-also copy those traces to a central Langfuse project. This is optional and is
-disabled unless you explicitly deploy the fanout collector.
+also copy those traces to a central Langfuse project. When trace payloads
+contain Langfuse media references, the gateway also copies the media upload to
+central and tenant Langfuse storage. This is optional and is disabled unless you
+explicitly deploy the fanout gateway.
+
+The deployment is a hybrid:
+
+- the Go gateway is the only endpoint LibreChat talks to;
+- trace requests are proxied to an internal OpenTelemetry collector;
+- the collector owns trace memory limiting, batching, routing, and export;
+- the Go gateway owns Langfuse media create/upload/patch fanout.

 ## How It Works

 - Agent traces use Langfuse OTLP ingestion.
- LibreChat sends tenant traces to the local collector when
+- LibreChat sends tenant traces to the local fanout gateway when
  `LANGFUSE_FANOUT_ENABLED=true` and `LANGFUSE_FANOUT_COLLECTOR_URL` points at
-  the collector.
+  the fanout gateway.
+- The gateway forwards trace requests to the internal OpenTelemetry collector
+  at `LANGFUSE_FANOUT_TRACE_COLLECTOR_URL`.
 - The collector exports every trace to the central Langfuse project using
  `LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER`. This prebuilt header is collector-only;
  the LibreChat app derives central score auth from `LANGFUSE_PUBLIC_KEY` and
  `LANGFUSE_SECRET_KEY`.
- The collector also exports the same trace to the tenant Langfuse project by
-  routing a LibreChat-stamped destination key to one of the configured tenant
-  Langfuse base URLs, then forwarding the tenant `Authorization` header that
-  LibreChat attaches to the OTLP request.
- Tenant export is conditional. LibreChat marks traces as tenant-exportable only
-  when tenant keys are configured and `LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED`
-  is not true; unmarked traces are still exported to central but are dropped by
-  the tenant pipeline.
- LibreChat's routing attributes are consumed by the collector and deleted
-  before central or tenant export, so they are not forwarded to Langfuse.
+- The collector also exports tenant-enabled traces to the tenant Langfuse
+  project by routing on `librechat.langfuse.destination`, then forwarding the
+  tenant `Authorization` header that LibreChat attaches to the OTLP request.
+- For tenant-exportable runs, LibreChat uses a destination-scoped gateway URL
+  like `http://langfuse-fanout-collector:4318/tenant/us`. Langfuse media upload
+  requests do not carry span attributes, so this path gives the gateway the
+  destination needed to copy media into the tenant's Langfuse region. For
+  traces on this path, the gateway restores the internal tenant routing
+  attributes before handing the request to the collector.
+- Before export, the collector deletes the internal `librechat.langfuse.*`
+  routing attributes from central and tenant traces.
+- Langfuse media upload is fanned out by calling `POST /api/public/media` on
+  central and tenant Langfuse, returning a one-time gateway upload URL, then
+  uploading the received bytes to each upstream presigned upload URL. The SDK's
+  `PATCH /api/public/media/{mediaId}` status call is also fanned out.
+- Tenant export is conditional. LibreChat uses a destination-scoped gateway URL
+  only when tenant keys are configured, the tenant base URL matches a configured
+  startup destination, and `LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED` is not true.
+  Other traces are still exported to central through the gateway without tenant
+  auth.
 - User feedback scores use Langfuse's direct REST API from the LibreChat API
  process. Central scores use LibreChat's normal central Langfuse env config;
  tenant scores use tenant app configuration when tenant fanout is enabled.

 Tenant Langfuse keys are expected to come from LibreChat app configuration, for
 example from an admin panel or another configuration data source. They are not
-defined in this collector config.
+defined in this gateway config.

 ## Limitations

 - Langfuse base URLs are startup configuration. `LANGFUSE_FANOUT_CENTRAL_BASE_URL`
-  and the tenant destination map must be known when LibreChat and the collector
-  start. Tenant app configuration may choose any configured tenant destination.
+  and `LANGFUSE_FANOUT_TENANT_DESTINATIONS` must be known when LibreChat and the
+  gateway start. Tenant app configuration may choose any configured tenant
+  destination.
 - Tenant Langfuse API keys can be added, changed, or disabled in tenant app
-  configuration at runtime without restarting LibreChat or the collector.
+  configuration at runtime without restarting LibreChat or the gateway.
 - Tenant app configuration must set a Langfuse base URL matching one of the
  startup destinations before tenant trace/score export is enabled; keys alone
  are treated as central-only.
 - `LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED=true` can be set on LibreChat as an
  emergency switch to stop tenant trace and score export while keeping central
-  collector export active. When omitted, false, or blank, tenant export remains
+  gateway export active. When omitted, false, or blank, tenant export remains
  available if tenant keys and a known destination are configured.
 - This supports Langfuse Cloud and self-hosted Langfuse as long as each allowed
-  tenant base URL is configured at LibreChat/collector startup. Runtime tenant
+  tenant base URL is configured at LibreChat/gateway startup. Runtime tenant
  config selects from those known destinations; it does not inject arbitrary
-  export URLs into the collector.
- The provided Compose collector config is a three-region Langfuse Cloud preset
-  (`eu`, `us`, `jp`). For self-hosted or additional destination keys,
-  use Helm values or update/generate the collector routing table and exporters
-  in lockstep with `LANGFUSE_FANOUT_TENANT_DESTINATIONS`.
+  export URLs into the gateway.
+- The provided Compose gateway config is a three-region Langfuse Cloud preset
+  (`eu`, `us`, `jp`). Compose's static collector config routes only those keys;
+  the gateway fails startup when `LANGFUSE_FANOUT_TENANT_DESTINATIONS` contains
+  a key outside `LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS`. For self-hosted or
+  additional destination keys, update the collector config too or use Helm.
+- Helm binds the internal collector receiver to `127.0.0.1:4319` because the
+  collector is a sidecar. Compose binds it to `0.0.0.0:4319` on the private
+  `langfuse-fanout` network. Do not publish the internal collector receiver
+  outside the fanout deployment; tenant routing validation happens in the
+  gateway before traces reach the collector.
+- The gateway stores short-lived one-time media upload plans in Redis. This lets
+  media create and byte-upload requests land on different gateway replicas.
+  Compose includes a private Redis container; Helm can derive the URI from the
+  bundled Redis chart or use an explicit `langfuseFanout.redis.uri`.
+- The gateway requires an explicit public/internal base URL for one-time upload
+  URLs. Compose sets `LANGFUSE_FANOUT_PUBLIC_URL` to its private gateway
+  service URL. Helm derives the fanout Service DNS name unless `publicUrl` is
+  set.
+- Media fanout is not transactional across central and tenant projects. If one
+  destination accepts `POST /api/public/media` and another fails, LibreChat sees
+  a gateway error and will not upload bytes, but the successful destination may
+  retain a short-lived, unused media record.
+- Trace batching is handled by the collector. By default it flushes after 128
+  items or 1 second, and tenant batches are separated by the request
+  `Authorization` metadata.
+- The gateway exposes Prometheus metrics at `/metrics` using the same bearer
+  token shape as LibreChat. Set `LANGFUSE_FANOUT_METRICS_SECRET`, or provide
+  `METRICS_SECRET` in the gateway environment. When neither is set, `/metrics`
+  returns 401.

 ## Docker Compose

@ -64,18 +111,29 @@ Set the central Langfuse destination in `.env`:
 # region as LANGFUSE_FANOUT_CENTRAL_BASE_URL when applicable.
 LANGFUSE_BASE_URL=https://cloud.langfuse.com

-# Used by the collector for central trace export.
+# Used by the gateway for central trace and media export.
 LANGFUSE_FANOUT_CENTRAL_BASE_URL=https://cloud.langfuse.com
 LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=Basic <base64-public-key-colon-secret-key>
-# Compose's included collector config supports these three destination keys.
-# Do not add a custom key unless the collector routing table also has a matching pipeline/exporter.
+# Compose's included gateway config supports these three destination keys.
 LANGFUSE_FANOUT_TENANT_DESTINATIONS=eu=https://cloud.langfuse.com,us=https://us.cloud.langfuse.com,jp=https://jp.cloud.langfuse.com
+LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS=eu,us,jp
 LANGFUSE_FANOUT_TENANT_EU_BASE_URL=https://cloud.langfuse.com
 LANGFUSE_FANOUT_TENANT_US_BASE_URL=https://us.cloud.langfuse.com
 LANGFUSE_FANOUT_TENANT_JP_BASE_URL=https://jp.cloud.langfuse.com
 LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED=false
+LANGFUSE_FANOUT_UPSTREAM_TIMEOUT=30s
+LANGFUSE_FANOUT_PUBLIC_URL=http://langfuse-fanout-collector:4318
+LANGFUSE_FANOUT_REDIS_URI=redis://langfuse-fanout-redis:6379
+LANGFUSE_FANOUT_REDIS_USERNAME=
+LANGFUSE_FANOUT_REDIS_PASSWORD=
+LANGFUSE_FANOUT_REDIS_KEY_PREFIX=langfuse-fanout
+LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT=0.0.0.0:4319
+LANGFUSE_FANOUT_METRICS_SECRET=<metrics-bearer-token>
+LANGFUSE_FANOUT_MEMORY_LIMIT_MIB=256
+LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB=64
 LANGFUSE_FANOUT_BATCH_TIMEOUT=1s
 LANGFUSE_FANOUT_BATCH_SEND_SIZE=128
+LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT=1000
 ```

 Langfuse Cloud base URL options:
@ -98,8 +156,9 @@ For the deployed compose stack:
 docker compose -f deploy-compose.yml -f deploy-compose.langfuse-fanout.yml up -d
 ```

-The override sets `LANGFUSE_FANOUT_ENABLED=true` and points LibreChat at
-`http://langfuse-fanout-collector:4318`.
+The override builds the fanout gateway image, sets `LANGFUSE_FANOUT_ENABLED=true`, and points LibreChat at
+`http://langfuse-fanout-collector:4318`. It also starts an internal
+`langfuse-fanout-otel` service on the private fanout network for trace export.

 ## Helm

@ -110,9 +169,13 @@ kubectl create secret generic langfuse-central \
  --from-literal=LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER='Basic <base64-public-key-colon-secret-key>'
 ```

-Enable the collector in values:
+Enable the gateway in values. Use either the bundled Redis chart as shown here
+or set `langfuseFanout.redis.uri` to an external Redis service.

 ```yaml
+redis:
+  enabled: true
+
 langfuseFanout:
  enabled: true
  central:
@ -120,6 +183,10 @@ langfuseFanout:
    authHeaderSecret:
      name: langfuse-central
      key: LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER
+  metrics:
+    secret:
+      name: librechat-metrics
+      key: METRICS_SECRET
  tenant:
    destinations:
      eu:
@ -128,30 +195,91 @@ langfuseFanout:
        baseUrl: https://us.cloud.langfuse.com
      jp:
        baseUrl: https://jp.cloud.langfuse.com
+  upstreamTimeout: 30s
+  publicUrl: ""
+  otelCollector:
+    receiverEndpoint: 127.0.0.1:4319
+  redis:
+    uri: ""
+    username: ""
+    passwordSecret:
+      name: ""
+      key: REDIS_PASSWORD
+    keyPrefix: langfuse-fanout
+  memoryLimitMiB: 256
+  memorySpikeLimitMiB: 64
  batchTimeout: 1s
  batchSendSize: 128
+  metadataCardinalityLimit: 1000
 ```

-The chart renders the collector Deployment, Service, and collector ConfigMap,
-and injects `LANGFUSE_FANOUT_ENABLED` plus `LANGFUSE_FANOUT_COLLECTOR_URL` into
-the LibreChat app ConfigMap when they are not already supplied in
-`librechat.configEnv`.
+The chart renders one fanout Deployment with two containers: the gateway on
+`4318` and an internal OpenTelemetry collector on `4319`. The Service exposes
+only the gateway. The chart also injects `LANGFUSE_FANOUT_ENABLED` plus
+`LANGFUSE_FANOUT_COLLECTOR_URL` into the LibreChat app ConfigMap when they are
+not already supplied in `librechat.configEnv`.
+
+Set `langfuseFanout.redis.uri` when using an external Redis service. If Redis
+requires auth, set `langfuseFanout.redis.username` and point
+`langfuseFanout.redis.passwordSecret.name`/`.key` at an existing Kubernetes
+Secret. When using the bundled Redis chart with auth enabled, create a password
+Secret for the gateway or provide an explicit authenticated URI.
+Prefer `passwordSecret` over embedding credentials in `redis.uri`, because the
+URI is rendered directly into the Deployment environment.
+Scale the gateway manually with `langfuseFanout.replicaCount`; the chart does
+not create a fanout HPA. The gateway container has configurable `/healthz`
+liveness and readiness probes under `langfuseFanout`.
+
+Useful gateway metrics include:
+
+- `langfuse_fanout_http_requests_total`
+- `langfuse_fanout_upstream_requests_total`
+- `langfuse_fanout_trace_exports_total`
+- `langfuse_fanout_media_upload_plans_created_total`
+- `langfuse_fanout_media_upload_plans_completed_total`
+- `langfuse_fanout_media_upload_plan_misses_total`
+- `langfuse_fanout_media_upload_plan_store_errors_total`
+- `langfuse_fanout_media_upload_bytes`
+- `langfuse_fanout_media_divergence_total`
+
+`langfuse_fanout_media_divergence_total{kind="media_id"}` is the correctness
+signal for trace/media token fanout. `kind="upload_url_presence"` records that
+some destinations returned an upload URL while others treated the media as
+already uploaded.

 ## Notes

- The collector only handles traces. Feedback scores go directly to Langfuse's
-  REST API from the LibreChat API process.
+- The gateway handles Langfuse media uploads and proxies traces to the internal
+  collector. Feedback scores go directly to Langfuse's REST API from the
+  LibreChat API process.
 - `LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER` must be a full Basic auth header and is
-  consumed by the collector only. The app does not use it for scores.
- `LANGFUSE_FANOUT_CENTRAL_BASE_URL` is also consumed by the collector only.
+  consumed by the fanout deployment only. The app does not use it for scores.
+- `LANGFUSE_FANOUT_CENTRAL_BASE_URL` is also consumed by the fanout deployment only.
  For non-EU central feedback scores, set LibreChat's normal `LANGFUSE_BASE_URL`
  to the same central Langfuse region.
 - Tenant destinations default to the three configured Langfuse Cloud regions. Add or
  override `langfuseFanout.tenant.destinations` in Helm for self-hosted or
-  custom destinations. Compose's included collector config is static; if you add
-  custom destination keys there, update or regenerate `otelcol.yaml` as well.
- `LANGFUSE_FANOUT_BATCH_TIMEOUT` and `LANGFUSE_FANOUT_BATCH_SEND_SIZE` tune
-  the OTel batch processors. The defaults (`1s`, `128`) favor low latency and
-  modest memory use; high-volume deployments may increase them for throughput.
- `LANGFUSE_FANOUT_COLLECTOR_URL` is the local collector URL used by LibreChat,
-  not a Langfuse Cloud base URL.
+  custom destinations.
+- `LANGFUSE_FANOUT_UPSTREAM_TIMEOUT` tunes the timeout for gateway calls to
+  Langfuse APIs and presigned media upload URLs.
+- `LANGFUSE_FANOUT_PUBLIC_URL` pins the base URL returned for the SDK's
+  one-time media upload. The gateway fails startup when it is unset or invalid;
+  this avoids trusting request `Host` or `X-Forwarded-Host` headers.
+- `LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS` is a startup guard that must contain
+  every key in `LANGFUSE_FANOUT_TENANT_DESTINATIONS`; this prevents media
+  fanout from accepting a destination the collector cannot route traces to.
+- `LANGFUSE_FANOUT_REDIS_URI`, optional `LANGFUSE_FANOUT_REDIS_USERNAME`,
+  optional `LANGFUSE_FANOUT_REDIS_PASSWORD`, and
+  `LANGFUSE_FANOUT_REDIS_KEY_PREFIX` configure the shared one-time media upload
+  plan store. The gateway fails startup without a Redis URI.
+- `LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT` controls the internal collector
+  receiver bind address.
+- `LANGFUSE_FANOUT_METRICS_SECRET` protects the gateway `/metrics` endpoint.
+  If unset, the gateway falls back to `METRICS_SECRET` when present.
+- `LANGFUSE_FANOUT_MEMORY_LIMIT_MIB`,
+  `LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB`, `LANGFUSE_FANOUT_BATCH_TIMEOUT`,
+  `LANGFUSE_FANOUT_BATCH_SEND_SIZE`, and
+  `LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT` tune the internal collector.
+- `LANGFUSE_FANOUT_COLLECTOR_URL` is the local gateway URL used by LibreChat.
+  The env name is kept for compatibility with the original collector shape; it
+  is not a Langfuse Cloud base URL.
--- a/otel/langfuse-fanout/cmd/langfuse-fanout/main.go
+++ b/otel/langfuse-fanout/cmd/langfuse-fanout/main.go
--- a/otel/langfuse-fanout/cmd/langfuse-fanout/main_test.go
+++ b/otel/langfuse-fanout/cmd/langfuse-fanout/main_test.go
@ -0,0 +1,888 @@
+package main
+
+import (
+	"bytes"
+	"compress/gzip"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	tracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1"
+	commonv1 "go.opentelemetry.io/proto/otlp/common/v1"
+	resourcev1 "go.opentelemetry.io/proto/otlp/resource/v1"
+	tracev1 "go.opentelemetry.io/proto/otlp/trace/v1"
+	"google.golang.org/protobuf/proto"
+)
+
+func TestLoadConfigRejectsDestinationsMissingTraceRoutes(t *testing.T) {
+	t.Setenv("LANGFUSE_FANOUT_CENTRAL_BASE_URL", "https://cloud.langfuse.com")
+	t.Setenv("LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER", "Basic central")
+	t.Setenv("LANGFUSE_FANOUT_PUBLIC_URL", "http://fanout.local:4318")
+	t.Setenv("LANGFUSE_FANOUT_TENANT_DESTINATIONS", "eu=https://cloud.langfuse.com,ca=https://example.com")
+	t.Setenv("LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS", "eu,us,jp")
+
+	_, err := loadConfig()
+	if err == nil || !strings.Contains(err.Error(), `tenant destination "ca"`) {
+		t.Fatalf("expected missing trace route error, got %v", err)
+	}
+}
+
+func TestLoadConfigRejectsTenantDestinationsWithoutTraceKeys(t *testing.T) {
+	t.Setenv("LANGFUSE_FANOUT_CENTRAL_BASE_URL", "https://cloud.langfuse.com")
+	t.Setenv("LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER", "Basic central")
+	t.Setenv("LANGFUSE_FANOUT_PUBLIC_URL", "http://fanout.local:4318")
+	t.Setenv("LANGFUSE_FANOUT_TENANT_DESTINATIONS", "eu=https://cloud.langfuse.com")
+	t.Setenv("LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS", "")
+
+	_, err := loadConfig()
+	if err == nil || !strings.Contains(err.Error(), "LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS is required") {
+		t.Fatalf("expected missing trace keys error, got %v", err)
+	}
+}
+
+func TestLoadConfigRejectsInvalidPublicURL(t *testing.T) {
+	t.Setenv("LANGFUSE_FANOUT_CENTRAL_BASE_URL", "https://cloud.langfuse.com")
+	t.Setenv("LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER", "Basic central")
+	t.Setenv("LANGFUSE_FANOUT_PUBLIC_URL", "ftp://example.com")
+
+	_, err := loadConfig()
+	if err == nil || !strings.Contains(err.Error(), "LANGFUSE_FANOUT_PUBLIC_URL") {
+		t.Fatalf("expected invalid public URL error, got %v", err)
+	}
+}
+
+func TestLoadConfigRequiresRedisURI(t *testing.T) {
+	t.Setenv("LANGFUSE_FANOUT_CENTRAL_BASE_URL", "https://cloud.langfuse.com")
+	t.Setenv("LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER", "Basic central")
+	t.Setenv("LANGFUSE_FANOUT_PUBLIC_URL", "http://fanout.local:4318")
+
+	_, err := loadConfig()
+	if err == nil || !strings.Contains(err.Error(), "LANGFUSE_FANOUT_REDIS_URI") {
+		t.Fatalf("expected missing Redis URI error, got %v", err)
+	}
+}
+
+func TestNormalizeBaseURLAllowsOnlyHTTPAndHTTPS(t *testing.T) {
+	if got := normalizeBaseURL("http://localhost:3000/path/"); got != "http://localhost:3000/path" {
+		t.Fatalf("http URL normalized to %q", got)
+	}
+	if got := normalizeBaseURL("https://cloud.langfuse.com/"); got != "https://cloud.langfuse.com" {
+		t.Fatalf("https URL normalized to %q", got)
+	}
+	if got := normalizeBaseURL("file:///tmp/langfuse"); got != "" {
+		t.Fatalf("file URL should be rejected, got %q", got)
+	}
+}
+
+func TestTraceProxyForwardsExistingRoutingAttributesToCollector(t *testing.T) {
+	t.Parallel()
+
+	var collectorTrace []byte
+	collector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != otelTracePath {
+			t.Fatalf("unexpected collector path %s", r.URL.Path)
+		}
+		if got := r.Header.Get("Authorization"); got != "Basic tenant" {
+			t.Fatalf("collector auth = %q", got)
+		}
+		collectorTrace, _ = io.ReadAll(r.Body)
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte("{}"))
+	}))
+	defer collector.Close()
+
+	gw := newTestGatewayWithCollector(collector.URL)
+	body := buildTraceRequest(t, map[string]string{
+		tenantExportAttribute: "true",
+		tenantDestAttribute:   "eu",
+		"kept":                "value",
+	})
+
+	req := httptest.NewRequest(http.MethodPost, otelTracePath, bytes.NewReader(body))
+	req.Header.Set("Content-Type", "application/x-protobuf")
+	req.Header.Set("Authorization", "Basic tenant")
+	resp := httptest.NewRecorder()
+
+	gw.handle(resp, req)
+	if resp.Code != http.StatusOK {
+		t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String())
+	}
+	if resp.Body.String() != "{}" {
+		t.Fatalf("expected collector response body, got %s", resp.Body.String())
+	}
+	if len(collectorTrace) == 0 {
+		t.Fatal("expected collector export")
+	}
+
+	attrs := parseTraceAttributes(t, collectorTrace)
+	if attrs[tenantExportAttribute] != "true" || attrs[tenantDestAttribute] != "eu" {
+		t.Fatalf("collector trace missing routing attrs: %#v", attrs)
+	}
+	if attrs["kept"] != "value" {
+		t.Fatalf("collector trace lost kept attr: %#v", attrs)
+	}
+}
+
+func TestGzipTraceProxyAddsRoutingAttributesFromPath(t *testing.T) {
+	t.Parallel()
+
+	var collectorTrace []byte
+	var collectorEncoding string
+	collector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		collectorEncoding = r.Header.Get("Content-Encoding")
+		reader, err := gzip.NewReader(r.Body)
+		if err != nil {
+			t.Fatalf("collector gzip reader: %v", err)
+		}
+		defer reader.Close()
+		collectorTrace, _ = io.ReadAll(reader)
+		_, _ = w.Write([]byte(`{"partialSuccess":{}}`))
+	}))
+	defer collector.Close()
+
+	var zipped bytes.Buffer
+	zipper := gzip.NewWriter(&zipped)
+	if _, err := zipper.Write(buildTraceRequest(t, map[string]string{"kept": "value"})); err != nil {
+		t.Fatal(err)
+	}
+	if err := zipper.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	gw := newTestGatewayWithCollector(collector.URL)
+	req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu"+otelTracePath, bytes.NewReader(zipped.Bytes()))
+	req.Header.Set("Content-Type", "application/x-protobuf")
+	req.Header.Set("Content-Encoding", "gzip")
+	req.Header.Set("Authorization", "Basic tenant")
+	resp := httptest.NewRecorder()
+
+	gw.handle(resp, req)
+	if resp.Code != http.StatusOK {
+		t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String())
+	}
+	if collectorEncoding != "gzip" {
+		t.Fatalf("collector encoding = %q", collectorEncoding)
+	}
+	if resp.Body.String() != `{"partialSuccess":{}}` {
+		t.Fatalf("expected collector response body, got %s", resp.Body.String())
+	}
+	attrs := parseTraceAttributes(t, collectorTrace)
+	if attrs[tenantExportAttribute] != "true" || attrs[tenantDestAttribute] != "eu" || attrs["kept"] != "value" {
+		t.Fatalf("collector trace attrs = %#v", attrs)
+	}
+}
+
+func TestJSONTraceProxyAddsRoutingAttributesFromPath(t *testing.T) {
+	t.Parallel()
+
+	var collectorTrace []byte
+	var collectorContentType string
+	collector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != otelTracePath {
+			t.Fatalf("unexpected collector path %s", r.URL.Path)
+		}
+		if got := r.Header.Get("Authorization"); got != "Basic tenant" {
+			t.Fatalf("collector auth = %q", got)
+		}
+		collectorContentType = r.Header.Get("Content-Type")
+		collectorTrace, _ = io.ReadAll(r.Body)
+		_, _ = w.Write([]byte("{}"))
+	}))
+	defer collector.Close()
+
+	gw := newTestGatewayWithCollector(collector.URL)
+	body := buildJSONTraceRequest(t, map[string]any{
+		"kept": "value",
+	})
+
+	req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu"+otelTracePath, bytes.NewReader(body))
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Authorization", "Basic tenant")
+	resp := httptest.NewRecorder()
+
+	gw.handle(resp, req)
+	if resp.Code != http.StatusOK {
+		t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String())
+	}
+	if len(collectorTrace) == 0 {
+		t.Fatal("expected collector export")
+	}
+	if collectorContentType != "application/json" {
+		t.Fatalf("content type = %q", collectorContentType)
+	}
+	attrs := parseJSONTraceAttributes(t, collectorTrace)
+	if attrs[tenantExportAttribute] != "true" || attrs[tenantDestAttribute] != "eu" {
+		t.Fatalf("collector trace missing routing attrs: %#v", attrs)
+	}
+	if attrs["kept"] != "value" {
+		t.Fatalf("collector trace lost kept attr: %#v", attrs)
+	}
+}
+
+func TestMediaUploadFansOutToCentralAndTenant(t *testing.T) {
+	t.Parallel()
+
+	var mu sync.Mutex
+	uploads := map[string]string{}
+	upstream := func(name string) *httptest.Server {
+		return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			switch {
+			case r.Method == http.MethodPost && r.URL.Path == mediaPath:
+				uploadURL := "http://" + r.Host + "/upload/" + name
+				writeJSON(w, http.StatusCreated, mediaUploadResponse{
+					MediaID:   "same-media-id",
+					UploadURL: &uploadURL,
+				})
+			case r.Method == http.MethodPut && r.URL.Path == "/upload/"+name:
+				body, _ := io.ReadAll(r.Body)
+				mu.Lock()
+				uploads[name] = string(body)
+				mu.Unlock()
+				w.WriteHeader(http.StatusOK)
+			case r.Method == http.MethodPatch && r.URL.Path == mediaPath+"/same-media-id":
+				w.WriteHeader(http.StatusNoContent)
+			default:
+				http.NotFound(w, r)
+			}
+		}))
+	}
+	central := upstream("central")
+	defer central.Close()
+	tenant := upstream("tenant")
+	defer tenant.Close()
+
+	store := newFakeUploadPlanStore()
+	createGateway := newTestGatewayWithStore(central.URL, map[string]string{"eu": tenant.URL}, store)
+	uploadGateway := newTestGatewayWithStore(central.URL, map[string]string{"eu": tenant.URL}, store)
+	createBody := `{"traceId":"trace","contentType":"image/png","contentLength":5,"sha256Hash":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","field":"input"}`
+	req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu"+mediaPath, strings.NewReader(createBody))
+	req.Header.Set("Authorization", "Basic tenant")
+	resp := httptest.NewRecorder()
+
+	createGateway.handle(resp, req)
+	if resp.Code != http.StatusCreated {
+		t.Fatalf("create status = %d, body = %s", resp.Code, resp.Body.String())
+	}
+	var create mediaUploadResponse
+	if err := json.NewDecoder(resp.Body).Decode(&create); err != nil {
+		t.Fatal(err)
+	}
+	if create.MediaID != "same-media-id" || create.UploadURL == nil || !strings.Contains(*create.UploadURL, mediaUploadProxyPath) {
+		t.Fatalf("unexpected create response: %#v", create)
+	}
+	uploadID := strings.TrimPrefix(newUploadURLPath(t, *create.UploadURL), mediaUploadProxyPath)
+	store.mu.Lock()
+	storedPlan := store.plans[uploadID]
+	store.mu.Unlock()
+	storedPlanJSON, err := json.Marshal(storedPlan)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if strings.Contains(string(storedPlanJSON), "Basic ") {
+		t.Fatalf("stored upload plan leaked authorization: %s", storedPlanJSON)
+	}
+
+	uploadReq := httptest.NewRequest(http.MethodPut, *create.UploadURL, strings.NewReader("hello"))
+	uploadReq.Header.Set("Content-Type", "image/png")
+	uploadResp := httptest.NewRecorder()
+	uploadGateway.handle(uploadResp, uploadReq)
+	if uploadResp.Code != http.StatusOK {
+		t.Fatalf("upload status = %d, body = %s", uploadResp.Code, uploadResp.Body.String())
+	}
+
+	patchReq := httptest.NewRequest(http.MethodPatch, tenantPrefix+"eu"+mediaPath+"/same-media-id", strings.NewReader(`{"uploadHttpStatus":200}`))
+	patchReq.Header.Set("Authorization", "Basic tenant")
+	patchResp := httptest.NewRecorder()
+	uploadGateway.handle(patchResp, patchReq)
+	if patchResp.Code != http.StatusNoContent {
+		t.Fatalf("patch status = %d, body = %s", patchResp.Code, patchResp.Body.String())
+	}
+
+	mu.Lock()
+	defer mu.Unlock()
+	if uploads["central"] != "hello" || uploads["tenant"] != "hello" {
+		t.Fatalf("uploads = %#v", uploads)
+	}
+}
+
+func TestMediaUploadRejectsInvalidIDBeforeReadingBody(t *testing.T) {
+	t.Parallel()
+
+	gw := newTestGateway("http://central.invalid", nil)
+	reader := &failingReader{}
+	req := httptest.NewRequest(http.MethodPut, mediaUploadProxyPath+"not-valid", reader)
+	resp := httptest.NewRecorder()
+
+	gw.handle(resp, req)
+	if resp.Code != http.StatusNotFound {
+		t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String())
+	}
+	if reader.read {
+		t.Fatal("invalid upload id should not read request body")
+	}
+}
+
+func TestMediaUploadIsOneTime(t *testing.T) {
+	t.Parallel()
+
+	var uploads int
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPut || r.URL.Path != "/upload" {
+			http.NotFound(w, r)
+			return
+		}
+		uploads++
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer upstream.Close()
+
+	store := newFakeUploadPlanStore()
+	uploadID := "abcdef1234"
+	store.Put(context.Background(), uploadID, uploadPlan{
+		ExpiresAt:     time.Now().Add(time.Hour),
+		ContentLength: 5,
+		Destinations: []uploadDestination{{
+			Name:      "central",
+			UploadURL: upstream.URL + "/upload",
+		}},
+	})
+	gw := newTestGatewayWithStore(upstream.URL, nil, store)
+
+	for index, expectedStatus := range []int{http.StatusOK, http.StatusNotFound} {
+		req := httptest.NewRequest(http.MethodPut, mediaUploadProxyPath+uploadID, strings.NewReader("hello"))
+		req.Header.Set("Content-Type", "image/png")
+		resp := httptest.NewRecorder()
+		gw.handle(resp, req)
+		if resp.Code != expectedStatus {
+			t.Fatalf("attempt %d status = %d, body = %s", index+1, resp.Code, resp.Body.String())
+		}
+	}
+	if uploads != 1 {
+		t.Fatalf("uploads = %d", uploads)
+	}
+}
+
+func TestMediaUploadOversizeRestoresPlanForRetry(t *testing.T) {
+	t.Parallel()
+
+	var uploads int
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPut || r.URL.Path != "/upload" {
+			http.NotFound(w, r)
+			return
+		}
+		uploads++
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer upstream.Close()
+
+	store := newFakeUploadPlanStore()
+	uploadID := "abcdef1234"
+	store.Put(context.Background(), uploadID, uploadPlan{
+		ExpiresAt:     time.Now().Add(time.Hour),
+		ContentLength: 1,
+		Destinations: []uploadDestination{{
+			Name:      "central",
+			UploadURL: upstream.URL + "/upload",
+		}},
+	})
+	gw := newTestGatewayWithStore(upstream.URL, nil, store)
+
+	oversizeReq := httptest.NewRequest(
+		http.MethodPut,
+		mediaUploadProxyPath+uploadID,
+		strings.NewReader(strings.Repeat("x", int(maxUploadBytes(1))+1)),
+	)
+	oversizeReq.Header.Set("Content-Type", "image/png")
+	oversizeResp := httptest.NewRecorder()
+	gw.handle(oversizeResp, oversizeReq)
+	if oversizeResp.Code != http.StatusBadRequest {
+		t.Fatalf("oversize status = %d, body = %s", oversizeResp.Code, oversizeResp.Body.String())
+	}
+
+	retryReq := httptest.NewRequest(http.MethodPut, mediaUploadProxyPath+uploadID, strings.NewReader("ok"))
+	retryReq.Header.Set("Content-Type", "image/png")
+	retryResp := httptest.NewRecorder()
+	gw.handle(retryResp, retryReq)
+	if retryResp.Code != http.StatusOK {
+		t.Fatalf("retry status = %d, body = %s", retryResp.Code, retryResp.Body.String())
+	}
+	if uploads != 1 {
+		t.Fatalf("uploads = %d", uploads)
+	}
+}
+
+func TestMediaUploadUnsupportedContentTypeRestoresPlanForRetry(t *testing.T) {
+	t.Parallel()
+
+	var uploads int
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPut || r.URL.Path != "/upload" {
+			http.NotFound(w, r)
+			return
+		}
+		uploads++
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer upstream.Close()
+
+	store := newFakeUploadPlanStore()
+	uploadID := "abcdef1234"
+	store.Put(context.Background(), uploadID, uploadPlan{
+		ExpiresAt:     time.Now().Add(time.Hour),
+		ContentLength: 5,
+		Destinations: []uploadDestination{{
+			Name:      "central",
+			UploadURL: upstream.URL + "/upload",
+		}},
+	})
+	gw := newTestGatewayWithStore(upstream.URL, nil, store)
+
+	badReq := httptest.NewRequest(http.MethodPut, mediaUploadProxyPath+uploadID, strings.NewReader("hello"))
+	badReq.Header.Set("Content-Type", "text/html")
+	badResp := httptest.NewRecorder()
+	gw.handle(badResp, badReq)
+	if badResp.Code != http.StatusBadGateway {
+		t.Fatalf("bad content-type status = %d, body = %s", badResp.Code, badResp.Body.String())
+	}
+
+	retryReq := httptest.NewRequest(http.MethodPut, mediaUploadProxyPath+uploadID, strings.NewReader("hello"))
+	retryReq.Header.Set("Content-Type", "image/png")
+	retryResp := httptest.NewRecorder()
+	gw.handle(retryResp, retryReq)
+	if retryResp.Code != http.StatusOK {
+		t.Fatalf("retry status = %d, body = %s", retryResp.Code, retryResp.Body.String())
+	}
+	if uploads != 1 {
+		t.Fatalf("uploads = %d", uploads)
+	}
+}
+
+func TestMediaCreateUsesConfiguredPublicUploadURL(t *testing.T) {
+	t.Parallel()
+
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost || r.URL.Path != mediaPath {
+			http.NotFound(w, r)
+			return
+		}
+		uploadURL := "http://" + r.Host + "/upload"
+		writeJSON(w, http.StatusCreated, mediaUploadResponse{
+			MediaID:   "same-media-id",
+			UploadURL: &uploadURL,
+		})
+	}))
+	defer upstream.Close()
+
+	gw := newTestGateway(upstream.URL, nil)
+	gw.cfg.publicURL = "https://fanout.example.com/base"
+	req := httptest.NewRequest(http.MethodPost, mediaPath, strings.NewReader(`{"contentLength":5}`))
+	req.Host = "attacker.example.com"
+	req.Header.Set("X-Forwarded-Host", "attacker.example.com")
+	resp := httptest.NewRecorder()
+
+	gw.handle(resp, req)
+	if resp.Code != http.StatusCreated {
+		t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String())
+	}
+	var create mediaUploadResponse
+	if err := json.NewDecoder(resp.Body).Decode(&create); err != nil {
+		t.Fatal(err)
+	}
+	if create.UploadURL == nil || !strings.HasPrefix(*create.UploadURL, "https://fanout.example.com/base/") {
+		t.Fatalf("unexpected upload URL: %#v", create.UploadURL)
+	}
+	if strings.Contains(*create.UploadURL, "attacker.example.com") {
+		t.Fatalf("upload URL trusted request host: %s", *create.UploadURL)
+	}
+}
+
+func TestMediaGetUsesTenantDestinationForTenantRoute(t *testing.T) {
+	t.Parallel()
+
+	var centralGets int
+	central := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		centralGets++
+		writeJSON(w, http.StatusOK, map[string]string{"url": "central"})
+	}))
+	defer central.Close()
+	var tenantGets int
+	tenant := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		tenantGets++
+		if r.URL.Path != mediaPath+"/media-id" {
+			t.Fatalf("unexpected tenant path %s", r.URL.Path)
+		}
+		if got := r.Header.Get("Authorization"); got != "Basic tenant" {
+			t.Fatalf("tenant auth = %q", got)
+		}
+		writeJSON(w, http.StatusOK, map[string]string{"url": "tenant"})
+	}))
+	defer tenant.Close()
+
+	gw := newTestGateway(central.URL, map[string]string{"eu": tenant.URL})
+	req := httptest.NewRequest(http.MethodGet, tenantPrefix+"eu"+mediaPath+"/media-id", nil)
+	req.Header.Set("Authorization", "Basic tenant")
+	resp := httptest.NewRecorder()
+
+	gw.handle(resp, req)
+	if resp.Code != http.StatusOK {
+		t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String())
+	}
+	if centralGets != 0 || tenantGets != 1 {
+		t.Fatalf("centralGets=%d tenantGets=%d", centralGets, tenantGets)
+	}
+	if !strings.Contains(resp.Body.String(), "tenant") {
+		t.Fatalf("unexpected body: %s", resp.Body.String())
+	}
+}
+
+func TestMediaGetUsesCentralDestinationForCentralRoute(t *testing.T) {
+	t.Parallel()
+
+	var centralGets int
+	central := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		centralGets++
+		if got := r.Header.Get("Authorization"); got != "Basic central" {
+			t.Fatalf("central auth = %q", got)
+		}
+		writeJSON(w, http.StatusOK, map[string]string{"url": "central"})
+	}))
+	defer central.Close()
+	var tenantGets int
+	tenant := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		tenantGets++
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer tenant.Close()
+
+	gw := newTestGateway(central.URL, map[string]string{"eu": tenant.URL})
+	req := httptest.NewRequest(http.MethodGet, mediaPath+"/media-id", nil)
+	resp := httptest.NewRecorder()
+
+	gw.handle(resp, req)
+	if resp.Code != http.StatusOK {
+		t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String())
+	}
+	if centralGets != 1 || tenantGets != 0 {
+		t.Fatalf("centralGets=%d tenantGets=%d", centralGets, tenantGets)
+	}
+	if !strings.Contains(resp.Body.String(), "central") {
+		t.Fatalf("unexpected body: %s", resp.Body.String())
+	}
+}
+
+func TestMediaCreateRejectsDifferentMediaIDs(t *testing.T) {
+	t.Parallel()
+
+	upstream := func(mediaID string) *httptest.Server {
+		return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if r.Method != http.MethodPost || r.URL.Path != mediaPath {
+				http.NotFound(w, r)
+				return
+			}
+			writeJSON(w, http.StatusCreated, mediaUploadResponse{MediaID: mediaID})
+		}))
+	}
+	central := upstream("central-id")
+	defer central.Close()
+	tenant := upstream("tenant-id")
+	defer tenant.Close()
+
+	gw := newTestGateway(central.URL, map[string]string{"eu": tenant.URL})
+	req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu"+mediaPath, strings.NewReader(`{"contentLength":0}`))
+	req.Header.Set("Authorization", "Basic tenant")
+	resp := httptest.NewRecorder()
+
+	gw.handle(resp, req)
+	if resp.Code != http.StatusBadGateway {
+		t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String())
+	}
+	if !strings.Contains(resp.Body.String(), "media IDs differ") {
+		t.Fatalf("unexpected body: %s", resp.Body.String())
+	}
+
+	metrics := scrapeMetrics(t, gw)
+	if !strings.Contains(metrics, `langfuse_fanout_media_divergence_total{destination="tenant_eu",kind="media_id"} 1`) {
+		t.Fatalf("missing media_id divergence metric:\n%s", metrics)
+	}
+}
+
+func TestMetricsEndpointRequiresBearerToken(t *testing.T) {
+	t.Parallel()
+
+	gw := newTestGateway("http://central.invalid", nil)
+
+	unauthorized := httptest.NewRecorder()
+	gw.handle(unauthorized, httptest.NewRequest(http.MethodGet, metricsPath, nil))
+	if unauthorized.Code != http.StatusUnauthorized {
+		t.Fatalf("unauthorized status = %d", unauthorized.Code)
+	}
+
+	wrong := httptest.NewRecorder()
+	wrongReq := httptest.NewRequest(http.MethodGet, metricsPath, nil)
+	wrongReq.Header.Set("Authorization", "Bearer wrong")
+	gw.handle(wrong, wrongReq)
+	if wrong.Code != http.StatusUnauthorized {
+		t.Fatalf("wrong token status = %d", wrong.Code)
+	}
+
+	authorized := httptest.NewRecorder()
+	authorizedReq := httptest.NewRequest(http.MethodGet, metricsPath, nil)
+	authorizedReq.Header.Set("Authorization", "Bearer test-secret")
+	gw.handle(authorized, authorizedReq)
+	if authorized.Code != http.StatusOK {
+		t.Fatalf("authorized status = %d, body = %s", authorized.Code, authorized.Body.String())
+	}
+	if !strings.Contains(authorized.Body.String(), "go_goroutines") {
+		t.Fatalf("missing gateway metrics:\n%s", authorized.Body.String())
+	}
+}
+
+func TestTraceProxyRecordsPrometheusMetrics(t *testing.T) {
+	t.Parallel()
+
+	collector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = w.Write([]byte("{}"))
+	}))
+	defer collector.Close()
+
+	gw := newTestGatewayWithCollector(collector.URL)
+	body := buildTraceRequest(t, nil)
+	req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu"+otelTracePath, bytes.NewReader(body))
+	req.Header.Set("Content-Type", "application/x-protobuf")
+	req.Header.Set("Authorization", "Basic tenant")
+	resp := httptest.NewRecorder()
+
+	gw.handle(resp, req)
+	if resp.Code != http.StatusOK {
+		t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String())
+	}
+
+	metrics := scrapeMetrics(t, gw)
+	if !strings.Contains(metrics, `langfuse_fanout_trace_exports_total{destination="tenant_eu",result="success"} 1`) {
+		t.Fatalf("missing trace export metric:\n%s", metrics)
+	}
+	if !strings.Contains(metrics, `langfuse_fanout_upstream_requests_total{destination="collector",operation="trace_collector",status_class="2xx"} 1`) {
+		t.Fatalf("missing upstream collector metric:\n%s", metrics)
+	}
+}
+
+func newTestGateway(centralURL string, tenants map[string]string) *gateway {
+	return newTestGatewayWithStore(centralURL, tenants, newFakeUploadPlanStore())
+}
+
+func newTestGatewayWithStore(centralURL string, tenants map[string]string, store uploadPlanStore) *gateway {
+	return newGateway(config{
+		traceCollectorURL: "http://collector.invalid",
+		publicURL:         "http://fanout.local:4318",
+		metricsSecret:     "test-secret",
+		central: destination{
+			name:          centralName,
+			baseURL:       centralURL,
+			authorization: "Basic central",
+		},
+		tenants:     tenants,
+		uploadStore: store,
+		client:      &http.Client{Timeout: 5 * time.Second},
+	})
+}
+
+func newTestGatewayWithCollector(collectorURL string) *gateway {
+	gateway := newTestGateway("http://central.invalid", map[string]string{"eu": "http://tenant.invalid"})
+	gateway.cfg.traceCollectorURL = collectorURL
+	return gateway
+}
+
+func buildTraceRequest(t *testing.T, attrs map[string]string) []byte {
+	t.Helper()
+	spanAttrs := make([]*commonv1.KeyValue, 0, len(attrs))
+	for key, value := range attrs {
+		spanAttrs = append(spanAttrs, &commonv1.KeyValue{
+			Key: key,
+			Value: &commonv1.AnyValue{
+				Value: &commonv1.AnyValue_StringValue{StringValue: value},
+			},
+		})
+	}
+	request := &tracepb.ExportTraceServiceRequest{
+		ResourceSpans: []*tracev1.ResourceSpans{{
+			Resource: &resourcev1.Resource{},
+			ScopeSpans: []*tracev1.ScopeSpans{{
+				Spans: []*tracev1.Span{{
+					TraceId:    []byte("1234567890123456"),
+					SpanId:     []byte("12345678"),
+					Name:       "test-span",
+					Attributes: spanAttrs,
+				}},
+			}},
+		}},
+	}
+	body, err := proto.Marshal(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return body
+}
+
+func buildJSONTraceRequest(t *testing.T, attrs map[string]any) []byte {
+	t.Helper()
+	spanAttrs := make([]map[string]any, 0, len(attrs))
+	for key, value := range attrs {
+		anyValue := map[string]any{}
+		switch typed := value.(type) {
+		case bool:
+			anyValue["boolValue"] = typed
+		default:
+			anyValue["stringValue"] = typed
+		}
+		spanAttrs = append(spanAttrs, map[string]any{
+			"key":   key,
+			"value": anyValue,
+		})
+	}
+	request := map[string]any{
+		"resourceSpans": []any{
+			map[string]any{
+				"resource": map[string]any{},
+				"scopeSpans": []any{
+					map[string]any{
+						"spans": []any{
+							map[string]any{
+								"traceId":    "31323334353637383930313233343536",
+								"spanId":     "3132333435363738",
+								"name":       "test-span",
+								"attributes": spanAttrs,
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	body, err := json.Marshal(request)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return body
+}
+
+func parseTraceAttributes(t *testing.T, body []byte) map[string]string {
+	t.Helper()
+	var request tracepb.ExportTraceServiceRequest
+	if err := proto.Unmarshal(body, &request); err != nil {
+		t.Fatal(err)
+	}
+	result := map[string]string{}
+	for _, resourceSpan := range request.ResourceSpans {
+		for _, scopeSpan := range resourceSpan.ScopeSpans {
+			for _, span := range scopeSpan.Spans {
+				for _, attr := range span.Attributes {
+					result[attr.Key] = stringValue(attr.Value)
+				}
+			}
+		}
+	}
+	return result
+}
+
+func parseJSONTraceAttributes(t *testing.T, body []byte) map[string]string {
+	t.Helper()
+	var request map[string]any
+	if err := json.Unmarshal(body, &request); err != nil {
+		t.Fatal(err)
+	}
+	result := map[string]string{}
+	resourceSpans, _ := request["resourceSpans"].([]any)
+	for _, resourceSpan := range resourceSpans {
+		resourceSpanMap, _ := resourceSpan.(map[string]any)
+		scopeSpans, _ := resourceSpanMap["scopeSpans"].([]any)
+		for _, scopeSpan := range scopeSpans {
+			scopeSpanMap, _ := scopeSpan.(map[string]any)
+			spans, _ := scopeSpanMap["spans"].([]any)
+			for _, span := range spans {
+				spanMap, _ := span.(map[string]any)
+				attrs, _ := spanMap["attributes"].([]any)
+				for _, attr := range attrs {
+					attrMap, _ := attr.(map[string]any)
+					valueMap, _ := attrMap["value"].(map[string]any)
+					if key, ok := attrMap["key"].(string); ok {
+						if value, ok := valueMap["stringValue"].(string); ok {
+							result[key] = value
+						}
+					}
+				}
+			}
+		}
+	}
+	return result
+}
+
+func scrapeMetrics(t *testing.T, gw *gateway) string {
+	t.Helper()
+	req := httptest.NewRequest(http.MethodGet, metricsPath, nil)
+	req.Header.Set("Authorization", "Bearer test-secret")
+	resp := httptest.NewRecorder()
+	gw.handle(resp, req)
+	if resp.Code != http.StatusOK {
+		t.Fatalf("metrics status = %d, body = %s", resp.Code, resp.Body.String())
+	}
+	return resp.Body.String()
+}
+
+func newUploadURLPath(t *testing.T, value string) string {
+	t.Helper()
+	parsed, err := url.Parse(value)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return parsed.Path
+}
+
+type fakeUploadPlanStore struct {
+	mu    sync.Mutex
+	plans map[string]uploadPlan
+}
+
+func newFakeUploadPlanStore() *fakeUploadPlanStore {
+	return &fakeUploadPlanStore{plans: map[string]uploadPlan{}}
+}
+
+func (s *fakeUploadPlanStore) Put(_ context.Context, uploadID string, plan uploadPlan) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.plans[uploadID] = plan
+	return nil
+}
+
+func (s *fakeUploadPlanStore) Take(_ context.Context, uploadID string) (uploadPlan, bool, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	plan, ok := s.plans[uploadID]
+	delete(s.plans, uploadID)
+	return plan, ok, nil
+}
+
+func (s *fakeUploadPlanStore) Ping(_ context.Context) error {
+	return nil
+}
+
+func (s *fakeUploadPlanStore) Close() error {
+	return nil
+}
+
+type failingReader struct {
+	read bool
+}
+
+func (r *failingReader) Read(_ []byte) (int, error) {
+	r.read = true
+	return 0, errors.New("read should not be called")
+}
--- a/otel/langfuse-fanout/cmd/langfuse-fanout/metrics.go
+++ b/otel/langfuse-fanout/cmd/langfuse-fanout/metrics.go
@ -0,0 +1,242 @@
+package main
+
+import (
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/collectors"
+)
+
+type gatewayMetrics struct {
+	registry              *prometheus.Registry
+	httpRequests          *prometheus.CounterVec
+	httpDuration          *prometheus.HistogramVec
+	upstreamRequests      *prometheus.CounterVec
+	upstreamDuration      *prometheus.HistogramVec
+	traceExports          *prometheus.CounterVec
+	mediaDivergence       *prometheus.CounterVec
+	uploadPlansCreated    prometheus.Counter
+	uploadPlansCompleted  prometheus.Counter
+	uploadPlanMisses      prometheus.Counter
+	uploadPlanStoreErrors *prometheus.CounterVec
+	uploadBytes           prometheus.Histogram
+}
+
+func newGatewayMetrics() *gatewayMetrics {
+	registry := prometheus.NewRegistry()
+	registry.MustRegister(
+		collectors.NewGoCollector(),
+		collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}),
+	)
+
+	metrics := &gatewayMetrics{
+		registry: registry,
+		httpRequests: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "langfuse_fanout_http_requests_total",
+			Help: "Total HTTP requests handled by the Langfuse fanout gateway.",
+		}, []string{"method", "path", "status"}),
+		httpDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "langfuse_fanout_http_request_duration_seconds",
+			Help:    "HTTP request duration for the Langfuse fanout gateway.",
+			Buckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30},
+		}, []string{"method", "path", "status"}),
+		upstreamRequests: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "langfuse_fanout_upstream_requests_total",
+			Help: "Total upstream requests made by the Langfuse fanout gateway.",
+		}, []string{"operation", "destination", "status_class"}),
+		upstreamDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "langfuse_fanout_upstream_request_duration_seconds",
+			Help:    "Upstream request duration for Langfuse and collector calls.",
+			Buckets: []float64{0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30},
+		}, []string{"operation", "destination", "status_class"}),
+		traceExports: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "langfuse_fanout_trace_exports_total",
+			Help: "Total trace export attempts through the Langfuse fanout gateway.",
+		}, []string{"destination", "result"}),
+		mediaDivergence: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "langfuse_fanout_media_divergence_total",
+			Help: "Media fanout upstream response divergence by kind. Values are counts only; no media IDs or URLs are exposed.",
+		}, []string{"kind", "destination"}),
+		uploadPlansCreated: prometheus.NewCounter(prometheus.CounterOpts{
+			Name: "langfuse_fanout_media_upload_plans_created_total",
+			Help: "Total media upload fanout plans stored for SDK byte upload.",
+		}),
+		uploadPlansCompleted: prometheus.NewCounter(prometheus.CounterOpts{
+			Name: "langfuse_fanout_media_upload_plans_completed_total",
+			Help: "Total media upload fanout plans consumed by SDK byte upload.",
+		}),
+		uploadPlanMisses: prometheus.NewCounter(prometheus.CounterOpts{
+			Name: "langfuse_fanout_media_upload_plan_misses_total",
+			Help: "Total media upload attempts that did not find a stored fanout plan.",
+		}),
+		uploadPlanStoreErrors: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "langfuse_fanout_media_upload_plan_store_errors_total",
+			Help: "Total Redis upload plan store errors by operation.",
+		}, []string{"operation"}),
+		uploadBytes: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Name:    "langfuse_fanout_media_upload_bytes",
+			Help:    "Configured media upload content length for fanout upload plans.",
+			Buckets: []float64{1_000, 10_000, 100_000, 1_000_000, 5_000_000, 10_000_000, 25_000_000, 50_000_000, 100_000_000, 250_000_000},
+		}),
+	}
+
+	registry.MustRegister(
+		metrics.httpRequests,
+		metrics.httpDuration,
+		metrics.upstreamRequests,
+		metrics.upstreamDuration,
+		metrics.traceExports,
+		metrics.mediaDivergence,
+		metrics.uploadPlansCreated,
+		metrics.uploadPlansCompleted,
+		metrics.uploadPlanMisses,
+		metrics.uploadPlanStoreErrors,
+		metrics.uploadBytes,
+	)
+	return metrics
+}
+
+func (m *gatewayMetrics) recordHTTP(method string, path string, status int, duration time.Duration) {
+	if m == nil {
+		return
+	}
+	labels := prometheus.Labels{
+		"method": method,
+		"path":   path,
+		"status": fmt.Sprintf("%d", status),
+	}
+	m.httpRequests.With(labels).Inc()
+	m.httpDuration.With(labels).Observe(duration.Seconds())
+}
+
+func (m *gatewayMetrics) recordUpstream(operation string, destination string, statusClass string, duration time.Duration) {
+	if m == nil {
+		return
+	}
+	labels := prometheus.Labels{
+		"operation":    operation,
+		"destination":  normalizeMetricLabel(destination),
+		"status_class": statusClass,
+	}
+	m.upstreamRequests.With(labels).Inc()
+	m.upstreamDuration.With(labels).Observe(duration.Seconds())
+}
+
+func (m *gatewayMetrics) recordTraceExport(destination string, result string) {
+	if m == nil {
+		return
+	}
+	m.traceExports.WithLabelValues(normalizeMetricLabel(destination), result).Inc()
+}
+
+func (m *gatewayMetrics) recordMediaDivergence(kind string, destination string) {
+	if m == nil {
+		return
+	}
+	m.mediaDivergence.WithLabelValues(normalizeMetricLabel(kind), normalizeMetricLabel(destination)).Inc()
+}
+
+func (m *gatewayMetrics) recordUploadPlanCreated(contentLength int64) {
+	if m == nil {
+		return
+	}
+	m.uploadPlansCreated.Inc()
+	if contentLength > 0 {
+		m.uploadBytes.Observe(float64(contentLength))
+	}
+}
+
+func (m *gatewayMetrics) recordUploadPlanCompleted() {
+	if m == nil {
+		return
+	}
+	m.uploadPlansCompleted.Inc()
+}
+
+func (m *gatewayMetrics) recordUploadPlanMiss() {
+	if m == nil {
+		return
+	}
+	m.uploadPlanMisses.Inc()
+}
+
+func (m *gatewayMetrics) recordUploadPlanStoreError(operation string) {
+	if m == nil {
+		return
+	}
+	m.uploadPlanStoreErrors.WithLabelValues(normalizeMetricLabel(operation)).Inc()
+}
+
+type statusRecorder struct {
+	http.ResponseWriter
+	status int
+}
+
+func (w *statusRecorder) WriteHeader(status int) {
+	w.status = status
+	w.ResponseWriter.WriteHeader(status)
+}
+
+func normalizeMetricPath(path string) string {
+	route := parseRoute(path)
+	switch {
+	case route.path == otelTracePath:
+		if route.destination == "" {
+			return otelTracePath
+		}
+		return tenantPrefix + "#destination" + otelTracePath
+	case route.path == mediaPath:
+		if route.destination == "" {
+			return mediaPath
+		}
+		return tenantPrefix + "#destination" + mediaPath
+	case strings.HasPrefix(route.path, mediaPath+"/"):
+		if route.destination == "" {
+			return mediaPath + "/#mediaId"
+		}
+		return tenantPrefix + "#destination" + mediaPath + "/#mediaId"
+	case strings.HasPrefix(path, mediaUploadProxyPath):
+		return mediaUploadProxyPath + "#uploadId"
+	case path == "/healthz":
+		return "/healthz"
+	default:
+		return "/#path"
+	}
+}
+
+func normalizeMetricLabel(value string) string {
+	value = strings.TrimSpace(strings.ToLower(value))
+	if value == "" {
+		return "unknown"
+	}
+	var builder strings.Builder
+	for _, r := range value {
+		switch {
+		case r >= 'a' && r <= 'z':
+			builder.WriteRune(r)
+		case r >= '0' && r <= '9':
+			builder.WriteRune(r)
+		case r == '_' || r == '-' || r == ':':
+			builder.WriteRune(r)
+		default:
+			builder.WriteRune('_')
+		}
+		if builder.Len() >= 80 {
+			break
+		}
+	}
+	if builder.Len() == 0 {
+		return "unknown"
+	}
+	return builder.String()
+}
+
+func statusClass(status int) string {
+	if status <= 0 {
+		return "error"
+	}
+	return fmt.Sprintf("%dxx", status/100)
+}
--- a/otel/langfuse-fanout/go.mod
+++ b/otel/langfuse-fanout/go.mod
@ -0,0 +1,29 @@
+module github.com/danny-avila/librechat/otel/langfuse-fanout
+
+go 1.23.0
+
+require (
+	github.com/prometheus/client_golang v1.23.2
+	github.com/redis/go-redis/v9 v9.17.2
+	go.opentelemetry.io/proto/otlp v1.9.0
+	google.golang.org/protobuf v1.36.10
+)
+
+require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect
+	github.com/kr/text v0.2.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/prometheus/client_model v0.6.2 // indirect
+	github.com/prometheus/common v0.66.1 // indirect
+	github.com/prometheus/procfs v0.16.1 // indirect
+	go.yaml.in/yaml/v2 v2.4.2 // indirect
+	golang.org/x/net v0.43.0 // indirect
+	golang.org/x/sys v0.35.0 // indirect
+	golang.org/x/text v0.28.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5 // indirect
+	google.golang.org/grpc v1.75.1 // indirect
+)
--- a/otel/langfuse-fanout/go.sum
+++ b/otel/langfuse-fanout/go.sum
@ -0,0 +1,90 @@
+github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
+github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
+github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
+github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
+github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 h1:8Tjv8EJ+pM1xP8mK6egEbD1OgnVTyacbefKhmbLhIhU=
+github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2/go.mod h1:pkJQ2tZHJ0aFOVEEot6oZmaVEZcRme73eIFmhiVuRWs=
+github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
+github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
+github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o=
+github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg=
+github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
+github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
+github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs=
+github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA=
+github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg=
+github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
+github.com/redis/go-redis/v9 v9.17.2 h1:P2EGsA4qVIM3Pp+aPocCJ7DguDHhqrXNhVcEp4ViluI=
+github.com/redis/go-redis/v9 v9.17.2/go.mod h1:u410H11HMLoB+TP67dz8rL9s6QW2j76l0//kSOd3370=
+github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
+github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
+go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
+go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ=
+go.opentelemetry.io/otel v1.37.0/go.mod h1:ehE/umFRLnuLa/vSccNq9oS1ErUlkkK71gMcN34UG8I=
+go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/WgbsdpcPoZE=
+go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E=
+go.opentelemetry.io/otel/sdk v1.37.0 h1:ItB0QUqnjesGRvNcmAcU0LyvkVyGJ2xftD29bWdDvKI=
+go.opentelemetry.io/otel/sdk v1.37.0/go.mod h1:VredYzxUvuo2q3WRcDnKDjbdvmO0sCzOvVAiY+yUkAg=
+go.opentelemetry.io/otel/sdk/metric v1.37.0 h1:90lI228XrB9jCMuSdA0673aubgRobVZFhbjxHHspCPc=
+go.opentelemetry.io/otel/sdk/metric v1.37.0/go.mod h1:cNen4ZWfiD37l5NhS+Keb5RXVWZWpRE+9WyVCpbo5ps=
+go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4=
+go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0=
+go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A=
+go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
+go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
+golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE=
+golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg=
+golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI=
+golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng=
+golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU=
+gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
+gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
+google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5 h1:BIRfGDEjiHRrk0QKZe3Xv2ieMhtgRGeLcZQ0mIVn4EY=
+google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5/go.mod h1:j3QtIyytwqGr1JUDtYXwtMXWPKsEa5LtzIFN1Wn5WvE=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5 h1:eaY8u2EuxbRv7c3NiGK0/NedzVsCcV6hDuU5qPX5EGE=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5/go.mod h1:M4/wBTSeyLxupu3W3tJtOgB14jILAS/XWPSSa3TAlJc=
+google.golang.org/grpc v1.75.1 h1:/ODCNEuf9VghjgO3rqLcfg8fiOP0nSluljWFlDxELLI=
+google.golang.org/grpc v1.75.1/go.mod h1:JtPAzKiq4v1xcAB2hydNlWI2RnF85XXcV0mhKXr2ecQ=
+google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
+google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
--- a/otel/langfuse-fanout/otelcol.yaml
+++ b/otel/langfuse-fanout/otelcol.yaml
@ -9,7 +9,7 @@ receivers:
  otlp:
    protocols:
      http:
-        endpoint: 0.0.0.0:4318
+        endpoint: ${env:LANGFUSE_FANOUT_OTEL_RECEIVER_ENDPOINT}
        include_metadata: true
        traces_url_path: /api/public/otel/v1/traces

@ -20,12 +20,12 @@ connectors:
      - context: span
        condition: attributes["librechat.langfuse.destination"] == "eu"
        pipelines: [traces/tenant_eu]
-      - context: span
-        condition: attributes["librechat.langfuse.destination"] == "us"
-        pipelines: [traces/tenant_us]
      - context: span
        condition: attributes["librechat.langfuse.destination"] == "jp"
        pipelines: [traces/tenant_jp]
+      - context: span
+        condition: attributes["librechat.langfuse.destination"] == "us"
+        pipelines: [traces/tenant_us]

 processors:
  memory_limiter:
@ -51,16 +51,17 @@ processors:
    send_batch_size: ${env:LANGFUSE_FANOUT_BATCH_SEND_SIZE}
    metadata_keys: [authorization]
    metadata_cardinality_limit: ${env:LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT}
-  batch/by_auth_us:
-    timeout: ${env:LANGFUSE_FANOUT_BATCH_TIMEOUT}
-    send_batch_size: ${env:LANGFUSE_FANOUT_BATCH_SEND_SIZE}
-    metadata_keys: [authorization]
-    metadata_cardinality_limit: ${env:LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT}
  batch/by_auth_jp:
    timeout: ${env:LANGFUSE_FANOUT_BATCH_TIMEOUT}
    send_batch_size: ${env:LANGFUSE_FANOUT_BATCH_SEND_SIZE}
    metadata_keys: [authorization]
    metadata_cardinality_limit: ${env:LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT}
+  batch/by_auth_us:
+    timeout: ${env:LANGFUSE_FANOUT_BATCH_TIMEOUT}
+    send_batch_size: ${env:LANGFUSE_FANOUT_BATCH_SEND_SIZE}
+    metadata_keys: [authorization]
+    metadata_cardinality_limit: ${env:LANGFUSE_FANOUT_METADATA_CARDINALITY_LIMIT}
+
 exporters:
  otlphttp/central:
    # Langfuse Cloud base URL options: https://cloud.langfuse.com (EU),
@ -75,18 +76,19 @@ exporters:
      authenticator: headers_setter/tenant_passthrough
    headers:
      x-langfuse-ingestion-version: '4'
-  otlphttp/tenant_us:
-    endpoint: '${env:LANGFUSE_FANOUT_TENANT_US_BASE_URL}/api/public/otel'
-    auth:
-      authenticator: headers_setter/tenant_passthrough
-    headers:
-      x-langfuse-ingestion-version: '4'
  otlphttp/tenant_jp:
    endpoint: '${env:LANGFUSE_FANOUT_TENANT_JP_BASE_URL}/api/public/otel'
    auth:
      authenticator: headers_setter/tenant_passthrough
    headers:
      x-langfuse-ingestion-version: '4'
+  otlphttp/tenant_us:
+    endpoint: '${env:LANGFUSE_FANOUT_TENANT_US_BASE_URL}/api/public/otel'
+    auth:
+      authenticator: headers_setter/tenant_passthrough
+    headers:
+      x-langfuse-ingestion-version: '4'
+
 service:
  extensions: [headers_setter/tenant_passthrough]
  pipelines:
@ -102,11 +104,11 @@ service:
      receivers: [routing/langfuse_tenant_destination]
      processors: [attributes/drop_librechat_routing, batch/by_auth_eu]
      exporters: [otlphttp/tenant_eu]
-    traces/tenant_us:
-      receivers: [routing/langfuse_tenant_destination]
-      processors: [attributes/drop_librechat_routing, batch/by_auth_us]
-      exporters: [otlphttp/tenant_us]
    traces/tenant_jp:
      receivers: [routing/langfuse_tenant_destination]
      processors: [attributes/drop_librechat_routing, batch/by_auth_jp]
      exporters: [otlphttp/tenant_jp]
+    traces/tenant_us:
+      receivers: [routing/langfuse_tenant_destination]
+      processors: [attributes/drop_librechat_routing, batch/by_auth_us]
+      exporters: [otlphttp/tenant_us]