Mask exported third party spans

docs/API_REFERENCE.md

@@ -58,6 +58,7 @@ Block receives a `Langfuse::Config` object with these properties:
 | `release`                      | String  | No       | `nil` (or `ENV["LANGFUSE_RELEASE"]` / common CI commit SHA env) | Default release identifier         |
 | `should_export_span`           | `#call` | No       | `nil`                          | Span export filter callback        |
 | `mask`                         | `#call` | No       | `nil`                          | Mask callable for input/output/metadata (receives `data:` keyword) |
+| `mask_otel_spans`              | `#call` | No       | `nil`                          | Export-stage span masking hook (receives `spans:` keyword; see [CONFIGURATION.md](CONFIGURATION.md#mask_otel_spans)) |
 
 **Example:**
 

docs/CONFIGURATION.md

@@ -393,6 +393,32 @@ config.mask = lambda { |data:|
 
 See [TRACING.md](TRACING.md#masking) for usage patterns and behavior details.
 
+#### `mask_otel_spans`
+
+- **Type:** `#call` (Proc, Lambda, or any object responding to `call`) or `nil`
+- **Default:** `nil` (export-stage masking disabled)
+- **Description:** Export-stage masking hook applied to every span batch exported to Langfuse, including spans created by third-party instrumentations. Receives a `spans:` keyword argument — a frozen Hash mapping stable span identifiers (`"<hex trace_id>:<hex span_id>"`) to immutable `Langfuse::OtelSpanSnapshot` values (trace/span IDs, name, instrumentation scope name, frozen span attributes, frozen resource attributes). Return `nil` to export the batch unchanged, or a sparse Hash of patches keyed by the same identifiers. Each patch is a Hash with optional `:delete` (Array of attribute keys) and `:set` (Hash of replacement attributes); deletes run before sets, so a replacement wins when a key appears in both.
+
+```ruby
+config.mask_otel_spans = lambda { |spans:|
+  spans.filter_map { |id, snapshot|
+    next unless snapshot.attributes.key?("gen_ai.prompt")
+
+    [id, { set: { "gen_ai.prompt" => "[REDACTED]" } }]
+  }.to_h
+}
+```
+
+Fail-closed behavior:
+
+- A hook exception or an invalid top-level result drops the whole Langfuse export batch.
+- A malformed per-span patch drops that span from the Langfuse export.
+- An invalid replacement attribute value omits that attribute rather than exporting its original value.
+
+Only the copy exported to Langfuse is transformed. Any other OpenTelemetry exporter receives the original, unmasked spans — Langfuse masking does not protect other telemetry backends. The hook is synchronous and runs on the batch export thread; it must not rely on request context, the current span, async work, or network calls.
+
+See [TRACING.md](TRACING.md#masking) for how the two masking boundaries relate.
+
 ## Tracing Behavior and OpenTelemetry Ownership
 
 There are three states worth documenting.
@@ -623,6 +649,7 @@ Validation rules:
 - `prompt_cache_observer` must respond to `#call` (if set)
 - `should_export_span` must respond to `#call` (if set)
 - `mask` must respond to `#call` (if set)
+- `mask_otel_spans` must respond to `#call` (if set)
 
 ## Accessing Current Configuration
 

docs/TRACING.md

@@ -316,3 +316,32 @@ end
 ```
 
 Masking applies to observation `input`, `output`, and `metadata`. The full configuration contract is in [CONFIGURATION.md](CONFIGURATION.md#mask).
+
+### Export-stage masking with `mask_otel_spans`
+
+`mask` runs while the SDK creates Langfuse-owned attributes; it never sees the raw
+attributes of third-party spans (e.g. `gen_ai.*` attributes set by an OpenAI or
+LangChain instrumentation). To transform those, configure the export-stage hook:
+
+```ruby
+Langfuse.configure do |config|
+  config.mask_otel_spans = lambda { |spans:|
+    spans.filter_map { |id, snapshot|
+      next unless snapshot.attributes.key?("gen_ai.prompt")
+
+      [id, { delete: ["gen_ai.completion"], set: { "gen_ai.prompt" => "[REDACTED]" } }]
+    }.to_h
+  }
+end
+```
+
+The hook runs after `should_export_span` selects spans and just before the batch
+is handed to the Langfuse OTLP exporter, so it sees exactly the spans this client
+will export — Langfuse-owned and third-party alike. Returning `nil` exports the
+batch unchanged. Errors fail closed: the batch (or the offending span) is dropped
+rather than exported unmasked.
+
+Only the Langfuse export copy is transformed. If you also send telemetry to
+another OpenTelemetry backend, that backend receives the original spans and needs
+its own masking. The two hooks are independent; applications may configure either
+or both. The full contract is in [CONFIGURATION.md](CONFIGURATION.md#mask_otel_spans).

lib/langfuse.rb

@@ -52,6 +52,7 @@ class UnauthorizedError < ApiError; end
 require_relative "langfuse/sampling"
 require_relative "langfuse/otel_setup"
 require_relative "langfuse/masking"
+require_relative "langfuse/masking_exporter"
 require_relative "langfuse/otel_attributes"
 require_relative "langfuse/propagation"
 require_relative "langfuse/span_processor"

lib/langfuse/config.rb

@@ -86,8 +86,26 @@ class Config
 
     # @return [#call, nil] Mask callable applied to input, output, and metadata before serialization.
     #   Receives `data:` keyword argument. nil disables masking.
+    #   This is a creation-time hook for Langfuse-owned attributes; it never sees
+    #   raw third-party span attributes. See {#mask_otel_spans} for those.
     attr_accessor :mask
 
+    # @return [#call, nil] Export-stage masking hook for spans exported to Langfuse.
+    #   Receives a `spans:` keyword argument — a frozen Hash mapping stable span
+    #   identifiers (`"<hex trace_id>:<hex span_id>"`) to {OtelSpanSnapshot}
+    #   values covering every span in one export batch, including spans created
+    #   by third-party instrumentations. Returns nil to export the batch
+    #   unchanged, or a sparse Hash of patches keyed by the same identifiers.
+    #   Each patch is a Hash with optional +:delete+ (Array of attribute keys)
+    #   and +:set+ (Hash of replacement attributes); deletes run before sets.
+    #   Only the copy exported to Langfuse is transformed — any other
+    #   OpenTelemetry exporter receives the original, unmasked spans, so
+    #   Langfuse masking does not protect other telemetry backends.
+    #   The hook is synchronous and must not rely on request context, the
+    #   current span, async work, or network calls. Exceptions and invalid
+    #   results fail closed by dropping the Langfuse export batch.
+    attr_accessor :mask_otel_spans
+
     # @return [String] Default Langfuse API base URL
     DEFAULT_BASE_URL = "https://cloud.langfuse.com"
 
@@ -193,10 +211,11 @@ def validate!
       validate_swr_config!
 
       validate_cache_backend!
-      validate_prompt_cache_observer!
       validate_sample_rate!
-      validate_should_export_span!
-      validate_mask!
+      validate_callable!(prompt_cache_observer, "prompt_cache_observer")
+      validate_callable!(should_export_span, "should_export_span")
+      validate_callable!(mask, "mask")
+      validate_callable!(mask_otel_spans, "mask_otel_spans")
     end
     # rubocop:enable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
 
@@ -242,6 +261,7 @@ def initialize_tracing_defaults
       self.sample_rate = env_value("LANGFUSE_SAMPLE_RATE") || DEFAULT_SAMPLE_RATE
       @should_export_span = nil
       @mask = nil
+      @mask_otel_spans = nil
     end
 
     def validate_cache_backend!
@@ -252,10 +272,10 @@ def validate_cache_backend!
             "cache_backend must be one of #{valid_backends.inspect}, got #{cache_backend.inspect}"
     end
 
-    def validate_prompt_cache_observer!
-      return if prompt_cache_observer.nil? || prompt_cache_observer.respond_to?(:call)
+    def validate_callable!(value, name)
+      return if value.nil? || value.respond_to?(:call)
 
-      raise ConfigurationError, "prompt_cache_observer must respond to #call"
+      raise ConfigurationError, "#{name} must respond to #call"
     end
 
     def validate_swr_config!
@@ -296,18 +316,6 @@ def validate_sample_rate!
       raise ConfigurationError, "sample_rate must be between 0.0 and 1.0"
     end
 
-    def validate_mask!
-      return if mask.nil? || mask.respond_to?(:call)
-
-      raise ConfigurationError, "mask must respond to #call"
-    end
-
-    def validate_should_export_span!
-      return if should_export_span.nil? || should_export_span.respond_to?(:call)
-
-      raise ConfigurationError, "should_export_span must respond to #call"
-    end
-
     def detect_release_from_ci_env
       COMMON_RELEASE_ENV_KEYS.each do |key|
         value = env_value(key)

lib/langfuse/masking_exporter.rb

@@ -0,0 +1,217 @@
+# frozen_string_literal: true
+
+require "opentelemetry/sdk"
+
+module Langfuse
+  # Immutable snapshot of one span in a Langfuse export batch.
+  #
+  # Passed to the {Config#mask_otel_spans} hook as the values of the +spans:+
+  # mapping. Exposes only the fields needed to make a masking decision; the
+  # mutable OpenTelemetry SDK span objects are never handed to the hook.
+  #
+  # @!attribute [r] trace_id
+  #   @return [String] 32-hex-character lowercase trace ID
+  # @!attribute [r] span_id
+  #   @return [String] 16-hex-character lowercase span ID
+  # @!attribute [r] name
+  #   @return [String] span name
+  # @!attribute [r] scope_name
+  #   @return [String, nil] instrumentation scope name (e.g. "ruby-openai")
+  # @!attribute [r] attributes
+  #   @return [Hash] frozen span attributes
+  # @!attribute [r] resource_attributes
+  #   @return [Hash] frozen resource attributes
+  OtelSpanSnapshot = Data.define(
+    :trace_id, :span_id, :name, :scope_name, :attributes, :resource_attributes
+  )
+
+  # Export-stage masking wrapper around the Langfuse OTLP exporter.
+  #
+  # Applies the {Config#mask_otel_spans} hook to every export batch after
+  # +should_export_span+ has selected spans and the batch processor has built
+  # the batch. Only the copy exported to Langfuse is transformed; original
+  # span data (and therefore any other OpenTelemetry exporter) is untouched.
+  #
+  # Fail-closed behavior:
+  # - A hook exception or an invalid top-level result drops the whole batch.
+  # - A malformed per-span patch drops that span from the Langfuse export.
+  # - An invalid replacement attribute value omits that attribute entirely
+  #   rather than exporting its original value.
+  #
+  # @api private
+  class MaskingExporter
+    SUCCESS = OpenTelemetry::SDK::Trace::Export::SUCCESS
+    FAILURE = OpenTelemetry::SDK::Trace::Export::FAILURE
+    private_constant :SUCCESS, :FAILURE
+
+    # Allowed keys of one sparse span patch.
+    PATCH_KEYS = %i[delete set].freeze
+    private_constant :PATCH_KEYS
+
+    # @param delegate [#export, #force_flush, #shutdown] Langfuse OTLP exporter
+    # @param hook [#call] The configured mask_otel_spans callable
+    # @param logger [Logger]
+    def initialize(delegate:, hook:, logger:)
+      @delegate = delegate
+      @hook = hook
+      @logger = logger
+    end
+
+    # Mask the batch and delegate export. Drops the whole batch (fail closed)
+    # when the hook raises or returns an invalid result.
+    #
+    # @param span_data [Enumerable<OpenTelemetry::SDK::Trace::SpanData>]
+    # @param timeout [Numeric, nil]
+    # @return [Integer] OpenTelemetry export result code
+    def export(span_data, timeout: nil)
+      batch = span_data.to_a
+      masked = mask_batch(batch)
+      return FAILURE unless masked
+
+      @delegate.export(masked, timeout: timeout)
+    end
+
+    # @return [Object] delegate's force_flush result
+    def force_flush(timeout: nil)
+      @delegate.force_flush(timeout: timeout)
+    end
+
+    # @return [Object] delegate's shutdown result
+    def shutdown(timeout: nil)
+      @delegate.shutdown(timeout: timeout)
+    end
+
+    private
+
+    # @return [Array<SpanData>, nil] the masked batch, or nil to drop it
+    def mask_batch(batch)
+      patches = @hook.call(spans: snapshot_batch(batch))
+      return batch if patches.nil?
+
+      unless patches.is_a?(Hash)
+        @logger.error(
+          "Langfuse mask_otel_spans returned #{patches.class} instead of nil or a Hash " \
+          "of patches; dropping the Langfuse export batch"
+        )
+        return nil
+      end
+
+      apply_patches(batch, patches)
+    rescue StandardError => e
+      @logger.error("Langfuse mask_otel_spans raised #{e.class}: #{e.message}; dropping the Langfuse export batch")
+      nil
+    end
+
+    def snapshot_batch(batch)
+      batch.to_h { |span| [span_identifier(span), snapshot(span)] }.freeze
+    end
+
+    # Stable identifier the hook uses to key sparse patches.
+    def span_identifier(span)
+      "#{span.hex_trace_id}:#{span.hex_span_id}"
+    end
+
+    def snapshot(span)
+      OtelSpanSnapshot.new(
+        trace_id: span.hex_trace_id,
+        span_id: span.hex_span_id,
+        name: span.name,
+        scope_name: span.instrumentation_scope&.name,
+        attributes: frozen_attributes(span.attributes),
+        resource_attributes: frozen_attributes(span.resource&.attribute_enumerator&.to_h)
+      )
+    end
+
+    # Copies mutable values so freezing the snapshot never freezes objects
+    # still referenced by the original span data.
+    def frozen_attributes(attributes)
+      (attributes || {}).transform_values { |value| frozen_copy(value) }.freeze
+    end
+
+    def frozen_copy(value)
+      case value
+      when String, Array then value.dup.freeze
+      else value
+      end
+    end
+
+    # Unpatched spans are passed through untouched; patched spans are cloned
+    # so the originals stay pristine for any other exporter.
+    def apply_patches(batch, patches)
+      batch.filter_map do |span|
+        patch = patches[span_identifier(span)]
+        next span unless patch
+
+        apply_patch(span, patch)
+      end
+    end
+
+    # @return [SpanData, nil] the cloned masked span, or nil to drop it
+    def apply_patch(span, patch)
+      unless valid_patch?(patch)
+        @logger.error(
+          "Langfuse mask_otel_spans produced a malformed patch for span '#{span.name}'; " \
+          "dropping the span from the Langfuse export"
+        )
+        return nil
+      end
+
+      clone_with_attributes(span, patched_attributes(span, patch))
+    end
+
+    def valid_patch?(patch)
+      patch.is_a?(Hash) &&
+        (patch.keys - PATCH_KEYS).empty? &&
+        (patch[:delete].nil? || patch[:delete].is_a?(Array)) &&
+        (patch[:set].nil? || patch[:set].is_a?(Hash))
+    end
+
+    # Deletes run before sets so a replacement wins when a key appears in both.
+    def patched_attributes(span, patch)
+      attributes = (span.attributes || {}).dup
+      Array(patch[:delete]).each { |key| attributes.delete(key.to_s) }
+      (patch[:set] || {}).each do |key, value|
+        apply_replacement(attributes, key.to_s, value)
+      end
+      attributes.freeze
+    end
+
+    # Invalid replacements delete the key: never export the original value of
+    # an attribute the hook intended to overwrite. The value itself is never
+    # logged.
+    def apply_replacement(attributes, key, value)
+      if valid_attribute_value?(value)
+        attributes[key] = value
+      else
+        attributes.delete(key)
+        @logger.warn(
+          "Langfuse mask_otel_spans replacement for attribute '#{key}' is not a valid " \
+          "OpenTelemetry attribute value (#{value.class}); omitting the attribute"
+        )
+      end
+    end
+
+    def valid_attribute_value?(value)
+      case value
+      when String, Integer, Float, true, false then true
+      when Array then homogeneous_scalar_array?(value)
+      else false
+      end
+    end
+
+    def homogeneous_scalar_array?(array)
+      return true if array.empty?
+      # Booleans mix TrueClass and FalseClass but form one OpenTelemetry type.
+      return array.all? { |element| [true, false].include?(element) } if [true, false].include?(array.first)
+
+      type = array.first.class
+      [String, Integer, Float].include?(type) && array.all? { |element| element.instance_of?(type) }
+    end
+
+    def clone_with_attributes(span, attributes)
+      copy = span.dup
+      copy.attributes = attributes
+      copy
+    end
+  end
+end