diff --git a/app/models/ci/build.rb b/app/models/ci/build.rb
index 8e36c676f0c7c4e87df7ae05ba069a16d93d5387..cfa26db8b8aabe46d2d80aa818fe09f8b82483a1 100644
--- a/app/models/ci/build.rb
+++ b/app/models/ci/build.rb
@@ -957,7 +957,7 @@ def supported_runner?(features)
     end
 
     def publishes_artifacts_reports?
-      options&.dig(:artifacts, :reports)&.any?
+      reports_definitions&.any?
     end
 
     def supports_artifacts_exclude?
@@ -1205,6 +1205,11 @@ def run_status_commit_hooks!
 
     private
 
+    def reports_definitions
+      options.dig(:artifacts, :reports)
+    end
+    strong_memoize_attr :reports_definitions
+
     def use_jwt_for_ci_cd_job_token?
       namespace&.root_ancestor&.namespace_settings&.jwt_ci_cd_job_token_enabled?
     end
diff --git a/app/services/ci/cancel_pipeline_service.rb b/app/services/ci/cancel_pipeline_service.rb
index a6a75b197d2d1ca93a57cc22c731b0497c9455e2..493b33eb4167fc2ebc12204077139d09b71e1cc1 100644
--- a/app/services/ci/cancel_pipeline_service.rb
+++ b/app/services/ci/cancel_pipeline_service.rb
@@ -94,17 +94,19 @@ def execute_async?
     def cancel_jobs(jobs)
       retries = 3
       retry_lock(jobs, retries, name: 'ci_pipeline_cancel_running') do |jobs_to_cancel|
-        preloaded_relations = [:project, :pipeline, :deployment, :taggings]
-
         jobs_to_cancel.find_in_batches do |batch|
           relation = CommitStatus.id_in(batch)
-          Preloaders::CommitStatusPreloader.new(relation).execute(preloaded_relations)
+          Preloaders::CommitStatusPreloader.new(relation).execute(build_preloads)
 
           relation.each { |job| cancel_job(job) }
         end
       end
     end
 
+    def build_preloads
+      [:project, :pipeline, :deployment, :taggings]
+    end
+
     def cancel_job(job)
       if @auto_canceled_by_pipeline
         job.auto_canceled_by_id = @auto_canceled_by_pipeline.id
@@ -151,3 +153,5 @@ def cancel_children
     end
   end
 end
+
+Ci::CancelPipelineService.prepend_mod
diff --git a/config/sidekiq_queues.yml b/config/sidekiq_queues.yml
index 2d39a2ca5b0145ae63dc90214f654d95cbe24117..7cdf800fcb63ccfd3d50d7d2c579f2f69dc89ee9 100644
--- a/config/sidekiq_queues.yml
+++ b/config/sidekiq_queues.yml
@@ -1009,6 +1009,8 @@
   - 1
 - - security_scans
   - 2
+- - security_scans_ingest_reports
+  - 1
 - - security_scans_purge_by_job_id
   - 1
 - - security_secret_detection_gitlab_token_verification
diff --git a/ee/app/events/ci/job_security_scan_completed_event.rb b/ee/app/events/ci/job_security_scan_completed_event.rb
new file mode 100644
index 0000000000000000000000000000000000000000..17d6194475f5aa500292dc5b0f178936cd0bef6e
--- /dev/null
+++ b/ee/app/events/ci/job_security_scan_completed_event.rb
@@ -0,0 +1,15 @@
+# frozen_string_literal: true
+
+module Ci
+  class JobSecurityScanCompletedEvent < ::Gitlab::EventStore::Event
+    def schema
+      {
+        'type' => 'object',
+        'required' => ['job_id'],
+        'properties' => {
+          'job_id' => { 'type' => 'integer' }
+        }
+      }
+    end
+  end
+end
diff --git a/ee/app/models/concerns/ee/enums/ci/job_artifact.rb b/ee/app/models/concerns/ee/enums/ci/job_artifact.rb
index b4feff680250666b9284f3f8a53f33ad77d3f8d7..46c9e9226264a99d184079517ee755bd8351683c 100644
--- a/ee/app/models/concerns/ee/enums/ci/job_artifact.rb
+++ b/ee/app/models/concerns/ee/enums/ci/job_artifact.rb
@@ -11,6 +11,8 @@ module JobArtifact
 
         SECURITY_REPORT_AND_CYCLONEDX_REPORT_FILE_TYPES = (SECURITY_REPORT_FILE_TYPES | %w[cyclonedx]).freeze
 
