feat(captain): trace Jev calls in Langfuse through a shared client (#16031)
Captain label and priority suggestions and conversation monitors call Jev outside RubyLLM. Those requests were missing from Langfuse. <img width="1732" height="1526" alt="CleanShot 2026-09-28 at 12 07 01@2x" src="https://github.com/user-attachments/assets/ce8735af-c360-4373-91d4-4e01822e65c0" /> ## What changed - Route both features through one System One HTTP client. - Record a Langfuse generation with the complete request and response, served model, reported token usage, cost, account, and conversation session. - Keep monitor credit accounting, response validation, scoring, and error mapping in its decision service. - Mark Jev's remote call as an OpenTelemetry client span and record its operation, returned model, conversation ID, and low-cardinality failure type. ## Langfuse v4 and OpenTelemetry review The existing exporter already uses OTLP/HTTP at `/api/public/otel/v1/traces`, Basic Auth, and `x-langfuse-ingestion-version: 4`. Each Jev call exports one completed generation span with its full request and response on the observation. The v4 migration does not require a new Langfuse API or SDK here. The new Jev fields follow the current OpenTelemetry GenAI client-span guidance while keeping the stable feature-specific span name. Jev's reported token counts are copied from its response. The full HTTP request body is captured, but Jev does not expose any internal representation it may build from that body. Higher token counts do not establish a specific hidden prompt or overhead. ## Verification - 103 focused RSpec examples passed for the shared Jev implementation; the four Jev client examples pass after the OpenTelemetry follow-up. RuboCop passes on the changed Ruby files. - Synthetic classifier and monitor calls succeeded locally. Langfuse returned one `GENERATION` observation for each call, with the request, response, model, usage, and cost. References: [Langfuse v4 custom ingestion](https://langfuse.com/integrations/native/opentelemetry/migration-to-v4), [Langfuse OTEL mapping](https://langfuse.com/integrations/native/opentelemetry), [OpenTelemetry GenAI client spans](https://github.com/open-telemetry/semantic-conventions-genai/blob/main/docs/gen-ai/gen-ai-spans.md).
This commit is contained in:
@@ -1,7 +1,5 @@
|
||||
class Captain::ConversationClassifierService
|
||||
MODEL = 'jev-latest'.freeze
|
||||
SYSTEM_ONE_PATH = '/v1/systemone'.freeze
|
||||
REQUEST_TIMEOUT = 15
|
||||
MESSAGE_LIMIT = 20
|
||||
STATE_TOKEN_BUDGET = 15_000
|
||||
CHARACTERS_PER_TOKEN = 4
|
||||
@@ -60,18 +58,11 @@ class Captain::ConversationClassifierService
|
||||
end
|
||||
|
||||
def ask(questions)
|
||||
response = HTTParty.post(
|
||||
"#{GlobalConfigService.load('CAPTAIN_OPENROUTER_DECISION_MODEL_ENDPOINT', nil)}#{SYSTEM_ONE_PATH}",
|
||||
headers: {
|
||||
'Authorization' => "Bearer #{GlobalConfigService.load('CAPTAIN_OPENROUTER_API_KEY', nil)}",
|
||||
'Content-Type' => 'application/json'
|
||||
},
|
||||
body: { model: MODEL, state: state, questions: questions }.to_json,
|
||||
timeout: REQUEST_TIMEOUT
|
||||
)
|
||||
raise Error, "Jev request failed with status #{response.code}: #{response.body}" unless response.success?
|
||||
|
||||
response.parsed_response['answers']
|
||||
body = Captain::JevClient.request_body(model: MODEL, state: state, questions: questions)
|
||||
Captain::JevClient.new(account_id: conversation.account_id, conversation_id: conversation.display_id,
|
||||
feature: 'captain_classifier').call(body: body)['answers']
|
||||
rescue Captain::JevClient::HTTPError => e
|
||||
raise Error, e.message
|
||||
end
|
||||
|
||||
def state
|
||||
|
||||
105
app/services/captain/jev_client.rb
Normal file
105
app/services/captain/jev_client.rb
Normal file
@@ -0,0 +1,105 @@
|
||||
class Captain::JevClient
|
||||
include Integrations::LlmInstrumentationConstants
|
||||
|
||||
ENDPOINT = 'https://openrouter.ai/api'.freeze
|
||||
SYSTEM_ONE_PATH = '/v1/systemone'.freeze
|
||||
|
||||
class HTTPError < StandardError
|
||||
attr_reader :status, :retry_after
|
||||
|
||||
def initialize(response)
|
||||
@status = response.status
|
||||
@retry_after = response.headers['retry-after']&.to_i
|
||||
super("Jev request failed with status #{status}: #{response.body}")
|
||||
end
|
||||
end
|
||||
|
||||
def initialize(account_id:, feature:, conversation_id: nil)
|
||||
@account_id = account_id
|
||||
@conversation_id = conversation_id
|
||||
@feature = feature
|
||||
end
|
||||
|
||||
def self.request_body(model:, state:, questions:)
|
||||
{ model: model, state: state, questions: questions }.to_json
|
||||
end
|
||||
|
||||
def self.api_key
|
||||
GlobalConfigService.load('CAPTAIN_OPENROUTER_API_KEY', nil)
|
||||
end
|
||||
|
||||
def self.endpoint
|
||||
base_url = GlobalConfigService.load('CAPTAIN_OPENROUTER_DECISION_MODEL_ENDPOINT', nil).presence || ENDPOINT
|
||||
"#{base_url.chomp('/')}#{SYSTEM_ONE_PATH}"
|
||||
end
|
||||
|
||||
def call(body:)
|
||||
trace_jev_call(body) do
|
||||
response = connection.post(self.class.endpoint, body)
|
||||
raise HTTPError, response unless response.success?
|
||||
|
||||
data = JSON.parse(response.body)
|
||||
yield(data) if block_given?
|
||||
data
|
||||
end
|
||||
end
|
||||
|
||||
private
|
||||
|
||||
def trace_jev_call(body, &)
|
||||
return yield unless ChatwootApp.otel_enabled?
|
||||
|
||||
traced_jev_call(body, &)
|
||||
end
|
||||
|
||||
# Tracing is best-effort: a tracing failure must not block the Jev call or discard a response that already used provider credit.
|
||||
def traced_jev_call(body, &)
|
||||
outcome = {}
|
||||
OpentelemetryConfig.tracer.in_span("llm.jev.#{@feature}", kind: :client) do |span|
|
||||
request_model = JSON.parse(body).fetch('model')
|
||||
set_request_attributes(span, body)
|
||||
call_in_span(span, outcome, &)
|
||||
set_response_attributes(span, outcome[:result], request_model)
|
||||
end
|
||||
outcome[:result]
|
||||
rescue StandardError => e
|
||||
raise outcome[:error] if outcome.key?(:error)
|
||||
|
||||
ChatwootExceptionTracker.new(e).capture_exception
|
||||
outcome.key?(:result) ? outcome[:result] : yield
|
||||
end
|
||||
|
||||
def call_in_span(span, outcome)
|
||||
outcome[:result] = yield
|
||||
rescue StandardError => e
|
||||
outcome[:error] = e
|
||||
span.status = OpenTelemetry::Trace::Status.error(e.message)
|
||||
raise
|
||||
end
|
||||
|
||||
def set_request_attributes(span, body)
|
||||
span.set_attribute(ATTR_LANGFUSE_OBSERVATION_TYPE, 'generation')
|
||||
span.set_attribute(ATTR_LANGFUSE_OBSERVATION_INPUT, body)
|
||||
span.set_attribute(ATTR_LANGFUSE_USER_ID, @account_id.to_s)
|
||||
span.set_attribute(ATTR_LANGFUSE_TAGS, [@feature].to_json)
|
||||
span.set_attribute(ATTR_LANGFUSE_SESSION_ID, "#{@account_id}_#{@conversation_id}") if @conversation_id
|
||||
end
|
||||
|
||||
def set_response_attributes(span, result, request_model)
|
||||
usage = result['usage'] || {}
|
||||
span.set_attribute(ATTR_LANGFUSE_OBSERVATION_OUTPUT, result.to_json)
|
||||
span.set_attribute(ATTR_LANGFUSE_OBSERVATION_MODEL_NAME, result['model'] || request_model)
|
||||
span.set_attribute(ATTR_GEN_AI_USAGE_INPUT_TOKENS, usage['input_tokens']) if usage['input_tokens']
|
||||
span.set_attribute(ATTR_GEN_AI_USAGE_OUTPUT_TOKENS, usage['output_tokens']) if usage['output_tokens']
|
||||
span.set_attribute(ATTR_LANGFUSE_OBSERVATION_COST_DETAILS, { total: usage['cost'] }.to_json) if usage['cost'].is_a?(Numeric)
|
||||
end
|
||||
|
||||
def connection
|
||||
Faraday.new do |client|
|
||||
client.options.open_timeout = 3
|
||||
client.options.timeout = 15
|
||||
client.headers['Authorization'] = "Bearer #{self.class.api_key}"
|
||||
client.headers['Content-Type'] = 'application/json'
|
||||
end
|
||||
end
|
||||
end
|
||||
@@ -47,8 +47,9 @@ class ConversationMonitors::PreviewJob < ApplicationJob
|
||||
|
||||
def matches?(conversation)
|
||||
state = ConversationMonitors::ContextBuilder.new(conversation).build
|
||||
response = ConversationMonitors::JevClient.new(account_id: conversation.account_id).evaluate(state: state, monitors: [@monitor])
|
||||
ConversationMonitors::JevClient.score(response['answers']['0']) >= ConversationMonitors::Configuration::THRESHOLD
|
||||
response = ConversationMonitors::DecisionService.new(account_id: conversation.account_id, conversation_id: conversation.display_id)
|
||||
.evaluate(state: state, monitors: [@monitor])
|
||||
ConversationMonitors::DecisionService.score(response['answers']['0']) >= ConversationMonitors::Configuration::THRESHOLD
|
||||
rescue CustomExceptions::MonitorEvaluationError => e
|
||||
raise unless %w[context_limit no_text].include?(e.code)
|
||||
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
class ConversationMonitors::Configuration
|
||||
MODEL = 'typesafe/jev-1.13'.freeze
|
||||
ENDPOINT = 'https://openrouter.ai/api'.freeze
|
||||
SYSTEM_ONE_PATH = '/v1/systemone'.freeze
|
||||
ENDPOINT = Captain::JevClient::ENDPOINT
|
||||
THRESHOLD = 0.6
|
||||
MAX_MONITORS = 20
|
||||
MONTHLY_CALL_LIMIT = 100_000
|
||||
@@ -19,12 +18,11 @@ class ConversationMonitors::Configuration
|
||||
end
|
||||
|
||||
def self.api_key
|
||||
GlobalConfigService.load('CAPTAIN_OPENROUTER_API_KEY', nil)
|
||||
Captain::JevClient.api_key
|
||||
end
|
||||
|
||||
def self.endpoint
|
||||
base_url = GlobalConfigService.load('CAPTAIN_OPENROUTER_DECISION_MODEL_ENDPOINT', nil).presence || ENDPOINT
|
||||
"#{base_url.chomp('/')}#{SYSTEM_ONE_PATH}"
|
||||
Captain::JevClient.endpoint
|
||||
end
|
||||
|
||||
def self.model
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
class ConversationMonitors::JevClient
|
||||
class ConversationMonitors::DecisionService
|
||||
# Translate the legacy model ID on existing monitors to OpenRouter's Jev 1.13 ID.
|
||||
MODEL_ALIASES = { 'jev-1.13.0' => 'typesafe/jev-1.13' }.freeze
|
||||
|
||||
def initialize(account_id:)
|
||||
def initialize(account_id:, conversation_id: nil)
|
||||
@account_id = account_id
|
||||
@conversation_id = conversation_id
|
||||
end
|
||||
|
||||
def evaluate(state:, monitors:)
|
||||
@@ -15,13 +16,16 @@ class ConversationMonitors::JevClient
|
||||
usage = ConversationMonitors::Usage.new(@account_id)
|
||||
usage.reserve!(body.bytesize)
|
||||
started = Process.clock_gettime(Process::CLOCK_MONOTONIC)
|
||||
response = connection.post(ConversationMonitors::Configuration.endpoint, body)
|
||||
# Rejected requests do not reach the model, but still consume a monthly call credit.
|
||||
reconcile_usage(usage, body.bytesize, 0) if response.status.between?(400, 499)
|
||||
data = parse_response(response)
|
||||
data = request_decision(body)
|
||||
reconcile_usage(usage, body.bytesize, data.fetch('usage').fetch('input_tokens'))
|
||||
instrument(data, started, monitors.size)
|
||||
data
|
||||
rescue Captain::JevClient::HTTPError => e
|
||||
# Rejected requests do not reach the model, but still consume a monthly call credit.
|
||||
reconcile_usage(usage, body.bytesize, 0) if e.status.between?(400, 499)
|
||||
raise monitor_error(e)
|
||||
rescue JSON::ParserError
|
||||
raise CustomExceptions::MonitorEvaluationError, 'invalid_response'
|
||||
rescue Faraday::Error
|
||||
raise CustomExceptions::MonitorEvaluationError, 'provider_unavailable'
|
||||
end
|
||||
@@ -48,16 +52,12 @@ class ConversationMonitors::JevClient
|
||||
|
||||
def request_body(state, monitors)
|
||||
model = self.class.resolve_model(monitors.first.model)
|
||||
{ model: model, state: state, questions: questions(monitors) }.to_json
|
||||
Captain::JevClient.request_body(model: model, state: state, questions: questions(monitors))
|
||||
end
|
||||
|
||||
def connection
|
||||
Faraday.new do |client|
|
||||
client.options.open_timeout = 3
|
||||
client.options.timeout = 15
|
||||
client.headers['Authorization'] = "Bearer #{ConversationMonitors::Configuration.api_key}"
|
||||
client.headers['Content-Type'] = 'application/json'
|
||||
end
|
||||
def request_decision(body)
|
||||
Captain::JevClient.new(account_id: @account_id, conversation_id: @conversation_id, feature: 'conversation_monitors')
|
||||
.call(body: body) { |response| validate_response!(response) }
|
||||
end
|
||||
|
||||
def questions(monitors)
|
||||
@@ -73,33 +73,24 @@ class ConversationMonitors::JevClient
|
||||
end
|
||||
end
|
||||
|
||||
def parse_response(response)
|
||||
check_status!(response)
|
||||
data = JSON.parse(response.body)
|
||||
def validate_response!(data)
|
||||
valid = data.is_a?(Hash) && data['answers'].is_a?(Hash) && data['model'].is_a?(String) && valid_usage?(data['usage'])
|
||||
raise CustomExceptions::MonitorEvaluationError, 'invalid_response' unless valid
|
||||
|
||||
data
|
||||
rescue JSON::ParserError
|
||||
raise CustomExceptions::MonitorEvaluationError, 'invalid_response'
|
||||
end
|
||||
|
||||
def valid_usage?(usage)
|
||||
usage.is_a?(Hash) && usage['input_tokens'].is_a?(Integer) && usage['input_tokens'] >= 0
|
||||
end
|
||||
|
||||
def check_status!(response)
|
||||
return if response.success?
|
||||
|
||||
code = case response.status
|
||||
def monitor_error(error)
|
||||
code = case error.status
|
||||
when 401, 403 then 'credentials_invalid'
|
||||
when 400, 404, 422 then 'invalid_request'
|
||||
when 402 then 'provider_credits_exhausted'
|
||||
when 429, 529 then 'provider_busy'
|
||||
else 'provider_unavailable'
|
||||
end
|
||||
retry_after = response.headers['retry-after']&.to_i
|
||||
raise CustomExceptions::MonitorEvaluationError.new(code, retry_after: retry_after)
|
||||
CustomExceptions::MonitorEvaluationError.new(code, retry_after: error.retry_after)
|
||||
end
|
||||
|
||||
def instrument(data, started, count)
|
||||
@@ -48,7 +48,7 @@ class ConversationMonitors::Evaluator
|
||||
def evaluate
|
||||
message_limit = ConversationMonitors::Configuration::LIVE_MESSAGE_LIMIT unless @snapshot[:full_history]
|
||||
state = ConversationMonitors::ContextBuilder.new(@work.conversation, message_limit: message_limit).build
|
||||
@monitors.group_by { |monitor| ConversationMonitors::JevClient.resolve_model(monitor.model) }.each_value do |monitors|
|
||||
@monitors.group_by { |monitor| ConversationMonitors::DecisionService.resolve_model(monitor.model) }.each_value do |monitors|
|
||||
monitors.each_slice(QUESTIONS_PER_REQUEST) { |batch| evaluate_batch(state, batch) }
|
||||
end
|
||||
rescue CustomExceptions::MonitorEvaluationError => e
|
||||
@@ -61,9 +61,10 @@ class ConversationMonitors::Evaluator
|
||||
monitors = monitors.select { |monitor| current_monitor?(monitor) }
|
||||
return if monitors.empty?
|
||||
|
||||
response = ConversationMonitors::JevClient.new(account_id: @work.account_id).evaluate(state: state, monitors: monitors)
|
||||
response = ConversationMonitors::DecisionService.new(account_id: @work.account_id, conversation_id: @work.conversation.display_id)
|
||||
.evaluate(state: state, monitors: monitors)
|
||||
monitors.each do |monitor|
|
||||
score = ConversationMonitors::JevClient.score(response['answers'][monitor.id.to_s])
|
||||
score = ConversationMonitors::DecisionService.score(response['answers'][monitor.id.to_s])
|
||||
writer.write(monitor, score: score, model: response['model'])
|
||||
rescue CustomExceptions::MonitorEvaluationError => e
|
||||
record_error([monitor], e)
|
||||
|
||||
@@ -30,4 +30,6 @@ module Integrations::LlmInstrumentationConstants
|
||||
ATTR_LANGFUSE_OBSERVATION_INPUT = 'langfuse.observation.input'
|
||||
ATTR_LANGFUSE_OBSERVATION_OUTPUT = 'langfuse.observation.output'
|
||||
ATTR_LANGFUSE_OBSERVATION_METADATA = 'langfuse.observation.metadata.%s'
|
||||
ATTR_LANGFUSE_OBSERVATION_MODEL_NAME = 'langfuse.observation.model.name'
|
||||
ATTR_LANGFUSE_OBSERVATION_COST_DETAILS = 'langfuse.observation.cost_details'
|
||||
end
|
||||
|
||||
@@ -13,7 +13,7 @@ RSpec.describe 'Monitors API', type: :request do
|
||||
|
||||
before do
|
||||
account.enable_features!('reports', 'conversation_monitors')
|
||||
create(:installation_config, name: 'CAPTAIN_OPENROUTER_API_KEY', value: 'test-key')
|
||||
InstallationConfig.find_or_initialize_by(name: 'CAPTAIN_OPENROUTER_API_KEY').update!(value: 'test-key')
|
||||
end
|
||||
|
||||
after { Redis::Alfred.delete(preview_key) }
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
require 'rails_helper'
|
||||
|
||||
RSpec.describe ConversationMonitors::JevClient do
|
||||
RSpec.describe ConversationMonitors::DecisionService do
|
||||
subject(:evaluate) { client.evaluate(state: state, monitors: [monitor]) }
|
||||
|
||||
let(:account) { create(:account) }
|
||||
@@ -111,6 +111,22 @@ RSpec.describe ConversationMonitors::JevClient do
|
||||
expect(ConversationMonitors::Configuration.model).to eq('typesafe/jev-1.13')
|
||||
end
|
||||
|
||||
it 'records the provider call as a Jev generation when Langfuse is enabled' do
|
||||
span = instance_double(OpenTelemetry::Trace::Span, set_attribute: nil)
|
||||
tracer = instance_double(OpenTelemetry::Trace::Tracer)
|
||||
allow(ChatwootApp).to receive(:otel_enabled?).and_return(true)
|
||||
allow(OpentelemetryConfig).to receive(:tracer).and_return(tracer)
|
||||
allow(tracer).to receive(:in_span).and_yield(span)
|
||||
|
||||
result = client.evaluate(state: state, monitors: [monitor])
|
||||
|
||||
expect(result).to eq(response_body.deep_stringify_keys)
|
||||
expect(tracer).to have_received(:in_span).with('llm.jev.conversation_monitors', kind: :client)
|
||||
expect(span).to have_received(:set_attribute).with('langfuse.observation.input', a_string_including('"model":"typesafe/jev-1.13"'))
|
||||
expect(span).to have_received(:set_attribute).with('gen_ai.usage.input_tokens', 100)
|
||||
expect(WebMock).to have_requested(:post, endpoint).once
|
||||
end
|
||||
|
||||
it 'translates existing monitor model IDs without changing their definition or historical results' do
|
||||
legacy_monitor = create(:conversation_monitor, account: account, model: 'jev-1.13.0')
|
||||
|
||||
@@ -14,7 +14,7 @@ RSpec.describe ConversationMonitors::Evaluator do
|
||||
let(:endpoint) { ConversationMonitors::Configuration.endpoint }
|
||||
|
||||
before do
|
||||
create(:installation_config, name: 'CAPTAIN_OPENROUTER_API_KEY', value: 'test-key')
|
||||
InstallationConfig.find_or_initialize_by(name: 'CAPTAIN_OPENROUTER_API_KEY').update!(value: 'test-key')
|
||||
account.enable_features!('reports', 'conversation_monitors')
|
||||
monitor
|
||||
second
|
||||
|
||||
@@ -7,7 +7,7 @@ RSpec.describe ConversationMonitors::Resume do
|
||||
let(:endpoint) { ConversationMonitors::Configuration.endpoint }
|
||||
|
||||
before do
|
||||
create(:installation_config, name: 'CAPTAIN_OPENROUTER_API_KEY', value: 'test-key')
|
||||
InstallationConfig.find_or_initialize_by(name: 'CAPTAIN_OPENROUTER_API_KEY').update!(value: 'test-key')
|
||||
account.enable_features!('reports', 'conversation_monitors')
|
||||
stub_request(:post, endpoint).to_return do |request|
|
||||
answers = JSON.parse(request.body)['questions'].keys.index_with { { type: 'noul', noul: 0.95 } }
|
||||
|
||||
@@ -7,7 +7,7 @@ RSpec.describe ConversationMonitors::Update do
|
||||
let(:endpoint) { ConversationMonitors::Configuration.endpoint }
|
||||
|
||||
before do
|
||||
create(:installation_config, name: 'CAPTAIN_OPENROUTER_API_KEY', value: 'test-key')
|
||||
InstallationConfig.find_or_initialize_by(name: 'CAPTAIN_OPENROUTER_API_KEY').update!(value: 'test-key')
|
||||
account.enable_features!('reports', 'conversation_monitors')
|
||||
end
|
||||
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
require 'rails_helper'
|
||||
|
||||
RSpec.describe Captain::ConversationClassifierService do
|
||||
let(:conversation) { create(:conversation) }
|
||||
let(:endpoint) { 'https://openrouter.ai/api/v1/systemone' }
|
||||
let(:response) do
|
||||
{ model: 'jev-1.13.0', answers: { priority: { type: 'choice', choice: 'medium', confidence: 0.8 } },
|
||||
usage: { input_tokens: 15, output_tokens: 0 } }
|
||||
end
|
||||
|
||||
before do
|
||||
allow(GlobalConfigService).to receive(:load).with('CAPTAIN_OPENROUTER_DECISION_MODEL_ENDPOINT', nil)
|
||||
.and_return('https://openrouter.ai/api')
|
||||
allow(GlobalConfigService).to receive(:load).with('CAPTAIN_OPENROUTER_API_KEY', nil).and_return('test-key')
|
||||
stub_request(:post, endpoint).to_return(status: 200, body: response.to_json, headers: { 'Content-Type' => 'application/json' })
|
||||
end
|
||||
|
||||
it 'records the classifier request as a Jev generation without changing its answer' do
|
||||
span = instance_double(OpenTelemetry::Trace::Span, set_attribute: nil)
|
||||
tracer = instance_double(OpenTelemetry::Trace::Tracer)
|
||||
allow(ChatwootApp).to receive(:otel_enabled?).and_return(true)
|
||||
allow(OpentelemetryConfig).to receive(:tracer).and_return(tracer)
|
||||
allow(tracer).to receive(:in_span).and_yield(span)
|
||||
|
||||
result = described_class.new(conversation: conversation).priority
|
||||
|
||||
expect(result).to eq(priority: 'medium', confidence: 0.8)
|
||||
expect(tracer).to have_received(:in_span).with('llm.jev.captain_classifier', kind: :client)
|
||||
expect(span).to have_received(:set_attribute).with('langfuse.session.id', "#{conversation.account_id}_#{conversation.display_id}")
|
||||
expect(span).to have_received(:set_attribute).with('gen_ai.usage.input_tokens', 15)
|
||||
expect(WebMock).to have_requested(:post, endpoint).once
|
||||
end
|
||||
|
||||
it 'preserves the classifier error on an unsuccessful Jev response' do
|
||||
stub_request(:post, endpoint).to_return(status: 401, body: 'invalid key')
|
||||
|
||||
expect { described_class.new(conversation: conversation).priority }
|
||||
.to raise_error(described_class::Error, 'Jev request failed with status 401: invalid key')
|
||||
expect(WebMock).to have_requested(:post, endpoint).once
|
||||
end
|
||||
end
|
||||
114
spec/services/captain/jev_client_spec.rb
Normal file
114
spec/services/captain/jev_client_spec.rb
Normal file
@@ -0,0 +1,114 @@
|
||||
require 'rails_helper'
|
||||
|
||||
RSpec.describe Captain::JevClient do
|
||||
let(:body) { described_class.request_body(model: 'jev-latest', state: { messages: [] }, questions: { test: { type: 'noul' } }) }
|
||||
let(:client) { described_class.new(account_id: 7, feature: 'captain_classifier') }
|
||||
|
||||
before do
|
||||
allow(ChatwootApp).to receive(:otel_enabled?).and_return(false)
|
||||
allow(GlobalConfigService).to receive(:load).with('CAPTAIN_OPENROUTER_API_KEY', nil).and_return('test-key')
|
||||
allow(GlobalConfigService).to receive(:load).with('CAPTAIN_OPENROUTER_DECISION_MODEL_ENDPOINT', nil).and_return(nil)
|
||||
end
|
||||
|
||||
it 'uses the shared default endpoint and returns parsed System One data' do
|
||||
response = { model: 'jev-1.13.0', answers: { test: { type: 'noul', noul: 0.9 } } }
|
||||
stub_request(:post, 'https://openrouter.ai/api/v1/systemone').to_return(status: 200, body: response.to_json)
|
||||
|
||||
expect(client.call(body: body)).to eq(response.deep_stringify_keys)
|
||||
expect(WebMock).to have_requested(:post, 'https://openrouter.ai/api/v1/systemone').with { |request|
|
||||
request.headers['Authorization'] == 'Bearer test-key' && JSON.parse(request.body)['model'] == 'jev-latest'
|
||||
}.once
|
||||
end
|
||||
|
||||
it 'exposes provider status and retry delay for monitor error handling' do
|
||||
stub_request(:post, 'https://openrouter.ai/api/v1/systemone')
|
||||
.to_return(status: 429, body: 'busy', headers: { 'Retry-After' => '12' })
|
||||
|
||||
expect { client.call(body: body) }.to raise_error(described_class::HTTPError) { |error|
|
||||
expect(error.status).to eq(429)
|
||||
expect(error.retry_after).to eq(12)
|
||||
}
|
||||
end
|
||||
|
||||
it 'records the exact request, response, model, usage, and cost in Langfuse' do
|
||||
response = { model: 'jev-1.13.0', answers: { test: { type: 'noul', noul: 0.9 } },
|
||||
usage: { input_tokens: 12, output_tokens: 0, cost: 0.000001 } }
|
||||
stub_request(:post, 'https://openrouter.ai/api/v1/systemone').to_return(status: 200, body: response.to_json)
|
||||
attributes = {}
|
||||
span = instance_double(OpenTelemetry::Trace::Span)
|
||||
tracer = instance_double(OpenTelemetry::Trace::Tracer)
|
||||
allow(span).to receive(:set_attribute) { |name, value| attributes[name] = value }
|
||||
allow(ChatwootApp).to receive(:otel_enabled?).and_return(true)
|
||||
allow(OpentelemetryConfig).to receive(:tracer).and_return(tracer)
|
||||
allow(tracer).to receive(:in_span).and_yield(span)
|
||||
|
||||
expect(described_class.new(account_id: 7, conversation_id: 42, feature: 'captain_classifier').call(body: body))
|
||||
.to eq(response.deep_stringify_keys)
|
||||
expect(tracer).to have_received(:in_span).with('llm.jev.captain_classifier', kind: :client)
|
||||
expect(attributes).to include(
|
||||
'langfuse.observation.input' => body,
|
||||
'langfuse.observation.output' => response.to_json,
|
||||
'langfuse.observation.model.name' => 'jev-1.13.0',
|
||||
'langfuse.observation.cost_details' => { total: 0.000001 }.to_json,
|
||||
'langfuse.session.id' => '7_42',
|
||||
'gen_ai.usage.input_tokens' => 12
|
||||
)
|
||||
expect(WebMock).to have_requested(:post, 'https://openrouter.ai/api/v1/systemone').with(body: body).once
|
||||
end
|
||||
|
||||
it 'marks a failed Jev call without retrying it' do
|
||||
stub_request(:post, 'https://openrouter.ai/api/v1/systemone').to_timeout
|
||||
attributes = {}
|
||||
span = instance_double(OpenTelemetry::Trace::Span, 'status=': nil)
|
||||
allow(span).to receive(:set_attribute) { |name, value| attributes[name] = value }
|
||||
tracer = instance_double(OpenTelemetry::Trace::Tracer)
|
||||
allow(ChatwootApp).to receive(:otel_enabled?).and_return(true)
|
||||
allow(OpentelemetryConfig).to receive(:tracer).and_return(tracer)
|
||||
allow(tracer).to receive(:in_span).and_yield(span)
|
||||
|
||||
expect { client.call(body: body) }.to raise_error(Faraday::ConnectionFailed)
|
||||
expect(span).to have_received(:status=).with(an_instance_of(OpenTelemetry::Trace::Status))
|
||||
expect(WebMock).to have_requested(:post, 'https://openrouter.ai/api/v1/systemone').once
|
||||
end
|
||||
|
||||
context 'when tracing fails' do
|
||||
let(:response) { { model: 'jev-1.13.0', answers: { test: { type: 'noul', noul: 0.9 } } } }
|
||||
let(:tracer) { instance_double(OpenTelemetry::Trace::Tracer) }
|
||||
let(:exception_tracker) { instance_double(ChatwootExceptionTracker, capture_exception: nil) }
|
||||
|
||||
before do
|
||||
stub_request(:post, 'https://openrouter.ai/api/v1/systemone').to_return(status: 200, body: response.to_json)
|
||||
allow(ChatwootApp).to receive(:otel_enabled?).and_return(true)
|
||||
allow(OpentelemetryConfig).to receive(:tracer).and_return(tracer)
|
||||
allow(ChatwootExceptionTracker).to receive(:new).and_return(exception_tracker)
|
||||
end
|
||||
|
||||
it 'keeps the provider response when recording it fails' do
|
||||
span = instance_double(OpenTelemetry::Trace::Span)
|
||||
allow(span).to receive(:set_attribute) { |name, _| raise 'exporter down' if name == 'langfuse.observation.output' }
|
||||
allow(tracer).to receive(:in_span).and_yield(span)
|
||||
|
||||
expect(client.call(body: body)).to eq(response.deep_stringify_keys)
|
||||
expect(exception_tracker).to have_received(:capture_exception)
|
||||
expect(WebMock).to have_requested(:post, 'https://openrouter.ai/api/v1/systemone').once
|
||||
end
|
||||
|
||||
it 'raises the provider error once when recording it fails' do
|
||||
stub_request(:post, 'https://openrouter.ai/api/v1/systemone').to_return(status: 429, body: 'busy')
|
||||
span = instance_double(OpenTelemetry::Trace::Span, set_attribute: nil)
|
||||
allow(span).to receive(:status=).and_raise('exporter down')
|
||||
allow(tracer).to receive(:in_span).and_yield(span)
|
||||
|
||||
expect { client.call(body: body) }.to raise_error(described_class::HTTPError)
|
||||
expect(WebMock).to have_requested(:post, 'https://openrouter.ai/api/v1/systemone').once
|
||||
end
|
||||
|
||||
it 'still calls Jev when the span cannot start' do
|
||||
allow(tracer).to receive(:in_span).and_raise('tracer down')
|
||||
|
||||
expect(client.call(body: body)).to eq(response.deep_stringify_keys)
|
||||
expect(exception_tracker).to have_received(:capture_exception)
|
||||
expect(WebMock).to have_requested(:post, 'https://openrouter.ai/api/v1/systemone').once
|
||||
end
|
||||
end
|
||||
end
|
||||
Reference in New Issue
Block a user