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