Merge branch 'feature/cw-7513' into feature/cw-7513-specs

spec/controllers/api/v1/accounts/agent_bots_controller_spec.rb

@@ -15,7 +15,7 @@ RSpec.describe 'Agent Bot API', type: :request do
       end
     end
 
-    context 'when it is an authenticated user' do
+    context 'when it is an authenticated agent' do
       it 'returns all the agent_bots in account along with global agent bots' do
         global_bot = create(:agent_bot)
         get "/api/v1/accounts/#{account.id}/agent_bots",
@@ -25,7 +25,7 @@ RSpec.describe 'Agent Bot API', type: :request do
         expect(response).to have_http_status(:success)
         expect(response.body).to include(agent_bot.name)
         expect(response.body).to include(global_bot.name)
-        expect(response.body).to include(agent_bot.access_token.token)
+        expect(response.body).not_to include(agent_bot.access_token.token)
         expect(response.body).not_to include(global_bot.access_token.token)
       end
 
@@ -54,6 +54,17 @@ RSpec.describe 'Agent Bot API', type: :request do
         expect(account_bot_response).to include('thumbnail')
       end
     end
+
+    context 'when it is an authenticated administrator' do
+      it 'returns the account bot access token' do
+        get "/api/v1/accounts/#{account.id}/agent_bots",
+            headers: admin.create_new_auth_token,
+            as: :json
+
+        expect(response).to have_http_status(:success)
+        expect(response.body).to include(agent_bot.access_token.token)
+      end
+    end
   end
 
   describe 'GET /api/v1/accounts/{account.id}/agent_bots/:id' do
@@ -65,7 +76,7 @@ RSpec.describe 'Agent Bot API', type: :request do
       end
     end
 
-    context 'when it is an authenticated user' do
+    context 'when it is an authenticated agent' do
       it 'shows the agent bot' do
         get "/api/v1/accounts/#{account.id}/agent_bots/#{agent_bot.id}",
             headers: agent.create_new_auth_token,
@@ -73,7 +84,7 @@ RSpec.describe 'Agent Bot API', type: :request do
 
         expect(response).to have_http_status(:success)
         expect(response.body).to include(agent_bot.name)
-        expect(response.body).to include(agent_bot.access_token.token)
+        expect(response.body).not_to include(agent_bot.access_token.token)
       end
 
       it 'will show a global agent bot' do
@@ -91,6 +102,17 @@ RSpec.describe 'Agent Bot API', type: :request do
         expect(response.parsed_body).not_to include('outgoing_url')
       end
     end
+
+    context 'when it is an authenticated administrator' do
+      it 'returns the account bot access token' do
+        get "/api/v1/accounts/#{account.id}/agent_bots/#{agent_bot.id}",
+            headers: admin.create_new_auth_token,
+            as: :json
+
+        expect(response).to have_http_status(:success)
+        expect(response.body).to include(agent_bot.access_token.token)
+      end
+    end
   end
 
   describe 'POST /api/v1/accounts/{account.id}/agent_bots' do

spec/controllers/api/v1/accounts/dashboard_apps_controller_spec.rb

@@ -70,8 +70,8 @@ RSpec.describe 'DashboardAppsController', type: :request do
       end
     end
 
-    context 'when it is an authenticated user' do
-      let(:user) { create(:user, account: account) }
+    context 'when it is an authenticated administrator' do
+      let(:user) { create(:user, account: account, role: :administrator) }
 
       it 'creates the dashboard app' do
         expect do
@@ -130,11 +130,26 @@ RSpec.describe 'DashboardAppsController', type: :request do
         expect(response).to have_http_status(:unprocessable_entity)
       end
     end
+
+    context 'when it is an authenticated agent' do
+      let(:agent) { create(:user, account: account, role: :agent) }
+
+      it 'does not create account-wide dashboard apps' do
+        expect do
+          post "/api/v1/accounts/#{account.id}/dashboard_apps",
+               headers: agent.create_new_auth_token,
+               params: payload,
+               as: :json
+        end.not_to change(DashboardApp, :count)
+
+        expect(response).to have_http_status(:unauthorized)
+      end
+    end
   end
 
   describe 'PATCH /api/v1/accounts/{account.id}/dashboard_apps/:id' do
     let(:payload) { { dashboard_app: { title: 'CRM Dashboard', content: [{ type: 'frame', url: 'https://link.com' }] } } }
-    let(:user) { create(:user, account: account) }
+    let(:user) { create(:user, account: account, role: :administrator) }
     let!(:dashboard_app) { create(:dashboard_app, user: user, account: account) }
 
     context 'when it is an unauthenticated user' do
@@ -160,10 +175,24 @@ RSpec.describe 'DashboardAppsController', type: :request do
         expect(json_response['content'][0]['type']).to eq payload[:dashboard_app][:content][0][:type]
       end
     end
+
+    context 'when it is an authenticated agent' do
+      let(:agent) { create(:user, account: account, role: :agent) }
+
+      it 'does not update account-wide dashboard apps' do
+        patch "/api/v1/accounts/#{account.id}/dashboard_apps/#{dashboard_app.id}",
+              headers: agent.create_new_auth_token,
+              params: payload,
+              as: :json
+
+        expect(response).to have_http_status(:unauthorized)
+        expect(dashboard_app.reload.title).not_to eq('CRM Dashboard')
+      end
+    end
   end
 
   describe 'DELETE /api/v1/accounts/{account.id}/dashboard_apps/:id' do
-    let(:user) { create(:user, account: account) }
+    let(:user) { create(:user, account: account, role: :administrator) }
     let!(:dashboard_app) { create(:dashboard_app, user: user, account: account) }
 
     context 'when it is an unauthenticated user' do
@@ -182,5 +211,18 @@ RSpec.describe 'DashboardAppsController', type: :request do
         expect(user.dashboard_apps.count).to be 0
       end
     end
+
+    context 'when it is an authenticated agent' do
+      let(:agent) { create(:user, account: account, role: :agent) }
+
+      it 'does not delete account-wide dashboard apps' do
+        delete "/api/v1/accounts/#{account.id}/dashboard_apps/#{dashboard_app.id}",
+               headers: agent.create_new_auth_token,
+               as: :json
+
+        expect(response).to have_http_status(:unauthorized)
+        expect(DashboardApp.exists?(dashboard_app.id)).to be(true)
+      end
+    end
   end
 end

spec/enterprise/controllers/api/v1/accounts/captain/documents_controller_spec.rb

@@ -51,6 +51,18 @@ RSpec.describe 'Api::V1::Accounts::Captain::Documents', type: :request do
           expect(json_response[:payload].length).to eq(5)
           expect(json_response[:meta]).to eq({ page: 2, total_count: 30 })
         end
+
+        it 'returns the generated FAQ count for each document' do
+          document = create(:captain_document, assistant: assistant, account: account)
+          create_list(:captain_assistant_response, 2,
+                      assistant: assistant, account: account, documentable: document)
+
+          get "/api/v1/accounts/#{account.id}/captain/documents",
+              headers: agent.create_new_auth_token, as: :json
+
+          matching_document = json_response[:payload].find { |item| item[:id] == document.id }
+          expect(matching_document[:responses_count]).to eq(2)
+        end
       end
 
       context 'when filtering by assistant_id' do
@@ -142,6 +154,10 @@ RSpec.describe 'Api::V1::Accounts::Captain::Documents', type: :request do
         expect(json_response[:external_link]).to eq(document.external_link)
       end
 
+      it 'returns the crawled content for the document' do
+        expect(json_response[:content]).to eq(document.content)
+      end
+
       it 'returns sync metadata when the document has been synced' do
         synced_at = 1.hour.ago
         document.update!(sync_status: :synced, last_synced_at: synced_at)

spec/enterprise/lib/captain/conversation_completion_service_spec.rb

@@ -68,6 +68,110 @@ RSpec.describe Captain::ConversationCompletionService do
       end
     end
 
+    context 'when building evaluation context' do
+      let(:captain_assistant) { create(:captain_assistant, account: account) }
+      let(:mock_response) do
+        instance_double(
+          RubyLLM::Message,
+          content: { 'complete' => false, 'reason' => 'Human follow-up is still pending' },
+          input_tokens: 100,
+          output_tokens: 20
+        )
+      end
+
+      it 'includes conversation status and speaker labels' do
+        conversation.update!(status: :pending, waiting_since: 2.hours.ago)
+        create(:message, conversation: conversation, inbox: inbox, account: account, message_type: :incoming, content: 'I need help with a refund')
+        create(
+          :message,
+          conversation: conversation,
+          inbox: inbox,
+          account: account,
+          message_type: :outgoing,
+          sender: captain_assistant,
+          content: 'I will transfer this to support for review.'
+        )
+
+        expect(mock_chat).to receive(:ask) do |content|
+          expect(content).to include(
+            'Conversation status: pending',
+            'Conversation transcript:',
+            'Customer: I need help with a refund',
+            'Captain: I will transfer this to support for review.'
+          )
+
+          mock_response
+        end
+
+        result = service.perform
+
+        expect(result[:complete]).to be false
+      end
+
+      it 'includes pending captain handoff evidence in the transcript' do
+        conversation.update!(status: :pending)
+        create(:message, conversation: conversation, inbox: inbox, account: account, message_type: :incoming, content: 'Please cancel my order')
+        create(
+          :message,
+          conversation: conversation,
+          inbox: inbox,
+          account: account,
+          message_type: :outgoing,
+          sender: captain_assistant,
+          content: 'I will transfer this to a specialist and they will follow up here.'
+        )
+
+        expect(mock_chat).to receive(:ask) do |content|
+          expect(content).to include(
+            'Conversation status: pending',
+            'Captain: I will transfer this to a specialist and they will follow up here.'
+          )
+
+          mock_response
+        end
+
+        result = service.perform
+
+        expect(result[:complete]).to be false
+      end
+
+      it 'reuses computed message content while formatting the transcript' do
+        content_for_llm_calls_by_message_id = Hash.new(0)
+        allow_any_instance_of(Message).to receive(:content_for_llm).and_wrap_original do |method, *args| # rubocop:disable RSpec/AnyInstance
+          content_for_llm_calls_by_message_id[method.receiver.id] += 1
+          method.call(*args)
+        end
+
+        incoming_message = create(
+          :message,
+          :with_attachment,
+          conversation: conversation,
+          inbox: inbox,
+          account: account,
+          message_type: :incoming,
+          content: nil
+        )
+        outgoing_message = create(
+          :message,
+          conversation: conversation,
+          inbox: inbox,
+          account: account,
+          message_type: :outgoing,
+          sender: captain_assistant,
+          content: 'What do you need help with?'
+        )
+
+        allow(mock_chat).to receive(:ask).and_return(mock_response)
+
+        service.perform
+
+        expect(content_for_llm_calls_by_message_id).to include(
+          incoming_message.id => 1,
+          outgoing_message.id => 1
+        )
+      end
+    end
+
     context 'when conversation has no messages' do
       it 'returns incomplete with appropriate reason' do
         result = service.perform

spec/enterprise/models/captain/agent_session_spec.rb

