From 38a317962b22f7512d00cb650a8b80bb8b99a1d8 Mon Sep 17 00:00:00 2001
From: Aakash Bakhle <48802744+aakashb95@users.noreply.github.com>
Date: Thu, 30 Jul 2026 15:58:54 +0530
Subject: [PATCH] fix: Captain handles message bursts with a single reply
 (#15133)

Captain now handles a burst of customer messages with one reply. If more
messages arrive before Captain replies, the latest job uses the full
conversation history. The change applies only to Captain V2 and works
across every channel that Captain supports.

Blocked on: https://github.com/chatwoot/chatwoot/pull/15212

## Closes

Closes https://github.com/chatwoot/chatwoot/issues/14545

## How to reproduce

1. Start a pending conversation with Captain V2.
2. Send several messages while Captain is preparing a reply.
3. Captain can generate and send a separate reply for each message.

## What changed

* Each Captain V2 job records the incoming message that started it.
* A job stops before generation if a newer message already exists.
* Captain discards a generated reply if a newer message arrived during
generation.
* The latest job replies using the full conversation history.
* Langfuse records whether a generation was discarded and whether a
customer credit was used.
* Captain V1 keeps its existing behavior.

## Tradeoffs

Discarded generations still cost money. Message bursts can also increase
background job work and model provider load. A continuous stream of
incoming messages can delay the reply until one generation finishes
without a newer message.

A small timing window remains if a message arrives after the final check
and before Captain saves the reply. A handoff also cannot be undone if a
newer message arrives after Captain has already changed the conversation
status.

## How to test

1. Enable Captain V2 and start a pending Captain conversation.
2. Send several messages while Captain is preparing a reply.
3. Confirm that Captain sends one reply based on the full message
history.
4. Confirm that Langfuse marks discarded runs with `discarded=true` and
`credit_used=false`.
5. Disable Captain V2 and confirm that Captain V1 behavior is unchanged.

---------

Co-authored-by: Sony Mathew <sony@chatwoot.com>
---
 app/models/conversation.rb                    |  6 +-
 .../conversation/response_builder_job.rb      | 89 +++++++++++------
 enterprise/app/models/enterprise/message.rb   | 14 +++
 .../captain/assistant/agent_runner_service.rb | 75 ++------------
 .../instrumentation_attribute_provider.rb     |  6 +-
 .../runner_instrumentation_helper.rb          | 99 +++++++++++++++++++
 .../captain/assistant/runner_state_helper.rb  |  1 +
 .../hook_execution_service.rb                 | 27 ++++-
 .../lib/captain/tools/base_public_tool.rb     | 30 ++++++
 .../lib/captain/tools/faq_lookup_tool.rb      |  4 +
 enterprise/lib/captain/tools/handoff_tool.rb  | 58 ++++++++---
 enterprise/lib/captain/tools/http_tool.rb     |  9 +-
 .../conversation/response_builder_job_spec.rb | 69 +++++++++++--
 .../tools/add_private_note_tool_spec.rb       | 30 ++++++
 .../lib/captain/tools/handoff_tool_spec.rb    | 94 +++++++++++++++++-
 .../assistant/agent_runner_service_spec.rb    | 98 +++++++++++++++++-
 .../hook_execution_service_spec.rb            | 62 +++++++++++-
 17 files changed, 635 insertions(+), 136 deletions(-)
 create mode 100644 enterprise/app/services/captain/assistant/runner_instrumentation_helper.rb

diff --git a/app/models/conversation.rb b/app/models/conversation.rb
index 9daa4256e..4d26dcf47 100644
--- a/app/models/conversation.rb
+++ b/app/models/conversation.rb
@@ -172,10 +172,14 @@ class Conversation < ApplicationRecord
     save
   end
 
-  def bot_handoff!
+  def bot_handoff!(dispatch_event: true)
     update(waiting_since: Time.current) if waiting_since.blank?
     self.assignee_agent_bot = nil
     open!
+    dispatch_bot_handoff_event if dispatch_event
+  end
+
+  def dispatch_bot_handoff_event
     dispatcher_dispatch(CONVERSATION_BOT_HANDOFF)
   end
 
diff --git a/enterprise/app/jobs/captain/conversation/response_builder_job.rb b/enterprise/app/jobs/captain/conversation/response_builder_job.rb
index 68367f532..b86f859f3 100644
--- a/enterprise/app/jobs/captain/conversation/response_builder_job.rb
+++ b/enterprise/app/jobs/captain/conversation/response_builder_job.rb
@@ -7,20 +7,20 @@ 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, responding_to_message_id = nil) # rubocop:disable Lint/UnusedMethodArgument
+  def perform(conversation, assistant, responding_to_message_id = nil)
     @conversation = conversation
     @inbox = conversation.inbox
     @assistant = assistant
+    @responding_to_message_id = responding_to_message_id if captain_v2_enabled?
 
     return unless conversation_pending?
 
     Current.executed_by = @assistant
 
-    if captain_v2_enabled?
-      generate_response_with_v2
-    else
-      generate_and_process_response
-    end
+    return generate_and_process_response unless captain_v2_enabled?
+    return if newer_customer_message_arrived?
+
+    generate_response_with_v2
   rescue ActiveStorage::FileNotFoundError, Faraday::BadRequestError => e
     handle_error(e)
     raise e
@@ -45,30 +45,27 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
   end
 
   def generate_response_with_v2
-    runner_service = Captain::Assistant::AgentRunnerService.new(assistant: @assistant, conversation: @conversation)
+    runner_service = v2_runner_service
     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
 
+    @v2_handoff_tool_completed = runner_service.handoff_completed?
+    return process_response if v2_handoff_tool_completed?
+    return if runner_service.response_discarded? || newer_customer_message_arrived?
+
     process_response
   end
 
