fix(captain): mark discarded generations

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!(

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