diff --git a/docs/API_REFERENCE.md b/docs/API_REFERENCE.md index b6a2107..68433df 100644 --- a/docs/API_REFERENCE.md +++ b/docs/API_REFERENCE.md @@ -909,6 +909,112 @@ trace = client.get_trace("trace-uuid-123") puts trace["name"] ``` +### `Client#list_observations` + +List observation rows with cursor-based pagination and field selection. + +> **Cloud-only.** Delegates to `GET /api/public/v2/observations`, which is only +> available on Langfuse Cloud. There is no fallback to legacy endpoints because +> their response and pagination semantics differ. Returns observation rows, not +> reconstructed trace objects — group rows by `traceId` when you need trace +> activity. + +**Signature:** + +```ruby +list_observations(from_start_time: nil, to_start_time: nil, trace_id: nil, **filters) # => Hash +``` + +**Parameters:** + +| Parameter | Type | Required | Description | +| ----------------------- | --------------------- | -------- | ---------------------------------------------------------------------------------------------------- | +| `from_start_time` | Time, String | Yes* | Inclusive lower bound on observation start time. *Required unless `trace_id` is provided. | +| `to_start_time` | Time, String | Yes* | Exclusive upper bound on observation start time. *Required unless `trace_id` is provided. | +| `trace_id` | String | No | Filter by trace ID | +| `fields` | String | No | Comma-separated field groups: `core`, `basic`, `time`, `io`, `metadata`, `model`, `usage`, `prompt`, `metrics`, `trace_context` | +| `cursor` | String | No | Cursor from the previous response's `meta` for the next page | +| `limit` | Integer | No | Items per page (max 1000, default 50) | +| `filter` | String | No | JSON string with structured filter conditions (takes precedence over individual filters) | +| `name` | String | No | Filter by observation name | +| `user_id` | String | No | Filter by user ID | +| `type` | String | No | Filter by observation type (e.g. `"GENERATION"`, `"SPAN"`) | +| `level` | String | No | Filter by level (e.g. `"DEFAULT"`, `"ERROR"`) | +| `parent_observation_id` | String | No | Filter by parent observation ID | +| `environment` | String | No | Filter by environment | +| `version` | String | No | Filter by observation version | +| `expand_metadata` | String | No | Comma-separated metadata keys to return non-truncated | + +**Returns:** `Hash` with `"data"` (observation rows) and `"meta"` (pagination cursor) + +**Raises:** + +- `ArgumentError` if the read is unbounded (no `trace_id` and missing start-time bounds) +- `UnauthorizedError` if authentication fails +- `ApiError` for other API errors (including non-Cloud deployments) + +**Examples:** + +```ruby +# Bounded read of recent generations +page = client.list_observations( + from_start_time: Time.now - 3600, + to_start_time: Time.now, + type: "GENERATION", + fields: "core,basic,usage" +) +page["data"].each { |obs| puts obs["id"] } + +# Next page via cursor +next_page = client.list_observations( + from_start_time: Time.now - 3600, + to_start_time: Time.now, + cursor: page.dig("meta", "cursor") +) + +# All observations for one trace +rows = client.list_observations(trace_id: "trace-uuid-123")["data"] +``` + +### `Client#query_metrics` + +Query aggregate metrics. + +> **Cloud-only.** Delegates to `GET /api/public/v2/metrics`. Supports the +> `observations`, `scores-numeric`, and `scores-categorical` views. + +**Signature:** + +```ruby +query_metrics(query:) # => Hash +``` + +**Parameters:** + +| Parameter | Type | Required | Description | +| --------- | ------------ | -------- | ------------------------------------------------------------------------------------------ | +| `query` | Hash, String | Yes | Metrics query. A Hash is JSON-encoded; a pre-encoded JSON String is passed through as-is. | + +**Returns:** `Hash` of metrics results + +**Raises:** + +- `ArgumentError` if query is neither a Hash nor a String +- `UnauthorizedError` if authentication fails +- `ApiError` for other API errors (including non-Cloud deployments) + +**Example:** + +```ruby +result = client.query_metrics(query: { + view: "observations", + metrics: [{ measure: "count", aggregation: "count" }], + dimensions: [{ field: "name" }], + fromTimestamp: "2026-07-01T00:00:00Z", + toTimestamp: "2026-07-02T00:00:00Z" +}) +``` + ## Scoring ### `Client#create_score` @@ -1012,6 +1118,62 @@ Langfuse.flush_scores See [SCORING.md](SCORING.md) for complete guide. +### `Client#list_scores` + +List scores with polymorphic values via the v3 scores API. + +**Signature:** + +```ruby +list_scores(**filters) # => Hash +``` + +**Parameters:** + +| Parameter | Type | Required | Description | +| ---------------- | ------------ | -------- | ------------------------------------------------------------------------------------------------ | +| `limit` | Integer | No | Items per page (max 100, default 50) | +| `cursor` | String | No | Cursor from the previous response's `meta` for the next page | +| `fields` | String | No | Comma-separated field groups in addition to core: `details`, `subject`, `annotation` | +| `id` | String | No | Comma-separated score IDs | +| `name` | String | No | Comma-separated score names | +| `source` | String | No | Comma-separated sources (e.g. `API`, `ANNOTATION`, `EVAL`) | +| `data_type` | String | No | Comma-separated data types: `NUMERIC`, `BOOLEAN`, `CATEGORICAL`, `TEXT`, `CORRECTION` | +| `environment` | String | No | Comma-separated environments | +| `config_id` | String | No | Comma-separated score config IDs | +| `queue_id` | String | No | Comma-separated annotation queue IDs | +| `author_user_id` | String | No | Comma-separated author user IDs | +| `value` | String | No | Comma-separated exact values (requires a single `NUMERIC`, `BOOLEAN`, or `CATEGORICAL` data type) | +| `value_min` | Numeric | No | Inclusive lower bound (requires `data_type: "NUMERIC"`) | +| `value_max` | Numeric | No | Inclusive upper bound (requires `data_type: "NUMERIC"`) | +| `trace_id` | String | No | Comma-separated trace IDs (mutually exclusive with `session_id`, `experiment_id`) | +| `session_id` | String | No | Comma-separated session IDs | +| `observation_id` | String | No | Comma-separated observation IDs (requires `trace_id`) | +| `experiment_id` | String | No | Comma-separated dataset run (experiment) IDs | +| `from_timestamp` | Time, String | No | Inclusive lower bound on score timestamp | +| `to_timestamp` | Time, String | No | Exclusive upper bound on score timestamp | + +**Returns:** `Hash` with `"data"` (score rows) and `"meta"` (pagination cursor). Score +values are polymorphic by `dataType`: `NUMERIC` scores return numbers, `BOOLEAN` +scores return booleans, and `CATEGORICAL`, `TEXT`, and `CORRECTION` scores return +strings. + +**Raises:** + +- `UnauthorizedError` if authentication fails +- `ApiError` for other API errors + +**Examples:** + +```ruby +# Scores for a trace +page = client.list_scores(trace_id: "trace-uuid-123") + +# Corrections with subject details +corrections = client.list_scores(data_type: "CORRECTION", fields: "subject,details") +corrections["data"].each { |score| puts score["value"] } +``` + ## Datasets ### `Client#create_dataset` diff --git a/lib/langfuse.rb b/lib/langfuse.rb index 5e5e28f..a947aa7 100644 --- a/lib/langfuse.rb +++ b/lib/langfuse.rb @@ -47,6 +47,7 @@ class UnauthorizedError < ApiError; end require_relative "langfuse/prompt_cache_coordinator" require_relative "langfuse/cache_warmer" require_relative "langfuse/prompt_cache_events" +require_relative "langfuse/read_api" require_relative "langfuse/api_client" require_relative "langfuse/span_filter" require_relative "langfuse/sampling" diff --git a/lib/langfuse/api_client.rb b/lib/langfuse/api_client.rb index 551910f..27d78fa 100644 --- a/lib/langfuse/api_client.rb +++ b/lib/langfuse/api_client.rb @@ -7,6 +7,7 @@ require "uri" require_relative "prompt_fetch_result" require_relative "prompt_cache_coordinator" +require_relative "read_api" module Langfuse # HTTP client for Langfuse API @@ -25,6 +26,7 @@ module Langfuse # class ApiClient # rubocop:disable Metrics/ClassLength include PromptCacheEvents + include ReadApi # @return [String] Langfuse public API key attr_reader :public_key diff --git a/lib/langfuse/client.rb b/lib/langfuse/client.rb index 1a82d17..1be841a 100644 --- a/lib/langfuse/client.rb +++ b/lib/langfuse/client.rb @@ -44,6 +44,9 @@ class Client # @!method validate_prompt_cache_backend! # @!method list_traces(**options) # @!method get_trace(id) + # @!method list_observations(**options) + # @!method query_metrics(query:) + # @!method list_scores(**options) # @!method list_datasets(page: nil, limit: nil) # @!method get_dataset_run(dataset_name:, run_name:) # @!method create_dataset_run_item(**) @@ -57,6 +60,9 @@ class Client :validate_prompt_cache_backend!, :list_traces, :get_trace, + :list_observations, + :query_metrics, + :list_scores, :list_datasets, :get_dataset_run, :create_dataset_run_item diff --git a/lib/langfuse/otel_setup.rb b/lib/langfuse/otel_setup.rb index e4bfaa1..ced6187 100644 --- a/lib/langfuse/otel_setup.rb +++ b/lib/langfuse/otel_setup.rb @@ -139,7 +139,7 @@ def build_tracer_provider(config) def build_exporter(config) OpenTelemetry::Exporter::OTLP::Exporter.new( endpoint: "#{config.base_url}/api/public/otel/v1/traces", - headers: build_headers(config.public_key, config.secret_key), + headers: build_headers(config), compression: "gzip" ) end @@ -170,10 +170,19 @@ def blank?(value) value.nil? || value.empty? end - def build_headers(public_key, secret_key) - credentials = "#{public_key}:#{secret_key}" + # Langfuse requires `x-langfuse-ingestion-version: 4` on direct OTLP + # exports for real-time Fast Preview processing; without it, data can be + # delayed on the new UI and v2 read surfaces. + def build_headers(config) + credentials = "#{config.public_key}:#{config.secret_key}" encoded = Base64.strict_encode64(credentials) - { "Authorization" => "Basic #{encoded}" } + { + "Authorization" => "Basic #{encoded}", + "x-langfuse-ingestion-version" => "4", + "x-langfuse-sdk-name" => "ruby", + "x-langfuse-sdk-version" => Langfuse::VERSION, + "x-langfuse-public-key" => config.public_key + } end def build_sampler(sample_rate) diff --git a/lib/langfuse/read_api.rb b/lib/langfuse/read_api.rb new file mode 100644 index 0000000..f48d9ac --- /dev/null +++ b/lib/langfuse/read_api.rb @@ -0,0 +1,231 @@ +# frozen_string_literal: true + +require "json" + +module Langfuse + # Read endpoints for the current Langfuse query surface. + # + # Implements the Cloud-only v2 observation and metrics reads plus the v3 + # scores read. Mixed into {ApiClient}, whose private +request+ helper + # provides HTTP transport and error handling. + # + # @note The v2 observations and v2 metrics endpoints are only available on + # Langfuse Cloud. There is no fallback to legacy endpoints because their + # response and pagination semantics differ. + module ReadApi + # Ruby keyword argument -> camelCase query parameter mappings. Start-time + # bounds are handled separately because they need ISO 8601 formatting. + OBSERVATION_QUERY_PARAMS = { + trace_id: :traceId, fields: :fields, cursor: :cursor, limit: :limit, + filter: :filter, name: :name, user_id: :userId, type: :type, + level: :level, parent_observation_id: :parentObservationId, + environment: :environment, version: :version, + expand_metadata: :expandMetadata + }.freeze + private_constant :OBSERVATION_QUERY_PARAMS + + SCORE_QUERY_PARAMS = { + limit: :limit, cursor: :cursor, fields: :fields, id: :id, name: :name, + source: :source, data_type: :dataType, environment: :environment, + config_id: :configId, queue_id: :queueId, author_user_id: :authorUserId, + value: :value, value_min: :valueMin, value_max: :valueMax, + trace_id: :traceId, session_id: :sessionId, + observation_id: :observationId, experiment_id: :experimentId + }.freeze + private_constant :SCORE_QUERY_PARAMS + + # List observations with cursor-based pagination and field selection + # + # Delegates to +GET /api/public/v2/observations+ (Langfuse Cloud only). + # Returns observation rows, not reconstructed trace objects. The full + # response envelope is preserved: +"data"+ holds the observation rows and + # +"meta"+ holds the pagination cursor for the next page. + # + # Broad reads must be bounded: unless +trace_id+ narrows the query, both + # +from_start_time+ and +to_start_time+ are required. + # + # @param from_start_time [Time, String, nil] Inclusive lower bound on observation start time + # @param to_start_time [Time, String, nil] Exclusive upper bound on observation start time + # @param trace_id [String, nil] Filter by trace ID + # @param fields [String, nil] Comma-separated field groups to include + # (core, basic, time, io, metadata, model, usage, prompt, metrics, trace_context) + # @param cursor [String, nil] Cursor from the previous response's meta for the next page + # @param limit [Integer, nil] Items per page (max 1000, default 50) + # @param filter [String, nil] JSON string with structured filter conditions; + # takes precedence over individual query parameter filters + # @param name [String, nil] Filter by observation name + # @param user_id [String, nil] Filter by user ID + # @param type [String, nil] Filter by observation type (e.g. "GENERATION", "SPAN") + # @param level [String, nil] Filter by level (e.g. "DEFAULT", "ERROR") + # @param parent_observation_id [String, nil] Filter by parent observation ID + # @param environment [String, nil] Filter by environment + # @param version [String, nil] Filter by observation version + # @param expand_metadata [String, nil] Comma-separated metadata keys to return non-truncated + # @return [Hash] Full response hash with "data" rows and "meta" cursor info + # @raise [ArgumentError] if the read is unbounded (no trace_id and missing start-time bounds) + # @raise [UnauthorizedError] if authentication fails + # @raise [ApiError] for other API errors (including non-Cloud deployments) + # + # @example Bounded read of recent generations + # page = api_client.list_observations( + # from_start_time: Time.now - 3600, + # to_start_time: Time.now, + # type: "GENERATION", + # fields: "core,basic,usage" + # ) + # page["data"].each { |obs| puts obs["id"] } + # next_cursor = page.dig("meta", "cursor") + # rubocop:disable Metrics/ParameterLists + def list_observations(from_start_time: nil, to_start_time: nil, trace_id: nil, + fields: nil, cursor: nil, limit: nil, filter: nil, + name: nil, user_id: nil, type: nil, level: nil, + parent_observation_id: nil, environment: nil, + version: nil, expand_metadata: nil) + validate_bounded_observation_read!(trace_id, from_start_time, to_start_time) + params = build_observations_params( + from_start_time: from_start_time, to_start_time: to_start_time, + trace_id: trace_id, fields: fields, cursor: cursor, limit: limit, + filter: filter, name: name, user_id: user_id, type: type, level: level, + parent_observation_id: parent_observation_id, environment: environment, + version: version, expand_metadata: expand_metadata + ) + request(:get, "/api/public/v2/observations", params: params) + end + # rubocop:enable Metrics/ParameterLists + + # Query aggregate metrics (Langfuse Cloud only) + # + # Delegates to +GET /api/public/v2/metrics+. Supports the +observations+, + # +scores-numeric+, and +scores-categorical+ views. + # + # @param query [Hash, String] Metrics query. A Hash is JSON-encoded into + # the endpoint's +query+ parameter; a pre-encoded JSON String is passed + # through unchanged. + # @return [Hash] The parsed metrics response + # @raise [ArgumentError] if query is neither a Hash nor a String + # @raise [UnauthorizedError] if authentication fails + # @raise [ApiError] for other API errors (including non-Cloud deployments) + # + # @example Count observations by name + # api_client.query_metrics(query: { + # view: "observations", + # metrics: [{ measure: "count", aggregation: "count" }], + # dimensions: [{ field: "name" }], + # fromTimestamp: "2026-07-01T00:00:00Z", + # toTimestamp: "2026-07-02T00:00:00Z" + # }) + def query_metrics(query:) + request(:get, "/api/public/v2/metrics", params: { query: encode_metrics_query(query) }) + end + + # List scores with polymorphic values (v3) + # + # Delegates to +GET /api/public/v3/scores+. The full response envelope is + # preserved: +"data"+ holds score rows and +"meta"+ holds the pagination + # cursor. Score values are polymorphic by +dataType+: NUMERIC scores return + # numbers, BOOLEAN scores return booleans, and CATEGORICAL, TEXT, and + # CORRECTION scores return strings. + # + # @param limit [Integer, nil] Items per page (max 100, default 50) + # @param cursor [String, nil] Cursor from the previous response's meta for the next page + # @param fields [String, nil] Comma-separated field groups in addition to core + # (details, subject, annotation) + # @param id [String, nil] Comma-separated score IDs to filter by + # @param name [String, nil] Comma-separated score names to filter by + # @param source [String, nil] Comma-separated score sources (e.g. API, ANNOTATION, EVAL) + # @param data_type [String, nil] Comma-separated data types + # (NUMERIC, BOOLEAN, CATEGORICAL, TEXT, CORRECTION) + # @param environment [String, nil] Comma-separated environments to filter by + # @param config_id [String, nil] Comma-separated score config IDs + # @param queue_id [String, nil] Comma-separated annotation queue IDs + # @param author_user_id [String, nil] Comma-separated author user IDs + # @param value [String, nil] Comma-separated exact values (requires a single + # NUMERIC, BOOLEAN, or CATEGORICAL data_type) + # @param value_min [Numeric, nil] Inclusive lower bound (requires data_type: "NUMERIC") + # @param value_max [Numeric, nil] Inclusive upper bound (requires data_type: "NUMERIC") + # @param trace_id [String, nil] Comma-separated trace IDs (mutually exclusive + # with session_id and experiment_id) + # @param session_id [String, nil] Comma-separated session IDs + # @param observation_id [String, nil] Comma-separated observation IDs (requires trace_id) + # @param experiment_id [String, nil] Comma-separated dataset run (experiment) IDs + # @param from_timestamp [Time, String, nil] Inclusive lower bound on score timestamp + # @param to_timestamp [Time, String, nil] Exclusive upper bound on score timestamp + # @return [Hash] Full response hash with "data" rows and "meta" cursor info + # @raise [UnauthorizedError] if authentication fails + # @raise [ApiError] for other API errors + # + # @example Read corrections for a trace + # page = api_client.list_scores(trace_id: trace_id, data_type: "CORRECTION", fields: "subject,details") + # page["data"].each { |score| puts score["value"] } + # rubocop:disable Metrics/ParameterLists + def list_scores(limit: nil, cursor: nil, fields: nil, id: nil, name: nil, + source: nil, data_type: nil, environment: nil, config_id: nil, + queue_id: nil, author_user_id: nil, value: nil, value_min: nil, + value_max: nil, trace_id: nil, session_id: nil, + observation_id: nil, experiment_id: nil, + from_timestamp: nil, to_timestamp: nil) + params = build_scores_params( + limit: limit, cursor: cursor, fields: fields, id: id, name: name, + source: source, data_type: data_type, environment: environment, + config_id: config_id, queue_id: queue_id, author_user_id: author_user_id, + value: value, value_min: value_min, value_max: value_max, + trace_id: trace_id, session_id: session_id, observation_id: observation_id, + experiment_id: experiment_id, from_timestamp: from_timestamp, to_timestamp: to_timestamp + ) + request(:get, "/api/public/v3/scores", params: params) + end + # rubocop:enable Metrics/ParameterLists + + private + + # v2 observation reads must always be bounded; an unbounded scan over the + # events table is rejected here rather than issued silently. Only values + # matching the documented query contract can satisfy the bound. + def validate_bounded_observation_read!(trace_id, from_start_time, to_start_time) + return if non_empty_string?(trace_id) + return if valid_query_time?(from_start_time) && valid_query_time?(to_start_time) + + raise ArgumentError, + "from_start_time and to_start_time are required unless trace_id is provided" + end + + def valid_query_time?(value) + non_empty_string?(format_query_time(value)) + end + + def non_empty_string?(value) + value.is_a?(String) && !value.strip.empty? + end + + def build_observations_params(**options) + camelize_params(OBSERVATION_QUERY_PARAMS, options).merge( + fromStartTime: format_query_time(options[:from_start_time]), + toStartTime: format_query_time(options[:to_start_time]) + ).compact + end + + def build_scores_params(**options) + camelize_params(SCORE_QUERY_PARAMS, options).merge( + fromTimestamp: format_query_time(options[:from_timestamp]), + toTimestamp: format_query_time(options[:to_timestamp]) + ).compact + end + + def camelize_params(mapping, options) + mapping.to_h { |ruby_key, api_key| [api_key, options[ruby_key]] } + end + + def encode_metrics_query(query) + case query + when Hash then JSON.generate(query) + when String then query + else raise ArgumentError, "query must be a Hash or JSON String, got #{query.class}" + end + end + + # Accepts Time-like values or pre-formatted ISO 8601 strings. + def format_query_time(value) + value.respond_to?(:iso8601) ? value.iso8601 : value + end + end +end diff --git a/spec/langfuse/otel_setup_spec.rb b/spec/langfuse/otel_setup_spec.rb index 7f76798..0570a5b 100644 --- a/spec/langfuse/otel_setup_spec.rb +++ b/spec/langfuse/otel_setup_spec.rb @@ -256,4 +256,25 @@ expect(exporter.finished_spans).to be_empty end end + + describe ".build_headers" do + let(:headers) { described_class.send(:build_headers, config) } + + it "includes Basic auth credentials" do + encoded = Base64.strict_encode64("pk_test_123:sk_test_456") + expect(headers["Authorization"]).to eq("Basic #{encoded}") + end + + it "requests v4 Fast Preview ingestion" do + expect(headers["x-langfuse-ingestion-version"]).to eq("4") + end + + it "identifies the SDK by name, version, and public key" do + expect(headers).to include( + "x-langfuse-sdk-name" => "ruby", + "x-langfuse-sdk-version" => Langfuse::VERSION, + "x-langfuse-public-key" => "pk_test_123" + ) + end + end end diff --git a/spec/langfuse/read_api_spec.rb b/spec/langfuse/read_api_spec.rb new file mode 100644 index 0000000..a85e182 --- /dev/null +++ b/spec/langfuse/read_api_spec.rb @@ -0,0 +1,406 @@ +# frozen_string_literal: true + +RSpec.describe Langfuse::ReadApi do + let(:base_url) { "https://cloud.langfuse.com" } + let(:api_client) do + Langfuse::ApiClient.new( + public_key: "pk_test_123", + secret_key: "sk_test_456", + base_url: base_url, + timeout: 10 + ) + end + + let(:from_time) { Time.utc(2026, 7, 1, 0, 0, 0) } + let(:to_time) { Time.utc(2026, 7, 2, 0, 0, 0) } + + describe "#list_observations" do + let(:observations_response) do + { + "data" => [ + { "id" => "obs-1", "traceId" => "trace-1", "type" => "GENERATION" }, + { "id" => "obs-2", "traceId" => "trace-1", "type" => "SPAN" } + ], + "meta" => { "cursor" => "bmV4dC1wYWdl", "limit" => 50 } + } + end + + context "with a bounded read" do + before do + stub_request(:get, "#{base_url}/api/public/v2/observations") + .with(query: { fromStartTime: from_time.iso8601, toStartTime: to_time.iso8601 }) + .to_return( + status: 200, + body: observations_response.to_json, + headers: { "Content-Type" => "application/json" } + ) + end + + it "preserves the full data and meta envelope" do + result = api_client.list_observations(from_start_time: from_time, to_start_time: to_time) + expect(result["data"].size).to eq(2) + expect(result.dig("meta", "cursor")).to eq("bmV4dC1wYWdl") + end + + it "serializes Time bounds to ISO 8601" do + api_client.list_observations(from_start_time: from_time, to_start_time: to_time) + expect( + a_request(:get, "#{base_url}/api/public/v2/observations") + .with(query: { fromStartTime: from_time.iso8601, toStartTime: to_time.iso8601 }) + ).to have_been_made.once + end + + it "passes pre-formatted string bounds through unchanged" do + api_client.list_observations( + from_start_time: from_time.iso8601, to_start_time: to_time.iso8601 + ) + expect( + a_request(:get, "#{base_url}/api/public/v2/observations") + .with(query: { fromStartTime: from_time.iso8601, toStartTime: to_time.iso8601 }) + ).to have_been_made.once + end + end + + context "with filters and field selection" do + let(:query) do + { + fromStartTime: from_time.iso8601, toStartTime: to_time.iso8601, + fields: "core,basic,usage", cursor: "Y3Vyc29y", limit: "100", + type: "GENERATION", userId: "user-1", name: "chat", level: "ERROR", + parentObservationId: "parent-1", environment: "production", + version: "1.0", expandMetadata: "key1,key2" + } + end + + before do + stub_request(:get, "#{base_url}/api/public/v2/observations") + .with(query: query) + .to_return( + status: 200, + body: observations_response.to_json, + headers: { "Content-Type" => "application/json" } + ) + end + + it "maps snake_case keywords to camelCase query params" do + api_client.list_observations( + from_start_time: from_time, to_start_time: to_time, + fields: "core,basic,usage", cursor: "Y3Vyc29y", limit: 100, + type: "GENERATION", user_id: "user-1", name: "chat", level: "ERROR", + parent_observation_id: "parent-1", environment: "production", + version: "1.0", expand_metadata: "key1,key2" + ) + expect( + a_request(:get, "#{base_url}/api/public/v2/observations").with(query: query) + ).to have_been_made.once + end + end + + context "with a structured filter" do + let(:filter_json) { '[{"type":"string","column":"id","operator":"=","value":"obs-1"}]' } + + before do + stub_request(:get, "#{base_url}/api/public/v2/observations") + .with(query: hash_including(filter: filter_json)) + .to_return( + status: 200, + body: observations_response.to_json, + headers: { "Content-Type" => "application/json" } + ) + end + + it "passes the filter JSON through to the query string" do + api_client.list_observations( + from_start_time: from_time, to_start_time: to_time, filter: filter_json + ) + expect( + a_request(:get, "#{base_url}/api/public/v2/observations") + .with(query: hash_including(filter: filter_json)) + ).to have_been_made.once + end + end + + context "with a trace-scoped read" do + before do + stub_request(:get, "#{base_url}/api/public/v2/observations") + .with(query: { traceId: "trace-1" }) + .to_return( + status: 200, + body: observations_response.to_json, + headers: { "Content-Type" => "application/json" } + ) + end + + it "allows omitting start-time bounds when trace_id is provided" do + result = api_client.list_observations(trace_id: "trace-1") + expect(result["data"].size).to eq(2) + end + end + + context "with an unbounded read" do + it "rejects a read with no bounds and no trace_id" do + expect { api_client.list_observations } + .to raise_error(ArgumentError, /from_start_time and to_start_time are required/) + end + + it "rejects a read missing one bound" do + expect { api_client.list_observations(from_start_time: from_time) } + .to raise_error(ArgumentError, /from_start_time and to_start_time are required/) + end + + it "rejects a blank trace_id as absent" do + expect { api_client.list_observations(trace_id: " ") } + .to raise_error(ArgumentError, /from_start_time and to_start_time are required/) + end + + it "rejects blank string bounds as absent" do + expect { api_client.list_observations(from_start_time: "", to_start_time: "") } + .to raise_error(ArgumentError, /from_start_time and to_start_time are required/) + end + + it "rejects false bounds as absent" do + expect { api_client.list_observations(from_start_time: false, to_start_time: false) } + .to raise_error(ArgumentError, /from_start_time and to_start_time are required/) + end + + it "rejects truthy non-string trace ids as absent" do + expect { api_client.list_observations(trace_id: []) } + .to raise_error(ArgumentError, /from_start_time and to_start_time are required/) + end + + it "rejects truthy non-time bounds as absent" do + expect { api_client.list_observations(from_start_time: [], to_start_time: {}) } + .to raise_error(ArgumentError, /from_start_time and to_start_time are required/) + end + + it "does not issue an HTTP request" do + begin + api_client.list_observations + rescue ArgumentError + nil + end + expect(a_request(:get, "#{base_url}/api/public/v2/observations")).not_to have_been_made + end + end + + context "when authentication fails" do + before do + stub_request(:get, "#{base_url}/api/public/v2/observations") + .with(query: hash_including({})) + .to_return(status: 401, body: { message: "Unauthorized" }.to_json) + end + + it "raises UnauthorizedError" do + expect { api_client.list_observations(from_start_time: from_time, to_start_time: to_time) } + .to raise_error(Langfuse::UnauthorizedError) + end + end + + context "when the endpoint is unavailable (self-hosted)" do + before do + stub_request(:get, "#{base_url}/api/public/v2/observations") + .with(query: hash_including({})) + .to_return(status: 404, body: { message: "Not found" }.to_json) + end + + it "raises NotFoundError rather than falling back to legacy endpoints" do + expect { api_client.list_observations(from_start_time: from_time, to_start_time: to_time) } + .to raise_error(Langfuse::NotFoundError) + expect(a_request(:get, "#{base_url}/api/public/traces")).not_to have_been_made + end + end + end + + describe "#query_metrics" do + let(:metrics_query) do + { + view: "observations", + metrics: [{ measure: "count", aggregation: "count" }], + fromTimestamp: from_time.iso8601, + toTimestamp: to_time.iso8601 + } + end + let(:metrics_response) { { "data" => [{ "count_count" => "42" }] } } + + before do + stub_request(:get, "#{base_url}/api/public/v2/metrics") + .with(query: { query: JSON.generate(metrics_query) }) + .to_return( + status: 200, + body: metrics_response.to_json, + headers: { "Content-Type" => "application/json" } + ) + end + + it "JSON-encodes a Hash query into the query parameter" do + result = api_client.query_metrics(query: metrics_query) + expect(result["data"]).to eq([{ "count_count" => "42" }]) + end + + it "passes a pre-encoded JSON string through unchanged" do + api_client.query_metrics(query: JSON.generate(metrics_query)) + expect( + a_request(:get, "#{base_url}/api/public/v2/metrics") + .with(query: { query: JSON.generate(metrics_query) }) + ).to have_been_made.once + end + + it "rejects non-Hash, non-String queries" do + expect { api_client.query_metrics(query: [1, 2]) } + .to raise_error(ArgumentError, /query must be a Hash or JSON String/) + end + + context "when authentication fails" do + before do + stub_request(:get, "#{base_url}/api/public/v2/metrics") + .with(query: hash_including({})) + .to_return(status: 401, body: { message: "Unauthorized" }.to_json) + end + + it "raises UnauthorizedError" do + expect { api_client.query_metrics(query: metrics_query) } + .to raise_error(Langfuse::UnauthorizedError) + end + end + end + + describe "#list_scores" do + let(:scores_response) do + { + "data" => [ + { "id" => "score-1", "name" => "quality", "dataType" => "NUMERIC", "value" => 0.85 }, + { "id" => "score-2", "name" => "passed", "dataType" => "BOOLEAN", "value" => true }, + { "id" => "score-3", "name" => "output", "dataType" => "CORRECTION", "value" => "fixed output" } + ], + "meta" => { "cursor" => "bmV4dA", "limit" => 50 } + } + end + + context "with a successful response" do + before do + stub_request(:get, "#{base_url}/api/public/v3/scores") + .with(query: hash_including({})) + .to_return( + status: 200, + body: scores_response.to_json, + headers: { "Content-Type" => "application/json" } + ) + end + + it "preserves the full data and meta envelope" do + result = api_client.list_scores + expect(result["data"].size).to eq(3) + expect(result.dig("meta", "cursor")).to eq("bmV4dA") + end + + it "preserves polymorphic score values" do + values = api_client.list_scores["data"].map { |score| score["value"] } + expect(values).to eq([0.85, true, "fixed output"]) + end + end + + context "with filters" do + let(:query) do + { + limit: "25", cursor: "Y3Vyc29y", fields: "subject,details", + id: "score-1,score-2", name: "quality", source: "API", + dataType: "CORRECTION", environment: "production", + configId: "cfg-1", queueId: "queue-1", authorUserId: "user-1", + traceId: "trace-1", observationId: "obs-1", + fromTimestamp: from_time.iso8601, toTimestamp: to_time.iso8601 + } + end + + before do + stub_request(:get, "#{base_url}/api/public/v3/scores") + .with(query: query) + .to_return( + status: 200, + body: scores_response.to_json, + headers: { "Content-Type" => "application/json" } + ) + end + + it "maps snake_case keywords to camelCase query params" do + api_client.list_scores( + limit: 25, cursor: "Y3Vyc29y", fields: "subject,details", + id: "score-1,score-2", name: "quality", source: "API", + data_type: "CORRECTION", environment: "production", + config_id: "cfg-1", queue_id: "queue-1", author_user_id: "user-1", + trace_id: "trace-1", observation_id: "obs-1", + from_timestamp: from_time, to_timestamp: to_time + ) + expect( + a_request(:get, "#{base_url}/api/public/v3/scores").with(query: query) + ).to have_been_made.once + end + end + + context "with numeric value bounds" do + before do + stub_request(:get, "#{base_url}/api/public/v3/scores") + .with(query: { dataType: "NUMERIC", valueMin: "0.5", valueMax: "1" }) + .to_return( + status: 200, + body: scores_response.to_json, + headers: { "Content-Type" => "application/json" } + ) + end + + it "sends valueMin and valueMax" do + api_client.list_scores(data_type: "NUMERIC", value_min: 0.5, value_max: 1) + expect( + a_request(:get, "#{base_url}/api/public/v3/scores") + .with(query: { dataType: "NUMERIC", valueMin: "0.5", valueMax: "1" }) + ).to have_been_made.once + end + end + + context "when authentication fails" do + before do + stub_request(:get, "#{base_url}/api/public/v3/scores") + .with(query: hash_including({})) + .to_return(status: 401, body: { message: "Unauthorized" }.to_json) + end + + it "raises UnauthorizedError" do + expect { api_client.list_scores }.to raise_error(Langfuse::UnauthorizedError) + end + end + end + + describe "Client delegation" do + let(:config) do + Langfuse::Config.new do |c| + c.public_key = "pk_test_123" + c.secret_key = "sk_test_456" + c.base_url = base_url + end + end + let(:client) { Langfuse::Client.new(config) } + + before do + stub_request(:get, %r{#{base_url}/api/public/(v2/observations|v2/metrics|v3/scores)}) + .to_return( + status: 200, + body: { "data" => [], "meta" => {} }.to_json, + headers: { "Content-Type" => "application/json" } + ) + end + + it "exposes list_observations on the flat client API" do + client.list_observations(from_start_time: from_time, to_start_time: to_time) + expect(a_request(:get, %r{/api/public/v2/observations})).to have_been_made.once + end + + it "exposes query_metrics on the flat client API" do + client.query_metrics(query: { view: "observations" }) + expect(a_request(:get, %r{/api/public/v2/metrics})).to have_been_made.once + end + + it "exposes list_scores on the flat client API" do + client.list_scores(trace_id: "trace-1") + expect(a_request(:get, %r{/api/public/v3/scores})).to have_been_made.once + end + end +end