diff --git a/app/services/data_imports/intercom/importer.rb b/app/services/data_imports/intercom/importer.rb
index 049508c66..fab61478d 100644
--- a/app/services/data_imports/intercom/importer.rb
+++ b/app/services/data_imports/intercom/importer.rb
@@ -18,7 +18,6 @@ class DataImports::Intercom::Importer
   TRUNCATED_PARTS_ERROR_CODE = 'DataImports::Intercom::TruncatedConversationParts'.freeze
   E164_REGEX = /\A\+[1-9]\d{1,14}\z/
   INTERCOM_NUMBER_REGEX = /\A[1-9]\d{1,14}\z/
-  REGULAR_MESSAGE_PART_TYPES = %w[comment note source].freeze
 
   def initialize(data_import:, run_id: nil)
     @data_import = data_import
@@ -164,8 +163,7 @@ class DataImports::Intercom::Importer
     if mapped_conversation && mapping.data_import_id != @data_import.id
       skip_already_imported_item(item, mapping, already_handled: already_handled)
       reconcile_item_stats('conversation') if already_handled
-      import_source_message(conversation, mapped_conversation, contact)
-      return unless import_conversation_parts(conversation, mapped_conversation, contact)
+      return unless import_conversation_messages(conversation, mapped_conversation, contact)
 
       update_conversation_activity(mapped_conversation)
       return
@@ -182,8 +180,7 @@ class DataImports::Intercom::Importer
       increment_stat('conversations', 'imported')
     end
 
-    import_source_message(conversation, chatwoot_conversation, contact)
-    return unless import_conversation_parts(conversation, chatwoot_conversation, contact)
+    return unless import_conversation_messages(conversation, chatwoot_conversation, contact)
 
     update_conversation_activity(chatwoot_conversation)
   rescue StandardError => e
@@ -397,61 +394,85 @@ class DataImports::Intercom::Importer
     end
   end
 
-  def import_source_message(conversation, chatwoot_conversation, contact)
-    source = conversation['source'].to_h
-    return unless source_message_importable?(source)
-
-    message_source_id = "conversation:#{source_id_for(conversation)}:source:#{source['id'].presence || 'initial'}"
-    source_part = source.merge('part_type' => 'source', 'created_at' => conversation['created_at'])
-    import_message(chatwoot_conversation, contact, source_part, message_source_id)
-  end
-
-  def import_conversation_parts(conversation, chatwoot_conversation, contact)
+  def import_conversation_messages(conversation, chatwoot_conversation, contact)
     parts_payload = conversation['conversation_parts'].to_h
     parts = Array(parts_payload['conversation_parts'])
+    batch_builder = DataImports::Intercom::MessageBatchBuilder.new(
+      data_import: @data_import,
+      conversation: chatwoot_conversation,
+      source_conversation: conversation
+    )
+    batch = begin
+      with_query_timeout_retry { batch_builder.perform }
+    rescue ActiveRecord::QueryCanceled
+      nil
+    end
+    return import_conversation_messages_individually(conversation, chatwoot_conversation, contact, batch_builder, parts.size) if batch.nil?
+
+    batch.source_entries.each { |entry| import_message(chatwoot_conversation, contact, entry) }
     record_truncated_conversation_parts(conversation, parts.size)
 
-    parts.each do |part|
+    batch.part_entries.each do |entry|
       return false unless continue_import_with_heartbeat?
 
-      message_source_id = "conversation:#{source_id_for(conversation)}:part:#{part['id']}"
-      import_message(chatwoot_conversation, contact, part, message_source_id)
+      import_message(chatwoot_conversation, contact, entry)
     end
     return false if import_stopped?
 
     true
   end
 
-  def import_message(conversation, contact, part, message_source_id)
+  def import_conversation_messages_individually(conversation, chatwoot_conversation, contact, batch_builder, parts_count)
+    source_entries, part_entries = batch_builder.unprepared_entries.partition { |entry| entry[:part]['part_type'] == 'source' }
+    source_entries.each { |entry| import_unprepared_message(chatwoot_conversation, contact, batch_builder, entry) }
+    record_truncated_conversation_parts(conversation, parts_count)
+
+    part_entries.each do |entry|
+      return false unless continue_import_with_heartbeat?
+
+      import_unprepared_message(chatwoot_conversation, contact, batch_builder, entry)
+    end
+    return false if import_stopped?
+
+    true
+  end
+
+  def import_unprepared_message(conversation, contact, batch_builder, source_entry)
+    entry = with_query_timeout_retry { batch_builder.perform([source_entry]).entries.first }
+    import_message(conversation, contact, entry)
+  rescue StandardError => e
+    fail_message(conversation, source_entry[:source_id], source_entry[:part], e)
+  end
+
+  def import_message(conversation, contact, entry)
     with_query_timeout_retry do