+  def v2_runner_service
+    runner_args = { assistant: @assistant, conversation: @conversation }
+    runner_args[:responding_to_message_id] = @responding_to_message_id if @responding_to_message_id.present?
+    Captain::Assistant::AgentRunnerService.new(**runner_args)
+  end
+
   def process_response
-    # Check V2 before V1: error_response can set both signals at once when HandoffTool
-    # fired before the runner errored. V2 must win — running V1 on top would duplicate
-    # OOO and re-dispatch the bot_handoff event.
     if v2_handoff_tool_fired?
-      if conversation_pending?
-        # HandoffTool flipped the flag without committing — its perform returned a
-        # failure string (e.g. "Conversation not found") before bot_handoff! ran. Fall
-        # back to a full V1 handoff so the customer still ends up with a human.
-        process_v1_handoff
-      else
-        # HandoffTool already opened the conversation inside the agent loop. All that's
-        # left is the customer-facing follow-up message.
-        process_v2_handoff
-      end
-      capture_assistant_session(result_message: @handoff_message, credits_consumed: 0.0)
+      process_v2_handoff_response
     elsif v1_handoff_requested?
       # V1 only signals via the response string — no state has been touched yet. If
       # the conversation isn't pending anymore, a human took over mid-run; bail out
@@ -77,16 +74,38 @@ 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 if captain_v2_enabled? && newer_customer_message_arrived?
+
+      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 process_v2_handoff_response
+    # Captain V1 infers completion from status. Captain V2 uses the completion
+    # marker set inside the locked handoff.
+    if captain_v2_enabled?
+      return unless v2_handoff_tool_completed? || conversation_pending?
+
+      v2_handoff_tool_completed? ? process_v2_handoff : process_v1_handoff
+    else
+      conversation_pending? ? process_v1_handoff : process_v2_handoff
+    end
+
+    capture_assistant_session(result_message: @handoff_message, credits_consumed: 0.0)
+  end
+
   def v1_handoff_requested?
     legacy_v1_handoff_token? || classifier_v1_handoff_requested?
   end
@@ -103,6 +122,8 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
     @response['handoff_tool_called']
   end
 
+  def v2_handoff_tool_completed? = @v2_handoff_tool_completed == true
+
   def process_v1_handoff
     I18n.with_locale(@assistant.account.locale) do
       Rails.logger.info(
@@ -152,7 +173,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? || !newer_customer_message_arrived?)
     true
   end
 
@@ -181,4 +202,14 @@ class Captain::Conversation::ResponseBuilderJob < ApplicationJob
     status = Conversation.uncached { Conversation.where(id: @conversation.id).pick(:status) }
     status == 'pending' || status == Conversation.statuses[:pending]
   end
+
+  def newer_customer_message_arrived?
+    return false if @responding_to_message_id.blank?
+
+    Conversation.uncached do
+      @conversation.messages
+                   .captain_response_triggering
+                   .exists?(['messages.id > ?', @responding_to_message_id])
+    end
+  end
 end
diff --git a/enterprise/app/models/enterprise/message.rb b/enterprise/app/models/enterprise/message.rb
index a6336fc55..11c65c6e5 100644
--- a/enterprise/app/models/enterprise/message.rb
+++ b/enterprise/app/models/enterprise/message.rb
@@ -4,6 +4,14 @@ module Enterprise::Message
       has_one :call, class_name: 'Call', foreign_key: :message_id, dependent: :nullify, inverse_of: :message
 
       scope :with_call, -> { includes(call: [:contact, { inbox: :channel }]) }
+      # Scheduling and freshness checks must share this scope so an email auto reply cannot cancel a pending response.
+      scope :captain_response_triggering, lambda {
+        incoming.joins(:inbox).where(
+          "((messages.content_attributes #>> '{}')::jsonb -> 'email' ->> 'auto_reply') IS DISTINCT FROM 'true' OR " \
+          "(messages.content_type != :incoming_email AND inboxes.channel_type != 'Channel::Email')",
+          incoming_email: content_types[:incoming_email]
+        )
+      }
     end
   end
 
@@ -13,6 +21,12 @@ module Enterprise::Message
     data
   end
 
+  def captain_response_triggering?
+    return incoming? && !auto_reply_email? unless persisted?
+
+    self.class.captain_response_triggering.exists?(id: id)
+  end
+
   private
 
   def mark_pending_conversation_as_open_for_human_response
diff --git a/enterprise/app/services/captain/assistant/agent_runner_service.rb b/enterprise/app/services/captain/assistant/agent_runner_service.rb
index ddec5d4b3..0d3d52e86 100644
--- a/enterprise/app/services/captain/assistant/agent_runner_service.rb
+++ b/enterprise/app/services/captain/assistant/agent_runner_service.rb
@@ -2,19 +2,21 @@ require 'agents'
 require 'agents/instrumentation'
 
 class Captain::Assistant::AgentRunnerService
-  include Integrations::LlmInstrumentationConstants
   include Captain::Assistant::RunnerCallbacksHelper
+  include Captain::Assistant::RunnerInstrumentationHelper
   include Captain::Assistant::TracePayloadHelper
   include Captain::Assistant::RunnerStateHelper
 
   attr_reader :last_run_result
 
-  def initialize(assistant:, conversation: nil, callbacks: {}, source: nil)
+  def initialize(assistant:, conversation: nil, callbacks: {}, source: nil, responding_to_message_id: nil)
     @assistant = assistant
     @conversation = conversation
     @callbacks = callbacks
     @source = source
+    @responding_to_message_id = responding_to_message_id
     @handoff_tool_called = false
+    @handoff_tool_completed = false
   end
 
   def generate_response(message_history: [])
@@ -35,6 +37,10 @@ class Captain::Assistant::AgentRunnerService
     error_response(e.message)
   end
 