@@ -0,0 +1,160 @@
+require 'rails_helper'
+
+RSpec.describe Captain::AgentSession, type: :model do
+  let(:account) { create(:account) }
+  let(:assistant) { create(:captain_assistant, account: account) }
+
+  describe 'associations' do
+    it { is_expected.to belong_to(:account) }
+    it { is_expected.to belong_to(:assistant).class_name('Captain::Assistant') }
+    it { is_expected.to belong_to(:user).optional }
+    it { is_expected.to belong_to(:subject) }
+    it { is_expected.to belong_to(:result).optional }
+  end
+
+  describe 'enums' do
+    it { is_expected.to define_enum_for(:session_type).with_values(assistant: 0, copilot: 1).with_prefix(:session) }
+  end
+
+  describe '#subject' do
+    it 'returns the conversation for an assistant session' do
+      conversation = create(:conversation, account: account)
+      session = create(:captain_agent_session, account: account, assistant: assistant, subject: conversation)
+
+      expect(session.subject).to eq(conversation)
+    end
+
+    it 'returns the copilot thread for a copilot session' do
+      user = create(:user, account: account)
+      copilot_thread = create(:captain_copilot_thread, account: account, user: user, assistant: assistant)
+      session = create(:captain_agent_session, :copilot, account: account, assistant: assistant, user: user, subject: copilot_thread)
+
+      expect(session.subject).to eq(copilot_thread)
+    end
+
+    it 'returns nil when the subject record no longer exists' do
+      conversation = create(:conversation, account: account)
+      session = create(:captain_agent_session, account: account, assistant: assistant, subject: conversation)
+      conversation.destroy
+
+      expect(session.reload.subject).to be_nil
+    end
+
+    it 'is not valid when the subject type does not match the session type' do
+      copilot_thread = create(:captain_copilot_thread, account: account, user: create(:user, account: account), assistant: assistant)
+      session = build(:captain_agent_session, account: account, assistant: assistant, subject: copilot_thread)
+
+      expect(session).not_to be_valid
+      expect(session.errors[:subject_type]).to be_present
+    end
+
+    it 'is not valid when the subject belongs to a different account' do
+      foreign_conversation = create(:conversation, account: create(:account))
+      session = build(:captain_agent_session, account: account, assistant: assistant, subject: foreign_conversation)
+
+      expect(session).not_to be_valid
+      expect(session.errors[:subject]).to be_present
+    end
+  end
+
+  describe '#result' do
+    it 'returns the message for an assistant session' do
+      conversation = create(:conversation, account: account)
+      message = create(:message, account: account, conversation: conversation)
+      session = create(:captain_agent_session, account: account, assistant: assistant, subject: conversation, result: message)
+
+      expect(session.result).to eq(message)
+    end
+
+    it 'returns the copilot message for a copilot session' do
+      user = create(:user, account: account)
+      copilot_thread = create(:captain_copilot_thread, account: account, user: user, assistant: assistant)
+      copilot_message = create(:captain_copilot_message, account: account, copilot_thread: copilot_thread)
+      session = create(:captain_agent_session, :copilot, account: account, assistant: assistant, user: user,
+                                                         subject: copilot_thread, result: copilot_message)
+
+      expect(session.result).to eq(copilot_message)
+    end
+
+    it 'returns nil when result_id is nil' do
+      session = create(:captain_agent_session, account: account, assistant: assistant)
+
+      expect(session.result).to be_nil
+    end
+
+    it 'is not valid when the result belongs to a different account' do
+      conversation = create(:conversation, account: account)
+      foreign_message = create(:message, account: create(:account))
+      session = build(:captain_agent_session, account: account, assistant: assistant, subject: conversation, result: foreign_message)
+
+      expect(session).not_to be_valid
+      expect(session.errors[:result]).to be_present
+    end
+
+    it 'is not valid when result_id/result_type are set directly for a different account' do
+      conversation = create(:conversation, account: account)
+      foreign_message = create(:message, account: create(:account))
+      session = build(:captain_agent_session, account: account, assistant: assistant, subject: conversation,
+                                              result_id: foreign_message.id, result_type: 'Message')
+
+      expect(session).not_to be_valid
+      expect(session.errors[:result]).to be_present
+    end
+
+    it 'is not valid when result_id/result_type are set directly for a stale id' do
+      conversation = create(:conversation, account: account)
+      session = build(:captain_agent_session, account: account, assistant: assistant, subject: conversation,
+                                              result_id: 0, result_type: 'Message')
+
+      expect(session).not_to be_valid
+      expect(session.errors[:result]).to be_present
+    end
+  end
+
+  describe 'account' do
+    it 'is derived from the assistant when created via the assistant association' do
+      conversation = create(:conversation, account: account)
+      session = assistant.agent_sessions.create!(subject: conversation, session_type: :assistant)
+
+      expect(session.account).to eq(account)
+    end
+
+    it 'overrides a mismatched explicit account with the assistant account' do
+      conversation = create(:conversation, account: account)
+      session = build(:captain_agent_session, account: create(:account), assistant: assistant, subject: conversation)
+
+      expect(session).to be_valid
+      expect(session.account).to eq(account)
+    end
+  end
+
+  describe 'defaults' do
+    it 'defaults faq_ids, document_ids, scenario_ids and run_context' do
+      session = create(:captain_agent_session, account: account, assistant: assistant)
+
+      expect(session.faq_ids).to eq([])
+      expect(session.document_ids).to eq([])
+      expect(session.scenario_ids).to eq([])
+      expect(session.run_context).to eq({})
+    end
+  end
+
+  describe 'factory' do
+    it 'builds a valid assistant session' do
+      session = create(:captain_agent_session, account: account, assistant: assistant)
+
+      expect(session).to be_valid
+      expect(session).to be_session_assistant
+      expect(session.subject).to be_a(Conversation)
+    end
+
+    it 'builds a valid copilot session' do
+      session = create(:captain_agent_session, :copilot, account: account, assistant: assistant)
+
+      expect(session).to be_valid
+      expect(session).to be_session_copilot
+      expect(session.subject).to be_a(CopilotThread)
+      expect(session.user).to be_present
+    end
+  end
+end

spec/enterprise/models/captain/assistant_spec.rb

@@ -0,0 +1,42 @@
+require 'rails_helper'
+
+RSpec.describe Captain::Assistant do
+  describe '#agent_tools' do
+    let(:account) { create(:account) }
+    let(:assistant) { create(:captain_assistant, account: account) }
+
+    it 'includes enabled custom tools from the assistant account' do
+      custom_tool = create(:captain_custom_tool, account: account)
+
+      tools = assistant.send(:agent_tools)
+
+      expect(tools.map(&:name)).to include(custom_tool.slug)
+      expect(tools.find { |tool| tool.name == custom_tool.slug }).to be_a(Captain::Tools::HttpTool)
+    end
+
+    it 'excludes disabled custom tools' do
+      custom_tool = create(:captain_custom_tool, :disabled, account: account)
+
+      tools = assistant.send(:agent_tools)
+
+      expect(tools.map(&:name)).not_to include(custom_tool.slug)
+    end
+
+    it 'excludes custom tools from other accounts' do
+      custom_tool = create(:captain_custom_tool)
+
+      tools = assistant.send(:agent_tools)
+
+      expect(tools.map(&:name)).not_to include(custom_tool.slug)
+    end
+
+    it 'keeps the built-in FAQ lookup and handoff tools' do
+      tools = assistant.send(:agent_tools)
+
+      expect(tools).to include(
+        an_instance_of(Captain::Tools::FaqLookupTool),
+        an_instance_of(Captain::Tools::HandoffTool)
+      )
+    end
+  end
+end

spec/enterprise/services/captain/assistant_migration/draft_applier_spec.rb

@@ -0,0 +1,116 @@
+require 'rails_helper'
+
+RSpec.describe Captain::AssistantMigration::DraftApplier do
+  let(:account) { create(:account) }
+  let(:assistant) do
+    create(
+      :captain_assistant,
+      account: account,
+      config: { 'product_name' => 'Test Product', 'instructions' => 'Legacy V1 custom instructions.' },
+      response_guidelines: [],
+      guardrails: []
+    )
+  end
+  let(:scenario_candidate) do
+    {
+      'title' => 'Billing Investigation',
+      'description' => 'Use when a customer reports an account-specific billing issue.',
+      'instruction' => 'Collect the invoice number and summarize the issue before escalating.',
+      'response_guideline' => 'For account-specific billing issues, collect the invoice number and summarize the issue before escalating.',
+      'tool_ids' => []
+    }
+  end
+  let(:faq_document_candidate) do
+    {
+      'question' => 'When is support available?',
+      'answer' => 'Support is available Monday to Friday.'
+    }
+  end
+  let(:draft) do
+    {
+      business_product_context: ['Support assistant for Test Product.'],
+      response_guidelines: ['Be concise.'],
+      guardrails: ['Do not guess.'],
+      conversation_messages: {},
+      scenario_candidates: [scenario_candidate],
+      faq_document_candidates: [faq_document_candidate],
+      needs_review: ['Pricing details are missing because factual details are absent from the source instructions.']
+    }
+  end
+
+  describe '#perform' do
+    it 'reports staged scenario candidates in dry run without writing to the assistant' do
+      result = described_class.new(assistant: assistant, draft: draft, dry_run: true).perform
+
+      expect(result.dig(:changes, :config, :to, 'assistant_migration', 'scenario_candidates')).to eq([scenario_candidate])
+      expect(result.dig(:changes, :response_guidelines, :to)).to include(
+        'For account-specific billing issues, collect the invoice number and summarize the issue before escalating.'
+      )
+      expect(assistant.reload.config).not_to have_key('assistant_migration')
+      expect(assistant.scenarios.count).to eq(0)
+    end
+
+    it 'stores scenario candidates in assistant config and flattens them into response guidelines' do
+      described_class.new(assistant: assistant, draft: draft, dry_run: false).perform
+
+      assistant.reload
+      expect(assistant.config.dig('assistant_migration', 'scenario_candidates')).to eq([scenario_candidate])
+      expect(assistant.config.dig('assistant_migration', 'faq_document_candidates')).to contain_exactly(faq_document_candidate)
+      expect(assistant.config.dig('assistant_migration', 'needs_review')).to contain_exactly(
+        'Pricing details are missing because factual details are absent from the source instructions.'
+      )
+      expect(assistant.response_guidelines).to include(
+        'For account-specific billing issues, collect the invoice number and summarize the issue before escalating.'
+      )
+      expect(assistant.response_guidelines).not_to include(faq_document_candidate['answer'])
+      expect(assistant.scenarios.count).to eq(0)
+    end
+
+    it 'rejects stale drafts whose FAQ candidates use the old string format' do
+      stale_draft = draft.merge(faq_document_candidates: ['Support is available Monday to Friday.'])
+
+      expect do
+        described_class.new(assistant: assistant, draft: stale_draft, dry_run: false).perform
+      end.to raise_error(ArgumentError, 'FAQ document candidates must be question and answer objects')
+
+      expect(assistant.reload.config).not_to have_key('assistant_migration')
+    end
+
+    it 'preserves original values in migration config before applying classifier output' do
+      assistant.update!(
+        description: 'Existing assistant description.',
+        response_guidelines: ['Use plain language.'],
+        guardrails: ['Do not disclose internal notes.']
+      )
+
+      described_class.new(assistant: assistant, draft: draft, dry_run: false).perform
+
+      assistant.reload
+      expect(assistant.description).to eq('Support assistant for Test Product.')
+      expect(assistant.response_guidelines).to include('Be concise.')
+      expect(assistant.guardrails).to eq(['Do not guess.'])
+      expect(assistant.config.dig('assistant_migration', 'original_values')).to include(
+        'name' => assistant.name,
+        'description' => 'Existing assistant description.',
+        'config' => { 'product_name' => 'Test Product', 'instructions' => 'Legacy V1 custom instructions.' },
+        'response_guidelines' => ['Use plain language.'],
+        'guardrails' => ['Do not disclose internal notes.']
+      )
+    end
+
+    it 'rejects an oversized assistant description from a stale draft' do
+      long_context = 'This assistant supports a very broad product surface with many long details. ' * 10
+      original_description = assistant.description
+
+      expect do
+        described_class.new(
+          assistant: assistant,
+          draft: draft.merge(business_product_context: [long_context]),
+          dry_run: false
+        ).perform
+      end.to raise_error(ArgumentError, 'Assistant description exceeds 500 characters')
+
+      expect(assistant.reload.description).to eq(original_description)
+    end
+  end
+end

spec/enterprise/services/enterprise/billing/reconcile_plan_features_service_spec.rb

@@ -0,0 +1,53 @@
+require 'rails_helper'
+
+describe Enterprise::Billing::ReconcilePlanFeaturesService do
+  let(:account) { create(:account) }
+
+  before do
+    create(:installation_config, {
+             name: 'CHATWOOT_CLOUD_PLANS',
+             value: [
+               { 'name' => 'Hacker', 'product_id' => ['plan_id_hacker'], 'price_ids' => ['price_hacker'] },
+               { 'name' => 'Startups', 'product_id' => ['plan_id_startups'], 'price_ids' => ['price_startups'] }
+             ]
+           })
+  end
+
+  describe '#perform' do
+    context 'with api_and_webhooks feature' do
+      it 'enables the feature for a paid plan with an active subscription' do
+        account.update!(custom_attributes: { 'plan_name' => 'Startups', 'subscription_status' => 'active' })
+
+        described_class.new(account: account).perform
+
+        expect(account.reload).to be_feature_enabled('api_and_webhooks')
+      end
+
+      it 'enables the feature for a paid plan on trial' do
+        account.update!(custom_attributes: { 'plan_name' => 'Startups', 'subscription_status' => 'trialing' })
+
+        described_class.new(account: account).perform
+
+        expect(account.reload).to be_feature_enabled('api_and_webhooks')
+      end
+
+      it 'disables the feature on the default plan' do
+        account.enable_features!('api_and_webhooks')
+        account.update!(custom_attributes: { 'plan_name' => 'Hacker', 'subscription_status' => 'active' })
+
+        described_class.new(account: account).perform
+
+        expect(account.reload).not_to be_feature_enabled('api_and_webhooks')
+      end
+
+      it 'keeps the feature enabled when manually managed' do
+        account.update!(custom_attributes: { 'plan_name' => 'Hacker', 'subscription_status' => 'trialing' })
+        Internal::Accounts::InternalAttributesService.new(account).manually_managed_features = ['api_and_webhooks']
+
+        described_class.new(account: account).perform
+
+        expect(account.reload).to be_feature_enabled('api_and_webhooks')
+      end
+    end
+  end
+end

spec/factories/captain/agent_session.rb

@@ -0,0 +1,14 @@
+FactoryBot.define do
+  factory :captain_agent_session, class: 'Captain::AgentSession' do
+    account
+    association :assistant, factory: :captain_assistant
+    session_type { :assistant }
+    subject { create(:conversation, account: account) }
+
+    trait :copilot do
+      session_type { :copilot }
+      user
+      subject { create(:captain_copilot_thread, account: account, user: user) }
+    end
+  end
+end

spec/factories/data_import.rb

@@ -3,5 +3,14 @@ FactoryBot.define do
     data_type { 'contacts' }
     import_file { Rack::Test::UploadedFile.new(Rails.root.join('spec/assets/contacts.csv'), 'text/csv') }
     account
+
+    trait :intercom do
+      data_type { 'intercom' }
+      source_type { 'api' }
+      source_provider { 'intercom' }
+      import_types { %w[contacts conversations] }
+      access_token { 'intercom-token' }
+      import_file { nil }
+    end
   end
 end

spec/finders/data_import_error_finder_spec.rb

@@ -0,0 +1,26 @@
+require 'rails_helper'
+
+RSpec.describe DataImportErrorFinder do
+  let(:data_import) { create(:data_import, :intercom) }
+
+  it 'returns only the latest five non-skip errors' do
+    6.times do |index|
+      data_import.import_errors.create!(
+        error_code: 'Intercom::Error',
+        source_object_id: "error_#{index}",
+        details: { kind: 'run_error' },
+        created_at: Time.zone.at(index)
+      )
+    end
+    data_import.import_errors.create!(
+      error_code: 'Intercom::Skipped',
+      source_object_id: 'skipped_error',
+      details: { kind: 'skipped' },
+      created_at: Time.zone.at(10)
+    )
+
+    errors = described_class.new(data_import).import_errors
+
+    expect(errors.pluck(:source_object_id)).to eq(%w[error_5 error_4 error_3 error_2 error_1])
+  end
+end

spec/finders/data_import_skip_log_finder_spec.rb