-      mapping = find_mapping('message', message_source_id)
-      if mapping && message_mapping_handled?(mapping, part)
-        if mapping.data_import_id == @data_import.id
-          reconcile_current_run_message_mapping(conversation, mapping, part)
-        else
-          skip_existing_message_mapping(conversation, mapping, part)
-        end
+      case entry.classification
+      when :current_import
+        reconcile_current_run_message_mapping(conversation, entry.mapping, entry.part)
+      when :previous_import
+        skip_existing_message_mapping(conversation, entry.mapping, entry.part)
+      when :repairable_stale_mapping, :existing_message, :new_message
+        create_message(conversation, contact, entry)
       else
-        create_message(conversation, contact, part, message_source_id)
+        raise ArgumentError, "Unsupported Intercom message classification: #{entry.classification}"
       end
     end
   rescue StandardError => e
-    fail_message(conversation, message_source_id, part, e)
+    fail_message(conversation, entry.source_id, entry.part, e)
   end
 
-  def create_message(conversation, contact, part, message_source_id)
-    content = content_for(part)
-    return record_skipped_message(conversation, message_source_id, part) if content.blank?
+  def create_message(conversation, contact, entry)
+    content = content_for(entry.part)
+    return record_skipped_message(conversation, entry) if content.blank?
 
-    attrs = message_attributes(conversation, contact, part, message_source_id, content)
-    message = nil
+    attrs = message_attributes(conversation, contact, entry.part, entry.source_id, content)
+    message = entry.message
     Message.transaction do
-      message = conversation.messages.find_by(source_id: attrs[:source_id])
       unless message
         result = Message.insert_all!([attrs], returning: %w[id])
         message = Message.find(result.rows.first.first)
       end
-      record_mapping('message', message_source_id, message, metadata: message_metadata(part))
+      record_message_mapping(entry, message)
     end
     increment_stat('messages', 'imported')
     reindex_message_for_search(message)
@@ -466,13 +487,12 @@ class DataImports::Intercom::Importer
     Rails.logger.warn("Intercom import message reindex failed for message #{message.id}: #{e.class} - #{e.message}")
   end
 
-  def record_skipped_message(conversation, message_source_id, part)
-    mapping = find_mapping('message', message_source_id)
-    if mapping
-      already_recorded = skip_log_recorded?('message', message_source_id, SKIPPED_MESSAGE_ERROR_CODE)
-      record_skipped_message_log(conversation, message_source_id, part)
+  def record_skipped_message(conversation, entry)
+    if entry.mapping
+      already_recorded = skip_log_recorded?('message', entry.source_id, SKIPPED_MESSAGE_ERROR_CODE)
+      record_skipped_message_log(conversation, entry.source_id, entry.part)
       increment_stat('messages', 'skipped') unless already_recorded
-      return mapping.chatwoot_record
+      return entry.message
     end
 
     DataImportMapping.create!(
@@ -480,15 +500,30 @@ class DataImports::Intercom::Importer
       data_import: @data_import,
       source_provider: PROVIDER,
       source_object_type: 'message',
-      source_object_id: message_source_id,
+      source_object_id: entry.source_id,
       chatwoot_record_type: 'Conversation',
       chatwoot_record_id: conversation.id,
-      metadata: message_metadata(part).merge(skipped: true, reason: 'blank_or_unsupported_intercom_part')
+      metadata: message_metadata(entry.part).merge(skipped: true, reason: 'blank_or_unsupported_intercom_part')
     )
-    record_skipped_message_log(conversation, message_source_id, part)
+    record_skipped_message_log(conversation, entry.source_id, entry.part)
     increment_stat('messages', 'skipped')
   end
 