+  def response_discarded? = @response_discarded == true
+
+  def handoff_completed? = @handoff_tool_completed == true
+
   private
 
   def build_context(message_history)
@@ -144,71 +150,6 @@ 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: Captain::Assistant::InstrumentationAttributeProvider.new(self)
-    )
-    register_trace_input_callback(runner)
-  end
-
-  def dynamic_trace_attributes(context_wrapper)
-    state = context_wrapper&.context&.dig(:state) || {}
-    conversation = state[:conversation] || {}
-    trace_input = context_wrapper&.context&.dig(:captain_v2_trace_input)
-
-    {
-      ATTR_LANGFUSE_USER_ID => state[:account_id],
-      format(ATTR_LANGFUSE_METADATA, 'assistant_id') => state[:assistant_id],
-      format(ATTR_LANGFUSE_METADATA, 'conversation_display_id') => conversation[:display_id],
-      format(ATTR_LANGFUSE_METADATA, 'channel_type') => state[:channel_type],
-      format(ATTR_LANGFUSE_METADATA, 'source') => state[:source],
-      ATTR_LANGFUSE_TRACE_INPUT => trace_input,
-      ATTR_LANGFUSE_OBSERVATION_INPUT => trace_input
-    }.compact.transform_values(&:to_s)
-  end
-
-  def add_usage_metadata_callback(runner)
-    handoff_tool_name = Captain::Tools::HandoffTool.new(@assistant).name
-
-    # Tool tracking always runs — process_response in the job consumes the resulting
-    # handoff_tool_called flag regardless of whether OTEL is enabled.
-    runner.on_tool_complete do |tool_name, _tool_result, context_wrapper|
-      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
-    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
-
-    # Mirror the flag onto the instance so error_response can surface it even when
-    # the runner raises before returning a result (the context is unreachable then).
-    context_wrapper.context[:captain_v2_handoff_tool_called] = true
-    @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
-
-    root_span.set_attribute(format(ATTR_LANGFUSE_METADATA, 'credit_used'), @handoff_tool_called ? 'false' : 'true')
-  end
-
   def runner
     @runner ||= begin
       configured_runner = Agents::Runner.with_agents(*build_and_wire_agents)
diff --git a/enterprise/app/services/captain/assistant/instrumentation_attribute_provider.rb b/enterprise/app/services/captain/assistant/instrumentation_attribute_provider.rb
index b9b812b0e..fa1a9f248 100644
--- a/enterprise/app/services/captain/assistant/instrumentation_attribute_provider.rb
+++ b/enterprise/app/services/captain/assistant/instrumentation_attribute_provider.rb
@@ -12,9 +12,13 @@ class Captain::Assistant::InstrumentationAttributeProvider
   end
 
   def generation_attributes(_context_wrapper, _chat, message)
-    {
+    attributes = {
       format(ATTR_LANGFUSE_OBSERVATION_METADATA, 'generation_stage') => generation_stage(message)
     }
+    if @service.send(:message_burst_protection_active?)
+      attributes[format(ATTR_LANGFUSE_OBSERVATION_METADATA, 'discarded')] = @service.send(:newer_customer_message_arrived?).to_s
+    end
+    attributes
   end
 
   private
diff --git a/enterprise/app/services/captain/assistant/runner_instrumentation_helper.rb b/enterprise/app/services/captain/assistant/runner_instrumentation_helper.rb
new file mode 100644
index 000000000..57562ec01
--- /dev/null
+++ b/enterprise/app/services/captain/assistant/runner_instrumentation_helper.rb
@@ -0,0 +1,99 @@
+module Captain::Assistant::RunnerInstrumentationHelper
+  include Integrations::LlmInstrumentationConstants
+
+  private
+
+  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: Captain::Assistant::InstrumentationAttributeProvider.new(self)
+    )
+    register_trace_input_callback(runner)
+  end
+
+  def dynamic_trace_attributes(context_wrapper)
+    state = context_wrapper&.context&.dig(:state) || {}
+    conversation = state[:conversation] || {}
+    trace_input = context_wrapper&.context&.dig(:captain_v2_trace_input)
+
+    {
+      ATTR_LANGFUSE_USER_ID => state[:account_id],
+      format(ATTR_LANGFUSE_METADATA, 'assistant_id') => state[:assistant_id],
+      format(ATTR_LANGFUSE_METADATA, 'conversation_display_id') => conversation[:display_id],
+      format(ATTR_LANGFUSE_METADATA, 'channel_type') => state[:channel_type],
+      format(ATTR_LANGFUSE_METADATA, 'source') => state[:source],
+      ATTR_LANGFUSE_TRACE_INPUT => trace_input,
+      ATTR_LANGFUSE_OBSERVATION_INPUT => trace_input
+    }.compact.transform_values(&:to_s)
+  end
+
+  def add_usage_metadata_callback(runner)
+    handoff_tool_name = Captain::Tools::HandoffTool.new(@assistant).name
+
+    # Tool tracking always runs — process_response in the job consumes the resulting
+    # handoff_tool_called flag regardless of whether OTEL is enabled.
+    runner.on_tool_complete do |tool_name, _tool_result, context_wrapper|
+      track_handoff_usage(tool_name, handoff_tool_name, context_wrapper)
+    end
+
+    if message_burst_protection_active?
+      runner.on_run_complete do |_agent_name, _result, context_wrapper|
+        @response_discarded = newer_customer_message_arrived?
+        write_run_metadata(context_wrapper) if ChatwootApp.otel_enabled?
+      end
+    elsif ChatwootApp.otel_enabled?
+      runner.on_run_complete do |_agent_name, _result, context_wrapper|
+        write_credits_used_metadata(context_wrapper)
+      end
+    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
+
+    # Mirror the flag onto the instance so error_response can surface it even when
+    # the runner raises before returning a result (the context is unreachable then).
+    context_wrapper.context[:captain_v2_handoff_tool_called] = true
+    @handoff_tool_called = true
+
+    return unless context_wrapper.context.dig(:state, :captain_v2_handoff_tool_completed)
+
+    @handoff_tool_completed = true
+  end
+
+  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, 'discarded'), response_discarded?.to_s)
+    root_span.set_attribute(format(ATTR_LANGFUSE_METADATA, 'credit_used'), (!@handoff_tool_called && !response_discarded?).to_s)
+  end
+
+  def write_credits_used_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')
+  end
+
+  def message_burst_protection_active? = @responding_to_message_id.present?
+
+  def newer_customer_message_arrived?
+    return false if @responding_to_message_id.blank? || @conversation.blank?
+
+    Conversation.uncached do
+      @conversation.messages
+                   .captain_response_triggering
+                   .exists?(['messages.id > ?', @responding_to_message_id])
+    end
+  end
+end
diff --git a/enterprise/app/services/captain/assistant/runner_state_helper.rb b/enterprise/app/services/captain/assistant/runner_state_helper.rb
index 2ab1b902a..7b7074b11 100644
--- a/enterprise/app/services/captain/assistant/runner_state_helper.rb
+++ b/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[:responding_to_message_id] = @responding_to_message_id if @responding_to_message_id.present?
 
     build_conversation_state(state) if @conversation
     state