@@ -0,0 +1,38 @@
+require 'rails_helper'
+
+RSpec.describe DataImportSkipLogFinder do
+  let(:data_import) { create(:data_import, :intercom) }
+
+  before do
+    6.times do |index|
+      data_import.import_errors.create!(
+        error_code: 'Intercom::Skipped',
+        source_object_type: 'message',
+        source_object_id: "message_#{index}",
+        details: { kind: 'skipped' },
+        created_at: Time.zone.at(index)
+      )
+    end
+    data_import.import_errors.create!(
+      error_code: 'Intercom::Skipped',
+      source_object_type: 'contact',
+      source_object_id: 'contact_1',
+      details: { kind: 'skipped' }
+    )
+  end
+
+  it 'filters skip logs and returns only the latest five', :aggregate_failures do
+    finder = described_class.new(data_import, skip_logs_type: 'message')
+
+    expect(finder.skip_logs.pluck(:source_object_id)).to eq(%w[message_5 message_4 message_3 message_2 message_1])
+    expect(finder.selected_source_object_type).to eq('message')
+    expect(finder.counts_by_type).to include('message' => 6, 'contact' => 1)
+  end
+
+  it 'ignores unsupported source object filters' do
+    finder = described_class.new(data_import, skip_logs_type: 'company')
+
+    expect(finder.selected_source_object_type).to be_nil
+    expect(finder.skip_logs.size).to eq(5)
+  end
+end

spec/jobs/data_imports/intercom/import_jobs_spec.rb

@@ -0,0 +1,218 @@
+require 'rails_helper'
+
+RSpec.describe DataImports::Intercom::ImportJob do
+  let(:account) { create(:account) }
+  let(:data_import) do
+    create(
+      :data_import, :intercom,
+      account: account,
+      import_types: %w[contacts conversations]
+    )
+  end
+  let(:importer) { instance_double(DataImports::Intercom::Importer) }
+  let(:run_id) { 'intercom-run-1' }
+
+  before do
+    account.enable_features!('data_import')
+    data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id })
+    allow(DataImports::Intercom::Importer).to receive(:new).with(data_import: data_import, run_id: run_id).and_return(importer)
+  end
+
+  describe DataImports::Intercom::BaseJob do
+    it 'checks rate limit retry before the generic client retry' do
+      expect(described_class.rescue_handlers.last.first).to eq('DataImports::Intercom::Client::RateLimitError')
+    end
+  end
+
+  describe DataImports::Intercom::ImportJob do
+    it 'starts the import and enqueues the first contacts page' do
+      allow(importer).to receive_messages(start!: true, import_contacts?: true, contacts_completed?: false, cursor_for: 'contact-cursor')
+
+      expect do
+        described_class.perform_now(data_import, run_id)
+      end.to have_enqueued_job(DataImports::Intercom::ContactsPageJob).with(data_import, 'contact-cursor', run_id).on_queue('low')
+
+      expect(importer).to have_received(:start!)
+    end
+
+    it 'resumes at conversations when contacts are already completed' do
+      allow(importer).to receive_messages(
+        start!: true,
+        import_contacts?: true,
+        contacts_completed?: true,
+        import_conversations?: true,
+        conversations_completed?: false
+      )
+      allow(importer).to receive(:cursor_for).with('conversations').and_return('conversation-cursor')
+
+      expect do
+        described_class.perform_now(data_import, run_id)
+      end.to have_enqueued_job(DataImports::Intercom::ConversationsPageJob).with(data_import, 'conversation-cursor', run_id)
+    end
+
+    it 'finishes immediately when every requested stage is already complete' do
+      allow(importer).to receive_messages(
+        start!: true,
+        import_contacts?: true,
+        contacts_completed?: true,
+        import_conversations?: true,
+        conversations_completed?: true,
+        finish!: true
+      )
+
+      expect do
+        described_class.perform_now(data_import, run_id)
+      end.not_to have_enqueued_job
+
+      expect(importer).to have_received(:finish!)
+    end
+
+    it 'skips stale import jobs from an earlier run' do
+      data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
+
+      expect(DataImports::Intercom::Importer).not_to receive(:new)
+
+      described_class.perform_now(data_import, 'old-run')
+    end
+  end
+
+  describe DataImports::Intercom::ContactsPageJob do
+    it 'hands off to conversations when a retry finds contacts already completed' do
+      allow(importer).to receive_messages(
+        contacts_completed?: true,
+        import_conversations?: true,
+        conversations_completed?: false
+      )
+      allow(importer).to receive(:cursor_for).with('conversations').and_return('conversation-cursor')
+      expect(importer).not_to receive(:import_contacts_page)
+
+      expect do
+        described_class.perform_now(data_import, 'completed-contact-cursor', run_id)
+      end.to have_enqueued_job(DataImports::Intercom::ConversationsPageJob).with(data_import, 'conversation-cursor', run_id)
+    end
+
+    it 'imports one contacts page and enqueues the next contacts page' do
+      result = DataImports::Intercom::Importer::PageResult.new(next_cursor: 'next-contact-cursor')
+      allow(importer).to receive_messages(contacts_completed?: false)
+      allow(importer).to receive(:import_contacts_page).with(starting_after: 'current-contact-cursor').and_return(result)
+
+      expect do
+        described_class.perform_now(data_import, 'current-contact-cursor', run_id)
+      end.to have_enqueued_job(described_class).with(data_import, 'next-contact-cursor', run_id)
+    end
+
+    it 'hands off to conversations after the final contacts page' do
+      result = DataImports::Intercom::Importer::PageResult.new(next_cursor: nil)
+      allow(importer).to receive_messages(
+        contacts_completed?: false,
+        import_conversations?: true,
+        conversations_completed?: false
+      )
+      allow(importer).to receive(:import_contacts_page).with(starting_after: nil).and_return(result)
+      allow(importer).to receive(:cursor_for).with('conversations').and_return(nil)
+
+      expect do
+        described_class.perform_now(data_import, nil, run_id)
+      end.to have_enqueued_job(DataImports::Intercom::ConversationsPageJob).with(data_import, nil, run_id)
+    end
+
+    it 'finishes after the final contacts page when conversations are not requested' do
+      result = DataImports::Intercom::Importer::PageResult.new(next_cursor: nil)
+      allow(importer).to receive_messages(contacts_completed?: false, import_conversations?: false, finish!: true)
+      allow(importer).to receive(:import_contacts_page).with(starting_after: nil).and_return(result)
+
+      expect do
+        described_class.perform_now(data_import, nil, run_id)
+      end.not_to have_enqueued_job
+
+      expect(importer).to have_received(:finish!)
+    end
+
+    it 'skips stale page jobs from an earlier run' do
+      data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
+
+      expect(DataImports::Intercom::Importer).not_to receive(:new)
+
+      described_class.perform_now(data_import, 'current-contact-cursor', 'old-run')
+    end
+
+    it 'skips failed page jobs from backend retries' do
+      data_import.update!(status: :failed)
+
+      expect(DataImports::Intercom::Importer).not_to receive(:new)
+
+      described_class.perform_now(data_import, 'current-contact-cursor', run_id)
+    end
+
+    it 'does not enqueue another stage when the page import becomes stale' do
+      result = DataImports::Intercom::Importer::PageResult.new(next_cursor: nil)
+      allow(importer).to receive_messages(contacts_completed?: false, finish!: true)
+      allow(importer).to receive(:import_contacts_page).with(starting_after: 'current-contact-cursor') do
+        data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
+        result
+      end
+
+      expect do
+        described_class.perform_now(data_import, 'current-contact-cursor', run_id)
+      end.not_to have_enqueued_job
+
+      expect(importer).not_to have_received(:finish!)
+    end
+  end
+
+  describe DataImports::Intercom::ConversationsPageJob do
+    it 'finishes when a retry finds conversations already completed' do
+      allow(importer).to receive_messages(conversations_completed?: true, finish!: true)
+      expect(importer).not_to receive(:import_conversations_page)
+
+      described_class.perform_now(data_import, 'completed-conversation-cursor', run_id)
+
+      expect(importer).to have_received(:finish!)
+    end
+
+    it 'imports one conversations page and enqueues the next conversations page' do
+      result = DataImports::Intercom::Importer::PageResult.new(next_cursor: 'next-conversation-cursor')
+      allow(importer).to receive_messages(conversations_completed?: false)
+      allow(importer).to receive(:import_conversations_page).with(starting_after: 'current-conversation-cursor').and_return(result)
+
+      expect do
+        described_class.perform_now(data_import, 'current-conversation-cursor', run_id)
+      end.to have_enqueued_job(described_class).with(data_import, 'next-conversation-cursor', run_id)
+    end
+
+    it 'finishes after the final conversations page' do
+      result = DataImports::Intercom::Importer::PageResult.new(next_cursor: nil)
+      allow(importer).to receive_messages(conversations_completed?: false, finish!: true)
+      allow(importer).to receive(:import_conversations_page).with(starting_after: nil).and_return(result)
+
+      expect do
+        described_class.perform_now(data_import, nil, run_id)
+      end.not_to have_enqueued_job
+
+      expect(importer).to have_received(:finish!)
+    end
+
+    it 'skips stale page jobs from an earlier run' do
+      data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
+
+      expect(DataImports::Intercom::Importer).not_to receive(:new)
+
+      described_class.perform_now(data_import, 'current-conversation-cursor', 'old-run')
+    end
+
+    it 'does not finish when the page import becomes stale' do
+      result = DataImports::Intercom::Importer::PageResult.new(next_cursor: nil)
+      allow(importer).to receive_messages(conversations_completed?: false, finish!: true)
+      allow(importer).to receive(:import_conversations_page).with(starting_after: 'current-conversation-cursor') do
+        data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
+        result
+      end
+
+      expect do
+        described_class.perform_now(data_import, 'current-conversation-cursor', run_id)
+      end.not_to have_enqueued_job
+
+      expect(importer).not_to have_received(:finish!)
+    end
+  end
+end

spec/models/account_spec.rb

@@ -108,6 +108,10 @@ RSpec.describe Account do
 
     it 'configures the account feature flag extension column' do
       expect(described_class.flag_columns).to include('feature_flags', 'feature_flags_ext_1')
+      expect(described_class.flag_mapping['feature_flags_ext_1']).to eq(feature_whatsapp_manual_transfer: 1, feature_data_import: 1 << 1,
+                                                                        feature_api_and_webhooks: 1 << 2)
+      expect(described_class.flag_mapping['feature_flags_ext_1'][:feature_whatsapp_manual_transfer]).to eq(1)
+      expect(described_class.flag_mapping['feature_flags_ext_1'][:feature_data_import]).to eq(2)
     end
 
     it 'keeps existing feature flags on the original column' do
@@ -116,15 +120,17 @@ RSpec.describe Account do
     end
 
     it 'keeps bulk selected feature assignment compatible with existing feature names' do
-      account.selected_feature_flags = [:feature_ip_lookup, :feature_assignment_v2, :feature_advanced_assignment]
+      account.selected_feature_flags = [:feature_ip_lookup, :feature_assignment_v2, :feature_advanced_assignment, :feature_data_import]
 
       expect(account).to be_feature_ip_lookup
       expect(account).to be_feature_assignment_v2
       expect(account).to be_feature_advanced_assignment
+      expect(account).to be_feature_data_import
       expect(account.selected_feature_flags).to contain_exactly(
         :feature_ip_lookup,
         :feature_assignment_v2,
-        :feature_advanced_assignment
+        :feature_advanced_assignment,
+        :feature_data_import
       )
     end
   end

spec/models/data_import_spec.rb

@@ -11,6 +11,18 @@ RSpec.describe DataImport do
     end
   end
 
+  describe 'access token encryption' do
+    it 'encrypts the Intercom access token at rest' do
+      skip('encryption keys missing; see run_mfa_spec workflow') unless Chatwoot.encryption_configured?
+
+      data_import = create(:data_import, :intercom, access_token: 'intercom-secret')
+      stored_value = data_import.reload.read_attribute_before_type_cast(:access_token).to_s
+
+      expect(stored_value).not_to include('intercom-secret')
+      expect(data_import.access_token).to eq('intercom-secret')
+    end
+  end
+
   describe 'callbacks' do
     let(:data_import) { build(:data_import) }
 
@@ -20,4 +32,36 @@ RSpec.describe DataImport do
       end.to have_enqueued_job(DataImportJob).with(data_import).on_queue('low')
     end
   end
+
+  describe '#abandon!' do
+    let(:account) { create(:account) }
+    let(:data_import) do
+      create(
+        :data_import, :intercom,
+        account: account,
+        status: :processing
+      )
+    end
+
+    before do
+      account.enable_features!('data_import')
+    end
+
+    it 'abandons active Intercom imports', :aggregate_failures do
+      data_import.abandon!
+
+      expect(data_import).to be_abandoned
+      expect(data_import.abandoned_at).to be_present
+    end
+
+    it 'does not overwrite terminal status from a stale instance', :aggregate_failures do
+      stale_import = described_class.find(data_import.id)
+      data_import.update!(status: :completed, completed_at: 1.minute.ago)
+
+      stale_import.abandon!
+
+      expect(data_import.reload).to be_completed
+      expect(data_import.abandoned_at).to be_nil
+    end
+  end
 end

spec/requests/api/v1/accounts/data_imports_spec.rb