+  def record_message_mapping(entry, message)
+    (entry.mapping || DataImportMapping.new(
+      account: @account,
+      source_provider: PROVIDER,
+      source_object_type: 'message',
+      source_object_id: entry.source_id
+    )).tap do |mapping|
+      mapping.data_import = @data_import
+      mapping.chatwoot_record_type = 'Message'
+      mapping.chatwoot_record_id = message.id
+      mapping.metadata = message_metadata(entry.part)
+      mapping.save!
+    end
+  end
+
   def message_attributes(conversation, contact, part, message_source_id, content)
     message_type = message_type_for(part)
     created_at = timestamp_for(part['created_at'])
@@ -529,8 +564,7 @@ class DataImports::Intercom::Importer
   end
 
   def activity_part?(part)
-    part_type = part['part_type'].to_s
-    part_type.present? && REGULAR_MESSAGE_PART_TYPES.exclude?(part_type)
+    DataImports::Intercom::MessageBatchBuilder.activity_part?(part)
   end
 
   def message_content(part)
@@ -713,12 +747,6 @@ class DataImports::Intercom::Importer
     end
   end
 
-  def message_mapping_handled?(mapping, part)
-    return false if mapping.metadata['skipped'] && activity_part?(part)
-
-    mapping.metadata['skipped'] || mapping.chatwoot_record.present?
-  end
-
   def fail_item(item, error)
     increment_stat('errors', 'count')
     item&.update!(status: :failed, last_error_code: error.class.name, last_error_message: error.message)
@@ -816,7 +844,7 @@ class DataImports::Intercom::Importer
   end
 
   def source_message_importable?(source)
-    source['body'].present? || source['subject'].present? || source['attachments'].present?
+    DataImports::Intercom::MessageBatchBuilder.source_message_importable?(source)
   end
 
   def skipped_message_log_message(part)
diff --git a/app/services/data_imports/intercom/message_batch_builder.rb b/app/services/data_imports/intercom/message_batch_builder.rb
new file mode 100644
index 000000000..852b51c3c
--- /dev/null
+++ b/app/services/data_imports/intercom/message_batch_builder.rb
@@ -0,0 +1,134 @@
+class DataImports::Intercom::MessageBatchBuilder
+  PROVIDER = 'intercom'.freeze
+  REGULAR_PART_TYPES = %w[comment note source].freeze
+
+  Entry = Struct.new(:source_id, :part, :position, :mapping, :message, :classification, keyword_init: true) do
+    def source?
+      part['part_type'] == 'source'
+    end
+  end
+
+  Batch = Struct.new(:items, keyword_init: true) do
+    def entries
+      items
+    end
+
+    def source_entries
+      items.select(&:source?)
+    end
+
+    def part_entries
+      items.reject(&:source?)
+    end
+  end
+
+  def self.activity_part?(part)
+    part_type = part['part_type'].to_s
+    part_type.present? && REGULAR_PART_TYPES.exclude?(part_type)
+  end
+
+  def self.source_message_importable?(source)
+    source['body'].present? || source['subject'].present? || source['attachments'].present?
+  end
+
+  def initialize(data_import:, conversation:, source_conversation:)
+    @data_import = data_import
+    @account = data_import.account
+    @conversation = conversation
+    @source_conversation = source_conversation
+  end
+
+  def perform(source_entries = unprepared_entries)
+    return Batch.new(items: []) if source_entries.empty?
+
+    mappings = message_mappings(source_entries)
+    messages = messages_for(source_entries, mappings)
+
+    Batch.new(items: source_entries.map do |source_entry|
+      build_entry(source_entry, source_entry[:position], mappings, messages)
+    end)
+  end
+
+  def unprepared_entries
+    ordered_source_entries.map.with_index { |entry, position| entry.merge(position: position) }
+  end
+
+  private
+
+  def ordered_source_entries
+    entries = []
+    source = @source_conversation['source'].to_h
+    if self.class.source_message_importable?(source)
+      entries << {
+        source_id: "conversation:#{source_conversation_id}:source:#{source['id'].presence || 'initial'}",
+        part: source.merge('part_type' => 'source', 'created_at' => @source_conversation['created_at'])
+      }
+    end
+
+    conversation_parts.each do |part|
+      entries << { source_id: "conversation:#{source_conversation_id}:part:#{part['id']}", part: part }
+    end
+    entries
+  end
+
+  def conversation_parts
+    Array(@source_conversation.dig('conversation_parts', 'conversation_parts'))
+  end
+
+  def source_conversation_id
+    @source_conversation['id'].presence || @source_conversation['external_id'].presence || @source_conversation['email'].presence
+  end
+
+  def message_mappings(source_entries)
+    DataImportMapping.where(
+      account: @account,
+      source_provider: PROVIDER,
+      source_object_type: 'message',
+      source_object_id: source_entries.pluck(:source_id)
+    ).index_by(&:source_object_id)
+  end
+
+  def messages_for(source_entries, mappings)
+    mapped_message_ids = mappings.values.filter_map do |mapping|
+      mapping.chatwoot_record_id if mapping.chatwoot_record_type == 'Message'
+    end
+    chatwoot_source_ids = source_entries.map { |entry| "intercom:#{entry[:source_id]}" }
+    messages = Message.where(id: mapped_message_ids).or(
+      Message.where(conversation_id: @conversation.id, source_id: chatwoot_source_ids)
+    ).to_a
+
+    {
+      by_id: messages.index_by(&:id),
+      by_source_id: messages.index_by(&:source_id)
+    }
+  end
+
+  def build_entry(source_entry, position, mappings, messages)
+    source_id = source_entry[:source_id]
+    mapping = mappings[source_id]
+    mapped_message = messages[:by_id][mapping.chatwoot_record_id] if mapping&.chatwoot_record_type == 'Message'
+    existing_message = messages[:by_source_id]["intercom:#{source_id}"]
+
+    Entry.new(
+      source_id: source_id,
+      part: source_entry[:part],
+      position: position,
+      mapping: mapping,
+      message: mapped_message || existing_message,
+      classification: classification_for(mapping, mapped_message, existing_message, source_entry[:part])
+    )
+  end
+
+  def classification_for(mapping, mapped_message, existing_message, part)
+    return existing_message.present? ? :existing_message : :new_message if mapping.blank?
+    return :repairable_stale_mapping unless mapping_handled?(mapping, mapped_message, part)
+
+    mapping.data_import_id == @data_import.id ? :current_import : :previous_import
+  end
+
+  def mapping_handled?(mapping, mapped_message, part)
+    return false if mapping.metadata['skipped'] && self.class.activity_part?(part)
+
+    mapping.metadata['skipped'] || mapped_message.present?
+  end
+end
diff --git a/spec/services/data_imports/intercom/importer_spec.rb b/spec/services/data_imports/intercom/importer_spec.rb
index 659a038e5..02e525a1b 100644
--- a/spec/services/data_imports/intercom/importer_spec.rb
+++ b/spec/services/data_imports/intercom/importer_spec.rb
@@ -367,8 +367,8 @@ RSpec.describe DataImports::Intercom::Importer do
       importer = described_class.new(data_import: data_import, run_id: run_id)
       allow(importer).to receive(:create_message).and_wrap_original do |method, *args|
         method.call(*args).tap do
