fix(captain): mark discarded generations

VERSION_CW

@@ -1 +1 @@
-4.16.1
+4.16.0

app/javascript/dashboard/components/widgets/conversation/MessagesView.vue

@@ -33,7 +33,9 @@ import {
 // constants
 import { BUS_EVENTS } from 'shared/constants/busEvents';
 import { REPLY_POLICY } from 'shared/constants/links';
-import wootConstants from 'dashboard/constants/globals';
+import wootConstants, {
+  META_RESTRICTION_STATUS_URL,
+} from 'dashboard/constants/globals';
 import { LOCAL_STORAGE_KEYS } from 'dashboard/constants/localStorage';
 import { INBOX_TYPES } from 'dashboard/helper/inbox';
 
@@ -93,6 +95,7 @@ export default {
       currentUserId: 'getCurrentUserID',
       listLoadingStatus: 'getAllMessagesLoaded',
       currentAccountId: 'getCurrentAccountId',
+      isOnChatwootCloud: 'globalConfig/isOnChatwootCloud',
     }),
     isOpen() {
       return this.currentChat?.status === wootConstants.STATUS_TYPE.OPEN;
@@ -170,6 +173,13 @@ export default {
         instagramInbox
       );
     },
+    isInstagramRestrictionBannerVisible() {
+      return this.isOnChatwootCloud && this.isAnInstagramChannel;
+    },
+    instagramRestrictionStatusUrl() {
+      return META_RESTRICTION_STATUS_URL;
+    },
+
     replyWindowBannerMessage() {
       if (this.isAWhatsAppChannel) {
         return this.$t('CONVERSATION.TWILIO_WHATSAPP_CAN_REPLY');
@@ -454,7 +464,15 @@ export default {
   >
     <div ref="topBannerRef">
       <Banner
-        v-if="!currentChat.can_reply"
+        v-if="isInstagramRestrictionBannerVisible"
+        color-scheme="warning"
+        class="mx-2 mt-2 overflow-hidden rounded-lg"
+        :banner-message="$t('CONVERSATION.INSTAGRAM_RESTRICTION_BANNER')"
+        :href-link="instagramRestrictionStatusUrl"
+        :href-link-text="$t('CONVERSATION.INSTAGRAM_RESTRICTION_STATUS_LINK')"
+      />
+      <Banner
+        v-else-if="!currentChat.can_reply"
         color-scheme="alert"
         class="mx-2 mt-2 overflow-hidden rounded-lg"
         :banner-message="replyWindowBannerMessage"

app/javascript/dashboard/constants/globals.js

@@ -78,3 +78,5 @@ export default {
   },
 };
 export const DEFAULT_REDIRECT_URL = '/app/';
+export const META_RESTRICTION_STATUS_URL =
+  'https://status.chatwoot.com/incident/948346';

app/javascript/dashboard/i18n/locale/en/conversation.json

@@ -44,6 +44,8 @@
     "TWILIO_WHATSAPP_CAN_REPLY": "You can only reply to this conversation using a template message due to",
     "TWILIO_WHATSAPP_24_HOURS_WINDOW": "24 hour message window restriction",
     "OLD_INSTAGRAM_INBOX_REPLY_BANNER": "This Instagram account was migrated to the new Instagram channel inbox. All new messages will show up there. You won’t be able to send messages from this conversation anymore.",
+    "INSTAGRAM_RESTRICTION_BANNER": "Instagram is currently restricted. Some messages or actions may be delayed or unavailable while we restore full support.",
+    "INSTAGRAM_RESTRICTION_STATUS_LINK": "View status update",
     "REPLYING_TO": "You are replying to:",
     "REMOVE_SELECTION": "Remove Selection",
     "DOWNLOAD": "Download",

app/javascript/dashboard/i18n/locale/en/inboxMgmt.json

@@ -58,7 +58,9 @@
         "ERROR_MESSAGE": "There was an error connecting to Instagram, please try again",
         "ERROR_AUTH": "There was an error connecting to Instagram, please try again",
         "NEW_INBOX_SUGGESTION": "This Instagram account was previously linked to a different inbox and has now been migrated here. All new messages will appear here. The old inbox will no longer be able to send or receive messages for this account.",
-        "DUPLICATE_INBOX_BANNER": "This Instagram account was migrated to the new Instagram channel inbox. You won’t be able to send/receive Instagram messages from this inbox anymore."
+        "DUPLICATE_INBOX_BANNER": "This Instagram account was migrated to the new Instagram channel inbox. You won’t be able to send/receive Instagram messages from this inbox anymore.",
+        "SETTINGS_RESTRICTED_WARNING": "Instagram is currently restricted. Some messages or actions may be delayed or unavailable while we restore full support.",
+        "STATUS_LINK": "View status update"
       },
       "TIKTOK": {
         "CONTINUE_WITH_TIKTOK": "Continue with TikTok",

app/javascript/dashboard/routes/dashboard/settings/inbox/Settings.vue

@@ -4,6 +4,8 @@ import { shouldBeUrl } from 'shared/helpers/Validators';
 import { useAlert } from 'dashboard/composables';
 import { useVuelidate } from '@vuelidate/core';
 import Avatar from 'next/avatar/Avatar.vue';
+import Banner from 'dashboard/components-next/banner/Banner.vue';
+import Icon from 'dashboard/components-next/icon/Icon.vue';
 import SettingIntroBanner from 'dashboard/components/widgets/SettingIntroBanner.vue';
 import SettingsToggleSection from 'dashboard/components-next/Settings/SettingsToggleSection.vue';
 import SettingsFieldSection from 'dashboard/components-next/Settings/SettingsFieldSection.vue';
@@ -44,9 +46,11 @@ import SelectInput from 'dashboard/components-next/select/Select.vue';
 import Widget from 'dashboard/modules/widget-preview/components/Widget.vue';
 import AccessToken from 'dashboard/routes/dashboard/settings/profile/AccessToken.vue';
 import { copyTextToClipboard } from 'shared/helpers/clipboard';
+import { META_RESTRICTION_STATUS_URL } from 'dashboard/constants/globals';
 
 export default {
   components: {
+    Banner,
     BotConfiguration,
     CollaboratorsPage,
     ConfigurationPage,
@@ -80,6 +84,7 @@ export default {
     WhatsappManualMigrationBanner,
     Widget,
     AccessToken,
+    Icon,
   },
   mixins: [inboxMixin],
   setup() {
@@ -343,6 +348,12 @@ export default {
     instagramUnauthorized() {
       return this.isAnInstagramChannel && this.inbox.reauthorization_required;
     },
+    showInstagramRestrictionSettingsBanner() {
+      return this.isOnChatwootCloud && this.isAnInstagramChannel;
+    },
+    metaRestrictionStatusUrl() {
+      return META_RESTRICTION_STATUS_URL;
+    },
     tiktokUnauthorized() {
       return this.isATiktokChannel && this.inbox.reauthorization_required;
     },
@@ -809,6 +820,29 @@ export default {
           :class="bannerMaxWidth"
           @start="openWhatsAppManualMigrationDialog"
         />
+        <Banner
+          v-if="showInstagramRestrictionSettingsBanner"
+          color="amber"
+          class="mx-6 mb-4 max-w-4xl"
+        >
+          <div class="flex items-start gap-3 text-start">
+            <Icon
+              icon="i-lucide-triangle-alert"
+              class="flex-shrink-0 size-4 mt-0.5"
+            />
+            <span>
+              {{ $t('INBOX_MGMT.ADD.INSTAGRAM.SETTINGS_RESTRICTED_WARNING') }}
+              <a
+                :href="metaRestrictionStatusUrl"
+                class="link underline"
+                rel="noopener noreferrer nofollow"
+                target="_blank"
+              >
+                {{ $t('INBOX_MGMT.ADD.INSTAGRAM.STATUS_LINK') }}
+              </a>
+            </span>
+          </div>
+        </Banner>
 
         <div
           v-if="selectedTabKey === 'inbox-settings'"

config/app.yml

@@ -1,5 +1,5 @@
 shared: &shared
-  version: '4.16.1'
+  version: '4.16.0'
 
 development:
   <<: *shared

enterprise/app/jobs/captain/conversation/response_builder_job.rb

@@ -7,16 +7,19 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
   retry_on ActiveStorage::FileNotFoundError, attempts: 3, wait: 2.seconds
   retry_on Faraday::BadRequestError, attempts: 3, wait: 2.seconds
 
-  def perform(conversation, assistant)
+  def perform(conversation, assistant, trigger_message_id = nil)
     @conversation = conversation
     @inbox = conversation.inbox
     @assistant = assistant
+    @trigger_message_id = trigger_message_id if captain_v2_enabled?
 
     return unless conversation_pending?
 
     Current.executed_by = @assistant
 
     if captain_v2_enabled?
+      return unless trigger_message_current?
+
       generate_response_with_v2
     else
       generate_and_process_response
@@ -45,11 +48,18 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
   end
 
   def generate_response_with_v2
-    runner_service = Captain::Assistant::AgentRunnerService.new(assistant: @assistant, conversation: @conversation)
+    runner_service = Captain::Assistant::AgentRunnerService.new(
+      assistant: @assistant,
+      conversation: @conversation,
+      trigger_message_id: @trigger_message_id
+    )
     message_history = Captain::Conversation::MessageHistoryBuilderService.new(conversation: @conversation).perform
     @response = runner_service.generate_response(message_history: message_history)
     @run_result = runner_service.last_run_result
 
+    return if runner_service.response_discarded?
+    return unless trigger_message_current?
+
     process_response
   end
 
@@ -77,16 +87,24 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
 
       process_v1_handoff
     elsif conversation_pending?
-      message = nil
-      ActiveRecord::Base.transaction do
-        message = create_messages
-        Rails.logger.info("[CAPTAIN][ResponseBuilderJob] Incrementing response usage for #{account.id}")
-        account.increment_response_usage
-      end
-      capture_assistant_session(result_message: message, credits_consumed: 1.0)
+      process_standard_response
     end
   end
 
+  def process_standard_response
+    message = nil
+    ActiveRecord::Base.transaction do
+      next unless trigger_message_current?
+
+      message = create_messages
+      Rails.logger.info("[CAPTAIN][ResponseBuilderJob] Incrementing response usage for #{account.id}")
+      account.increment_response_usage
+    end
+    return unless message
+
+    capture_assistant_session(result_message: message, credits_consumed: 1.0)
+  end
+
   def v1_handoff_requested?
     legacy_v1_handoff_token? || classifier_v1_handoff_requested?
   end
@@ -117,6 +135,8 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
   end
 
   def process_v2_handoff
+    return unless trigger_message_current?
+
     # HandoffTool already ran bot_handoff! + OOO inside the agent loop. Preserve
     # waiting_since so this message doesn't clear the timestamp it left in place.
     I18n.with_locale(@assistant.account.locale) do
@@ -152,7 +172,7 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
     @response ||= {}
     @response['action_source'] ||= 'error'
     @response['action_reason'] ||= error_action_reason(error)
-    process_v1_handoff if conversation_pending?
+    process_v1_handoff if conversation_pending? && (!captain_v2_enabled? || trigger_message_current?)
     true
   end
 
@@ -181,4 +201,14 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
     status = Conversation.uncached { Conversation.where(id: @conversation.id).pick(:status) }
     status == 'pending' || status == Conversation.statuses[:pending]
   end
+
+  def trigger_message_current?
+    return true if @trigger_message_id.blank?
+
+    latest_incoming_message_id = Conversation.uncached do
+      @conversation.messages.incoming.maximum(:id)
+    end
+
+    latest_incoming_message_id == @trigger_message_id
+  end
 end

enterprise/app/services/captain/assistant/agent_runner_service.rb

@@ -9,11 +9,12 @@ class Captain::Assistant::AgentRunnerService
 
   attr_reader :last_run_result
 
-  def initialize(assistant:, conversation: nil, callbacks: {}, source: nil)
+  def initialize(assistant:, conversation: nil, callbacks: {}, source: nil, trigger_message_id: nil)
     @assistant = assistant
     @conversation = conversation
     @callbacks = callbacks
     @source = source
+    @trigger_message_id = trigger_message_id
     @handoff_tool_called = false
   end
 
@@ -31,6 +32,10 @@ class Captain::Assistant::AgentRunnerService
     error_response(e.message)
   end
 
+  def response_discarded?
+    @response_discarded == true
+  end
+
   private
 
   def build_context(message_history)
@@ -151,10 +156,9 @@ class Captain::Assistant::AgentRunnerService
       track_handoff_usage(tool_name, handoff_tool_name, context_wrapper)
     end
 
-    if ChatwootApp.otel_enabled?
-      runner.on_run_complete do |_agent_name, _result, context_wrapper|
-        write_credits_used_metadata(context_wrapper)
-      end
+    runner.on_run_complete do |_agent_name, _result, context_wrapper|
+      @response_discarded = trigger_message_stale?
+      write_run_metadata(context_wrapper) if ChatwootApp.otel_enabled?
     end
     runner
   end
@@ -169,11 +173,22 @@ class Captain::Assistant::AgentRunnerService
     @handoff_tool_called = true
   end
 
-  def write_credits_used_metadata(context_wrapper)
+  def write_run_metadata(context_wrapper)
     root_span = context_wrapper&.context&.dig(:__otel_tracing, :root_span)
     return unless root_span
 
-    root_span.set_attribute(format(ATTR_LANGFUSE_METADATA, 'credit_used'), @handoff_tool_called ? 'false' : 'true')
+    root_span.set_attribute(format(ATTR_LANGFUSE_METADATA, 'discarded'), response_discarded?.to_s)
+    root_span.set_attribute(format(ATTR_LANGFUSE_METADATA, 'credit_used'), (!@handoff_tool_called && !response_discarded?).to_s)
+  end
+
+  def trigger_message_stale?
+    return false if @trigger_message_id.blank? || @conversation.blank?
+
+    latest_incoming_message_id = Conversation.uncached do
+      @conversation.messages.incoming.maximum(:id)
+    end
+
+    latest_incoming_message_id != @trigger_message_id
   end
 
   def runner

enterprise/app/services/captain/assistant/instrumentation_attribute_provider.rb

@@ -13,7 +13,8 @@ class Captain::Assistant::InstrumentationAttributeProvider
 
   def generation_attributes(_context_wrapper, _chat, message)
     {
-      format(ATTR_LANGFUSE_OBSERVATION_METADATA, 'generation_stage') => generation_stage(message)
+      format(ATTR_LANGFUSE_OBSERVATION_METADATA, 'generation_stage') => generation_stage(message),
+      format(ATTR_LANGFUSE_OBSERVATION_METADATA, 'discarded') => @service.send(:trigger_message_stale?).to_s
     }
   end
 

enterprise/app/services/captain/assistant/runner_state_helper.rb

@@ -23,6 +23,7 @@ module Captain::Assistant::RunnerStateHelper
       timezone: @conversation&.inbox&.timezone.presence || 'UTC'
     }
     state[:source] = @source if @source.present?
+    state[:trigger_message_id] = @trigger_message_id if @trigger_message_id.present?
 
     build_conversation_state(state) if @conversation
     state

enterprise/app/services/enterprise/message_templates/hook_execution_service.rb

@@ -31,6 +31,7 @@ module Enterprise::MessageTemplates::HookExecutionService
 
   def schedule_captain_response
     job_args = [conversation, conversation.inbox.captain_assistant]
+    job_args << message.id if conversation.account.feature_enabled?('captain_integration_v2')
 
     if message.attachments.blank?
       Captain::Conversation::ResponseBuilderJob.perform_later(*job_args)

enterprise/lib/captain/tools/handoff_tool.rb

@@ -5,6 +5,7 @@ class Captain::Tools::HandoffTool < Captain::Tools::BasePublicTool
   def perform(tool_context, reason: nil)
     conversation = find_conversation(tool_context.state)
     return 'Conversation not found' unless conversation
+    return 'Handoff skipped because a newer customer message arrived' unless trigger_message_current?(tool_context.state, conversation)
 
     # Log the handoff with reason
     log_tool_usage('tool_handoff', {
@@ -23,6 +24,17 @@ class Captain::Tools::HandoffTool < Captain::Tools::BasePublicTool
 
   private
 
+  def trigger_message_current?(state, conversation)
+    trigger_message_id = state&.dig(:trigger_message_id)
+    return true if trigger_message_id.blank?
+
+    latest_incoming_message_id = Conversation.uncached do
+      conversation.messages.incoming.maximum(:id)
+    end
+
+    latest_incoming_message_id == trigger_message_id
+  end
+
   def trigger_handoff(tool_context, conversation, reason)
     # post the reason as a private note
     note = conversation.messages.create!(

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@chatwoot/chatwoot",
-  "version": "4.16.1",
+  "version": "4.16.0",
   "license": "MIT",
   "scripts": {
     "eslint": "eslint app/**/*.{js,vue}",

spec/enterprise/jobs/captain/conversation/response_builder_job_spec.rb

@@ -13,6 +13,7 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
     let(:mock_action_classifier_service) { instance_double(Captain::Llm::AssistantActionClassifierService) }
     let(:mock_false_promise_service) { instance_double(Captain::Llm::AssistantFalsePromiseService) }
     let(:assistant_model) { Llm::Models.default_model_for('assistant') }
+    let(:trigger_message) { conversation.messages.find_by!(content: 'Hello') }
 
     before do
       create(:message, conversation: conversation, content: 'Hello', message_type: :incoming)
@@ -23,6 +24,7 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
       allow(Captain::Assistant::AgentRunnerService).to receive(:new).and_return(mock_agent_runner_service)
       allow(mock_agent_runner_service).to receive(:generate_response).and_return({ 'response' => 'Hey, welcome to Captain V2' })
       allow(mock_agent_runner_service).to receive(:last_run_result).and_return(nil)
+      allow(mock_agent_runner_service).to receive(:response_discarded?).and_return(false)
       allow(Captain::Llm::AssistantActionClassifierService).to receive(:new).and_return(mock_action_classifier_service)
       allow(mock_action_classifier_service).to receive(:classify).and_return({ 'action' => 'continue' })
       allow(Captain::Llm::AssistantFalsePromiseService).to receive(:new).and_return(mock_false_promise_service)
@@ -358,14 +360,28 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
       it 'uses Captain::Assistant::AgentRunnerService' do
         expect(Captain::Assistant::AgentRunnerService).to receive(:new).with(
           assistant: assistant,
-          conversation: conversation
+          conversation: conversation,
+          trigger_message_id: trigger_message.id
         )
         expect(Captain::Llm::AssistantChatService).not_to receive(:new)
 
-        described_class.perform_now(conversation, assistant)
+        described_class.perform_now(conversation, assistant, trigger_message.id)
         expect(conversation.messages.last.content).to eq('Hey, welcome to Captain V2')
       end
 
+      it 'discards a response when a newer message arrives during generation' do
+        allow(mock_agent_runner_service).to receive(:generate_response) do
+          create(:message, conversation: conversation, content: 'New context', message_type: :incoming)
+          { 'response' => 'Stale response', 'handoff_tool_called' => false }
+        end
+        allow(mock_agent_runner_service).to receive(:response_discarded?).and_return(true)
+
+        described_class.perform_now(conversation, assistant, trigger_message.id)
+
+        expect(conversation.messages.outgoing.count).to eq(0)
+        expect(account.reload.usage_limits[:captain][:responses][:consumed]).to eq(0)
+      end
+
       it 'passes message history with resolution markers to agent runner service' do
         same_second = Time.current.change(usec: 0)
         conversation.messages.find_by!(content: 'Hello').update!(created_at: same_second, updated_at: same_second)

spec/enterprise/lib/captain/tools/handoff_tool_spec.rb

@@ -28,6 +28,37 @@ RSpec.describe Captain::Tools::HandoffTool, type: :model do
 
   describe '#perform' do
     context 'when conversation exists' do
+      context 'with a trigger message id' do
+        let(:trigger_message) do
+          create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming, created_at: same_second)
+        end
+        let(:same_second) { Time.current.change(usec: 0) }
+        let(:tool_context) do
+          Struct.new(:state).new({ conversation: { id: conversation.id }, trigger_message_id: trigger_message.id })
+        end
+
+        it 'hands off when the trigger message is still the latest message' do
+          trigger_message
+
+          expect do
+            result = tool.perform(tool_context, reason: 'Customer needs specialized support')
+            expect(result).to include('Conversation handed off')
+          end.to change(Message, :count).by(1)
+        end
+
+        it 'skips the handoff when a newer message has arrived' do
+          trigger_message
+          conversation.update!(status: :pending)
+          create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming, created_at: same_second)
+
+          expect do
+            result = tool.perform(tool_context, reason: 'Customer needs specialized support')
+            expect(result).to eq('Handoff skipped because a newer customer message arrived')
+          end.not_to change(Message, :count)
+          expect(conversation.reload.status).to eq('pending')
+        end
+      end
+
       context 'with reason provided' do
         it 'creates a private note with reason and hands off conversation' do
           reason = 'Customer needs specialized support'

spec/enterprise/services/captain/assistant/agent_runner_service_spec.rb

@@ -52,6 +52,12 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
 
       expect(service.instance_variable_get(:@callbacks)).to eq(callbacks)
     end
+
+    it 'accepts the trigger message id' do
+      service = described_class.new(assistant: assistant, conversation: conversation, trigger_message_id: 123)
+
+      expect(service.instance_variable_get(:@trigger_message_id)).to eq(123)
+    end
   end
 
   describe '#generate_response' do
@@ -99,6 +105,18 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
       service.generate_response(message_history: message_history)
     end
 
+    it 'adds the trigger message id to the runner state' do
+      service = described_class.new(assistant: assistant, conversation: conversation, trigger_message_id: 123)
+
+      expect(mock_runner).to receive(:run).with(
+        'I need help with my account',
+        context: hash_including(state: hash_including(trigger_message_id: 123)),
+        max_turns: 10
+      )
+
+      service.generate_response(message_history: message_history)
+    end
+
     context 'when the latest user message is multimodal' do
       let(:multimodal_message_history) do
         [
@@ -446,6 +464,7 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
       attributes = provider.generation_attributes(nil, nil, message)
 
       expect(attributes['langfuse.observation.metadata.generation_stage']).to eq('final_response')
+      expect(attributes['langfuse.observation.metadata.discarded']).to eq('false')
     end
 
     it 'marks tool call generations separately from final responses' do
@@ -456,6 +475,18 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
 
       expect(attributes['langfuse.observation.metadata.generation_stage']).to eq('tool_call')
     end
+
+    it 'marks a generation as discarded when a newer message has arrived' do
+      trigger_message = create(:message, conversation: conversation, message_type: :incoming)
+      runner_service = described_class.new(assistant: assistant, conversation: conversation, trigger_message_id: trigger_message.id)
+      attribute_provider = Captain::Assistant::InstrumentationAttributeProvider.new(runner_service)
+      message = instance_double(RubyLLM::Message, tool_calls: {})
+      create(:message, conversation: conversation, message_type: :incoming)
+
+      attributes = attribute_provider.generation_attributes(nil, nil, message)
+
+      expect(attributes['langfuse.observation.metadata.discarded']).to eq('true')
+    end
   end
 
   describe '#build_state' do
@@ -574,6 +605,7 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
 
       tool_complete_callback.call(Captain::Tools::HandoffTool.new(assistant).name, 'ok', context_wrapper)
 
+      expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.discarded', 'false')
       expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.credit_used', 'false')
       run_complete_callback.call('assistant', nil, context_wrapper)
     end
@@ -588,6 +620,7 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
         tool_complete_callback = block
         runner
       end
+      allow(runner).to receive(:on_run_complete).and_return(runner)
 
       service.send(:add_usage_metadata_callback, runner)
 
@@ -599,15 +632,24 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
       expect(context_wrapper.context[:captain_v2_handoff_tool_called]).to be true
     end
 
-    it 'does not register OTEL run callback when OTEL is disabled' do
-      service = described_class.new(assistant: assistant, conversation: conversation)
+    it 'tracks discarded responses when OTEL is disabled' do
+      trigger_message = create(:message, conversation: conversation, message_type: :incoming)
+      service = described_class.new(assistant: assistant, conversation: conversation, trigger_message_id: trigger_message.id)
       runner = instance_double(Agents::AgentRunner)
+      run_complete_callback = nil
 
       allow(ChatwootApp).to receive(:otel_enabled?).and_return(false)
       allow(runner).to receive(:on_tool_complete).and_return(runner)
-      expect(runner).not_to receive(:on_run_complete)
+      allow(runner).to receive(:on_run_complete) do |&block|
+        run_complete_callback = block
+        runner
+      end
 
       service.send(:add_usage_metadata_callback, runner)
+      create(:message, conversation: conversation, message_type: :incoming)
+      run_complete_callback.call('assistant', nil, Struct.new(:context).new({}))
+
+      expect(service.response_discarded?).to be true
     end
 
     it 'sets credit_used=true when handoff tool is not used' do
@@ -629,9 +671,38 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
 
       service.send(:add_usage_metadata_callback, runner)
 
+      expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.discarded', 'false')
       expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.credit_used', 'true')
       run_complete_callback.call('assistant', nil, context_wrapper)
     end
+
+    it 'marks the trace discarded and does not use credit when a newer message arrived' do
+      trigger_message = create(:message, conversation: conversation, message_type: :incoming)
+      service = described_class.new(assistant: assistant, conversation: conversation, trigger_message_id: trigger_message.id)
+      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)
+      create(:message, conversation: conversation, message_type: :incoming)
+
+      expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.discarded', 'true')
+      expect(root_span).to receive(:set_attribute).with('langfuse.trace.metadata.credit_used', 'false')
+      run_complete_callback.call('assistant', nil, context_wrapper)
+
+      expect(service.response_discarded?).to be true
+    end
   end
 
   describe 'constants' do

spec/enterprise/services/enterprise/message_templates/hook_execution_service_spec.rb

@@ -22,9 +22,13 @@ RSpec.describe MessageTemplates::HookExecutionService do
       end
 
       it 'schedules captain response job for incoming messages on pending conversations' do
-        expect(Captain::Conversation::ResponseBuilderJob).to receive(:perform_later).with(conversation, assistant)
+        allow(account).to receive(:feature_enabled?).and_call_original
+        allow(account).to receive(:feature_enabled?).with('captain_integration_v2').and_return(true)
+        allow(Captain::Conversation::ResponseBuilderJob).to receive(:perform_later)
 
-        create(:message, conversation: conversation, message_type: :incoming, account: account)
+        message = create(:message, conversation: conversation, message_type: :incoming, account: account)
+
+        expect(Captain::Conversation::ResponseBuilderJob).to have_received(:perform_later).with(conversation, assistant, message.id)
       end
     end