@@ -0,0 +1,438 @@
+require 'rails_helper'
+
+RSpec.describe 'Data Imports API', type: :request do
+  let(:account) { create(:account) }
+  let(:admin) { create(:user, account: account, role: :administrator) }
+  let(:validator) { instance_double(DataImports::Intercom::CredentialsValidator, perform: { 'contacts' => 12, 'conversations' => 8 }) }
+
+  before do
+    account.enable_features!('data_import')
+    allow(DataImports::Intercom::CredentialsValidator).to receive(:new).and_return(validator)
+  end
+
+  describe 'POST /api/v1/accounts/:account_id/data_imports/validate_source' do
+    it 'validates the selected Intercom source and returns discovered totals' do
+      post validate_source_api_v1_account_data_imports_url(account_id: account.id),
+           params: {
+             source_provider: 'intercom', access_token: 'intercom-token', import_types: %w[contacts conversations]
+           },
+           headers: admin.create_new_auth_token,
+           as: :json
+
+      expect(response).to have_http_status(:ok)
+      expect(response.parsed_body).to eq('valid' => true, 'totals' => { 'contacts' => 12, 'conversations' => 8 })
+    end
+
+    it 'returns a safe validation error' do
+      allow(validator).to receive(:perform).and_raise(DataImports::Intercom::Client::AuthenticationError, 'provider response')
+
+      post validate_source_api_v1_account_data_imports_url(account_id: account.id),
+           params: { source_provider: 'intercom', access_token: 'invalid', import_types: %w[contacts] },
+           headers: admin.create_new_auth_token,
+           as: :json
+
+      expect(response).to have_http_status(:unprocessable_entity)
+      expect(response.parsed_body).to eq(
+        'valid' => false,
+        'message' => 'We could not validate this Intercom access key. Check the key and its permissions.'
+      )
+    end
+  end
+
+  describe 'POST /api/v1/accounts/:account_id/data_imports' do
+    it 'returns unauthorized and does not enqueue imports when data import is disabled' do
+      account.disable_features!('data_import')
+
+      expect do
+        post api_v1_account_data_imports_url(account_id: account.id),
+             params: {
+               name: 'Migration run', source_provider: 'intercom', access_token: 'intercom-token',
+               import_types: %w[contacts conversations]
+             },
+             headers: admin.create_new_auth_token,
+             as: :json
+      end.not_to have_enqueued_job(DataImports::Intercom::ImportJob)
+
+      expect(response).to have_http_status(:unauthorized)
+      expect(account.data_imports).to be_empty
+    end
+
+    it 'creates and enqueues an Intercom import', :aggregate_failures do
+      expect do
+        post api_v1_account_data_imports_url(account_id: account.id),
+             params: {
+               name: 'Migration run', source_provider: 'intercom', access_token: 'intercom-token',
+               import_types: %w[contacts conversations]
+             },
+             headers: admin.create_new_auth_token,
+             as: :json
+      end.to have_enqueued_job(DataImports::Intercom::ImportJob)
+
+      expect(response).to have_http_status(:ok)
+      data_import = account.data_imports.last
+      expect(data_import).to have_attributes(
+        name: 'Migration run',
+        data_type: 'intercom',
+        source_type: 'api',
+        source_provider: 'intercom',
+        initiated_by_id: admin.id
+      )
+      expect(data_import.access_token).to eq('intercom-token')
+      expect(data_import.import_types).to eq(%w[contacts conversations])
+      expect(data_import.stats).to include(
+        'contacts' => include('total' => 12),
+        'conversations' => include('total' => 8)
+      )
+      expect(response.parsed_body['source_provider']).to eq('intercom')
+      expect(response.parsed_body).not_to have_key('access_token')
+    end
+
+    it 'rejects creation while another Intercom import is active' do
+      active_import = create(
+        :data_import, :intercom,
+        account: account,
+        status: :processing
+      )
+
+      expect do
+        post api_v1_account_data_imports_url(account_id: account.id),
+             params: {
+               name: 'Second run', source_provider: 'intercom', access_token: 'intercom-token',
+               import_types: %w[contacts conversations]
+             },
+             headers: admin.create_new_auth_token,
+             as: :json
+      end.not_to have_enqueued_job(DataImports::Intercom::ImportJob)
+
+      expect(response).to have_http_status(:unprocessable_entity)
+      expect(response.parsed_body['message']).to eq('Another data import is already in progress.')
+      expect(account.data_imports.where(data_type: 'intercom', source_provider: 'intercom').count).to eq(1)
+      expect(active_import.reload).to be_processing
+    end
+
+    it 'rejects unsupported import types instead of silently importing everything' do
+      allow(validator).to receive(:perform).and_raise(ArgumentError, 'Unsupported import types: companies')
+
+      expect do
+        post api_v1_account_data_imports_url(account_id: account.id),
+             params: {
+               name: 'Migration run', source_provider: 'intercom', access_token: 'intercom-token', import_types: %w[companies]
+             },
+             headers: admin.create_new_auth_token,
+             as: :json
+      end.not_to have_enqueued_job(DataImports::Intercom::ImportJob)
+
+      expect(response).to have_http_status(:unprocessable_entity)
+      expect(response.parsed_body['message']).to eq('Unsupported import types: companies')
+      expect(account.data_imports).to be_empty
+    end
+  end
+
+  describe 'POST /api/v1/accounts/:account_id/data_imports/:id/start' do
+    let(:data_import) { create(:data_import, :intercom, account: account) }
+
+    it 'restarts abandoned imports' do
+      data_import.update!(
+        status: :abandoned,
+        abandoned_at: 1.hour.ago,
+        source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'previous-run' }
+      )
+      data_import.import_errors.create!(
+        error_code: 'StandardError',
+        message: 'old run error',
+        details: { kind: 'run_error' }
+      )
+      data_import.import_errors.create!(
+        error_code: DataImports::Intercom::Importer::ALREADY_IMPORTED_ERROR_CODE,
+        message: 'old skip log',
+        details: { kind: 'skipped' }
+      )
+
+      expect do
+        post start_api_v1_account_data_import_url(account_id: account.id, id: data_import.id),
+             headers: admin.create_new_auth_token,
+             as: :json
+      end.to have_enqueued_job(DataImports::Intercom::ImportJob).with(data_import, a_kind_of(String))
+
+      expect(response).to have_http_status(:ok)
+      expect(data_import.reload).to be_pending
+      expect(data_import.abandoned_at).to be_nil
+      expect(data_import.started_at).to be_nil
+      expect(data_import.active_intercom_import_run_id).not_to eq('previous-run')
+      expect(data_import.import_errors.pluck(:error_code)).to eq([DataImports::Intercom::Importer::ALREADY_IMPORTED_ERROR_CODE])
+    end
+
+    it 'does not enqueue duplicate jobs for active imports' do
+      data_import.update!(status: :processing)
+
+      expect do
+        post start_api_v1_account_data_import_url(account_id: account.id, id: data_import.id),
+             headers: admin.create_new_auth_token,
+             as: :json
+      end.not_to have_enqueued_job(DataImports::Intercom::ImportJob)
+
+      expect(response).to have_http_status(:ok)
+      expect(data_import.reload).to be_processing
+    end
+
+    it 'returns the active Intercom import instead of restarting another import' do
+      data_import.update!(status: :abandoned, abandoned_at: 1.hour.ago)
+      active_import = create(
+        :data_import, :intercom,
+        account: account,
+        status: :processing
+      )
+
+      expect do
+        post start_api_v1_account_data_import_url(account_id: account.id, id: data_import.id),
+             headers: admin.create_new_auth_token,
+             as: :json
+      end.not_to have_enqueued_job(DataImports::Intercom::ImportJob)
+
+      expect(response).to have_http_status(:ok)
+      expect(response.parsed_body['id']).to eq(active_import.id)
+      expect(data_import.reload).to be_abandoned
+    end
+
+    it 'does not restart imports when the stored access key is unavailable' do
+      data_import.update!(status: :abandoned, abandoned_at: 1.hour.ago, access_token: nil)
+
+      expect do
+        post start_api_v1_account_data_import_url(account_id: account.id, id: data_import.id),
+             headers: admin.create_new_auth_token,
+             as: :json
+      end.not_to have_enqueued_job(DataImports::Intercom::ImportJob)
+
+      expect(response).to have_http_status(:unprocessable_entity)
+      expect(response.parsed_body['message']).to eq('The Intercom access key for this import is unavailable.')
+      expect(data_import.reload).to be_abandoned
+    end
+  end
+
+  describe 'POST /api/v1/accounts/:account_id/data_imports/:id/abandon' do
+    let(:data_import) { create(:data_import, :intercom, account: account) }
+
+    it 'abandons active imports' do
+      data_import.update!(status: :processing)
+
+      post abandon_api_v1_account_data_import_url(account_id: account.id, id: data_import.id),
+           headers: admin.create_new_auth_token,
+           as: :json
+
+      expect(response).to have_http_status(:ok)
+      expect(data_import.reload).to be_abandoned
+      expect(data_import.abandoned_at).to be_present
+    end
+
+    it 'does not rewrite completed imports as abandoned' do
+      data_import.update!(status: :completed, completed_at: 1.hour.ago)
+
+      post abandon_api_v1_account_data_import_url(account_id: account.id, id: data_import.id),
+           headers: admin.create_new_auth_token,
+           as: :json
+
+      expect(response).to have_http_status(:ok)
+      expect(data_import.reload).to be_completed
+      expect(data_import.abandoned_at).to be_nil
+    end
+
+    it 'does not abandon legacy contact imports' do
+      legacy_import = create(:data_import, account: account, status: :processing)
+
+      post abandon_api_v1_account_data_import_url(account_id: account.id, id: legacy_import.id),
+           headers: admin.create_new_auth_token,
+           as: :json
+
+      expect(response).to have_http_status(:ok)
+      expect(legacy_import.reload).to be_processing
+      expect(legacy_import.abandoned_at).to be_nil
+    end
+  end
+
+  describe 'GET /api/v1/accounts/:account_id/data_imports/:id' do
+    let(:data_import) do
+      create(
+        :data_import, :intercom,
+        account: account,
+        name: 'July Intercom migration',
+        initiated_by: admin
+      )
+    end
+
+    it 'returns import details with recent errors' do
+      data_import.import_errors.create!(
+        error_code: 'Intercom::RateLimited',
+        message: 'Rate limited',
+        source_object_type: 'conversation',
+        source_object_id: 'conversation_1'
+      )
+      data_import.import_errors.create!(
+        error_code: 'DataImports::Intercom::SkippedMessage',
+        message: 'Skipped blank message',
+        source_object_type: 'message',
+        source_object_id: 'conversation:conversation_1:part:blank_part',
+        details: { kind: 'skipped', reason: 'blank_or_unsupported_intercom_part' }
+      )
+
+      get api_v1_account_data_import_url(account_id: account.id, id: data_import.id),
+          headers: admin.create_new_auth_token,
+          as: :json
+
+      expect(response).to have_http_status(:ok)
+      expect(response.parsed_body).to include(
+        'id' => data_import.id,
+        'name' => 'July Intercom migration',
+        'source_provider' => 'intercom',
+        'import_errors_count' => 1,
+        'skip_logs_count' => 1
+      )
+      expect(response.parsed_body['import_errors'].first).to include(
+        'error_code' => 'Intercom::RateLimited',
+        'message' => 'Rate limited',
+        'source_object_type' => 'conversation',
+        'source_object_id' => 'conversation_1'
+      )
+      expect(response.parsed_body['skip_logs'].first).to include(
+        'kind' => 'skipped',
+        'error_code' => 'DataImports::Intercom::SkippedMessage',
+        'source_object_type' => 'message',
+        'source_object_id' => 'conversation:conversation_1:part:blank_part'
+      )
+    end
+
+    it 'returns the latest five skip logs' do
+      16.times do |index|
+        data_import.import_errors.create!(
+          error_code: 'DataImports::Intercom::AlreadyImported',
+          message: 'Already imported in a previous import.',
+          source_object_type: 'message',
+          source_object_id: "message_#{index}",
+          details: { kind: 'skipped', reason: 'already_imported' },
+          created_at: Time.zone.at(index)
+        )
+      end
+
+      get api_v1_account_data_import_url(account_id: account.id, id: data_import.id),
+          headers: admin.create_new_auth_token,
+          as: :json
+
+      expect(response).to have_http_status(:ok)
+      expect(response.parsed_body['skip_logs'].pluck('source_object_id')).to eq(%w[message_15 message_14 message_13 message_12 message_11])
+      expect(response.parsed_body).not_to have_key('skip_logs_pagination')
+    end
+
+    it 'filters skip logs by source object type with counts for each type' do
+      3.times do |index|
+        data_import.import_errors.create!(
+          error_code: 'DataImports::Intercom::AlreadyImported',
+          message: 'Already imported in a previous import.',
+          source_object_type: 'contact',
+          source_object_id: "contact_#{index}",
+          details: { kind: 'skipped', reason: 'already_imported' }
+        )
+      end
+      2.times do |index|
+        data_import.import_errors.create!(
+          error_code: 'DataImports::Intercom::AlreadyImported',
+          message: 'Already imported in a previous import.',
+          source_object_type: 'message',
+          source_object_id: "message_#{index}",
+          details: { kind: 'skipped', reason: 'already_imported' }
+        )
+      end
+
+      get api_v1_account_data_import_url(account_id: account.id, id: data_import.id, skip_logs_type: 'contact'),
+          headers: admin.create_new_auth_token,
+          as: :json
+
+      expect(response).to have_http_status(:ok)
+      expect(response.parsed_body['skip_logs'].pluck('source_object_type').uniq).to eq(['contact'])
+      expect(response.parsed_body['skip_logs_filters']).to include(
+        'selected_source_object_type' => 'contact',
+        'counts_by_type' => include('contact' => 3, 'message' => 2)
+      )
+    end
+
+    it 'returns the latest five error logs' do
+      16.times do |index|
+        data_import.import_errors.create!(
+          error_code: 'Intercom::RateLimited',
+          message: 'Rate limited',
+          source_object_type: 'conversation',
+          source_object_id: "conversation_#{index}",
+          created_at: Time.zone.at(index)
+        )
+      end
+
+      get api_v1_account_data_import_url(account_id: account.id, id: data_import.id),
+          headers: admin.create_new_auth_token,
+          as: :json
+
+      expect(response).to have_http_status(:ok)
+      expect(response.parsed_body['import_errors'].pluck('source_object_id')).to eq(
+        %w[conversation_15 conversation_14 conversation_13 conversation_12 conversation_11]
+      )
+      expect(response.parsed_body).not_to have_key('import_errors_pagination')
+    end
+  end
+
+  describe 'GET /api/v1/accounts/:account_id/data_imports/:id/error_logs.csv' do
+    let(:data_import) do
+      create(
+        :data_import, :intercom,
+        account: account,
+        initiated_by: admin
+      )
+    end
+
+    it 'downloads all error logs as CSV' do
+      6.times do |index|
+        data_import.import_errors.create!(
+          error_code: 'Intercom::RateLimited',
+          message: 'Rate limited',
+          source_object_type: 'conversation',
+          source_object_id: "conversation_#{index}",
+          details: { kind: 'run_error' }
+        )
+      end
+
+      get error_logs_api_v1_account_data_import_url(account_id: account.id, id: data_import.id, format: :csv),
+          headers: admin.create_new_auth_token
+
+      expect(response).to have_http_status(:ok)
+      expect(response.media_type).to eq('text/csv')
+      expect(response.body).to include('source_object_type,source_object_id')
+      expect(response.body).to include('conversation,conversation_0,Intercom::RateLimited,Rate limited')
+      expect(response.body).to include('conversation,conversation_5,Intercom::RateLimited,Rate limited')
+      expect(response.body.lines.size).to eq(7)
+    end
+  end
+
+  describe 'GET /api/v1/accounts/:account_id/data_imports/:id/skip_logs.csv' do
+    let(:data_import) do
+      create(
+        :data_import, :intercom,
+        account: account,
+        initiated_by: admin
+      )
+    end
+
+    it 'downloads skip logs as CSV' do
+      data_import.import_errors.create!(
+        error_code: 'DataImports::Intercom::SkippedMessage',
+        message: 'Skipped blank message',
+        source_object_type: 'message',
+        source_object_id: 'conversation:conversation_1:part:blank_part',
+        details: { kind: 'skipped', reason: 'blank_or_unsupported_intercom_part' }
+      )
+
+      get skip_logs_api_v1_account_data_import_url(account_id: account.id, id: data_import.id, format: :csv),
+          headers: admin.create_new_auth_token
+
+      expect(response).to have_http_status(:ok)
+      expect(response.media_type).to eq('text/csv')
+      expect(response.body).to include('source_object_type,source_object_id')
+      expect(response.body).to include('message,conversation:conversation_1:part:blank_part')
+    end
+  end
+end