-          part = args[2]
-          next unless part['id'] == 'part_2'
+          entry = args[2]
+          next unless entry.part['id'] == 'part_2'
 
           DataImport.find(data_import.id).update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
         end
@@ -441,11 +441,7 @@ RSpec.describe DataImports::Intercom::Importer do
 
     it 'rolls back a newly inserted message when mapping persistence fails', :aggregate_failures do
       importer = described_class.new(data_import: data_import)
-      allow(importer).to receive(:record_mapping).and_wrap_original do |method, object_type, source_id, record, metadata:|
-        raise StandardError, 'mapping failed' if object_type == 'message'
-
-        method.call(object_type, source_id, record, metadata: metadata)
-      end
+      allow(importer).to receive(:record_message_mapping).and_raise(StandardError, 'mapping failed')
 
       importer.import_conversations_page
 
@@ -486,13 +482,13 @@ RSpec.describe DataImports::Intercom::Importer do
       mapping_attempts = 0
       target_source_id = 'conversation:conversation_1:part:part_1'
       allow(importer).to receive(:sleep)
-      allow(importer).to receive(:record_mapping).and_wrap_original do |method, object_type, source_id, record, metadata:|
-        if object_type == 'message' && source_id == target_source_id
+      allow(importer).to receive(:record_message_mapping).and_wrap_original do |method, entry, message|
+        if entry.source_id == target_source_id
           mapping_attempts += 1
           raise ActiveRecord::QueryCanceled, 'statement timeout' if mapping_attempts == 1
         end
 
-        method.call(object_type, source_id, record, metadata: metadata)
+        method.call(entry, message)
       end
 
       importer.import_conversations_page