diff --git a/enterprise/app/services/enterprise/message_templates/hook_execution_service.rb b/enterprise/app/services/enterprise/message_templates/hook_execution_service.rb
index 56dbc7245..d32968384 100644
--- a/enterprise/app/services/enterprise/message_templates/hook_execution_service.rb
+++ b/enterprise/app/services/enterprise/message_templates/hook_execution_service.rb
@@ -31,17 +31,34 @@ module Enterprise::MessageTemplates::HookExecutionService
 
   def schedule_captain_response
     job_args = [conversation, conversation.inbox.captain_assistant]
+    captain_v2_enabled = conversation.account.feature_enabled?('captain_integration_v2')
+    job_args << message.id if captain_v2_enabled
+    wait_time = attachment_wait_time(captain_v2_enabled)
 
-    if message.attachments.blank?
+    if wait_time.zero?
       Captain::Conversation::ResponseBuilderJob.perform_later(*job_args)
     else
-      wait_time = calculate_attachment_wait_time
       Captain::Conversation::ResponseBuilderJob.set(wait: wait_time).perform_later(*job_args)
     end
   end
 
-  def calculate_attachment_wait_time
-    attachment_count = message.attachments.size
+  def attachment_wait_time(captain_v2_enabled)
+    attachment_count = captain_v2_enabled ? recent_attachment_count : message.attachments.size
+    return 0.seconds if attachment_count.zero?
+
+    calculate_attachment_wait_time(attachment_count)
+  end
+
+  def recent_attachment_count
+    maximum_wait = (MAX_ATTACHMENT_WAIT_SECONDS + 1).seconds
+
+    conversation.messages.incoming
+                .joins(:attachments)
+                .where(attachments: { created_at: maximum_wait.ago.. })
+                .count
+  end
+
+  def calculate_attachment_wait_time(attachment_count)
     base_wait = 1.second
 
     # Wait longer for more attachments or larger files
@@ -50,7 +67,7 @@ module Enterprise::MessageTemplates::HookExecutionService
   end
 
   def should_process_captain_response?
-    conversation.pending? && message.incoming? && inbox.captain_assistant.present?
+    conversation.pending? && message.captain_response_triggering? && inbox.captain_assistant.present?
   end
 
   def perform_handoff
diff --git a/enterprise/lib/captain/tools/base_public_tool.rb b/enterprise/lib/captain/tools/base_public_tool.rb
index e1f779b36..16a5def4d 100644
--- a/enterprise/lib/captain/tools/base_public_tool.rb
+++ b/enterprise/lib/captain/tools/base_public_tool.rb
@@ -6,6 +6,14 @@ class Captain::Tools::BasePublicTool < Agents::Tool
     super()
   end
 
+  def execute(tool_context, **params)
+    return super unless captain_v2_enabled?
+    return super if safe_to_run_after_new_customer_message?
+    return 'Tool skipped because a newer customer message arrived' if newer_customer_message_arrived?(tool_context.state)
+
+    super
+  end
+
   def active?
     # Public tools are always active
     true
@@ -37,6 +45,28 @@ class Captain::Tools::BasePublicTool < Agents::Tool
     account_scoped(::Contact).find_by(id: contact_id)
   end
 
+  def safe_to_run_after_new_customer_message?
+    false
+  end
+
+  def captain_v2_enabled?
+    @assistant.account.feature_enabled?('captain_integration_v2')
+  end
+
+  def newer_customer_message_arrived?(state)
+    responding_to_message_id = state&.dig(:responding_to_message_id)
+    return false if responding_to_message_id.blank?
+
+    conversation_id = state&.dig(:conversation, :id)
+
+    ::Message.uncached do
+      account_scoped(::Message)
+        .where(conversation_id: conversation_id)
+        .captain_response_triggering
+        .exists?(['messages.id > ?', responding_to_message_id])
+    end
+  end
+
   def log_tool_usage(action, details = {})
     Rails.logger.info do
       "#{self.class.name}: #{action} for assistant #{@assistant&.id} - #{details.inspect}"
diff --git a/enterprise/lib/captain/tools/faq_lookup_tool.rb b/enterprise/lib/captain/tools/faq_lookup_tool.rb
index 2a16e7e99..087e35672 100644
--- a/enterprise/lib/captain/tools/faq_lookup_tool.rb
+++ b/enterprise/lib/captain/tools/faq_lookup_tool.rb
@@ -20,6 +20,10 @@ class Captain::Tools::FaqLookupTool < Captain::Tools::BasePublicTool
 
   private
 