spec/services/data_imports/intercom/activity_content_builder_spec.rb

@@ -0,0 +1,51 @@
+require 'rails_helper'
+
+event_content = {
+  'assignment' => 'Avery assigned the conversation to Support',
+  'assign_and_reopen' => 'Avery assigned the conversation to Support and reopened it',
+  'open' => 'Avery opened the conversation',
+  'close' => 'Avery closed the conversation',
+  'snoozed' => 'Avery snoozed the conversation',
+  'participant_added' => 'Avery added Support as a participant',
+  'participant_removed' => 'Avery removed Support as a participant',
+  'conversation_attribute_updated_by_admin' => 'Avery updated conversation attributes',
+  'conversation_attribute_updated_by_user' => 'Avery updated conversation attributes',
+  'conversation_attribute_updated_by_workflow' => 'Avery updated conversation attributes',
+  'ticket_attribute_updated_by_admin' => 'Avery updated ticket attributes',
+  'ticket_state_updated_by_admin' => 'Avery updated the ticket state',
+  'custom_action_started' => 'Avery started a custom action',
+  'custom_action_finished' => 'Avery finished a custom action',
+  'quick_reply' => 'Avery used a quick reply'
+}.freeze
+
+RSpec.describe DataImports::Intercom::ActivityContentBuilder do
+  event_content.each do |part_type, expected_content|
+    it "builds readable content for #{part_type}" do
+      part = {
+        'part_type' => part_type,
+        'author' => { 'type' => 'admin', 'name' => 'Avery' },
+        'assigned_to' => { 'name' => 'Support' }
+      }
+
+      expect(described_class.new(part).perform).to eq(expected_content)
+    end
+  end
+
+  it 'uses a humanized fallback for unknown future event types' do
+    part = { 'part_type' => 'journey_stage_changed', 'author' => { 'type' => 'bot' } }
+
+    expect(described_class.new(part).perform).to eq('Intercom automation recorded journey stage changed')
+  end
+
+  it 'appends sanitized body context' do
+    part = {
+      'part_type' => 'close',
+      'author' => { 'type' => 'admin' },
+      'body' => '<p>Customer confirmed <strong>resolution</strong></p><script>alert(1)</script>'
+    }
+
+    expect(described_class.new(part).perform).to eq(
+      'Intercom teammate closed the conversation: Customer confirmed resolution'
+    )
+  end
+end

spec/services/data_imports/intercom/client_spec.rb

@@ -0,0 +1,16 @@
+require 'rails_helper'
+
+RSpec.describe DataImports::Intercom::Client do
+  let(:client) { described_class.new(access_token: 'intercom-token') }
+
+  describe '#list_contacts' do
+    it 'wraps transport failures in a retryable client error', :aggregate_failures do
+      allow(HTTParty).to receive(:get).and_raise(SocketError, 'getaddrinfo failed')
+
+      expect { client.list_contacts }.to raise_error(DataImports::Intercom::Client::Error) do |error|
+        expect(error.message).to eq('Intercom API request failed before receiving a response: getaddrinfo failed')
+        expect(error.body).to include(transport_error_class: 'SocketError')
+      end
+    end
+  end
+end

spec/services/data_imports/intercom/creation_service_spec.rb

@@ -0,0 +1,52 @@
+require 'rails_helper'
+
+RSpec.describe DataImports::Intercom::CreationService do
+  let(:account) { create(:account) }
+  let(:user) { create(:user, account: account) }
+  let(:validator) { instance_double(DataImports::Intercom::CredentialsValidator, perform: { 'contacts' => 12 }) }
+
+  before do
+    allow(DataImports::Intercom::CredentialsValidator).to receive(:new).and_return(validator)
+  end
+
+  it 'validates and creates an import with its credentials and totals', :aggregate_failures do
+    data_import = described_class.new(
+      account: account,
+      initiated_by: user,
+      source_params: {
+        name: 'Migration run',
+        source_provider: 'intercom',
+        access_token: ' intercom-token ',
+        import_types: %w[contacts]
+      }
+    ).perform
+
+    expect(data_import).to have_attributes(
+      name: 'Migration run',
+      source_type: 'api',
+      source_provider: 'intercom',
+      import_types: %w[contacts],
+      access_token: 'intercom-token',
+      initiated_by_id: user.id
+    )
+    expect(data_import.stats.dig('contacts', 'total')).to eq(12)
+    expect(data_import.active_intercom_import_run_id).to be_present
+  end
+
+  it 'returns no import without validating when another import is active' do
+    create(:data_import, :intercom, account: account, status: :processing)
+
+    data_import = described_class.new(
+      account: account,
+      initiated_by: user,
+      source_params: {
+        name: 'Second run',
+        source_provider: 'intercom',
+        access_token: 'intercom-token'
+      }
+    ).perform
+
+    expect(data_import).to be_nil
+    expect(validator).not_to have_received(:perform)
+  end
+end

spec/services/data_imports/intercom/credentials_validator_spec.rb

@@ -0,0 +1,54 @@
+require 'rails_helper'
+
+RSpec.describe DataImports::Intercom::CredentialsValidator do
+  let(:client) { instance_double(DataImports::Intercom::Client) }
+
+  before do
+    allow(DataImports::Intercom::Client).to receive(:new).with(access_token: 'intercom-token').and_return(client)
+    allow(client).to receive(:list_contacts)
+    allow(client).to receive(:list_conversations)
+  end
+
+  it 'validates and counts only contacts when conversations are not selected' do
+    allow(client).to receive(:list_contacts).with(per_page: 1).and_return('total_count' => 42)
+
+    totals = described_class.new(access_token: ' intercom-token ', import_types: %w[contacts]).perform
+
+    expect(totals).to eq('contacts' => 42)
+    expect(client).not_to have_received(:list_conversations)
+  end
+
+  it 'validates contact access and counts only conversations when contacts are not selected' do
+    allow(client).to receive(:list_contacts).with(per_page: 1).and_return('total_count' => 42)
+    allow(client).to receive(:list_conversations).with(per_page: 1).and_return('total_count' => 17)
+
+    totals = described_class.new(access_token: 'intercom-token', import_types: %w[conversations]).perform
+
+    expect(totals).to eq('conversations' => 17)
+    expect(client).to have_received(:list_contacts).with(per_page: 1)
+  end
+
+  it 'keeps an undiscovered total absent' do
+    allow(client).to receive(:list_contacts).with(per_page: 1).and_return('data' => [])
+
+    totals = described_class.new(access_token: 'intercom-token', import_types: %w[contacts]).perform
+
+    expect(totals).to be_empty
+  end
+
+  it 'preserves a known zero total' do
+    allow(client).to receive(:list_contacts).with(per_page: 1).and_return('total_count' => 0)
+
+    totals = described_class.new(access_token: 'intercom-token', import_types: %w[contacts]).perform
+
+    expect(totals).to eq('contacts' => 0)
+  end
+
+  it 'rejects an empty access key before calling Intercom' do
+    expect do
+      described_class.new(access_token: '', import_types: %w[contacts]).perform
+    end.to raise_error(ArgumentError, 'Intercom access key is required.')
+
+    expect(DataImports::Intercom::Client).not_to have_received(:new)
+  end
+end

spec/services/data_imports/intercom/importer_spec.rb