@@ -511,13 +507,13 @@ RSpec.describe DataImports::Intercom::Importer do
       mapping_attempts = 0
       target_source_id = 'conversation:conversation_1:part:part_1'
       allow(importer).to receive(:sleep)
-      allow(importer).to receive(:record_mapping).and_wrap_original do |method, object_type, source_id, record, metadata:|
-        if object_type == 'message' && source_id == target_source_id
+      allow(importer).to receive(:record_message_mapping).and_wrap_original do |method, entry, message|
+        if entry.source_id == target_source_id
           mapping_attempts += 1
           raise ActiveRecord::QueryCanceled, 'statement timeout'
         end
 
-        method.call(object_type, source_id, record, metadata: metadata)
+        method.call(entry, message)
       end
 
       importer.import_conversations_page
@@ -694,6 +690,33 @@ RSpec.describe DataImports::Intercom::Importer do
       expect(message_mappings.filter_map(&:chatwoot_record).count).to eq(3)
     end
 
+    it 'repairs a missing mapping without recreating the existing message', :aggregate_failures do
+      described_class.new(data_import: data_import).import_conversations_page
+      conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
+      message = conversation.messages.find_by!(source_id: 'intercom:conversation:conversation_1:part:part_1')
+      DataImportMapping.find_by!(
+        account: account,
+        source_provider: 'intercom',
+        source_object_type: 'message',
+        source_object_id: 'conversation:conversation_1:part:part_1'
+      ).destroy!
+      importer = described_class.new(data_import: data_import)
+      allow(importer).to receive(:find_mapping).and_call_original
+
+      importer.import_conversations_page
+
+      repaired_mapping = DataImportMapping.find_by!(
+        account: account,
+        source_provider: 'intercom',
+        source_object_type: 'message',
+        source_object_id: 'conversation:conversation_1:part:part_1'
+      )
+      expect(conversation.messages.where(source_id: message.source_id).count).to eq(1)
+      expect(repaired_mapping.chatwoot_record).to eq(message)
+      expect(importer).not_to have_received(:find_mapping).with('message', anything)
+      expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
+    end
+
     it 'updates conversation activity when a later import adds new messages to the mapped conversation', :aggregate_failures do
       new_part = {
         'id' => 'part_3',
diff --git a/spec/services/data_imports/intercom/message_batch_builder_spec.rb b/spec/services/data_imports/intercom/message_batch_builder_spec.rb
new file mode 100644
index 000000000..dc61cbabb
--- /dev/null
+++ b/spec/services/data_imports/intercom/message_batch_builder_spec.rb
@@ -0,0 +1,178 @@
+require 'rails_helper'
+
+RSpec.describe DataImports::Intercom::MessageBatchBuilder do
+  let(:account) { create(:account) }
+  let(:data_import) { create(:data_import, :intercom, account: account) }
+  let(:conversation) { create(:conversation, account: account) }
+  let(:source_conversation) do
+    {
+      'id' => 'conversation_1',
+      'created_at' => 1_700_000_000,
+      'source' => {
+        'id' => 'source_1',
+        'part_type' => 'conversation',
+        'body' => '<p>Initial message</p>',
+        'created_at' => 1_700_000_000
+      },
+      'conversation_parts' => {
+        'conversation_parts' => [
+          {
+            'id' => 'part_1',
+            'part_type' => 'comment',
+            'body' => '<p>First reply</p>',
+            'created_at' => 1_700_000_000
+          },
+          {
+            'id' => 'part_2',
+            'part_type' => 'note',
+            'body' => '<p>Internal note</p>',
+            'created_at' => 1_700_000_100
+          }
+        ]
+      }
+    }
+  end
+  let(:builder) do
+    described_class.new(
+      data_import: data_import,
+      conversation: conversation,
+      source_conversation: source_conversation
+    )
+  end
+
+  it 'preserves source order and prefetches mappings and messages once per batch', :aggregate_failures do
+    sql_queries = []
+    subscriber = lambda do |_name, _start, _finish, _id, payload|
+      sql_queries << payload unless payload[:name] == 'SCHEMA'
+    end
+    batch_builder = builder
+
+    batch = ActiveSupport::Notifications.subscribed(subscriber, 'sql.active_record') { batch_builder.perform }
+
+    expect(batch.entries.map(&:source_id)).to eq(
+      %w[
+        conversation:conversation_1:source:source_1
+        conversation:conversation_1:part:part_1
+        conversation:conversation_1:part:part_2
+      ]
+    )
+    expect(batch.entries.map(&:position)).to eq([0, 1, 2])
+    expect(batch.entries.map(&:classification)).to all(eq(:new_message))
+    expect(sql_queries.count { |query| query[:name] == 'DataImportMapping Load' }).to eq(1)
+    message_queries = sql_queries.select { |query| query[:name] == 'Message Load' }
+    expect(message_queries.size).to eq(1), message_queries.pluck(:sql).join("\n")
+  end
+
+  it 'returns an empty batch without prefetch queries when the conversation has no messages' do
+    source_conversation['source'] = nil
+    source_conversation['conversation_parts']['conversation_parts'] = []
+
+    expect(DataImportMapping).not_to receive(:where)
+    expect(Message).not_to receive(:where)
+
+    expect(builder.perform.entries).to be_empty
+  end
+
+  it 'classifies a live mapping from the current import as already handled' do
+    message = create(
+      :message,
+      account: account,
+      conversation: conversation,
+      inbox: conversation.inbox,
+      source_id: 'intercom:conversation:conversation_1:part:part_1'
+    )
+    DataImportMapping.create!(
+      account: account,
+      data_import: data_import,
+      source_provider: 'intercom',
+      source_object_type: 'message',
+      source_object_id: 'conversation:conversation_1:part:part_1',
+      chatwoot_record_type: 'Message',
+      chatwoot_record_id: message.id,
+      metadata: {}
+    )
+
+    entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
+
+    expect(entry).to have_attributes(classification: :current_import, message: message)
+  end
+
+  it 'classifies a live mapping from a previous import without changing its owner' do
+    previous_import = create(:data_import, :intercom, account: account)
+    message = create(
+      :message,
+      account: account,
+      conversation: conversation,
+      inbox: conversation.inbox,
+      source_id: 'intercom:conversation:conversation_1:part:part_1'
+    )
+    mapping = DataImportMapping.create!(
+      account: account,
+      data_import: previous_import,
+      source_provider: 'intercom',
+      source_object_type: 'message',
+      source_object_id: 'conversation:conversation_1:part:part_1',
+      chatwoot_record_type: 'Message',
+      chatwoot_record_id: message.id,
+      metadata: {}
+    )
+
+    entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
+
+    expect(entry).to have_attributes(classification: :previous_import, mapping: mapping, message: message)
+    expect(mapping.reload.data_import).to eq(previous_import)
+  end
+
+  it 'classifies a mapping whose message was deleted as repairable' do
+    mapping = DataImportMapping.create!(
+      account: account,
+      data_import: data_import,
+      source_provider: 'intercom',
+      source_object_type: 'message',
+      source_object_id: 'conversation:conversation_1:part:part_1',
+      chatwoot_record_type: 'Message',
+      chatwoot_record_id: 0,
+      metadata: {}
+    )
+
+    entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
+
+    expect(entry).to have_attributes(classification: :repairable_stale_mapping, mapping: mapping, message: nil)
+  end
+
+  it 'classifies an existing conversation message without a mapping for repair' do
+    message = create(
+      :message,
+      account: account,
+      conversation: conversation,
+      inbox: conversation.inbox,
+      source_id: 'intercom:conversation:conversation_1:part:part_1'
+    )
+
+    entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
+
+    expect(entry).to have_attributes(classification: :existing_message, mapping: nil, message: message)
+  end
+
+  it 'repairs a skipped mapping when the source part is now an activity' do
+    source_conversation.dig('conversation_parts', 'conversation_parts').first.merge!(
+      'part_type' => 'assignment',
+      'body' => nil,
+      'assigned_to' => { 'name' => 'Support' }
+    )
+    mapping = DataImportMapping.create!(
+      account: account,
+      data_import: data_import,
+      source_provider: 'intercom',
+      source_object_type: 'message',
+      source_object_id: 'conversation:conversation_1:part:part_1',
+      chatwoot_record_type: 'Conversation',
+      chatwoot_record_id: conversation.id,
+      metadata: { skipped: true }
+    )
+
+    entry = builder.perform.entries.find { |batch_entry| batch_entry.source_id.end_with?('part:part_1') }
+
+    expect(entry).to have_attributes(classification: :repairable_stale_mapping, mapping: mapping, message: nil)
+  end
+end