+  def safe_to_run_after_new_customer_message?
+    true
+  end
+
   def record_retrieved_sources(tool_context, responses)
     return if responses.empty?
 
diff --git a/enterprise/lib/captain/tools/handoff_tool.rb b/enterprise/lib/captain/tools/handoff_tool.rb
index f3ca8b1fe..73a76c2d0 100644
--- a/enterprise/lib/captain/tools/handoff_tool.rb
+++ b/enterprise/lib/captain/tools/handoff_tool.rb
@@ -13,7 +13,9 @@ class Captain::Tools::HandoffTool < Captain::Tools::BasePublicTool
                    })
 
     # Use existing handoff mechanism from ResponseBuilderJob
-    trigger_handoff(tool_context, conversation, reason)
+    handoff_result = trigger_handoff(tool_context, conversation, reason)
+    return 'Handoff skipped because a newer customer message arrived' if handoff_result == :stale
+    return 'Handoff skipped because the conversation changed' unless handoff_result == :completed
 
     "Conversation handed off to human support team#{" (Reason: #{reason})" if reason}"
   rescue StandardError => e
@@ -24,30 +26,54 @@ class Captain::Tools::HandoffTool < Captain::Tools::BasePublicTool
   private
 
   def trigger_handoff(tool_context, conversation, reason)
-    # post the reason as a private note
-    note = conversation.messages.create!(
-      message_type: :outgoing,
-      private: true,
-      sender: @assistant,
-      account: conversation.account,
-      inbox: conversation.inbox,
-      content: reason
-    )
+    return trigger_legacy_handoff(tool_context, conversation, reason) unless captain_v2_enabled?
+
+    note = nil
+    handoff_result = conversation.with_lock do
+      next :changed unless conversation.pending?
+      next :stale if newer_customer_message_arrived?(tool_context.state)
+
+      # post the reason as a private note
+      note = conversation.messages.create!(
+        message_type: :outgoing, private: true, sender: @assistant,
+        account: conversation.account, inbox: conversation.inbox, content: reason
+      )
+
+      conversation.bot_handoff!(dispatch_event: false)
+      :completed
+    end
+
+    return handoff_result unless handoff_result == :completed
 
     # Session capture attributes the run to this note so agents can inspect the
     # generation path on the handoff reason instead of the canned follow-up message.
     # A reason-less note has no content and never renders in the dashboard, so
     # leave it unrecorded and let capture fall back to the follow-up message.
-    if reason.present?
-      metadata = tool_context.state[:cw_metadata] ||= {}
-      metadata[:handoff_note_id] = note.id
-    end
+    record_handoff_note(tool_context, note) if reason.present?
 
-    # Trigger the bot handoff (sets status to open + dispatches events)
-    conversation.bot_handoff!
+    tool_context.state[:captain_v2_handoff_tool_completed] = true
+    # Queue the event after the state change commits so notification jobs always see the open conversation.
+    conversation.dispatch_bot_handoff_event
 
     # Send out of office message if applicable (since template messages were suppressed while Captain was handling)
     send_out_of_office_message_if_applicable(conversation)
+    :completed
+  end
+
+  def trigger_legacy_handoff(tool_context, conversation, reason)
+    note = conversation.messages.create!(
+      message_type: :outgoing, private: true, sender: @assistant,
+      account: conversation.account, inbox: conversation.inbox, content: reason
+    )
+    record_handoff_note(tool_context, note) if reason.present?
+    conversation.bot_handoff!
+    send_out_of_office_message_if_applicable(conversation)
+    :completed
+  end
+
+  def record_handoff_note(tool_context, note)
+    metadata = tool_context.state[:cw_metadata] ||= {}
+    metadata[:handoff_note_id] = note.id
   end
 
   def send_out_of_office_message_if_applicable(conversation)
diff --git a/enterprise/lib/captain/tools/http_tool.rb b/enterprise/lib/captain/tools/http_tool.rb
index 70576fb21..4fce42dc9 100644
--- a/enterprise/lib/captain/tools/http_tool.rb
+++ b/enterprise/lib/captain/tools/http_tool.rb
@@ -1,10 +1,9 @@
 require 'agents'
 
-class Captain::Tools::HttpTool < Agents::Tool
+class Captain::Tools::HttpTool < Captain::Tools::BasePublicTool
   def initialize(assistant, custom_tool)
-    @assistant = assistant
     @custom_tool = custom_tool
-    super()
+    super(assistant)
   end
 
   def active?
@@ -24,6 +23,10 @@ class Captain::Tools::HttpTool < Agents::Tool
 
   private
 
+  def safe_to_run_after_new_customer_message?
+    @custom_tool.http_method == 'GET'
+  end
+
   # Limit response size to prevent memory exhaustion and match LLM token limits
   # 1MB of text ≈ 250K tokens, which exceeds most LLM context windows
   MAX_RESPONSE_SIZE = 1.megabyte
diff --git a/spec/enterprise/jobs/captain/conversation/response_builder_job_spec.rb b/spec/enterprise/jobs/captain/conversation/response_builder_job_spec.rb
index 08bb2a50d..946fda149 100644
--- a/spec/enterprise/jobs/captain/conversation/response_builder_job_spec.rb
+++ b/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(:responding_to_message) { conversation.messages.find_by!(content: 'Hello') }
 
     before do
       create(:message, conversation: conversation, content: 'Hello', message_type: :incoming)
@@ -23,6 +24,8 @@ 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(mock_agent_runner_service).to receive(:handoff_completed?).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)
@@ -355,15 +358,57 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
         allow(account).to receive(:feature_enabled?).with('captain_integration_v2').and_return(true)
       end
 