@@ -0,0 +1,968 @@
+require 'rails_helper'
+
+RSpec.describe DataImports::Intercom::Importer do
+  let(:account) { create(:account) }
+  let(:data_import) do
+    create(
+      :data_import, :intercom,
+      account: account
+    )
+  end
+  let(:client) { instance_double(DataImports::Intercom::Client) }
+  let(:contact_payload) do
+    {
+      'id' => 'contact_1',
+      'external_id' => 'external_1',
+      'email' => 'CUSTOMER@Example.com',
+      'phone' => '15551234567',
+      'name' => 'Customer One',
+      'created_at' => 1_700_000_000,
+      'updated_at' => 1_700_000_100
+    }
+  end
+  let(:conversation_payload) do
+    {
+      'id' => 'conversation_1',
+      'created_at' => 1_700_000_000,
+      'updated_at' => 1_700_000_200,
+      'state' => 'closed',
+      'open' => false,
+      'admin_assignee_id' => 123,
+      'team_assignee_id' => 456,
+      'contacts' => { 'contacts' => [{ 'id' => 'contact_1' }] },
+      'source' => {
+        'id' => 'source_1',
+        'type' => 'email',
+        'delivered_as' => 'customer_initiated',
+        'subject' => 'Need help',
+        'body' => '<p>Hello there</p>',
+        'author' => { 'type' => 'user', 'id' => 'contact_1', 'email' => 'CUSTOMER@example.com' }
+      },
+      'conversation_parts' => {
+        'conversation_parts' => [
+          {
+            'id' => 'part_1',
+            'part_type' => 'comment',
+            'body' => '<p>Admin reply</p>',
+            'created_at' => 1_700_000_100,
+            'updated_at' => 1_700_000_100,
+            'author' => { 'type' => 'admin', 'id' => 'admin_1' },
+            'attachments' => []
+          },
+          {
+            'id' => 'part_2',
+            'part_type' => 'note',
+            'body' => '<strong>Internal note</strong>',
+            'created_at' => 1_700_000_150,
+            'updated_at' => 1_700_000_150,
+            'author' => { 'type' => 'admin', 'id' => 'admin_1' },
+            'attachments' => []
+          }
+        ]
+      }
+    }
+  end
+
+  before do
+    account.enable_features!('data_import')
+    allow(DataImports::Intercom::Client).to receive(:new).with(access_token: 'intercom-token').and_return(client)
+    allow(client).to receive(:list_contacts).with(starting_after: nil).and_return(
+      'data' => [contact_payload],
+      'total_count' => 1,
+      'pages' => { 'next' => nil }
+    )
+    allow(client).to receive(:list_conversations).with(starting_after: nil).and_return(
+      'conversations' => [{ 'id' => 'conversation_1' }],
+      'total_count' => 1,
+      'pages' => { 'next' => nil }
+    )
+    allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(conversation_payload)
+    allow(client).to receive(:retrieve_contact).with('contact_1').and_return(contact_payload)
+  end
+
+  it 'imports contacts, conversations, messages, and source-bucket inboxes without normal message creation callbacks', :aggregate_failures do
+    described_class.new(data_import: data_import).perform
+
+    contact = account.contacts.find_by!(email: 'customer@example.com')
+    expect(contact.name).to eq('Customer One')
+    expect(contact.phone_number).to eq('+15551234567')
+    expect(contact).to be_lead
+    expect(contact.custom_attributes).to include('intercom_contact_id' => 'contact_1')
+
+    inbox = account.inboxes.find_by!(name: 'Intercom Import - Email')
+    expect(inbox.channel.additional_attributes).to include('source_bucket' => 'email', 'import_placeholder' => true)
+
+    conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
+    expect(conversation).to have_attributes(
+      status: 'resolved',
+      inbox_id: inbox.id,
+      contact_id: contact.id
+    )
+    expect(conversation.additional_attributes.dig('source', 'routing_method')).to eq('source_bucket_api_inbox')
+
+    expect(conversation.messages.order(:created_at).pluck(:content)).to eq(["Need help\n\nHello there", 'Admin reply', 'Internal note'])
+    expect(conversation.messages.order(:created_at).map(&:message_type)).to eq(%w[incoming outgoing outgoing])
+    expect(conversation.messages.order(:created_at).last.private).to be(true)
+
+    expect(data_import.reload).to be_completed
+    expect(data_import.stats).to include(
+      'contacts' => include('imported' => 1, 'skipped' => 0, 'total' => 1),
+      'conversations' => include('imported' => 1, 'skipped' => 0, 'total' => 1),
+      'messages' => include('imported' => 3, 'skipped' => 0, 'total' => 3),
+      'errors' => { 'count' => 0 }
+    )
+    expect(data_import.processed_records).to eq(5)
+    expect(data_import.items.imported.count).to eq(2)
+    expect(DataImportMapping.where(data_import: data_import).count).to eq(5)
+  end
+
+  it 'imports historical records without dispatching record events or outbound side effects', :aggregate_failures do
+    dispatched_events = []
+    allow(Rails.configuration.dispatcher).to receive(:dispatch) do |event_name, *_args|
+      dispatched_events << event_name
+    end
+    clear_enqueued_jobs
+
+    described_class.new(data_import: data_import).perform
+
+    record_events = [
+      Events::Types::CONTACT_CREATED,
+      Events::Types::CONTACT_UPDATED,
+      Events::Types::CONVERSATION_CREATED,
+      Events::Types::CONVERSATION_UPDATED,
+      Events::Types::CONVERSATION_STATUS_CHANGED,
+      Events::Types::ASSIGNEE_CHANGED,
+      Events::Types::TEAM_CHANGED,
+      Events::Types::MESSAGE_CREATED,
+      Events::Types::FIRST_REPLY_CREATED,
+      Events::Types::REPLY_CREATED
+    ]
+    side_effect_jobs = [SendReplyJob, EventDispatcherJob, ActionCableBroadcastJob, WebhookJob, HookJob]
+
+    expect(dispatched_events & record_events).to be_empty
+    expect(enqueued_jobs.pluck(:job) & side_effect_jobs).to be_empty
+    expect(Notification.where(account: account)).to be_empty
+  end
+
+  context 'when Intercom contact activity timestamps are available' do
+    let(:contact_payload) do
+      super().merge('last_seen_at' => 1_700_000_050, 'last_replied_at' => 1_700_000_090)
+    end
+
+    it 'prefers last_seen_at for contact activity' do
+      described_class.new(data_import: data_import).import_contacts_page
+
+      contact = account.contacts.find_by!(email: 'customer@example.com')
+      expect(contact.last_activity_at).to eq(Time.zone.at(1_700_000_050))
+    end
+  end
+
+  context 'when Intercom contact last_seen_at is unavailable' do
+    let(:contact_payload) do
+      super().merge('last_seen_at' => nil, 'last_replied_at' => 1_700_000_090)
+    end
+
+    it 'falls back to last_replied_at for contact activity' do
+      described_class.new(data_import: data_import).import_contacts_page
+
+      contact = account.contacts.find_by!(email: 'customer@example.com')
+      expect(contact.last_activity_at).to eq(Time.zone.at(1_700_000_090))
+    end
+  end
+
+  it 'leaves contact activity blank when Intercom activity timestamps are unavailable' do
+    described_class.new(data_import: data_import).import_contacts_page
+
+    contact = account.contacts.find_by!(email: 'customer@example.com')
+    expect(contact.last_activity_at).to be_nil
+  end
+
+  it 'updates message totals by delta when a conversation page is retried' do
+    importer = described_class.new(data_import: data_import)
+
+    importer.import_conversations_page
+    importer.import_conversations_page
+
+    expect(data_import.reload.stats.dig('messages', 'total')).to eq(3)
+    item = data_import.items.find_by!(source_object_type: 'conversation', source_object_id: 'conversation_1')
+    expect(item.metadata['message_total_contribution']).to eq(3)
+  end
+
+  it 'reconciles imported message stats from same-run mappings on retry' do
+    described_class.new(data_import: data_import).import_conversations_page
+    stats = data_import.reload.stats.deep_dup
+    stats['messages']['imported'] = 0
+    data_import.update!(stats: stats)
+
+    described_class.new(data_import: data_import).import_conversations_page
+
+    expect(data_import.reload.stats.dig('messages', 'imported')).to eq(3)
+  end
+
+  it 'indexes imported messages for advanced search' do
+    allow(ChatwootApp).to receive(:advanced_search_allowed?).and_return(true)
+    allow(ChatwootApp).to receive(:chatwoot_cloud?).and_return(false)
+    reindexed_message_ids = []
+    original_reindex_for_search = Message.instance_method(:reindex_for_search)
+    Message.define_method(:reindex_for_search) { reindexed_message_ids << id }
+    Message.__send__(:private, :reindex_for_search)
+
+    described_class.new(data_import: data_import).perform
+
+    expect(reindexed_message_ids).to match_array(Message.where(account_id: account.id).pluck(:id))
+  ensure
+    Message.define_method(:reindex_for_search, original_reindex_for_search)
+    Message.__send__(:private, :reindex_for_search)
+  end
+
+  it 'keeps imported messages successful when search reindexing fails', :aggregate_failures do
+    allow(ChatwootApp).to receive(:advanced_search_allowed?).and_return(true)
+    allow(ChatwootApp).to receive(:chatwoot_cloud?).and_return(false)
+    # rubocop:disable RSpec/AnyInstance
+    allow_any_instance_of(Message).to receive(:reindex_for_search).and_raise(StandardError, 'search unavailable')
+    # rubocop:enable RSpec/AnyInstance
+
+    described_class.new(data_import: data_import).perform
+
+    message = account.messages.find_by!(source_id: 'intercom:conversation:conversation_1:source:source_1')
+    mapping = data_import.mappings.find_by!(source_object_type: 'message', source_object_id: 'conversation:conversation_1:source:source_1')
+    expect(mapping.chatwoot_record).to eq(message)
+    expect(data_import.reload).to be_completed
+    expect(data_import.import_errors.exists?).to be(false)
+    expect(data_import.stats.dig('messages', 'imported')).to eq(3)
+  end
+
+  describe '#start!' do
+    it 'does not overwrite an import abandoned by another process', :aggregate_failures do
+      importer = described_class.new(data_import: data_import)
+
+      DataImport.find(data_import.id).update!(
+        status: :abandoned,
+        abandoned_at: Time.current
+      )
+
+      expect(importer.start!).to be_nil
+      expect(data_import.reload).to be_abandoned
+      expect(data_import.started_at).to be_nil
+    end
+  end
+
+  describe '#perform' do
+    it 'stops when the import was abandoned before processing starts' do
+      importer = described_class.new(data_import: data_import)
+      DataImport.find(data_import.id).update!(
+        status: :abandoned,
+        abandoned_at: Time.current
+      )
+
+      expect(client).not_to receive(:list_contacts)
+
+      importer.perform
+
+      expect(data_import.reload).to be_abandoned
+    end
+  end
+
+  describe '#import_conversations_page' do
+    it 'stops an in-flight page when a newer import run takes over', :aggregate_failures do
+      run_id = 'intercom-run-1'
+      data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => run_id })
+      allow(client).to receive(:list_conversations).with(starting_after: nil).and_return(
+        'conversations' => [{ 'id' => 'conversation_1' }, { 'id' => 'conversation_2' }],
+        'pages' => { 'next' => { 'starting_after' => 'next-conversation-cursor' } }
+      )
+      allow(client).to receive(:retrieve_conversation).with('conversation_1') do
+        data_import.update!(source_metadata: { DataImport::ACTIVE_INTERCOM_IMPORT_RUN_ID_KEY => 'new-run' })
+        conversation_payload
+      end
+
+      result = described_class.new(data_import: data_import, run_id: run_id).import_conversations_page
+
+      expect(result).to be_done
+      expect(client).not_to have_received(:retrieve_conversation).with('conversation_2')
+      expect(account.conversations.where(identifier: 'intercom:conversation_1')).to be_empty
+      expect(account.contacts.where(email: 'customer@example.com')).to be_empty
+      expect(data_import.reload.cursor.dig('conversations', 'starting_after')).to be_nil
+    end
+
+    it 'rolls back a newly inserted conversation 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 == 'conversation'
+
+        method.call(object_type, source_id, record, metadata: metadata)
+      end
+
+      importer.import_conversations_page
+
+      expect(account.conversations.where(identifier: 'intercom:conversation_1')).to be_empty
+      item = data_import.items.find_by!(source_object_type: 'conversation', source_object_id: 'conversation_1')
+      expect(item).to be_failed
+      expect(item.last_error_message).to eq('mapping failed')
+    end
+
+    it 'rolls back a newly inserted contact when mapping persistence fails', :aggregate_failures do
+      sparse_contact = contact_payload.slice('id', 'name', 'created_at', 'updated_at')
+      allow(client).to receive(:retrieve_contact).with('contact_1').and_return(sparse_contact)
+      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 == 'contact'
+
+        method.call(object_type, source_id, record, metadata: metadata)
+      end
+
+      importer.import_conversations_page
+
+      expect(account.contacts.where(name: 'Customer One')).to be_empty
+      expect(data_import.mappings.where(source_object_type: 'contact', source_object_id: 'contact_1')).to be_empty
+      contact_item = data_import.items.find_by!(source_object_type: 'contact', source_object_id: 'contact_1')
+      expect(contact_item).to be_failed
+      expect(contact_item.last_error_message).to eq('mapping failed')
+    end
+
+    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
+
+      importer.import_conversations_page
+
+      conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
+      expect(conversation.messages.where(source_id: 'intercom:conversation:conversation_1:source:source_1')).to be_empty
+      error = data_import.import_errors.find_by!(
+        source_object_type: 'message',
+        source_object_id: 'conversation:conversation_1:source:source_1'
+      )
+      expect(error).to have_attributes(error_code: 'StandardError', message: 'mapping failed')
+    end
+  end
+
+  describe '#finish!' do
+    it 'does not overwrite an import abandoned by another process' do
+      data_import.update!(status: :processing)
+      importer = described_class.new(data_import: data_import)
+
+      DataImport.find(data_import.id).update!(
+        status: :abandoned,
+        abandoned_at: Time.current
+      )
+
+      importer.finish!
+
+      expect(data_import.reload).to be_abandoned
+      expect(data_import.completed_at).to be_nil
+    end
+  end
+
+  describe '#fail!' do
+    it 'does not overwrite an import abandoned by another process', :aggregate_failures do
+      data_import.update!(status: :processing)
+      importer = described_class.new(data_import: data_import)
+
+      DataImport.find(data_import.id).update!(
+        status: :abandoned,
+        abandoned_at: Time.current
+      )
+
+      importer.fail!(StandardError.new('boom'))
+
+      expect(data_import.reload).to be_abandoned
+      expect(data_import.last_error_at).to be_nil
+      expect(data_import.import_errors.exists?).to be(false)
+    end
+  end
+
+  context 'when the Intercom records were imported by an earlier run' do
+    let(:next_data_import) do
+      create(
+        :data_import, :intercom,
+        account: account
+      )
+    end
+
+    it 'records the already mapped records as skipped for the current import run', :aggregate_failures do
+      described_class.new(data_import: data_import).perform
+
+      described_class.new(data_import: next_data_import).perform
+
+      expect(next_data_import.reload.stats).to include(
+        'contacts' => include('imported' => 0, 'skipped' => 1, 'total' => 1),
+        'conversations' => include('imported' => 0, 'skipped' => 1, 'total' => 1),
+        'messages' => include('imported' => 0, 'skipped' => 3, 'total' => 3),
+        'errors' => { 'count' => 0 }
+      )
+      expect(next_data_import).to be_completed
+      expect(next_data_import.total_records).to eq(5)
+      expect(next_data_import.processed_records).to eq(0)
+      expect(next_data_import.items.skipped.count).to eq(2)
+      expect(next_data_import.import_errors.skip_logs.group(:source_object_type).count).to eq(
+        'contact' => 1,
+        'conversation' => 1,
+        'message' => 3
+      )
+      expect(next_data_import.import_errors.skip_logs.pluck(:details).map { |details| details['reason'] }.uniq).to eq(['already_imported'])
+    end
+
+    it 'recreates messages when existing message mappings point to deleted records', :aggregate_failures do
+      described_class.new(data_import: data_import).perform
+      conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
+      Message.where(conversation_id: conversation.id).delete_all
+
+      described_class.new(data_import: next_data_import).perform
+
+      expect(conversation.reload.messages.pluck(:source_id)).to match_array(
+        %w[
+          intercom:conversation:conversation_1:source:source_1
+          intercom:conversation:conversation_1:part:part_1
+          intercom:conversation:conversation_1:part:part_2
+        ]
+      )
+      expect(next_data_import.reload.stats).to include(
+        'contacts' => include('imported' => 0, 'skipped' => 1, 'total' => 1),
+        'conversations' => include('imported' => 0, 'skipped' => 1, 'total' => 1),
+        'messages' => include('imported' => 3, 'skipped' => 0, 'total' => 3),
+        'errors' => { 'count' => 0 }
+      )
+      expect(next_data_import.import_errors.skip_logs.where(source_object_type: 'message')).to be_empty
+      message_mappings = DataImportMapping.where(account: account, source_provider: 'intercom', source_object_type: 'message')
+      expect(message_mappings.filter_map(&:chatwoot_record).count).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',
+        'part_type' => 'comment',
+        'body' => '<p>Follow-up reply</p>',
+        'created_at' => 1_700_000_300,
+        'updated_at' => 1_700_000_300,
+        'author' => { 'type' => 'admin', 'id' => 'admin_1' },
+        'attachments' => []
+      }
+      updated_conversation_payload = conversation_payload.deep_dup
+      updated_conversation_payload['updated_at'] = 1_700_000_300
+      updated_conversation_payload['conversation_parts']['conversation_parts'] << new_part
+      allow(client).to receive(:retrieve_conversation).with('conversation_1').and_return(
+        conversation_payload,
+        updated_conversation_payload
+      )
+
+      described_class.new(data_import: data_import).perform
+      conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
+
+      described_class.new(data_import: next_data_import).perform
+
+      expect(conversation.reload.last_activity_at).to eq(Time.zone.at(1_700_000_300))
+      expect(conversation.messages.find_by!(source_id: 'intercom:conversation:conversation_1:part:part_3').content).to eq('Follow-up reply')
+    end
+  end
+
+  context 'when a conversation references an already mapped contact' do
+    it 'reuses the mapped contact without hydrating the sparse reference' do
+      described_class.new(data_import: data_import).import_contacts_page
+
+      expect(client).not_to receive(:retrieve_contact)
+
+      described_class.new(data_import: data_import).import_conversations_page
+    end
+  end
+
+  context 'when a same-run contact mapping outlives its item progress' do
+    let!(:mapped_contact) { create(:contact, account: account) }
+
+    before do
+      DataImportMapping.create!(
+        account: account,
+        data_import: data_import,
+        source_provider: 'intercom',
+        source_object_type: 'contact',
+        source_object_id: 'contact_1',
+        chatwoot_record_type: 'Contact',
+        chatwoot_record_id: mapped_contact.id,
+        metadata: {}
+      )
+      data_import.items.create!(
+        source_provider: 'intercom',
+        source_object_type: 'contact',
+        source_object_id: 'contact_1',
+        status: :processing,
+        metadata: contact_payload
+      )
+    end
+
+    it 'repairs the item and imported count on retry', :aggregate_failures do
+      described_class.new(data_import: data_import).import_contacts_page
+
+      item = data_import.items.find_by!(source_object_type: 'contact', source_object_id: 'contact_1')
+      expect(item).to be_imported
+      expect(item).to have_attributes(chatwoot_record_type: 'Contact', chatwoot_record_id: mapped_contact.id)
+      expect(data_import.reload.stats.dig('contacts', 'imported')).to eq(1)
+    end
+  end
+
+  context 'when an existing contact has the same email but a different external id' do
+    let(:contact_payload) do
+      super().merge('last_replied_at' => 1_700_000_090)
+    end
+    let!(:existing_contact) { create(:contact, account: account, email: 'customer@example.com', identifier: nil) }
+
+    it 'updates the existing contact instead of creating a duplicate', :aggregate_failures do
+      described_class.new(data_import: data_import).import_contacts_page
+
+      expect(existing_contact.reload.identifier).to eq('external_1')
+      expect(existing_contact.last_activity_at).to eq(Time.zone.at(1_700_000_090))
+      expect(account.contacts.where(email: 'customer@example.com').count).to eq(1)
+      item = data_import.items.imported.find_by!(source_object_type: 'contact', source_object_id: 'contact_1')
+      expect(item).to have_attributes(chatwoot_record_type: 'Contact', chatwoot_record_id: existing_contact.id)
+    end
+  end
+
+  context 'when an existing contact has the same phone but a different external id' do
+    let(:contact_payload) do
+      super().merge('email' => nil)
+    end
+    let!(:existing_contact) { create(:contact, account: account, phone_number: '+15551234567', identifier: nil) }
+
+    it 'updates the existing contact instead of creating a duplicate', :aggregate_failures do
+      described_class.new(data_import: data_import).import_contacts_page
+
+      expect(existing_contact.reload.identifier).to eq('external_1')
+      expect(account.contacts.where(phone_number: '+15551234567').count).to eq(1)
+      item = data_import.items.imported.find_by!(source_object_type: 'contact', source_object_id: 'contact_1')
+      expect(item).to have_attributes(chatwoot_record_type: 'Contact', chatwoot_record_id: existing_contact.id)
+    end
+  end
+
+  context 'when an existing contact has the same phone but Intercom sends a new email' do
+    let!(:existing_contact) { create(:contact, account: account, phone_number: '+15551234567', identifier: nil) }
+
+    it 'falls through to the phone match after the email lookup misses', :aggregate_failures do
+      described_class.new(data_import: data_import).import_contacts_page
+
+      expect(existing_contact.reload.email).to eq('customer@example.com')
+      expect(existing_contact.identifier).to eq('external_1')
+      expect(account.contacts.where(phone_number: '+15551234567').count).to eq(1)
+      item = data_import.items.imported.find_by!(source_object_type: 'contact', source_object_id: 'contact_1')
+      expect(item).to have_attributes(chatwoot_record_type: 'Contact', chatwoot_record_id: existing_contact.id)
+    end
+  end
+
+  context 'when an existing visitor contact matches the Intercom external id' do
+    let!(:existing_contact) { create(:contact, account: account, identifier: 'external_1') }
+
+    it 'promotes the contact to a lead when adding email or phone', :aggregate_failures do
+      expect(existing_contact).to be_visitor
+
+      described_class.new(data_import: data_import).import_contacts_page
+
+      expect(existing_contact.reload).to be_lead
+      expect(existing_contact.email).to eq('customer@example.com')
+      expect(existing_contact.phone_number).to eq('+15551234567')
+    end
+  end
+
+  context 'when an identifier match has contact details owned by another contact' do
+    let!(:existing_contact) { create(:contact, account: account, identifier: 'external_1') }
+    let!(:email_owner) { create(:contact, account: account, email: 'customer@example.com') }
+    let!(:phone_owner) { create(:contact, account: account, phone_number: '+15551234567') }
+
+    it 'does not copy the conflicting email or phone number', :aggregate_failures do
+      described_class.new(data_import: data_import).import_contacts_page
+
+      expect(existing_contact.reload.email).to be_nil
+      expect(existing_contact.phone_number).to be_nil
+      expect(existing_contact).to be_visitor
+      expect(email_owner.reload.email).to eq('customer@example.com')
+      expect(phone_owner.reload.phone_number).to eq('+15551234567')
+      expect(account.contacts.where(email: 'customer@example.com').count).to eq(1)
+      expect(account.contacts.where(phone_number: '+15551234567').count).to eq(1)
+
+      item = data_import.items.imported.find_by!(source_object_type: 'contact', source_object_id: 'contact_1')
+      expect(item).to have_attributes(chatwoot_record_type: 'Contact', chatwoot_record_id: existing_contact.id)
+    end
+  end
+
+  context 'when Intercom rate limits a conversation detail request' do
+    before do
+      allow(client).to receive(:retrieve_conversation).with('conversation_1').and_raise(
+        DataImports::Intercom::Client::RateLimitError.new('rate limited', status: 429)
+      )
+    end
+
+    it 're-raises the provider error so the page job can retry', :aggregate_failures do
+      expect { described_class.new(data_import: data_import).import_conversations_page }
+        .to raise_error(DataImports::Intercom::Client::RateLimitError)
+
+      item = data_import.items.find_by!(source_object_type: 'conversation', source_object_id: 'conversation_1')
+      expect(item).to be_processing
+      expect(data_import.import_errors.exists?).to be(false)
+    end
+  end
+
+  context 'when Intercom rate limits a contact hydration request' do
+    before do
+      allow(client).to receive(:retrieve_contact).with('contact_1').and_raise(
+        DataImports::Intercom::Client::RateLimitError.new('rate limited', status: 429)
+      )
+    end
+
+    it 're-raises the provider error instead of importing a sparse contact', :aggregate_failures do
+      expect { described_class.new(data_import: data_import).import_conversations_page }
+        .to raise_error(DataImports::Intercom::Client::RateLimitError)
+
+      expect(data_import.items.exists?(source_object_type: 'contact')).to be(false)
+      expect(data_import.import_errors.exists?).to be(false)
+    end
+  end
+
+  context 'when Intercom no longer has a sparse contact referenced by a conversation' do
+    before do
+      allow(client).to receive(:retrieve_contact).with('contact_1').and_raise(
+        DataImports::Intercom::Client::Error.new('not found', status: 404)
+      )
+    end
+
+    it 'falls back to the conversation contact reference', :aggregate_failures do
+      expect { described_class.new(data_import: data_import).import_conversations_page }.not_to raise_error
+
+      expect(data_import.items.imported.exists?(source_object_type: 'contact', source_object_id: 'contact_1')).to be(true)
+      expect(data_import.import_errors.exists?).to be(false)
+    end
+  end
+
+  context 'when the Intercom source message only has attachments' do
+    let(:conversation_payload) do
+      super().deep_merge(
+        'source' => {
+          'subject' => nil,
+          'body' => nil,
+          'attachments' => [{ 'name' => 'invoice.pdf', 'url' => 'https://example.com/invoice.pdf' }]
+        },
+        'conversation_parts' => {
+          'conversation_parts' => []
+        }
+      )
+    end
+
+    it 'imports the source message attachment placeholder', :aggregate_failures do
+      described_class.new(data_import: data_import).perform
+
+      conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
+      expect(conversation.messages.pluck(:content)).to eq(['[Intercom attachment skipped: 1]'])
+      expect(conversation.messages.first.additional_attributes.dig('source', 'attachments')).to eq(
+        [{ 'name' => 'invoice.pdf', 'url' => 'https://example.com/invoice.pdf' }]
+      )
+      expect(data_import.reload.stats.dig('messages', 'imported')).to eq(1)
+    end
+  end
+
+  context 'when the Intercom source message has text and attachments' do
+    let(:conversation_payload) do
+      super().deep_merge(
+        'source' => {
+          'attachments' => [{ 'name' => 'invoice.pdf', 'url' => 'https://example.com/invoice.pdf' }]
+        }
+      )
+    end
+
+    it 'adds an attachment omission marker to the imported message', :aggregate_failures do
+      described_class.new(data_import: data_import).perform
+
+      message = account.messages.find_by!(source_id: 'intercom:conversation:conversation_1:source:source_1')
+      expect(message.content).to eq("Need help\n\nHello there\n\n[Intercom attachment skipped: 1]")
+      expect(message.additional_attributes.dig('source', 'attachments')).to eq(
+        [{ 'name' => 'invoice.pdf', 'url' => 'https://example.com/invoice.pdf' }]
+      )
+      expect(data_import.reload.stats.dig('messages', 'skipped')).to eq(0)
+    end
+  end
+
+  context 'when Intercom omits the conversation source' do
+    let(:conversation_payload) do
+      super().merge(
+        'source' => nil,
+        'first_contact_reply' => {
+          'type' => 'whatsapp',
+          'created_at' => 1_700_000_000,
+          'url' => nil
+        }
+      )
+    end
+
+    it 'routes the conversation from the first contact reply type', :aggregate_failures do
+      described_class.new(data_import: data_import).perform
+
+      inbox = account.inboxes.find_by!(name: 'Intercom Import - WhatsApp')
+      conversation = account.conversations.find_by!(identifier: 'intercom:conversation_1')
+
+      expect(conversation.inbox).to eq(inbox)
+      expect(conversation.additional_attributes.dig('source', 'source_type')).to eq('whatsapp')
+    end
+  end
+
+  context 'when an Intercom chat message part cannot be imported' do
+    let(:conversation_payload) do
+      super().deep_merge(
+        'conversation_parts' => {
+          'conversation_parts' => [
+            {
+              'id' => 'blank_part',
+              'part_type' => 'comment',
+              'body' => nil,
+              'created_at' => 1_700_000_175,
+              'updated_at' => 1_700_000_175,
+              'author' => { 'type' => 'admin', 'id' => 'admin_1' },
+              'attachments' => []
+            }
+          ]
+        }
+      )
+    end
+
+    it 'records a skip log with the Intercom message source id', :aggregate_failures do
+      described_class.new(data_import: data_import).perform
+
+      skip_log = data_import.import_errors.skip_logs.find_by!(source_object_type: 'message')
+      expect(skip_log).to have_attributes(
+        source_object_id: 'conversation:conversation_1:part:blank_part',
+        error_code: 'DataImports::Intercom::SkippedMessage',
+        message: 'Skipped Intercom comment event blank_part: no message body or attachments to import.'
+      )
+      expect(skip_log.details).to include(
+        'kind' => 'skipped',
+        'reason' => 'blank_or_unsupported_intercom_part',
+        'reason_details' => 'no message body or attachments to import',
+        'event_name' => 'comment',
+        'event_type' => 'comment',
+        'author_type' => 'admin'
+      )
+      expect(data_import.reload.stats.dig('messages', 'skipped')).to eq(1)
+    end
+
+    it 'records the skip log again for a later import run', :aggregate_failures do
+      described_class.new(data_import: data_import).perform
+      next_data_import = create(
+        :data_import, :intercom,
+        account: account
+      )
+
+      described_class.new(data_import: next_data_import).perform
+
+      skip_log = next_data_import.import_errors.skip_logs.find_by!(
+        source_object_type: 'message',
+        source_object_id: 'conversation:conversation_1:part:blank_part',
+        error_code: 'DataImports::Intercom::SkippedMessage'
+      )
+      expect(skip_log).to have_attributes(
+        source_object_id: 'conversation:conversation_1:part:blank_part',
+        error_code: 'DataImports::Intercom::SkippedMessage'
+      )
+      expect(next_data_import.reload.stats.dig('messages', 'skipped')).to eq(2)
+    end
+
+    it 'reconciles a same-run skipped mapping and missing skip log on retry', :aggregate_failures do
+      described_class.new(data_import: data_import).import_conversations_page
+      data_import.import_errors.where(source_object_type: 'message').delete_all
+      stats = data_import.reload.stats.deep_dup
+      stats['messages']['skipped'] = 0
+      data_import.update!(stats: stats)
+
+      described_class.new(data_import: data_import).import_conversations_page
+
+      expect(data_import.reload.stats.dig('messages', 'skipped')).to eq(1)
+      expect(data_import.import_errors.skip_logs.exists?(source_object_id: 'conversation:conversation_1:part:blank_part')).to be(true)
+    end
+
+    it 'repairs a previously skipped mapping when the part is now an activity', :aggregate_failures do
+      described_class.new(data_import: data_import).perform
+      previous_skip_log = data_import.import_errors.skip_logs.find_by!(source_object_id: 'conversation:conversation_1:part:blank_part')
+      conversation_payload.dig('conversation_parts', 'conversation_parts').first.merge!(
+        'part_type' => 'assignment',
+        'assigned_to' => { 'name' => 'Support' }
+      )
+      next_data_import = create(:data_import, :intercom, account: account)
+
+      described_class.new(data_import: next_data_import).perform
+
+      activity = account.messages.find_by!(source_id: 'intercom:conversation:conversation_1:part:blank_part')
+      mapping = DataImportMapping.find_by!(
+        account: account,
+        source_provider: 'intercom',
+        source_object_type: 'message',
+        source_object_id: 'conversation:conversation_1:part:blank_part'
+      )
+      expect(activity).to be_activity
+      expect(activity.content).to eq('Intercom teammate assigned the conversation to Support')
+      expect(mapping.chatwoot_record).to eq(activity)
+      expect(data_import.import_errors.skip_logs).to include(previous_skip_log)
+      expect(next_data_import.import_errors.skip_logs.where(source_object_id: mapping.source_object_id)).to be_empty
+    end
+  end
+
+  context 'when Intercom returns bodyless lifecycle events' do
+    let(:conversation_payload) do
+      super().deep_merge(
+        'conversation_parts' => {
+          'total_count' => 1,
+          'conversation_parts' => [
+            {
+              'id' => 'assignment_part',
+              'part_type' => 'assignment',
+              'body' => nil,
+              'created_at' => 1_700_000_175,
+              'author' => { 'type' => 'admin', 'name' => 'Avery' },
+              'assigned_to' => { 'type' => 'team', 'name' => 'Support' },
+              'state' => 'open',
+              'tags' => { 'tags' => [{ 'name' => 'priority' }] },
+              'event_details' => { 'source' => 'workflow' },
+              'app_package_code' => 'workflow'
+            }
+          ]
+        }
+      )
+    end
+
+    it 'imports events as public activity messages with source metadata', :aggregate_failures do
+      described_class.new(data_import: data_import).perform
+
+      activity = account.messages.find_by!(source_id: 'intercom:conversation:conversation_1:part:assignment_part')
+      expect(activity).to have_attributes(
+        message_type: 'activity',
+        content: 'Avery assigned the conversation to Support',
+        private: false,
+        sender: nil,
+        created_at: Time.zone.at(1_700_000_175)
+      )
+      expect(activity.additional_attributes['source']).to include(
+        'part_type' => 'assignment',
+        'assigned_to' => include('name' => 'Support'),
+        'state' => 'open',
+        'event_details' => include('source' => 'workflow'),
+        'app_package_code' => 'workflow'
+      )
+      expect(data_import.reload.stats['messages']).to include('imported' => 2, 'skipped' => 0, 'total' => 2)
+      expect(data_import.import_errors.skip_logs).to be_empty
+    end
+  end
+
+  context 'when Intercom omits older conversation parts from the retrieved conversation' do
+    let(:conversation_payload) do
+      super().deep_merge(
+        'conversation_parts' => {
+          'total_count' => 503
+        },
+        'statistics' => {
+          'count_conversation_parts' => 503
+        }
+      )
+    end
+
+    it 'records an incomplete import error and completes with errors', :aggregate_failures do
+      described_class.new(data_import: data_import).perform
+
+      error = data_import.import_errors.non_skip_logs.find_by!(
+        source_object_type: 'conversation',
+        source_object_id: 'conversation_1',
+        error_code: 'DataImports::Intercom::TruncatedConversationParts'
+      )
+      expect(error.message).to eq('Intercom returned 2 of 503 conversation parts.')
+      expect(error.details).to include(
+        'kind' => 'incomplete',
+        'imported_parts_count' => 2,
+        'total_parts_count' => 503
+      )
+      expect(data_import.reload).to be_completed_with_errors
+      expect(data_import.stats.dig('errors', 'count')).to eq(1)
+    end
+  end
+
+  context 'when the conversation parts total matches the returned parts' do
+    let(:conversation_payload) do
+      super().deep_merge(
+        'conversation_parts' => {
+          'total_count' => 2
+        },
+        'statistics' => {
+          'count_conversation_parts' => 2
+        }
+      )
+    end
+
+    it 'does not record a truncated parts error', :aggregate_failures do
+      described_class.new(data_import: data_import).perform
+
+      expect(data_import.import_errors.non_skip_logs).to be_empty
+      expect(data_import.reload).to be_completed
+      expect(data_import.stats.dig('errors', 'count')).to eq(0)
+    end
+  end
+
+  context 'when Intercom statistics count is higher than the conversation parts total' do
+    let(:conversation_payload) do
+      super().deep_merge(
+        'source' => {},
+        'conversation_parts' => {
+          'total_count' => 2
+        },
+        'statistics' => {
+          'count_conversation_parts' => 3
+        }
+      )
+    end
+
+    it 'trusts the returned conversation parts total over the statistics counter', :aggregate_failures do
+      described_class.new(data_import: data_import).perform
+
+      expect(data_import.import_errors.non_skip_logs).to be_empty
+      expect(data_import.reload).to be_completed
+      expect(data_import.stats.dig('errors', 'count')).to eq(0)
+    end
+  end
+
+  context 'when a specific Intercom message part fails to persist' do
+    let(:conversation_payload) do
+      super().deep_merge(
+        'conversation_parts' => {
+          'conversation_parts' => [
+            {
+              'id' => 'bad_part',
+              'part_type' => 'comment',
+              'body' => '<p>Message that cannot be stored</p>',
+              'created_at' => 1_700_000_175,
+              'updated_at' => 1_700_000_175,
+              'author' => { 'type' => 'admin', 'id' => 'admin_1' },
+              'attachments' => []
+            }
+          ]
+        }
+      )
+    end
+
+    before do
+      allow(Message).to receive(:insert_all!).and_wrap_original do |method, records, **kwargs|
+        raise ActiveRecord::StatementInvalid, 'bad message' if records.first[:source_id] == 'intercom:conversation:conversation_1:part:bad_part'
+
+        method.call(records, **kwargs)
+      end
+    end
+
+    it 'records a skip log with the Intercom message part id', :aggregate_failures do
+      described_class.new(data_import: data_import).perform
+
+      skip_log = data_import.import_errors.skip_logs.find_by!(source_object_type: 'message')
+      expect(skip_log).to have_attributes(
+        source_object_id: 'conversation:conversation_1:part:bad_part',
+        error_code: 'ActiveRecord::StatementInvalid',
+        message: 'bad message'
+      )
+      expect(skip_log.details).to include(
+        'kind' => 'failed',
+        'conversation_id' => 'intercom:conversation_1'
+      )
+      expect(data_import.reload).to be_completed_with_errors
+      expect(data_import.stats.dig('errors', 'count')).to eq(1)
+    end
+  end
+end