+        SECURITY_ALL_REPORT_FILE_TYPES = (SECURITY_REPORT_AND_CYCLONEDX_REPORT_FILE_TYPES | %w[license_scanning]).freeze
+
         EE_REPORT_FILE_TYPES = {
           license_scanning: %w[license_scanning].freeze,
           dependency_list: %w[dependency_scanning].freeze,
@@ -34,6 +36,10 @@ def self.security_report_and_cyclonedx_report_file_types
           SECURITY_REPORT_AND_CYCLONEDX_REPORT_FILE_TYPES
         end
 
+        def self.all_security_report_file_types
+          SECURITY_ALL_REPORT_FILE_TYPES
+        end
+
         def self.ee_report_file_types
           EE_REPORT_FILE_TYPES
         end
diff --git a/ee/app/models/ee/ci/build.rb b/ee/app/models/ee/ci/build.rb
index 68bb62d6bfa1af84c5968019ba50163739ccff90..de676488cbe3766ee6b92fc344e0094acf2f2157 100644
--- a/ee/app/models/ee/ci/build.rb
+++ b/ee/app/models/ee/ci/build.rb
@@ -73,6 +73,17 @@ module Build
               ::Ci::Minutes::UpdateBuildMinutesService.new(build.project, nil).execute(build)
             end
           end
+
+          after_transition any => ::Ci::Build.completed_statuses do |build|
+            if ::Feature.enabled?(:ingest_sec_reports_when_sec_jobs_completed, build.project) &&
+                build.security_job?
+              build.run_after_commit do
+                ::Gitlab::EventStore.publish(
+                  ::Ci::JobSecurityScanCompletedEvent.new(data: { job_id: build.id })
+                )
+              end
+            end
+          end
         end
       end
 
@@ -203,8 +214,19 @@ def pages
       end
       strong_memoize_attr :pages
 
+      def security_job?
+        return false unless publishes_artifacts_reports?
+
+        ::EE::Enums::Ci::JobArtifact.security_report_and_cyclonedx_report_file_types.intersect?(reports_file_types)
+      end
+      strong_memoize_attr :security_job?
+
       private
 
+      def reports_file_types
+        reports_definitions&.keys&.map(&:to_s)
+      end
+
       def expand_pages_variables
         pages_config
           .slice(:path_prefix, :expire_in)
diff --git a/ee/app/models/ee/ci/pipeline.rb b/ee/app/models/ee/ci/pipeline.rb
index 48acf7741e4bac9a6cd6b5c31bb5efeab3473e68..822d572914169501554cd186b3fcd2acf83f6b17 100644
--- a/ee/app/models/ee/ci/pipeline.rb
+++ b/ee/app/models/ee/ci/pipeline.rb
@@ -74,13 +74,15 @@ def self.latest_limited_pipeline_ids_per_source(pipelines, sha)
           end
 
           after_transition any => ::Ci::Pipeline.completed_with_manual_statuses do |pipeline|
-            pipeline.run_after_commit do
-              if pipeline.can_store_security_reports?
-                ::Security::StoreScansWorker.perform_async(pipeline.id)
-                ::Security::ProcessScanEventsWorker.perform_async(pipeline.id)
-              else
-                ::Sbom::ScheduleIngestReportsService.new(pipeline).execute
-                ::Ci::CompareSecurityReportsService.set_security_mr_widget_to_ready(pipeline_id: pipeline.id)
+            if ::Feature.disabled?(:ingest_sec_reports_when_sec_jobs_completed, pipeline.project)
+              pipeline.run_after_commit do
+                if pipeline.can_store_security_reports?
+                  ::Security::StoreScansWorker.perform_async(pipeline.id)
+                  ::Security::ProcessScanEventsWorker.perform_async(pipeline.id)
+                else
+                  ::Sbom::ScheduleIngestReportsService.new(pipeline).execute
+                  ::Ci::CompareSecurityReportsService.set_security_mr_widget_to_ready(pipeline_id: pipeline.id)
+                end
               end
             end
           end
@@ -329,11 +331,17 @@ def security_scans_created_at
       end
 
       def has_security_reports?
-        security_and_license_scanning_file_types = EE::Enums::Ci::JobArtifact.security_report_and_cyclonedx_report_file_types | %w[license_scanning]
+        security_and_license_scanning_file_types = EE::Enums::Ci::JobArtifact.all_security_report_file_types
 
         complete_or_manual_and_has_reports?(::Ci::JobArtifact.with_file_types(security_and_license_scanning_file_types))
       end
 
+      def all_security_jobs_complete?
+        security_builds = builds.select(&:security_job?)
+        blocking_manual_jobs = security_builds.select(&:manual?).reject(&:allow_failure?)
+        security_builds.reject(&:manual?).all?(&:complete?) && blocking_manual_jobs.empty?
+      end
+
       def has_all_security_policies_reports?
         can_store_security_reports? && can_ingest_sbom_reports?
       end
diff --git a/ee/app/services/ee/ci/cancel_pipeline_service.rb b/ee/app/services/ee/ci/cancel_pipeline_service.rb
new file mode 100644
index 0000000000000000000000000000000000000000..aa55c1f0f94db7b7c9875ce581343d16829020cd
--- /dev/null
+++ b/ee/app/services/ee/ci/cancel_pipeline_service.rb
@@ -0,0 +1,16 @@
+# frozen_string_literal: true
+
+module EE
+  module Ci
+    module CancelPipelineService
+      extend ::Gitlab::Utils::Override
+
+      private
+
+      override :build_preloads
+      def build_preloads
+        super + [:metadata]
+      end
+    end
+  end
+end
diff --git a/ee/app/services/security/scans/ingest_reports_service.rb b/ee/app/services/security/scans/ingest_reports_service.rb
new file mode 100644
index 0000000000000000000000000000000000000000..3d8edfe2e453292a2740d4c4e22357b11dcede36
--- /dev/null
+++ b/ee/app/services/security/scans/ingest_reports_service.rb
@@ -0,0 +1,52 @@
+# frozen_string_literal: true
+
+module Security
+  module Scans
+    class IngestReportsService
+      include Gitlab::ExclusiveLeaseHelpers
+
+      TTL_REPORT_INGESTION = 1.hour
+
+      def self.execute(pipeline)
+        new(pipeline).execute
+      end
+
+      def initialize(pipeline)
+        @pipeline = pipeline
+      end
+
+      def execute
+        return unless pipeline.all_security_jobs_complete?
+
+        return if already_ingested?
+
+        if pipeline.project.can_store_security_reports?
+          ::Security::StoreScansWorker.perform_async(pipeline.id)
+          ::Security::ProcessScanEventsWorker.perform_async(pipeline.id)
+        else
+          ::Sbom::ScheduleIngestReportsService.new(pipeline).execute
+          ::Ci::CompareSecurityReportsService.set_security_mr_widget_to_ready(pipeline_id: pipeline.id)
+        end
+      end
+
+      private
+
+      attr_reader :pipeline
+
+      def scans_cache_key
+        sha = pipeline.latest_builds.select(&:security_job?)
+              .sort_by(&:id)
+              .map { |build| "#{build.id}:#{build.updated_at}" }
+              .join('|')
+              .then { |value| Digest::SHA512.hexdigest(value) }
+        "security:report:ingest:#{sha}"
+      end
+
+      def already_ingested?
+        ::Gitlab::Redis::SharedState.with do |redis|
+          !redis.set(scans_cache_key, 'OK', nx: true, ex: TTL_REPORT_INGESTION)
+        end
+      end
+    end
+  end
+end
diff --git a/ee/app/workers/all_queues.yml b/ee/app/workers/all_queues.yml
index 4a7bb91eff3c9f1f7b6130a1682535164ac969f9..67c5d0f70fa760f0f53ee0156f89f1855f6a6a67 100644
--- a/ee/app/workers/all_queues.yml
+++ b/ee/app/workers/all_queues.yml
@@ -3814,6 +3814,16 @@
   :idempotent: true
   :tags: []
   :queue_namespace:
+- :name: security_scans_ingest_reports
+  :worker_name: Security::Scans::IngestReportsWorker
+  :feature_category: :vulnerability_management
+  :has_external_dependencies: false
+  :urgency: :low
+  :resource_boundary: :cpu
+  :weight: 1
+  :idempotent: true
+  :tags: []
+  :queue_namespace:
 - :name: security_scans_purge_by_job_id
   :worker_name: Security::Scans::PurgeByJobIdWorker
   :feature_category: :vulnerability_management
diff --git a/ee/app/workers/security/scans/ingest_reports_worker.rb b/ee/app/workers/security/scans/ingest_reports_worker.rb
new file mode 100644
index 0000000000000000000000000000000000000000..05a002191dd228fd78007493ff5163275f7a9896
--- /dev/null
+++ b/ee/app/workers/security/scans/ingest_reports_worker.rb
@@ -0,0 +1,25 @@
+# frozen_string_literal: true
+
+module Security
+  module Scans
+    class IngestReportsWorker
+      include ApplicationWorker
+      include Gitlab::EventStore::Subscriber
+
+      feature_category :vulnerability_management
+      urgency :low
+      worker_resource_boundary :cpu
+      data_consistency :sticky
+
+      defer_on_database_health_signal :gitlab_sec, [:vulnerability_occurences], 1.minute
+      idempotent!
+
+      def handle_event(event)
+        build = ::Ci::Build.find_by_id(event.data[:job_id])
+        return unless build
+
+        ::Security::Scans::IngestReportsService.execute(build.pipeline)
+      end
+    end
+  end
+end
diff --git a/ee/app/workers/security/store_scans_worker.rb b/ee/app/workers/security/store_scans_worker.rb
index 292a5bf962ae7f3c8fbd904ceb9260efbedfa2d4..0d66b0c00a634b47dc02d5d5c907082c7abe6958 100644
--- a/ee/app/workers/security/store_scans_worker.rb
+++ b/ee/app/workers/security/store_scans_worker.rb
@@ -14,7 +14,7 @@ class StoreScansWorker # rubocop:disable Scalability/IdempotentWorker
 
     def perform(pipeline_id)
       ::Ci::Pipeline.find_by_id(pipeline_id).try do |pipeline|
-        break unless pipeline.can_store_security_reports?
+        break unless pipeline.project.can_store_security_reports?
 
         Security::StoreScansService.execute(pipeline)
       end
diff --git a/ee/config/feature_flags/gitlab_com_derisk/ingest_sec_reports_when_sec_jobs_completed.yml b/ee/config/feature_flags/gitlab_com_derisk/ingest_sec_reports_when_sec_jobs_completed.yml
new file mode 100644
index 0000000000000000000000000000000000000000..b67d6ecad2e183a4c62af5d8d27b28e3570d1ac4
--- /dev/null
+++ b/ee/config/feature_flags/gitlab_com_derisk/ingest_sec_reports_when_sec_jobs_completed.yml
@@ -0,0 +1,10 @@
+---
+name: ingest_sec_reports_when_sec_jobs_completed
+description:
+feature_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/513326
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/195012
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/554222
+milestone: '18.3'
+group: group::security infrastructure
+type: gitlab_com_derisk
+default_enabled: false
diff --git a/ee/lib/ee/gitlab/event_store.rb b/ee/lib/ee/gitlab/event_store.rb
index f9707b5cbe1444b1a7a1ed760d7e848fdea36da2..d5ac078d2d3d21c66f850100df1715673c364179 100644
--- a/ee/lib/ee/gitlab/event_store.rb
+++ b/ee/lib/ee/gitlab/event_store.rb
@@ -24,6 +24,7 @@ def configure!(store)
           # Add EE only subscriptions here:
 
           store.subscribe ::Security::Scans::PurgeByJobIdWorker, to: ::Ci::JobArtifactsDeletedEvent
+          store.subscribe ::Security::Scans::IngestReportsWorker, to: ::Ci::JobSecurityScanCompletedEvent
           store.subscribe ::Geo::CreateRepositoryUpdatedEventWorker,
             to: ::Repositories::KeepAroundRefsCreatedEvent,
             if: ->(_) { ::Gitlab::Geo.primary? }
diff --git a/ee/spec/lib/ee/gitlab/event_store_spec.rb b/ee/spec/lib/ee/gitlab/event_store_spec.rb
index d23052e5d52d71bedc3d47c5709611beed901f70..eff64061cddd43f620699bf2d7e4926f97a30109 100644
--- a/ee/spec/lib/ee/gitlab/event_store_spec.rb
+++ b/ee/spec/lib/ee/gitlab/event_store_spec.rb
@@ -13,6 +13,7 @@
         Ai::ActiveContext::Code::SaasInitialIndexingEvent,
         ::Ci::JobArtifactsDeletedEvent,
         ::Ci::PipelineCreatedEvent,
+        ::Ci::JobSecurityScanCompletedEvent,
         ::Repositories::KeepAroundRefsCreatedEvent,
         ::MergeRequests::ApprovedEvent,
         ::MergeRequests::MergedEvent,
diff --git a/ee/spec/models/ci/build_spec.rb b/ee/spec/models/ci/build_spec.rb
index fe23214e32bc9774b0814abf6b45caa24dc121bf..cfac99ec682531414f812c33dd26678472021f5b 100644
--- a/ee/spec/models/ci/build_spec.rb
+++ b/ee/spec/models/ci/build_spec.rb
@@ -372,6 +372,28 @@
     end
   end
 
+  describe '#security_job?' do
+    subject { job.security_job? }
+
+    context 'when build does not publish artifacts reports' do
+      it { is_expected.to be false }
+    end
+
+    context 'when build publishes artifacts reports' do
+      context 'when build has multiple security report artifact' do
+        let(:job) { create(:ci_build, :sast, pipeline: pipeline) }
+
+        it { is_expected.to be true }
+      end
+
+      context 'when build has no security report artifacts' do
+        let(:job) { create(:ci_build, :coverage_report_cobertura, pipeline: pipeline) }
+
+        it { is_expected.to be false }
+      end
+    end
+  end
+
   describe '#unmerged_security_reports' do
     subject(:security_reports) { job.unmerged_security_reports }
 
@@ -1265,4 +1287,80 @@
       end
     end
   end
+
+  describe 'JobSecurityScanCompletedEvent publishing' do
+    let(:build) { create(:ci_build, :running, pipeline: pipeline) }
+
+    context 'when build is a security job' do
+      before do
+        allow(build).to receive(:security_job?).and_return(true)
+      end
+
+      context "when actions aren't complete statuses" do
+        where(:transition_method, :transition_description) do
+          [
+            [:process, 'created'],
+            [:enqueue, 'pending'],
+            [:run, 'running']
+          ]
+        end
+
+        with_them do
+          context "when transitioning to #{params[:transition_description]}" do
+            it 'does not publish the JobSecurityScanCompletedEvent' do
+              expect(::Gitlab::EventStore).not_to receive(:publish)
+
+              build.public_send(transition_method)
+            end
+          end
+        end
+      end
+
+      context "when actions are complete statuses" do
+        where(:transition_method, :transition_description) do
+          [
+            [:success, 'success'],
+            [:drop, 'failed'],
+            [:cancel, 'canceled']
+          ]
+        end
+
+        with_them do
+          context "when transitioning to #{params[:transition_description]}" do
+            it 'publishes JobSecurityScanCompletedEvent' do
+              expect(::Gitlab::EventStore).to receive(:publish) do |event|
+                expect(event).to be_an_instance_of(::Ci::JobSecurityScanCompletedEvent)
+                expect(event.data).to eq({ "job_id" => build.id })
+              end
+              build.public_send(transition_method)
+            end
+          end
+        end
+      end
+    end
+
+    context 'when build is not a security job' do
+      before do
+        allow(build).to receive(:security_job?).and_return(false)
+      end
+
+      where(:transition_method, :transition_description) do
+        [
+          [:success, 'success'],
+          [:drop, 'failed'],
+          [:cancel, 'canceled']
+        ]
+      end
+
+      with_them do
+        context "when transitioning to #{params[:transition_description]}" do
+          it 'does not publish the JobSecurityScanCompletedEvent' do
+            expect(::Gitlab::EventStore).not_to receive(:publish)
+
+            build.public_send(transition_method)
+          end
+        end
+      end
+    end
+  end
 end
diff --git a/ee/spec/models/ci/pipeline_spec.rb b/ee/spec/models/ci/pipeline_spec.rb
index 43844931e05c62af4456d51ef42fd79c4f85f150..f4d7ca0130b0b31f7bd0060223e2a6bc8ff0e48a 100644
--- a/ee/spec/models/ci/pipeline_spec.rb
+++ b/ee/spec/models/ci/pipeline_spec.rb
@@ -218,26 +218,46 @@
         allow(redis_spy).to receive(:ttl).and_return(10) # to allow event tracking Redis call
       end
 
-      it "sets the polling redis key for mr security widget when transitioning to: #{transition}" do
-        expect(Gitlab::Redis::SharedState).to receive(:with).and_yield(redis_spy).at_least(:once)
+      context "when ingest_sec_reports_when_sec_jobs_completed set to true" do
+        before do
+          stub_feature_flags(ingest_sec_reports_when_sec_jobs_completed: true)
+        end
 
-        transition_pipeline
+        it "sets the polling redis key for mr security widget when transitioning to: #{transition}" do
+          expect(Gitlab::Redis::SharedState).to receive(:with).and_yield(redis_spy).at_least(:once)
 
-        expect(redis_spy).to have_received(:set).with(cache_key, pipeline_id, ex: kind_of(Integer))
+          transition_pipeline
+
+          expect(redis_spy).to have_received(:set).with(cache_key, pipeline_id, ex: kind_of(Integer))
+        end
       end
 
-      context 'when the security scans can not be stored for the pipeline' do
+      context "when ingest_sec_reports_when_sec_jobs_completed set to false" do
         before do
-          allow(pipeline).to receive(:can_store_security_reports?).and_return(false)
+          stub_feature_flags(ingest_sec_reports_when_sec_jobs_completed: false)
         end
 
-        it 'deletes the polling cache key' do
-          expect(Gitlab::Redis::SharedState).to receive(:with).and_yield(redis_spy).at_least(:twice)
+        it "sets the polling redis key for mr security widget when transitioning to: #{transition}" do
+          expect(Gitlab::Redis::SharedState).to receive(:with).and_yield(redis_spy).at_least(:once)
 
           transition_pipeline
 
-          expect(redis_spy).to have_received(:set).with(cache_key, pipeline_id, ex: kind_of(Integer)).once
-          expect(redis_spy).to have_received(:del).with(cache_key).once
+          expect(redis_spy).to have_received(:set).with(cache_key, pipeline_id, ex: kind_of(Integer))
+        end
+
+        context 'when the security scans can not be stored for the pipeline' do
+          before do
+            allow(pipeline).to receive(:can_store_security_reports?).and_return(false)
+          end
+
+          it 'deletes the polling cache key' do
+            expect(Gitlab::Redis::SharedState).to receive(:with).and_yield(redis_spy).at_least(:twice)
+
+            transition_pipeline
+
+            expect(redis_spy).to have_received(:set).with(cache_key, pipeline_id, ex: kind_of(Integer)).once
+            expect(redis_spy).to have_received(:del).with(cache_key).once
+          end
         end
       end
     end
@@ -258,23 +278,55 @@
         allow(pipeline).to receive(:can_store_security_reports?).and_return(can_store_security_reports)
       end
 
-      context 'when the security scans can be stored for the pipeline' do
-        let(:can_store_security_reports) { true }
+      context 'when ingest_sec_reports_when_sec_jobs_completed flag is disabled' do
+        before do
+          stub_feature_flags(ingest_sec_reports_when_sec_jobs_completed: false)
+        end
 
-        it 'schedules store security scans job' do
-          transition_pipeline
+        context 'when the security scans can be stored for the pipeline' do
+          let(:can_store_security_reports) { true }
 
-          expect(::Security::StoreScansWorker).to have_received(:perform_async).with(pipeline.id)
+          it 'schedules store security scans job' do
+            transition_pipeline
+
+            expect(::Security::StoreScansWorker).to have_received(:perform_async).with(pipeline.id)
+          end
+        end
+
+        context 'when the security scans can not be stored for the pipeline' do
+          let(:can_store_security_reports) { false }
+
+          it 'does not schedule store security scans job' do
+            transition_pipeline
+
+            expect(::Security::StoreScansWorker).not_to have_received(:perform_async)
+          end
         end
       end
 
-      context 'when the security scans can not be stored for the pipeline' do
-        let(:can_store_security_reports) { false }
+      context 'when ingest_sec_reports_when_sec_jobs_completed flag is enabled' do
+        before do
+          stub_feature_flags(ingest_sec_reports_when_sec_jobs_completed: true)
+        end
 
-        it 'does not schedule store security scans job' do
-          transition_pipeline
+        context 'when the security scans can be stored for the pipeline' do
+          let(:can_store_security_reports) { true }
+
+          it 'does not schedule store security scans job' do
+            transition_pipeline
+
+            expect(::Security::StoreScansWorker).not_to have_received(:perform_async)
+          end
+        end
+
+        context 'when the security scans can not be stored for the pipeline' do
+          let(:can_store_security_reports) { false }
+
+          it 'does not schedule store security scans job' do
+            transition_pipeline
 
-          expect(::Security::StoreScansWorker).not_to have_received(:perform_async)
+            expect(::Security::StoreScansWorker).not_to have_received(:perform_async)
+          end
         end
       end
     end
@@ -349,57 +401,123 @@
       allow(::Sbom::ScheduleIngestReportsService).to receive(:new).with(pipeline).and_return(sbom_ingestion_scheduler)
     end
 
-    shared_examples_for 'ingesting sbom reports' do
-      context 'when security reports are available' do
-        before do
-          allow(pipeline).to receive(:can_store_security_reports?).and_return(true)
+    context 'when ingest_sec_reports_when_sec_jobs_completed flag is disabled' do
+      before do
+        stub_feature_flags(ingest_sec_reports_when_sec_jobs_completed: false)
+      end
+
+      shared_examples_for 'ingesting sbom reports' do
+        context 'when security reports are available' do
+          before do
+            allow(pipeline).to receive(:can_store_security_reports?).and_return(true)
+          end
+
+          it 'does not try to ingest the SBOM reports' do
+            transition_pipeline
+
+            expect(::Sbom::ScheduleIngestReportsService).not_to have_received(:new)
+          end
         end
 
-        it 'does not try to ingest the SBOM reports' do
-          transition_pipeline
+        context 'when security reports are not available' do
+          before do
+            allow(pipeline).to receive(:can_store_security_reports?).and_return(false)
+          end
+
+          it 'tries to ingest sbom reports' do
+            transition_pipeline
 
-          expect(::Sbom::ScheduleIngestReportsService).not_to have_received(:new)
+            expect(::Sbom::ScheduleIngestReportsService).to have_received(:new).with(pipeline)
+            expect(sbom_ingestion_scheduler).to have_received(:execute)
+          end
         end
       end
 
-      context 'when security reports are not available' do
-        before do
-          allow(pipeline).to receive(:can_store_security_reports?).and_return(false)
+      context 'when transitioning to completed or blocked status' do
+        where(:transition) { %i[succeed drop skip cancel block] }
+
+        with_them do
+          it_behaves_like 'ingesting sbom reports'
         end
+      end
 
-        it 'tries to ingest sbom reports' do
-          transition_pipeline
+      context 'when transitioning to a non-completed status except block' do
+        where(:transition) do
+          %i[
+            enqueue
+            request_resource
+            prepare
+            run
+            delay
+          ]
+        end
+
+        with_them do
+          it 'does not try to ingest sbom reports' do
+            transition_pipeline
 
-          expect(::Sbom::ScheduleIngestReportsService).to have_received(:new).with(pipeline)
-          expect(sbom_ingestion_scheduler).to have_received(:execute)
+            expect(::Sbom::ScheduleIngestReportsService).not_to have_received(:new)
+          end
         end
       end
     end
 
-    context 'when transitioning to completed or blocked status' do
-      where(:transition) { %i[succeed drop skip cancel block] }
+    context 'when ingest_sec_reports_when_sec_jobs_completed flag is enabled' do
+      before do
+        stub_feature_flags(ingest_sec_reports_when_sec_jobs_completed: true)
+      end
 
-      with_them do
-        it_behaves_like 'ingesting sbom reports'
+      shared_examples_for 'ingesting sbom reports' do
+        context 'when security reports are available' do
+          before do
+            allow(pipeline).to receive(:can_store_security_reports?).and_return(true)
+          end
+
+          it 'does not try to ingest the SBOM reports' do
+            transition_pipeline
+
+            expect(::Sbom::ScheduleIngestReportsService).not_to have_received(:new)
+          end
+        end
+
+        context 'when security reports are not available' do
+          before do
+            allow(pipeline).to receive(:can_store_security_reports?).and_return(false)
+          end
+
+          it 'tries to ingest sbom reports' do
+            transition_pipeline
+
+            expect(::Sbom::ScheduleIngestReportsService).not_to have_received(:new)
+          end
+        end
       end
-    end
 
-    context 'when transitioning to a non-completed status except block' do
-      where(:transition) do
-        %i[
-          enqueue
-          request_resource
-          prepare
-          run
-          delay
-        ]
+      context 'when transitioning to completed or blocked status' do
+        where(:transition) { %i[succeed drop skip cancel block] }
+
+        with_them do
+          it_behaves_like 'ingesting sbom reports'
+        end
       end
 
-      with_them do
-        it 'does not try to ingest sbom reports' do
-          transition_pipeline
+      context 'when transitioning to a non-completed status except block' do
+        where(:transition) do
+          %i[
+            enqueue
+            request_resource
+            prepare
+            run
+            delay
+          ]
+        end
 
-          expect(::Sbom::ScheduleIngestReportsService).not_to have_received(:new)
+        with_them do
+          it 'does not try to ingest sbom reports' do
+            transition_pipeline
+
+            expect(::Sbom::ScheduleIngestReportsService).not_to have_received(:new)
+          end
         end
       end
     end
diff --git a/ee/spec/services/security/scans/ingest_reports_service_spec.rb b/ee/spec/services/security/scans/ingest_reports_service_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..f3307160e1694fe4e8b3e888fe83ed6ace594499
--- /dev/null
+++ b/ee/spec/services/security/scans/ingest_reports_service_spec.rb
@@ -0,0 +1,97 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Security::Scans::IngestReportsService, :clean_gitlab_redis_shared_state, feature_category: :vulnerability_management do
+  let_it_be(:user) { create(:user) }
+
+  describe '.execute' do
+    let_it_be(:pipeline) { create(:ci_pipeline, user: user) }
+    let(:mock_service_object) { instance_double(described_class, execute: true) }
+
+    subject(:execute) { described_class.execute(pipeline) }
+
+    before do
+      allow(described_class).to receive(:new).with(pipeline).and_return(mock_service_object)
+    end
+
+    it 'delegates the call to an instance of `Security::Scans::IngestReportsService`' do
+      execute
+
+      expect(described_class).to have_received(:new).with(pipeline)
+      expect(mock_service_object).to have_received(:execute)
+    end
+  end
+
+  describe '#execute' do
+    let(:service_object) { described_class.new(pipeline) }
+    let(:mock_sbom_ingestion_service) { instance_double(::Sbom::ScheduleIngestReportsService, execute: nil) }
+
+    subject(:ingest_security_scans) { service_object.execute }
+
+    before do
+      allow(::Security::StoreScansWorker).to receive(:perform_async)
+      allow(::Sbom::ScheduleIngestReportsService).to receive(:new)
+        .with(pipeline).and_return(mock_sbom_ingestion_service)
+    end
+
+    context 'when the security scans not completed' do
+      let_it_be(:pipeline) { create(:ci_pipeline, user: user) }
+      let_it_be(:job) { create(:ci_build, :sast, pipeline: pipeline, status: 'running') }
+
+      it 'does not schedule store security scans job and to ingests sbom reports' do
+        ingest_security_scans
+
+        expect(::Security::StoreScansWorker).not_to have_received(:perform_async)
+        expect(::Sbom::ScheduleIngestReportsService).not_to have_received(:new)
+      end
+    end
+
+    context 'when the security scans can be stored for the pipeline' do
+      let_it_be(:pipeline) { create(:ci_pipeline, user: user) }
+      let_it_be(:job) { create(:ci_build, :sast, pipeline: pipeline, status: 'success') }
+      let_it_be(:mock_cache_key) { SecureRandom.uuid }
+
+      before do
+        allow(pipeline).to receive_message_chain(:project,
+          :can_store_security_reports?).and_return(true)
+        allow(service_object).to receive(:scans_cache_key).and_return(mock_cache_key)
+      end
+
+      it 'schedules store security scans job and does not ingest the SBOM reports' do
+        ingest_security_scans
+
+        expect(::Security::StoreScansWorker).to have_received(:perform_async).with(pipeline.id)
+        expect(::Sbom::ScheduleIngestReportsService).not_to have_received(:new)
+      end
+
+      it 'already ingested' do
+        ::Gitlab::Redis::SharedState.with do |redis|
+          !redis.set(mock_cache_key, 'OK', nx: true, ex: described_class::TTL_REPORT_INGESTION)
+        end
+        ingest_security_scans
+
+        expect(::Security::StoreScansWorker).not_to have_received(:perform_async)
+        expect(::Sbom::ScheduleIngestReportsService).not_to have_received(:new)
+      end
+    end
+
+    context 'when the security scans can not be stored for the pipeline' do
+      before do
+        allow(pipeline).to receive_message_chain(:project,
+          :can_store_security_reports?).and_return(false)
+      end
+
+      let_it_be(:pipeline) { create(:ci_pipeline, user: user) }
+      let_it_be(:job) { create(:ci_build, :sast, pipeline: pipeline, status: 'success') }
+
+      it 'does not schedule store security scans job and to ingests sbom reports' do
+        ingest_security_scans
+
+        expect(::Security::StoreScansWorker).not_to have_received(:perform_async)
+        expect(::Sbom::ScheduleIngestReportsService).to have_received(:new).with(pipeline)
+        expect(mock_sbom_ingestion_service).to have_received(:execute)
+      end
+    end
+  end
+end
diff --git a/ee/spec/workers/security/scans/ingest_reports_worker_spec.rb b/ee/spec/workers/security/scans/ingest_reports_worker_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..e21ade76a51d9063bc8963647baa31265bac9464
--- /dev/null
+++ b/ee/spec/workers/security/scans/ingest_reports_worker_spec.rb
@@ -0,0 +1,25 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe Security::Scans::IngestReportsWorker, feature_category: :vulnerability_management do
+  let(:pipeline) { create(:ci_pipeline) }
+  let(:job) { create(:ci_build, :sast, pipeline: pipeline, status: 'success') }
+  let(:event) { ::Ci::JobSecurityScanCompletedEvent.new(data: { job_id: job.id }) }
+
+  subject(:handle_event) { consume_event(subscriber: described_class, event: event) }
+
+  before do
+    allow(::Security::Scans::IngestReportsService).to receive(:execute)
+  end
+
+  it_behaves_like 'subscribes to event'
+
+  describe '.handle_event' do
+    it 'handle_event calls service' do
+      handle_event
+
+      expect(::Security::Scans::IngestReportsService).to have_received(:execute).with(pipeline)
+    end
+  end
+end
diff --git a/ee/spec/workers/security/store_scans_worker_spec.rb b/ee/spec/workers/security/store_scans_worker_spec.rb
index 41f7815230fc1297aee062089955b7e91a3535a7..d0513d33c377a8ba128a8d09103b260bc430aace 100644
--- a/ee/spec/workers/security/store_scans_worker_spec.rb
+++ b/ee/spec/workers/security/store_scans_worker_spec.rb
@@ -13,7 +13,8 @@
     before do
       allow(Security::StoreScansService).to receive(:execute)
       allow_next_found_instance_of(Ci::Pipeline) do |record|
-        allow(record).to receive(:can_store_security_reports?).and_return(can_store_security_reports)
+        allow(record).to receive_message_chain(:project,
+          :can_store_security_reports?).and_return(can_store_security_reports)
       end
     end