-      it 'uses Captain::Assistant::AgentRunnerService' do
-        expect(Captain::Assistant::AgentRunnerService).to receive(:new).with(
-          assistant: assistant,
-          conversation: conversation
-        )
-        expect(Captain::Llm::AssistantChatService).not_to receive(:new)
+      context 'with message burst protection' do
+        it 'passes the responding message id to the runner' do
+          expect(Captain::Assistant::AgentRunnerService).to receive(:new).with(
+            assistant: assistant,
+            conversation: conversation,
+            responding_to_message_id: responding_to_message.id
+          )
 
-        described_class.perform_now(conversation, assistant)
-        expect(conversation.messages.last.content).to eq('Hey, welcome to Captain V2')
+          described_class.perform_now(conversation, assistant, responding_to_message.id)
+        end
+
+        it 'discards a response when the runner sees a newer customer message' 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, responding_to_message.id)
+
+          expect(conversation.messages.outgoing.count).to eq(0)
+          expect(account.reload.usage_limits[:captain][:responses][:consumed]).to eq(0)
+        end
+
+        it 'checks freshness itself after 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(false)
+
+          described_class.perform_now(conversation, assistant, responding_to_message.id)
+
+          expect(conversation.messages.outgoing.count).to eq(0)
+          expect(account.reload.usage_limits[:captain][:responses][:consumed]).to eq(0)
+        end
+
+        it 'keeps the pending response fresh when an email auto reply arrives' do
+          create(
+            :message,
+            conversation: conversation,
+            message_type: :incoming,
+            content_type: :incoming_email,
+            content_attributes: { email: { auto_reply: true } }
+          )
+
+          described_class.perform_now(conversation, assistant, responding_to_message.id)
+
+          expect(conversation.messages.outgoing.last.content).to eq('Hey, welcome to Captain V2')
+          expect(account.reload.usage_limits[:captain][:responses][:consumed]).to eq(1)
+        end
       end
 
       it 'passes message history with resolution markers to agent runner service' do
@@ -417,9 +462,11 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
       before do
         allow(account).to receive(:feature_enabled?).and_return(false)
         allow(account).to receive(:feature_enabled?).with('captain_integration_v2').and_return(true)
+        allow(mock_agent_runner_service).to receive(:handoff_completed?).and_return(true)
       end
 
-      it 'creates a public handoff message visible to the customer' do
+      it 'creates a public handoff message after the generated response is discarded' do
+        allow(mock_agent_runner_service).to receive(:response_discarded?).and_return(true)
         allow(mock_agent_runner_service).to receive(:generate_response) do
           conversation.update!(status: :open)
           { 'response' => 'Let me connect you', 'handoff_tool_called' => true }
@@ -469,6 +516,7 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
       end
 
       it 'does not hand off when handoff_tool_called is false' do