spec/services/data_imports/intercom/placeholder_inbox_builder_spec.rb

@@ -0,0 +1,33 @@
+require 'rails_helper'
+
+RSpec.describe DataImports::Intercom::PlaceholderInboxBuilder do
+  let(:account) { create(:account) }
+
+  describe '#inbox_for' do
+    it 'creates a source-bucket API inbox for an Intercom conversation source' do
+      inbox = described_class.new(account: account).inbox_for('email')
+
+      expect(inbox.name).to eq('Intercom Import - Email')
+      expect(inbox.channel).to be_a(Channel::Api)
+      expect(inbox.enable_auto_assignment).to be(false)
+      expect(inbox.allow_messages_after_resolved).to be(false)
+      expect(inbox.channel.additional_attributes).to include(
+        'source_provider' => 'intercom',
+        'source_bucket' => 'email',
+        'import_placeholder' => true,
+        'agent_reply_time_window' => 1
+      )
+    end
+
+    it 'reuses an existing placeholder inbox for the same source bucket' do
+      builder = described_class.new(account: account)
+
+      first_inbox = builder.inbox_for('phone_call')
+      expect(account).not_to receive(:inboxes)
+      second_inbox = builder.inbox_for('phone_switch')
+
+      expect(second_inbox).to eq(first_inbox)
+      expect(Inbox.where(account: account, channel_type: 'Channel::Api').count).to eq(1)
+    end
+  end
+end

