diff --git a/Gemfile b/Gemfile index 2023c32b1..e6c5a5250 100644 --- a/Gemfile +++ b/Gemfile @@ -191,7 +191,7 @@ gem 'reverse_markdown' gem 'iso-639' gem 'ruby-openai' -gem 'ai-agents', '>= 0.7.0' +gem 'ai-agents' # TODO: Move this gem as a dependency of ai-agents gem 'ruby_llm', '>= 1.8.2' diff --git a/Gemfile.lock b/Gemfile.lock index db9b59c66..cc1f9e253 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -126,8 +126,8 @@ GEM jbuilder (~> 2) rails (>= 4.2, < 7.2) selectize-rails (~> 0.6) - ai-agents (0.7.0) - ruby_llm (~> 1.8.2) + ai-agents (0.9.0) + ruby_llm (~> 1.9.1) annotaterb (4.20.0) activerecord (>= 6.0.0) activesupport (>= 6.0.0) @@ -314,7 +314,7 @@ GEM faraday-net_http_persistent (2.1.0) faraday (~> 2.5) net-http-persistent (~> 4.0) - faraday-retry (2.2.1) + faraday-retry (2.4.0) faraday (~> 2.0) faraday_middleware-aws-sigv4 (1.0.1) aws-sigv4 (~> 1.0) @@ -540,7 +540,7 @@ GEM net-imap net-pop net-smtp - marcel (1.0.4) + marcel (1.1.0) maxminddb (0.1.22) meta_request (0.8.5) rack-contrib (>= 1.1, < 3) @@ -559,7 +559,7 @@ GEM multi_json (1.15.0) multi_xml (0.8.0) bigdecimal (>= 3.1, < 5) - multipart-post (2.3.0) + multipart-post (2.4.1) mutex_m (0.3.0) neighbor (0.2.3) activerecord (>= 5.2) @@ -825,7 +825,7 @@ GEM ruby2ruby (2.5.0) ruby_parser (~> 3.1) sexp_processor (~> 4.6) - ruby_llm (1.8.2) + ruby_llm (1.9.2) base64 event_stream_parser (~> 1) faraday (>= 1.10.0) @@ -1004,7 +1004,7 @@ GEM working_hours (1.4.1) activesupport (>= 3.2) tzinfo - zeitwerk (2.6.17) + zeitwerk (2.7.4) PLATFORMS arm64-darwin-20 @@ -1024,7 +1024,7 @@ DEPENDENCIES administrate (>= 0.20.1) administrate-field-active_storage (>= 1.0.3) administrate-field-belongs_to_search (>= 0.9.0) - ai-agents (>= 0.7.0) + ai-agents annotaterb attr_extras audited (~> 5.4, >= 5.4.1) diff --git a/app/javascript/dashboard/components-next/NewConversation/ComposeConversation.vue b/app/javascript/dashboard/components-next/NewConversation/ComposeConversation.vue index ee71bf51a..8e24f3d50 100644 --- a/app/javascript/dashboard/components-next/NewConversation/ComposeConversation.vue +++ b/app/javascript/dashboard/components-next/NewConversation/ComposeConversation.vue @@ -1,5 +1,5 @@ diff --git a/app/javascript/dashboard/composables/captain/constants.js b/app/javascript/dashboard/composables/captain/constants.js new file mode 100644 index 000000000..98629a94f --- /dev/null +++ b/app/javascript/dashboard/composables/captain/constants.js @@ -0,0 +1,12 @@ +export const CAPTAIN_ERROR_TYPES = Object.freeze({ + ABORTED: 'aborted', + API_ERROR: 'api_error', + HTTP_PREFIX: 'http_', + ABORT_ERROR: 'AbortError', + CANCELED_ERROR: 'CanceledError', +}); + +export const CAPTAIN_GENERATION_FAILURE_REASONS = Object.freeze({ + EMPTY_RESPONSE: 'empty_response', + EXCEPTION: 'exception', +}); diff --git a/app/javascript/dashboard/composables/chatlist/useBulkActions.js b/app/javascript/dashboard/composables/chatlist/useBulkActions.js index a32c4c512..45421b978 100644 --- a/app/javascript/dashboard/composables/chatlist/useBulkActions.js +++ b/app/javascript/dashboard/composables/chatlist/useBulkActions.js @@ -102,6 +102,28 @@ export function useBulkActions() { } } + // Only used in context menu + async function onRemoveLabels(labelsToRemove, conversationId = null) { + try { + await store.dispatch('bulkActions/process', { + type: 'Conversation', + ids: conversationId || selectedConversations.value, + labels: { + remove: labelsToRemove, + }, + }); + + useAlert( + t('CONVERSATION.CARD_CONTEXT_MENU.API.LABEL_REMOVAL.SUCCESFUL', { + labelName: labelsToRemove[0], + conversationId, + }) + ); + } catch (err) { + useAlert(t('CONVERSATION.CARD_CONTEXT_MENU.API.LABEL_REMOVAL.FAILED')); + } + } + async function onAssignTeamsForBulk(team) { try { await store.dispatch('bulkActions/process', { @@ -189,6 +211,7 @@ export function useBulkActions() { isConversationSelected, onAssignAgent, onAssignLabels, + onRemoveLabels, onAssignTeamsForBulk, onUpdateConversations, }; diff --git a/app/javascript/dashboard/composables/useCaptain.js b/app/javascript/dashboard/composables/useCaptain.js index f7c72e04f..b9648fdd0 100644 --- a/app/javascript/dashboard/composables/useCaptain.js +++ b/app/javascript/dashboard/composables/useCaptain.js @@ -11,6 +11,7 @@ import { useAlert } from 'dashboard/composables'; import { useI18n } from 'vue-i18n'; import { FEATURE_FLAGS } from 'dashboard/featureFlags'; import TasksAPI from 'dashboard/api/captain/tasks'; +import { CAPTAIN_ERROR_TYPES } from 'dashboard/composables/captain/constants'; export function useCaptain() { const store = useStore(); @@ -69,7 +70,10 @@ export function useCaptain() { * @param {Error} error - The error object from the API call. */ const handleAPIError = error => { - if (error.name === 'AbortError' || error.name === 'CanceledError') { + if ( + error.name === CAPTAIN_ERROR_TYPES.ABORT_ERROR || + error.name === CAPTAIN_ERROR_TYPES.CANCELED_ERROR + ) { return; } const errorMessage = @@ -78,6 +82,24 @@ export function useCaptain() { useAlert(errorMessage); }; + /** + * Classifies API error types for downstream analytics. + * @param {Error} error + * @returns {string} + */ + const getErrorType = error => { + if ( + error.name === CAPTAIN_ERROR_TYPES.ABORT_ERROR || + error.name === CAPTAIN_ERROR_TYPES.CANCELED_ERROR + ) { + return CAPTAIN_ERROR_TYPES.ABORTED; + } + if (error.response?.status) { + return `${CAPTAIN_ERROR_TYPES.HTTP_PREFIX}${error.response.status}`; + } + return CAPTAIN_ERROR_TYPES.API_ERROR; + }; + // === Task Methods === /** * Rewrites content with a specific operation. @@ -103,7 +125,7 @@ export function useCaptain() { return { message: generatedMessage, followUpContext }; } catch (error) { handleAPIError(error); - return { message: '' }; + return { message: '', errorType: getErrorType(error) }; } }; @@ -125,7 +147,7 @@ export function useCaptain() { return { message: generatedMessage, followUpContext }; } catch (error) { handleAPIError(error); - return { message: '' }; + return { message: '', errorType: getErrorType(error) }; } }; @@ -147,7 +169,7 @@ export function useCaptain() { return { message: generatedMessage, followUpContext }; } catch (error) { handleAPIError(error); - return { message: '' }; + return { message: '', errorType: getErrorType(error) }; } }; @@ -171,7 +193,11 @@ export function useCaptain() { return { message: generatedMessage, followUpContext: updatedContext }; } catch (error) { handleAPIError(error); - return { message: '', followUpContext }; + return { + message: '', + followUpContext, + errorType: getErrorType(error), + }; } }; diff --git a/app/javascript/dashboard/composables/useCopilotReply.js b/app/javascript/dashboard/composables/useCopilotReply.js index 492bcb43e..42f506b16 100644 --- a/app/javascript/dashboard/composables/useCopilotReply.js +++ b/app/javascript/dashboard/composables/useCopilotReply.js @@ -3,6 +3,10 @@ import { useCaptain } from 'dashboard/composables/useCaptain'; import { useUISettings } from 'dashboard/composables/useUISettings'; import { useTrack } from 'dashboard/composables'; import { CAPTAIN_EVENTS } from 'dashboard/helper/AnalyticsHelper/events'; +import { + CAPTAIN_ERROR_TYPES, + CAPTAIN_GENERATION_FAILURE_REASONS, +} from 'dashboard/composables/captain/constants'; // Actions that map to REWRITE events (with operation attribute) const REWRITE_ACTIONS = [ @@ -52,6 +56,20 @@ function buildPayload(action, conversationId, followUpCount = undefined) { return payload; } +function trackGenerationFailure({ + action, + conversationId, + followUpCount = undefined, + stage, + reason, +}) { + useTrack(CAPTAIN_EVENTS.GENERATION_FAILED, { + ...buildPayload(action, conversationId, followUpCount), + stage, + reason, + }); +} + /** * Composable for managing Copilot reply generation state and actions. * Extracts copilot-related logic from ReplyBox for cleaner code organization. @@ -146,7 +164,8 @@ export function useCopilotReply() { // Reset without tracking dismiss (starting new action) reset(false); - abortController.value = new AbortController(); + const requestController = new AbortController(); + abortController.value = requestController; isGenerating.value = true; isContentReady.value = false; currentAction.value = action; @@ -154,28 +173,66 @@ export function useCopilotReply() { trackedConversationId.value = conversationId.value; try { - const { message: content, followUpContext: newContext } = - await processEvent(action, data, { - signal: abortController.value.signal, - }); + const { + message: content, + followUpContext: newContext, + errorType, + } = await processEvent(action, data, { + signal: requestController.signal, + }); - if (!abortController.value?.signal.aborted) { - generatedContent.value = content; - followUpContext.value = newContext; - if (content) { - showEditor.value = true; - // Track "Used" event on successful generation - const eventKey = `${getEventPrefix(action)}_USED`; - useTrack( - CAPTAIN_EVENTS[eventKey], - buildPayload(action, trackedConversationId.value) - ); + if (requestController.signal.aborted) return; + if (errorType === CAPTAIN_ERROR_TYPES.ABORTED) { + if (abortController.value === requestController) { + isGenerating.value = false; } - isGenerating.value = false; + return; } - } catch { - if (!abortController.value?.signal.aborted) { - isGenerating.value = false; + + generatedContent.value = content; + followUpContext.value = newContext; + if (content) { + showEditor.value = true; + // Track "Used" event on successful generation + const eventKey = `${getEventPrefix(action)}_USED`; + useTrack( + CAPTAIN_EVENTS[eventKey], + buildPayload(action, trackedConversationId.value) + ); + } else if (errorType && errorType !== CAPTAIN_ERROR_TYPES.ABORTED) { + trackGenerationFailure({ + action, + conversationId: trackedConversationId.value, + stage: 'initial', + reason: errorType, + }); + } else { + trackGenerationFailure({ + action, + conversationId: trackedConversationId.value, + stage: 'initial', + reason: CAPTAIN_GENERATION_FAILURE_REASONS.EMPTY_RESPONSE, + }); + } + isGenerating.value = false; + } catch (error) { + if ( + requestController.signal.aborted || + error?.name === CAPTAIN_ERROR_TYPES.ABORT_ERROR || + error?.name === CAPTAIN_ERROR_TYPES.CANCELED_ERROR + ) { + return; + } + trackGenerationFailure({ + action, + conversationId: trackedConversationId.value, + stage: 'initial', + reason: error?.name || CAPTAIN_GENERATION_FAILURE_REASONS.EXCEPTION, + }); + isGenerating.value = false; + } finally { + if (abortController.value === requestController) { + abortController.value = null; } } } @@ -187,7 +244,8 @@ export function useCopilotReply() { async function sendFollowUp(message) { if (!followUpContext.value || !message.trim()) return; - abortController.value = new AbortController(); + const requestController = new AbortController(); + abortController.value = requestController; isGenerating.value = true; isContentReady.value = false; @@ -198,24 +256,65 @@ export function useCopilotReply() { followUpCount.value += 1; try { - const { message: content, followUpContext: updatedContext } = - await followUp({ - followUpContext: followUpContext.value, - message, - signal: abortController.value.signal, - }); + const { + message: content, + followUpContext: updatedContext, + errorType, + } = await followUp({ + followUpContext: followUpContext.value, + message, + signal: requestController.signal, + }); - if (!abortController.value?.signal.aborted) { - if (content) { - generatedContent.value = content; - followUpContext.value = updatedContext; - showEditor.value = true; + if (requestController.signal.aborted) return; + if (errorType === CAPTAIN_ERROR_TYPES.ABORTED) { + if (abortController.value === requestController) { + isGenerating.value = false; } - isGenerating.value = false; + return; } - } catch { - if (!abortController.value?.signal.aborted) { - isGenerating.value = false; + + if (content) { + generatedContent.value = content; + followUpContext.value = updatedContext; + showEditor.value = true; + } else if (errorType && errorType !== CAPTAIN_ERROR_TYPES.ABORTED) { + trackGenerationFailure({ + action: currentAction.value, + conversationId: trackedConversationId.value, + followUpCount: followUpCount.value, + stage: 'follow_up', + reason: errorType, + }); + } else { + trackGenerationFailure({ + action: currentAction.value, + conversationId: trackedConversationId.value, + followUpCount: followUpCount.value, + stage: 'follow_up', + reason: CAPTAIN_GENERATION_FAILURE_REASONS.EMPTY_RESPONSE, + }); + } + isGenerating.value = false; + } catch (error) { + if ( + requestController.signal.aborted || + error?.name === CAPTAIN_ERROR_TYPES.ABORT_ERROR || + error?.name === CAPTAIN_ERROR_TYPES.CANCELED_ERROR + ) { + return; + } + trackGenerationFailure({ + action: currentAction.value, + conversationId: trackedConversationId.value, + followUpCount: followUpCount.value, + stage: 'follow_up', + reason: error?.name || CAPTAIN_GENERATION_FAILURE_REASONS.EXCEPTION, + }); + isGenerating.value = false; + } finally { + if (abortController.value === requestController) { + abortController.value = null; } } } diff --git a/app/javascript/dashboard/helper/AnalyticsHelper/events.js b/app/javascript/dashboard/helper/AnalyticsHelper/events.js index 0b6e85d77..c9fefb129 100644 --- a/app/javascript/dashboard/helper/AnalyticsHelper/events.js +++ b/app/javascript/dashboard/helper/AnalyticsHelper/events.js @@ -85,6 +85,11 @@ export const PORTALS_EVENTS = Object.freeze({ }); export const CAPTAIN_EVENTS = Object.freeze({ + // Editor funnel events + EDITOR_AI_MENU_OPENED: 'Captain: Editor AI menu opened', + GENERATION_FAILED: 'Captain: Generation failed', + AI_ASSISTED_MESSAGE_SENT: 'Captain: AI-assisted message sent', + // Rewrite events (with operation attribute in payload) REWRITE_USED: 'Captain: Rewrite used', REWRITE_APPLIED: 'Captain: Rewrite applied', diff --git a/app/javascript/dashboard/i18n/locale/en/conversation.json b/app/javascript/dashboard/i18n/locale/en/conversation.json index 99e0bb072..951a46993 100644 --- a/app/javascript/dashboard/i18n/locale/en/conversation.json +++ b/app/javascript/dashboard/i18n/locale/en/conversation.json @@ -174,6 +174,10 @@ "SUCCESFUL": "Assigned label #{labelName} to conversation id {conversationId}", "FAILED": "Couldn't assign label. Please try again." }, + "LABEL_REMOVAL": { + "SUCCESFUL": "Removed label #{labelName} from conversation id {conversationId}", + "FAILED": "Couldn't remove label. Please try again." + }, "TEAM_ASSIGNMENT": { "SUCCESFUL": "Assigned team \"{team}\" to conversation id {conversationId}", "FAILED": "Couldn't assign team. Please try again." @@ -186,6 +190,8 @@ "DISABLE_SIGN_TOOLTIP": "Disable signature", "MSG_INPUT": "Shift + enter for new line. Start with '/' to select a Canned Response.", "PRIVATE_MSG_INPUT": "Shift + enter for new line. This will be visible only to Agents", + "MESSAGING_RESTRICTED": "You cannot reply to this conversation", + "MESSAGING_RESTRICTED_WHATSAPP": "You can only reply using a template message due to 24-hour message window restriction", "MESSAGE_SIGNATURE_NOT_CONFIGURED": "Message signature is not configured, please configure it in profile settings.", "COPILOT_MSG_INPUT": "Give copilot additional prompts, or ask anything else... Press enter to send follow-up", "CLICK_HERE": "Click here to update", diff --git a/app/services/whatsapp/incoming_message_base_service.rb b/app/services/whatsapp/incoming_message_base_service.rb index a90814bdc..a8ad176b6 100644 --- a/app/services/whatsapp/incoming_message_base_service.rb +++ b/app/services/whatsapp/incoming_message_base_service.rb @@ -28,19 +28,19 @@ class Whatsapp::IncomingMessageBaseService # if the webhook event is a reaction or an ephermal message or an unsupported message. return if unprocessable_message_type?(message_type) - # Multiple webhook event can be received against the same message due to misconfigurations in the Meta - # business manager account. While we have not found the core reason yet, the following line ensure that - # there are no duplicate messages created. - return if find_message_by_source_id(messages_data.first[:id]) || message_under_process? + # Multiple webhook events can be received for the same message due to + # misconfigurations in the Meta business manager account. + # We use an atomic Redis SET NX to prevent concurrent workers from both + # processing the same message simultaneously. + return if find_message_by_source_id(messages_data.first[:id]) + return unless lock_message_source_id! - cache_message_source_id_in_redis set_contact return unless @contact ActiveRecord::Base.transaction do set_conversation create_messages - clear_message_source_id_from_redis end end diff --git a/app/services/whatsapp/incoming_message_service_helpers.rb b/app/services/whatsapp/incoming_message_service_helpers.rb index c803d61bd..bac6f6222 100644 --- a/app/services/whatsapp/incoming_message_service_helpers.rb +++ b/app/services/whatsapp/incoming_message_service_helpers.rb @@ -69,20 +69,9 @@ module Whatsapp::IncomingMessageServiceHelpers @message = Message.find_by(source_id: source_id) end - def message_under_process? - key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: messages_data.first[:id]) - Redis::Alfred.get(key) - end + def lock_message_source_id! + return false if messages_data.blank? - def cache_message_source_id_in_redis - return if messages_data.blank? - - key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: messages_data.first[:id]) - ::Redis::Alfred.setex(key, true) - end - - def clear_message_source_id_from_redis - key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: messages_data.first[:id]) - ::Redis::Alfred.delete(key) + Whatsapp::MessageDedupLock.new(messages_data.first[:id]).acquire! end end diff --git a/app/services/whatsapp/message_dedup_lock.rb b/app/services/whatsapp/message_dedup_lock.rb new file mode 100644 index 000000000..4a863a4a4 --- /dev/null +++ b/app/services/whatsapp/message_dedup_lock.rb @@ -0,0 +1,19 @@ +# Atomic dedup lock for WhatsApp incoming messages. +# +# Meta can deliver the same webhook event multiple times. This lock uses +# Redis SET NX EX to ensure only one worker processes a given source_id. +class Whatsapp::MessageDedupLock + KEY_PREFIX = Redis::RedisKeys::MESSAGE_SOURCE_KEY + DEFAULT_TTL = 1.day.to_i + + def initialize(source_id, ttl: DEFAULT_TTL) + @key = format(KEY_PREFIX, id: source_id) + @ttl = ttl + end + + # Returns true when the lock is acquired (caller should proceed). + # Returns false when another worker already holds the lock. + def acquire! + ::Redis::Alfred.set(@key, true, nx: true, ex: @ttl) + end +end diff --git a/enterprise/app/helpers/captain/chat_response_helper.rb b/enterprise/app/helpers/captain/chat_response_helper.rb index e0323996f..bfb8adc11 100644 --- a/enterprise/app/helpers/captain/chat_response_helper.rb +++ b/enterprise/app/helpers/captain/chat_response_helper.rb @@ -1,10 +1,13 @@ module Captain::ChatResponseHelper + include Integrations::LlmInstrumentationConstants + private def build_response(response) Rails.logger.debug { "#{self.class.name} Assistant: #{@assistant.id}, Received response #{response}" } parsed = parse_json_response(response.content) + apply_credit_usage_metadata(parsed) persist_message(parsed, 'assistant') parsed @@ -19,6 +22,26 @@ module Captain::ChatResponseHelper { 'content' => content } end + def apply_credit_usage_metadata(parsed_response) + return unless captain_v1_assistant? + + OpenTelemetry::Trace.current_span.set_attribute( + format(ATTR_LANGFUSE_METADATA, 'credit_used'), + credit_used_for_response?(parsed_response).to_s + ) + rescue StandardError => e + Rails.logger.warn "#{self.class.name} Assistant: #{@assistant.id}, Failed to set credit usage metadata: #{e.message}" + end + + def credit_used_for_response?(parsed_response) + response = parsed_response['response'] + response.present? && response != 'conversation_handoff' + end + + def captain_v1_assistant? + feature_name == 'assistant' && !@assistant.account.feature_enabled?('captain_integration_v2') + end + def persist_thinking_message(tool_call) return if @copilot_thread.blank? diff --git a/enterprise/app/jobs/messages/audio_transcription_job.rb b/enterprise/app/jobs/messages/audio_transcription_job.rb index ce35405c8..5daf1e160 100644 --- a/enterprise/app/jobs/messages/audio_transcription_job.rb +++ b/enterprise/app/jobs/messages/audio_transcription_job.rb @@ -1,6 +1,15 @@ class Messages::AudioTranscriptionJob < ApplicationJob queue_as :low + discard_on Faraday::BadRequestError do |job, error| + log_context = { + attachment_id: job.arguments.first, + job_id: job.job_id, + status_code: error.response&.dig(:status) + } + + Rails.logger.warn("Discarding audio transcription job due to bad request: #{log_context}") + end retry_on ActiveStorage::FileNotFoundError, wait: 2.seconds, attempts: 3 def perform(attachment_id) diff --git a/enterprise/app/services/captain/assistant/agent_runner_service.rb b/enterprise/app/services/captain/assistant/agent_runner_service.rb index 9c4e56841..a40e096e6 100644 --- a/enterprise/app/services/captain/assistant/agent_runner_service.rb +++ b/enterprise/app/services/captain/assistant/agent_runner_service.rb @@ -1,6 +1,9 @@ require 'agents' +require 'agents/instrumentation' class Captain::Assistant::AgentRunnerService + include Integrations::LlmInstrumentationConstants + CONVERSATION_STATE_ATTRIBUTES = %i[ id display_id inbox_id contact_id status priority label_list custom_attributes additional_attributes @@ -22,7 +25,9 @@ class Captain::Assistant::AgentRunnerService context = build_context(message_history) message_to_process = extract_last_user_message(message_history) runner = Agents::Runner.with_agents(*agents) + runner = add_usage_metadata_callback(runner) runner = add_callbacks_to_runner(runner) if @callbacks.any? + install_instrumentation(runner) result = runner.run(message_to_process, context: context, max_turns: 100) process_agent_result(result) @@ -50,6 +55,7 @@ class Captain::Assistant::AgentRunnerService end { + session_id: "#{@assistant.account_id}_#{@conversation&.display_id}", conversation_history: conversation_history, state: build_state } @@ -124,6 +130,31 @@ class Captain::Assistant::AgentRunnerService [assistant_agent] + scenario_agents end + def install_instrumentation(runner) + return unless ChatwootApp.otel_enabled? + + Agents::Instrumentation.install( + runner, + tracer: OpentelemetryConfig.tracer, + trace_name: 'llm.captain_v2', + span_attributes: { + ATTR_LANGFUSE_TAGS => ['captain_v2'].to_json + }, + attribute_provider: ->(context_wrapper) { dynamic_trace_attributes(context_wrapper) } + ) + end + + def dynamic_trace_attributes(context_wrapper) + state = context_wrapper&.context&.dig(:state) || {} + conversation = state[:conversation] || {} + { + ATTR_LANGFUSE_USER_ID => state[:account_id], + format(ATTR_LANGFUSE_METADATA, 'assistant_id') => state[:assistant_id], + format(ATTR_LANGFUSE_METADATA, 'conversation_id') => conversation[:id], + format(ATTR_LANGFUSE_METADATA, 'conversation_display_id') => conversation[:display_id] + }.compact.transform_values(&:to_s) + end + def add_callbacks_to_runner(runner) runner = add_agent_thinking_callback(runner) if @callbacks[:on_agent_thinking] runner = add_tool_start_callback(runner) if @callbacks[:on_tool_start] @@ -132,6 +163,36 @@ class Captain::Assistant::AgentRunnerService runner end + def add_usage_metadata_callback(runner) + return runner unless ChatwootApp.otel_enabled? + + handoff_tool_name = Captain::Tools::HandoffTool.new(@assistant).name + + runner.on_tool_complete do |tool_name, _tool_result, context_wrapper| + track_handoff_usage(tool_name, handoff_tool_name, context_wrapper) + end + + runner.on_run_complete do |_agent_name, _result, context_wrapper| + write_credits_used_metadata(context_wrapper) + end + runner + end + + def track_handoff_usage(tool_name, handoff_tool_name, context_wrapper) + return unless context_wrapper&.context + return unless tool_name.to_s == handoff_tool_name + + context_wrapper.context[:captain_v2_handoff_tool_called] = true + end + + def write_credits_used_metadata(context_wrapper) + root_span = context_wrapper&.context&.dig(:__otel_tracing, :root_span) + return unless root_span + + credits_used = !context_wrapper.context[:captain_v2_handoff_tool_called] + root_span.set_attribute(format(ATTR_LANGFUSE_METADATA, 'credits_used'), credits_used) + end + def add_agent_thinking_callback(runner) runner.on_agent_thinking do |*args| @callbacks[:on_agent_thinking].call(*args) diff --git a/enterprise/app/services/messages/audio_transcription_service.rb b/enterprise/app/services/messages/audio_transcription_service.rb index 1676dd862..4aa156f47 100644 --- a/enterprise/app/services/messages/audio_transcription_service.rb +++ b/enterprise/app/services/messages/audio_transcription_service.rb @@ -31,12 +31,20 @@ class Messages::AudioTranscriptionService< Llm::LegacyBaseOpenAiService end def fetch_audio_file + blob = attachment.file.blob temp_dir = Rails.root.join('tmp/uploads/audio-transcriptions') FileUtils.mkdir_p(temp_dir) - temp_file_path = File.join(temp_dir, "#{attachment.file.blob.key}-#{attachment.file.filename}") + temp_file_name = "#{blob.key}-#{blob.filename}" + + if blob.filename.extension_without_delimiter.blank? + extension = extension_from_content_type(blob.content_type) + temp_file_name = "#{temp_file_name}.#{extension}" if extension.present? + end + + temp_file_path = File.join(temp_dir, temp_file_name) File.open(temp_file_path, 'wb') do |file| - attachment.file.blob.open do |blob_file| + blob.open do |blob_file| IO.copy_stream(blob_file, file) end end @@ -49,13 +57,12 @@ class Messages::AudioTranscriptionService< Llm::LegacyBaseOpenAiService return transcribed_text if transcribed_text.present? temp_file_path = fetch_audio_file - transcribed_text = nil File.open(temp_file_path, 'rb') do |file| response = @client.audio.transcribe( parameters: { - model: 'whisper-1', + model: WHISPER_MODEL, file: file, temperature: 0.4 } @@ -63,10 +70,10 @@ class Messages::AudioTranscriptionService< Llm::LegacyBaseOpenAiService transcribed_text = response['text'] end - FileUtils.rm_f(temp_file_path) - update_transcription(transcribed_text) transcribed_text + ensure + FileUtils.rm_f(temp_file_path) if temp_file_path.present? end def instrumentation_params(file_path) @@ -90,4 +97,15 @@ class Messages::AudioTranscriptionService< Llm::LegacyBaseOpenAiService message.reindex end + + def extension_from_content_type(content_type) + subtype = content_type.to_s.downcase.split(';').first.to_s.split('/').last.to_s + return if subtype.blank? + + { + 'x-m4a' => 'm4a', + 'x-wav' => 'wav', + 'x-mp3' => 'mp3' + }.fetch(subtype, subtype) + end end diff --git a/lib/captain/tool_instrumentation.rb b/lib/captain/tool_instrumentation.rb index af79a3fca..157aab829 100644 --- a/lib/captain/tool_instrumentation.rb +++ b/lib/captain/tool_instrumentation.rb @@ -15,6 +15,7 @@ module Captain::ToolInstrumentation response = yield executed = true span.set_attribute(ATTR_LANGFUSE_OBSERVATION_OUTPUT, response[:message] || response.to_json) + set_tool_session_error_attributes(span, response) if response.is_a?(Hash) end response rescue StandardError => e @@ -29,6 +30,14 @@ module Captain::ToolInstrumentation span.set_attribute(ATTR_LANGFUSE_OBSERVATION_INPUT, params[:messages].to_json) end + def set_tool_session_error_attributes(span, response) + error = response[:error] || response['error'] + return if error.blank? + + span.set_attribute(ATTR_GEN_AI_RESPONSE_ERROR, error.to_json) + span.status = OpenTelemetry::Trace::Status.error(error.to_s.truncate(1000)) + end + def record_generation(chat, message, model) return unless ChatwootApp.otel_enabled? return unless message.respond_to?(:role) && message.role.to_s == 'assistant' diff --git a/lib/integrations/llm_instrumentation.rb b/lib/integrations/llm_instrumentation.rb index c3baf291e..326bb901e 100644 --- a/lib/integrations/llm_instrumentation.rb +++ b/lib/integrations/llm_instrumentation.rb @@ -37,6 +37,7 @@ module Integrations::LlmInstrumentation result = yield executed = true span.set_attribute(ATTR_LANGFUSE_OBSERVATION_OUTPUT, result.to_json) + set_error_attributes(span, result) if result.is_a?(Hash) result end rescue StandardError => e @@ -50,9 +51,11 @@ module Integrations::LlmInstrumentation return yield unless ChatwootApp.otel_enabled? tracer.in_span(format(TOOL_SPAN_NAME, tool_name)) do |span| + span.set_attribute(ATTR_LANGFUSE_OBSERVATION_TYPE, 'tool') span.set_attribute(ATTR_LANGFUSE_OBSERVATION_INPUT, arguments.to_json) result = yield span.set_attribute(ATTR_LANGFUSE_OBSERVATION_OUTPUT, result.to_json) + set_error_attributes(span, result) if result.is_a?(Hash) result end end diff --git a/lib/integrations/llm_instrumentation_constants.rb b/lib/integrations/llm_instrumentation_constants.rb index 6ce296ee9..dfe1e7704 100644 --- a/lib/integrations/llm_instrumentation_constants.rb +++ b/lib/integrations/llm_instrumentation_constants.rb @@ -26,6 +26,7 @@ module Integrations::LlmInstrumentationConstants ATTR_LANGFUSE_METADATA = 'langfuse.trace.metadata.%s' ATTR_LANGFUSE_TRACE_INPUT = 'langfuse.trace.input' ATTR_LANGFUSE_TRACE_OUTPUT = 'langfuse.trace.output' + ATTR_LANGFUSE_OBSERVATION_TYPE = 'langfuse.observation.type' ATTR_LANGFUSE_OBSERVATION_INPUT = 'langfuse.observation.input' ATTR_LANGFUSE_OBSERVATION_OUTPUT = 'langfuse.observation.output' end diff --git a/lib/integrations/llm_instrumentation_spans.rb b/lib/integrations/llm_instrumentation_spans.rb index 824b6aa4a..85ea599f8 100644 --- a/lib/integrations/llm_instrumentation_spans.rb +++ b/lib/integrations/llm_instrumentation_spans.rb @@ -39,6 +39,7 @@ module Integrations::LlmInstrumentationSpans tool_name = tool_call.name.to_s span = tracer.start_span(format(TOOL_SPAN_NAME, tool_name)) + span.set_attribute(ATTR_LANGFUSE_OBSERVATION_TYPE, 'tool') span.set_attribute(ATTR_LANGFUSE_OBSERVATION_INPUT, tool_call.arguments.to_json) @pending_tool_spans ||= [] diff --git a/spec/enterprise/services/captain/assistant/agent_runner_service_spec.rb b/spec/enterprise/services/captain/assistant/agent_runner_service_spec.rb index 2c05860e2..04fa0f967 100644 --- a/spec/enterprise/services/captain/assistant/agent_runner_service_spec.rb +++ b/spec/enterprise/services/captain/assistant/agent_runner_service_spec.rb @@ -75,6 +75,7 @@ RSpec.describe Captain::Assistant::AgentRunnerService do it 'runs agent with extracted user message and context' do expected_context = { + session_id: "#{account.id}_#{conversation.display_id}", conversation_history: [ { role: :user, content: 'Hello there', agent_name: nil }, { role: :assistant, content: 'Hi! How can I help you?', agent_name: 'Assistant' }, @@ -306,6 +307,60 @@ RSpec.describe Captain::Assistant::AgentRunnerService do end end + describe '#add_usage_metadata_callback' do + it 'sets credits_used=false when handoff tool is used' do + service = described_class.new(assistant: assistant, conversation: conversation) + runner = instance_double(Agents::AgentRunner) + tool_complete_callback = nil + run_complete_callback = nil + span_class = Class.new do + def set_attribute(*); end + end + root_span = instance_double(span_class) + context_wrapper = Struct.new(:context).new({ __otel_tracing: { root_span: root_span } }) + + allow(ChatwootApp).to receive(:otel_enabled?).and_return(true) + allow(runner).to receive(:on_tool_complete) do |&block| + tool_complete_callback = block + runner + end + allow(runner).to receive(:on_run_complete) do |&block| + run_complete_callback = block + runner + end + + service.send(:add_usage_metadata_callback, runner) + + tool_complete_callback.call(Captain::Tools::HandoffTool.new(assistant).name, 'ok', context_wrapper) + + expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.credits_used', false) + run_complete_callback.call('assistant', nil, context_wrapper) + end + + it 'sets credits_used=true when handoff tool is not used' do + service = described_class.new(assistant: assistant, conversation: conversation) + runner = instance_double(Agents::AgentRunner) + run_complete_callback = nil + span_class = Class.new do + def set_attribute(*); end + end + root_span = instance_double(span_class) + context_wrapper = Struct.new(:context).new({ __otel_tracing: { root_span: root_span } }) + + allow(ChatwootApp).to receive(:otel_enabled?).and_return(true) + allow(runner).to receive(:on_tool_complete).and_return(runner) + allow(runner).to receive(:on_run_complete) do |&block| + run_complete_callback = block + runner + end + + service.send(:add_usage_metadata_callback, runner) + + expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.credits_used', true) + run_complete_callback.call('assistant', nil, context_wrapper) + end + end + describe 'constants' do it 'defines conversation state attributes' do expect(described_class::CONVERSATION_STATE_ATTRIBUTES).to include( diff --git a/spec/enterprise/services/messages/audio_transcription_service_spec.rb b/spec/enterprise/services/messages/audio_transcription_service_spec.rb index 41a4cae83..7ece2540a 100644 --- a/spec/enterprise/services/messages/audio_transcription_service_spec.rb +++ b/spec/enterprise/services/messages/audio_transcription_service_spec.rb @@ -8,8 +8,8 @@ RSpec.describe Messages::AudioTranscriptionService, type: :service do before do # Create required installation configs - create(:installation_config, name: 'CAPTAIN_OPEN_AI_API_KEY', value: 'test-api-key') - create(:installation_config, name: 'CAPTAIN_OPEN_AI_MODEL', value: 'gpt-4o-mini') + InstallationConfig.find_or_create_by!(name: 'CAPTAIN_OPEN_AI_API_KEY') { |config| config.value = 'test-api-key' } + InstallationConfig.find_or_create_by!(name: 'CAPTAIN_OPEN_AI_MODEL') { |config| config.value = 'gpt-4o-mini' } # Mock usage limits for transcription to be available allow(account).to receive(:usage_limits).and_return({ captain: { responses: { current_available: 100 } } }) @@ -64,4 +64,24 @@ RSpec.describe Messages::AudioTranscriptionService, type: :service do end end end + + describe '#fetch_audio_file' do + let(:service) { described_class.new(attachment) } + + before do + attachment.file.attach( + io: File.open(Rails.public_path.join('audio/widget/ding.mp3')), + filename: 'speech', + content_type: 'audio/mpeg' + ) + end + + it 'adds extension from content type when filename has no extension' do + temp_file_path = service.send(:fetch_audio_file) + + expect(File.extname(temp_file_path)).to eq('.mpeg') + ensure + FileUtils.rm_f(temp_file_path) if temp_file_path.present? + end + end end diff --git a/spec/services/whatsapp/incoming_message_service_spec.rb b/spec/services/whatsapp/incoming_message_service_spec.rb index 6c23e9b71..2ecf60acb 100644 --- a/spec/services/whatsapp/incoming_message_service_spec.rb +++ b/spec/services/whatsapp/incoming_message_service_spec.rb @@ -6,6 +6,14 @@ describe Whatsapp::IncomingMessageService do stub_request(:post, 'https://waba.360dialog.io/v1/configs/webhook') end + after do + # The atomic dedup lock lives in Redis and is not rolled back by + # transactional fixtures. Clean up any keys created during the test. + Redis::Alfred.scan_each(match: 'MESSAGE_SOURCE_KEY::*') do |key| + Redis::Alfred.delete(key) + end + end + let!(:whatsapp_channel) { create(:channel_whatsapp, sync_templates: false) } let(:wa_id) { '2423423243' } let!(:params) do @@ -393,8 +401,8 @@ describe Whatsapp::IncomingMessageService do end end - describe 'when message processing is in progress' do - it 'ignores the current message creation request' do + describe 'when another worker already holds the dedup lock' do + it 'skips message creation' do params = { 'contacts' => [{ 'profile' => { 'name' => 'Kedar' }, 'wa_id' => '919746334593' }], 'messages' => [{ 'from' => '919446284490', 'id' => 'wamid.SDFADSf23sfasdafasdfa', @@ -409,17 +417,14 @@ describe Whatsapp::IncomingMessageService do 'phones' => [{ 'phone' => '+1 (415) 341-8386' }] } ] }] }.with_indifferent_access - expect(Message.find_by(source_id: 'wamid.SDFADSf23sfasdafasdfa')).not_to be_present - key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: 'wamid.SDFADSf23sfasdafasdfa') - - Redis::Alfred.setex(key, true) - expect(Redis::Alfred.get(key)).to be_truthy + # Simulate another worker holding the lock + lock = Whatsapp::MessageDedupLock.new('wamid.SDFADSf23sfasdafasdfa') + expect(lock.acquire!).to be_truthy described_class.new(inbox: whatsapp_channel.inbox, params: params).perform expect(whatsapp_channel.inbox.messages.count).to eq(0) - expect(Message.find_by(source_id: 'wamid.SDFADSf23sfasdafasdfa')).not_to be_present - - expect(Redis::Alfred.get(key)).to be_truthy + ensure + key = format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: 'wamid.SDFADSf23sfasdafasdfa') Redis::Alfred.delete(key) end end diff --git a/spec/services/whatsapp/incoming_message_whatsapp_cloud_service_spec.rb b/spec/services/whatsapp/incoming_message_whatsapp_cloud_service_spec.rb index 6112ddc7b..4b6841811 100644 --- a/spec/services/whatsapp/incoming_message_whatsapp_cloud_service_spec.rb +++ b/spec/services/whatsapp/incoming_message_whatsapp_cloud_service_spec.rb @@ -2,6 +2,10 @@ require 'rails_helper' describe Whatsapp::IncomingMessageWhatsappCloudService do describe '#perform' do + after do + Redis::Alfred.scan_each(match: 'MESSAGE_SOURCE_KEY::*') { |key| Redis::Alfred.delete(key) } + end + let!(:whatsapp_channel) { create(:channel_whatsapp, provider: 'whatsapp_cloud', sync_templates: false, validate_provider_config: false) } let(:params) do { diff --git a/spec/services/whatsapp/message_dedup_lock_spec.rb b/spec/services/whatsapp/message_dedup_lock_spec.rb new file mode 100644 index 000000000..f3009b2ff --- /dev/null +++ b/spec/services/whatsapp/message_dedup_lock_spec.rb @@ -0,0 +1,43 @@ +require 'rails_helper' + +describe Whatsapp::MessageDedupLock do + let(:source_id) { "wamid.test_#{SecureRandom.hex(8)}" } + let(:lock) { described_class.new(source_id) } + let(:redis_key) { format(Redis::RedisKeys::MESSAGE_SOURCE_KEY, id: source_id) } + + after { Redis::Alfred.delete(redis_key) } + + describe '#acquire!' do + it 'returns truthy on first acquire' do + expect(lock.acquire!).to be_truthy + end + + it 'returns falsy on second acquire for the same source_id' do + lock.acquire! + expect(described_class.new(source_id).acquire!).to be_falsy + end + + it 'allows different source_ids to acquire independently' do + lock.acquire! + other = described_class.new("wamid.other_#{SecureRandom.hex(8)}") + expect(other.acquire!).to be_truthy + end + + it 'lets exactly one thread win when two race for the same source_id' do + results = Concurrent::Array.new + barrier = Concurrent::CyclicBarrier.new(2) + + threads = Array.new(2) do + Thread.new do + barrier.wait + results << described_class.new(source_id).acquire! + end + end + + threads.each(&:join) + + wins = results.count { |r| r } + expect(wins).to eq(1), "Expected exactly 1 winner but got #{wins}. Results: #{results.inspect}" + end + end +end