+        allow(mock_agent_runner_service).to receive(:handoff_completed?).and_return(false)
         allow(mock_agent_runner_service).to receive(:generate_response).and_return({
                                                                                      'response' => 'Hi! How can I help you?',
                                                                                      'handoff_tool_called' => false
@@ -482,6 +530,7 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
       end
 
       it 'falls back to a full V1 handoff when HandoffTool fired but failed to commit' do
+        allow(mock_agent_runner_service).to receive(:handoff_completed?).and_return(false)
         allow(mock_agent_runner_service).to receive(:generate_response).and_return({
                                                                                      'response' => 'I tried to hand off',
                                                                                      'handoff_tool_called' => true
@@ -548,6 +597,7 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
       end
 
       it 'creates a zero-credit session when the handoff tool fired' do
+        allow(mock_agent_runner_service).to receive(:handoff_completed?).and_return(true)
         allow(mock_agent_runner_service).to receive(:generate_response) do
           conversation.update!(status: :open)
           { 'response' => 'Let me connect you', 'handoff_tool_called' => true }
@@ -562,6 +612,7 @@ RSpec.describe Captain::Conversation::ResponseBuilderJob, type: :job do
       end
 
       it 'attributes the handoff session to the private reason note when the tool recorded one' do
+        allow(mock_agent_runner_service).to receive(:handoff_completed?).and_return(true)
         handoff_note = create(:message, conversation: conversation, account: account, message_type: :outgoing,
                                         private: true, sender: assistant, content: 'Needs a human')
         run_context[:state][:cw_metadata][:handoff_note_id] = handoff_note.id
diff --git a/spec/enterprise/lib/captain/tools/add_private_note_tool_spec.rb b/spec/enterprise/lib/captain/tools/add_private_note_tool_spec.rb
index cfce1a7d1..5c0c5321d 100644
--- a/spec/enterprise/lib/captain/tools/add_private_note_tool_spec.rb
+++ b/spec/enterprise/lib/captain/tools/add_private_note_tool_spec.rb
@@ -116,6 +116,36 @@ RSpec.describe Captain::Tools::AddPrivateNoteTool, type: :model do
     end
   end
 
+  describe '#execute' do
+    let(:responding_to_message) do
+      create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming)
+    end
+    let(:tool_context) do
+      Struct.new(:state).new({ conversation: { id: conversation.id }, responding_to_message_id: responding_to_message.id })
+    end
+
+    before do
+      responding_to_message
+      create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming)
+    end
+
+    it 'keeps legacy public-tool side effects for Captain V1' do
+      expect do
+        result = tool.execute(tool_context, note: 'Keep the legacy side effect')
+        expect(result).to eq('Private note added successfully')
+      end.to change(Message, :count).by(1)
+    end
+
+    it 'skips stale public-tool side effects for Captain V2' do
+      account.enable_features!(:captain_integration_v2)
+
+      expect do
+        result = tool.execute(tool_context, note: 'Do not create this note')
+        expect(result).to eq('Tool skipped because a newer customer message arrived')
+      end.not_to change(Message, :count)
+    end
+  end
+
   describe '#active?' do
     it 'returns true for public tools' do
       expect(tool.active?).to be true
diff --git a/spec/enterprise/lib/captain/tools/handoff_tool_spec.rb b/spec/enterprise/lib/captain/tools/handoff_tool_spec.rb
index db5ee462c..2047a2da1 100644
--- a/spec/enterprise/lib/captain/tools/handoff_tool_spec.rb
+++ b/spec/enterprise/lib/captain/tools/handoff_tool_spec.rb
@@ -7,7 +7,7 @@ RSpec.describe Captain::Tools::HandoffTool, type: :model do
   let(:user) { create(:user, account: account) }
   let(:inbox) { create(:inbox, account: account) }
   let(:contact) { create(:contact, account: account) }
-  let(:conversation) { create(:conversation, account: account, inbox: inbox, contact: contact) }
+  let(:conversation) { create(:conversation, account: account, inbox: inbox, contact: contact, status: :pending) }
   let(:tool_context) { Struct.new(:state).new({ conversation: { id: conversation.id } }) }
 
   describe '#description' do
@@ -28,6 +28,98 @@ RSpec.describe Captain::Tools::HandoffTool, type: :model do
 
   describe '#perform' do
     context 'when conversation exists' do
+      context 'when Captain is responding to a customer message' do
+        let(:responding_to_message) do
+          create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming)
+        end
+        let(:tool_context) do
+          Struct.new(:state).new({ conversation: { id: conversation.id }, responding_to_message_id: responding_to_message.id })
+        end
+
+        before do
+          account.enable_features!(:captain_integration_v2)
+          responding_to_message
+        end
+
+        it 'hands off when no newer customer message has arrived' do
+          found_conversation = Conversation.find(conversation.id)
+          scoped_conversations = Conversation.where(account_id: assistant.account_id)
+          allow(Conversation).to receive(:where).with(account_id: assistant.account_id).and_return(scoped_conversations)
+          allow(scoped_conversations).to receive(:find_by).with(id: conversation.id).and_return(found_conversation)
+          expect(found_conversation).to receive(:with_lock).and_call_original
+
+          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)
+          expect(tool_context.state[:captain_v2_handoff_tool_completed]).to be true
+        end
+
+        it 'dispatches the handoff event after leaving the lock transaction' do
+          found_conversation = Conversation.find(conversation.id)
+          scoped_conversations = Conversation.where(account_id: assistant.account_id)
+          allow(Conversation).to receive(:where).with(account_id: assistant.account_id).and_return(scoped_conversations)
+          allow(scoped_conversations).to receive(:find_by).with(id: conversation.id).and_return(found_conversation)
+          open_transactions_before_handoff = ActiveRecord::Base.connection.open_transactions
+
+          expect(found_conversation).to receive(:dispatch_bot_handoff_event) do
+            expect(ActiveRecord::Base.connection.open_transactions).to eq(open_transactions_before_handoff)
+          end
+
+          tool.perform(tool_context, reason: 'Customer needs specialized support')
+        end
+
+        it 'notifies inbox members after the committed handoff' do
+          create(:inbox_member, user: user, inbox: inbox)
+          notification_setting = user.notification_settings.find_by!(account: account)
+          notification_setting.selected_email_flags = [:email_conversation_creation]
+          notification_setting.selected_push_flags = []
+          notification_setting.save!
+
+          perform_enqueued_jobs do
+            tool.perform(tool_context, reason: 'Customer needs specialized support')
+          end
+
+          expect(user.notifications.find_by(primary_actor: conversation, notification_type: :conversation_creation)).to be_present
+        end
+
+        it 'skips the handoff when a newer message has arrived' do
+          conversation.update!(status: :pending)
+          create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming)
+
+          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 Captain V1' do
+        let(:tool_context) do
+          Struct.new(:state).new({ conversation: { id: conversation.id }, responding_to_message_id: responding_to_message.id })
+        end
+        let(:responding_to_message) do
+          create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming)
+        end
+
+        it 'uses the legacy handoff without a lock or stale-message guard' do
+          responding_to_message
+          create(:message, conversation: conversation, account: account, inbox: inbox, message_type: :incoming)
+          found_conversation = Conversation.find(conversation.id)
+          scoped_conversations = Conversation.where(account_id: assistant.account_id)
+          allow(Conversation).to receive(:where).with(account_id: assistant.account_id).and_return(scoped_conversations)
+          allow(scoped_conversations).to receive(:find_by).with(id: conversation.id).and_return(found_conversation)
+          expect(found_conversation).not_to receive(:with_lock)
+
+          result = tool.perform(tool_context, reason: 'Customer needs specialized support')
+
+          expect(result).to include('Conversation handed off')
+          expect(conversation.reload.status).to eq('open')
+          expect(tool_context.state).not_to have_key(:captain_v2_handoff_tool_completed)
+        end
+      end
+
       context 'with reason provided' do
         it 'creates a private note with reason and hands off conversation' do
           reason = 'Customer needs specialized support'
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 e88d040b3..304a7dd34 100644
--- a/spec/enterprise/services/captain/assistant/agent_runner_service_spec.rb
+++ b/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 message id it is responding to' do
+      service = described_class.new(assistant: assistant, conversation: conversation, responding_to_message_id: 123)
+
+      expect(service.instance_variable_get(:@responding_to_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 responding message id to the runner state' do
+      service = described_class.new(assistant: assistant, conversation: conversation, responding_to_message_id: 123)
+
+      expect(mock_runner).to receive(:run).with(
+        'I need help with my account',
+        context: hash_including(state: hash_including(responding_to_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,19 @@ 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).not_to have_key('langfuse.observation.metadata.discarded')
+    end
+
+    it 'marks a protected generation as not discarded when no newer message has arrived' do
+      responding_to_message = create(:message, conversation: conversation, message_type: :incoming)
+      runner_service = described_class.new(assistant: assistant, conversation: conversation,
+                                           responding_to_message_id: responding_to_message.id)
+      attribute_provider = Captain::Assistant::InstrumentationAttributeProvider.new(runner_service)
+      message = instance_double(RubyLLM::Message, tool_calls: {})
+
+      attributes = attribute_provider.generation_attributes(nil, nil, message)
+
+      expect(attributes['langfuse.observation.metadata.discarded']).to eq('false')
     end
 
     it 'marks tool call generations separately from final responses' do
@@ -456,6 +487,19 @@ 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
+      responding_to_message = create(:message, conversation: conversation, message_type: :incoming)
+      runner_service = described_class.new(assistant: assistant, conversation: conversation,
+                                           responding_to_message_id: responding_to_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
@@ -588,18 +632,40 @@ 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)
 
-      context_wrapper = Struct.new(:context).new({})
+      context_wrapper = Struct.new(:context).new({ state: { captain_v2_handoff_tool_completed: true } })
 
       expect(tool_complete_callback).not_to be_nil
       tool_complete_callback.call(Captain::Tools::HandoffTool.new(assistant).name, 'ok', context_wrapper)
 
       expect(context_wrapper.context[:captain_v2_handoff_tool_called]).to be true
+      expect(service.handoff_completed?).to be true
     end
 
-    it 'does not register OTEL run callback when OTEL is disabled' do
+    it 'tracks discarded responses when OTEL is disabled' do
+      responding_to_message = create(:message, conversation: conversation, message_type: :incoming)
+      service = described_class.new(assistant: assistant, conversation: conversation, responding_to_message_id: responding_to_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)
+      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 'does not register a run callback when OTEL and burst protection are disabled' do
       service = described_class.new(assistant: assistant, conversation: conversation)
       runner = instance_double(Agents::AgentRunner)
 
@@ -632,6 +698,34 @@ RSpec.describe Captain::Assistant::AgentRunnerService do
       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
+      responding_to_message = create(:message, conversation: conversation, message_type: :incoming)
+      service = described_class.new(assistant: assistant, conversation: conversation, responding_to_message_id: responding_to_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
diff --git a/spec/enterprise/services/enterprise/message_templates/hook_execution_service_spec.rb b/spec/enterprise/services/enterprise/message_templates/hook_execution_service_spec.rb
index 9eacb0ba1..94894dd6e 100644
--- a/spec/enterprise/services/enterprise/message_templates/hook_execution_service_spec.rb
+++ b/spec/enterprise/services/enterprise/message_templates/hook_execution_service_spec.rb
@@ -21,10 +21,68 @@ 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)
+      it 'keeps the legacy job arguments for Captain V1' do
+        allow(Captain::Conversation::ResponseBuilderJob).to receive(:perform_later)
 
         create(:message, conversation: conversation, message_type: :incoming, account: account)
+
+        expect(Captain::Conversation::ResponseBuilderJob).to have_received(:perform_later).with(conversation, assistant)
+      end
+
+      it 'passes the responding message id for Captain V2' do
+        account.enable_features!(:captain_integration_v2)
+        allow(Captain::Conversation::ResponseBuilderJob).to receive(:perform_later)
+
+        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
+
+      it 'does not lock or schedule a job for an email auto reply' do
+        account.enable_features!(:captain_integration_v2)
+        allow(Captain::Conversation::ResponseBuilderJob).to receive(:perform_later)
+
+        customer_message = create(:message, conversation: conversation, message_type: :incoming, account: account)
+        auto_reply = build(
+          :message,
+          conversation: conversation,
+          message_type: :incoming,
+          content_type: :incoming_email,
+          content_attributes: { email: { auto_reply: true } },
+          account: account
+        )
+        auto_reply.save!
+
+        expect(Captain::Conversation::ResponseBuilderJob).to have_received(:perform_later).once
+        expect(Captain::Conversation::ResponseBuilderJob).to have_received(:perform_later).with(conversation, assistant, customer_message.id)
+        expect(conversation.messages.captain_response_triggering).to contain_exactly(customer_message)
+        expect(conversation.messages.captain_response_triggering).not_to include(auto_reply)
+      end
+    end
+
+    context 'when calculating attachment wait time' do
+      let(:configured_job) { instance_double(ActiveJob::ConfiguredJob, perform_later: true) }
+
+      before do
+        allow(Captain::Conversation::ResponseBuilderJob).to receive(:set).and_return(configured_job)
+      end
+
+      it 'uses only the current message attachments for Captain V1' do
+        create(:message, :with_attachment, conversation: conversation, message_type: :incoming, account: account)
+        create(:message, :with_attachment, conversation: conversation, message_type: :incoming, account: account)
+
+        expect(Captain::Conversation::ResponseBuilderJob).to have_received(:set).with(wait: 2.seconds).twice
+        expect(Captain::Conversation::ResponseBuilderJob).not_to have_received(:set).with(wait: 3.seconds)
+      end
+
+      it 'recalculates the wait from recent burst attachments for Captain V2' do
+        account.enable_features!(:captain_integration_v2)
+
+        create(:message, :with_attachment, conversation: conversation, message_type: :incoming, account: account)
+        create(:message, :with_attachment, conversation: conversation, message_type: :incoming, account: account)
+
+        expect(Captain::Conversation::ResponseBuilderJob).to have_received(:set).with(wait: 2.seconds).once
+        expect(Captain::Conversation::ResponseBuilderJob).to have_received(:set).with(wait: 3.seconds).once
       end
     end