spec/services/data_imports/intercom/restart_service_spec.rb

@@ -0,0 +1,64 @@
+require 'rails_helper'
+
+RSpec.describe DataImports::Intercom::RestartService do
+  let(:account) { create(:account) }
+  let(:data_import) { create(:data_import, :intercom, account: account, status: :abandoned, abandoned_at: 1.hour.ago) }
+
+  it 'prepares a failed or abandoned import for another run', :aggregate_failures do
+    data_import.update!(
+      stats: {
+        'contacts' => { 'imported' => 1, 'skipped' => 9, 'total' => 10 },
+        'conversations' => { 'imported' => 2, 'skipped' => 8, 'total' => 10 },
+        'messages' => { 'imported' => 3, 'skipped' => 7, 'total' => 10 },
+        'errors' => { 'count' => 6 }
+      }
+    )
+    data_import.import_errors.create!(error_code: 'StandardError', message: 'old run error')
+    data_import.import_errors.create!(
+      error_code: 'ContactFailed',
+      message: 'old contact error',
+      source_object_type: 'contact',
+      details: { kind: 'failed' }
+    )
+    retained_skip_log = data_import.import_errors.create!(
+      error_code: DataImports::Intercom::Importer::ALREADY_IMPORTED_ERROR_CODE,
+      message: 'old skip log',
+      source_object_type: 'contact',
+      details: { kind: 'skipped' }
+    )
+    previous_run_id = data_import.assign_active_intercom_import_run_id
+    data_import.save!
+    service = described_class.new(account: account, data_import: data_import)
+
+    expect(service.perform).to eq(:enqueue)
+    expect(service.data_import).to be_pending
+    expect(service.data_import.abandoned_at).to be_nil
+    expect(service.data_import.started_at).to be_nil
+    expect(service.data_import.active_intercom_import_run_id).not_to eq(previous_run_id)
+    expect(service.data_import.import_errors).to contain_exactly(retained_skip_log)
+    expect(service.data_import.stats).to eq(
+      'contacts' => { 'imported' => 1, 'skipped' => 1, 'total' => 10 },
+      'conversations' => { 'imported' => 2, 'skipped' => 0, 'total' => 10 },
+      'messages' => { 'imported' => 3, 'skipped' => 0, 'total' => 10 },
+      'errors' => { 'count' => 0 }
+    )
+  end
+
+  it 'returns the active import instead of restarting another import', :aggregate_failures do
+    active_import = create(:data_import, :intercom, account: account, status: :processing)
+    service = described_class.new(account: account, data_import: data_import)
+
+    expect(service.perform).to eq(:render_show)
+    expect(service.data_import).to eq(active_import)
+    expect(data_import.reload).to be_abandoned
+  end
+
+  it 'does not restart when the stored access token is missing' do
+    data_import.update!(access_token: nil)
+
+    result = described_class.new(account: account, data_import: data_import).perform
+
+    expect(result).to eq(:access_token_missing)
+    expect(data_import.reload).to be_abandoned
+  end
+end

spec/services/data_imports/intercom/source_bucket_spec.rb

@@ -0,0 +1,17 @@
+require 'rails_helper'
+
+RSpec.describe DataImports::Intercom::SourceBucket do
+  describe '.for' do
+    it 'maps Intercom source types to Chatwoot inbox buckets' do
+      expect(described_class.for('email')).to eq({ key: 'email', name: 'Email' })
+      expect(described_class.for('phone_switch')).to eq({ key: 'phone', name: 'Phone' })
+      expect(described_class.for('inapp')).to eq({ key: 'messenger', name: 'Messenger' })
+      expect(described_class.for('messenger')).to eq({ key: 'messenger', name: 'Messenger' })
+      expect(described_class.for('push')).to eq({ key: 'messenger', name: 'Messenger' })
+    end
+
+    it 'uses an unknown bucket for unsupported source types' do
+      expect(described_class.for('unsupported_source')).to eq({ key: 'unknown', name: 'Unknown' })
+    end
+  end
+end

spec/services/whatsapp/webhook_teardown_service_spec.rb

@@ -51,18 +51,41 @@ RSpec.describe Whatsapp::WebhookTeardownService do
       end
     end
 
-    context 'when channel is whatsapp_cloud but not embedded_signup' do
+    context 'when channel is whatsapp_cloud with manual setup' do
       before do
+        allow(channel).to receive(:setup_webhooks).and_return(true)
+
         channel.update!(
           provider: 'whatsapp_cloud',
-          provider_config: { 'source' => 'manual' }
+          provider_config: {
+            'source' => 'manual',
+            'phone_number_id' => 'manual_phone_id',
+            'business_account_id' => 'manual_waba_id',
+            'api_key' => 'manual_api_key'
+          }
         )
       end
 
-      it 'does not attempt to unsubscribe webhook' do
-        expect(Whatsapp::FacebookApiClient).not_to receive(:new)
+      it 'clears the phone number callback override' do
+        api_client = instance_double(Whatsapp::FacebookApiClient)
+        allow(Whatsapp::FacebookApiClient).to receive(:new).with('manual_api_key').and_return(api_client)
+        allow(api_client).to receive(:clear_phone_number_callback_override).with('manual_phone_id')
 
         service.perform
+
+        expect(api_client).to have_received(:clear_phone_number_callback_override).with('manual_phone_id')
+      end
+
+      # The manual token belongs to the customer's own Meta app, so its WABA subscription is not ours to remove.
+      it 'does not unsubscribe the app from the WABA' do
+        api_client = instance_double(Whatsapp::FacebookApiClient)
+        allow(Whatsapp::FacebookApiClient).to receive(:new).and_return(api_client)
+        allow(api_client).to receive(:clear_phone_number_callback_override)
+        allow(api_client).to receive(:unsubscribe_app_from_waba)
+
+        service.perform
+
+        expect(api_client).not_to have_received(:unsubscribe_app_from_waba)
       end
     end