diff --git a/app/finders/clusters/agents/authorizations/ci_access/finder.rb b/app/finders/clusters/agents/authorizations/ci_access/finder.rb
index e55cc6ec11d1960b9760d913f7de66868ba0792a..846c80bb37ca73eb14cb51df8471889f3017d655 100644
--- a/app/finders/clusters/agents/authorizations/ci_access/finder.rb
+++ b/app/finders/clusters/agents/authorizations/ci_access/finder.rb
@@ -5,8 +5,9 @@ module Agents
     module Authorizations
       module CiAccess
         class Finder
-          def initialize(project)
+          def initialize(project, agent: nil)
             @project = project
+            @agent = agent
           end
 
           def execute
@@ -17,10 +18,12 @@ def execute
 
           private
 
-          attr_reader :project
+          attr_reader :project, :agent
 
           def implicit_authorizations
-            project.cluster_agents.map do |agent|
+            agents = agent&.project_id == project.id ? [agent] : project.cluster_agents
+
+            agents.map do |agent|
               Clusters::Agents::Authorizations::CiAccess::ImplicitAuthorization.new(agent: agent)
             end
           end
@@ -29,13 +32,15 @@ def implicit_authorizations
           def project_authorizations
             namespace_ids = project.group ? all_namespace_ids : project.namespace_id
 
-            Clusters::Agents::Authorizations::CiAccess::ProjectAuthorization
+            query = Clusters::Agents::Authorizations::CiAccess::ProjectAuthorization
               .where(project_id: project.id)
               .joins(agent: :project)
               .preload(agent: :project)
               .where(cluster_agents: { projects: { namespace_id: namespace_ids } })
               .with_available_ci_access_fields(project)
-              .to_a
+
+            query = query.where(agent_id: agent.id) if agent
+            query.to_a
           end
 
           def group_authorizations
@@ -52,7 +57,7 @@ def group_authorizations
               authorizations[:group_id].eq(ordered_ancestors_cte.table[:id])
             ).join_sources
 
-            Clusters::Agents::Authorizations::CiAccess::GroupAuthorization
+            query = Clusters::Agents::Authorizations::CiAccess::GroupAuthorization
               .with(ordered_ancestors_cte.to_arel)
               .joins(cte_join_sources)
               .joins(agent: :project)
@@ -66,7 +71,9 @@ def group_authorizations
               )
               .select('DISTINCT ON (agent_id) agent_group_authorizations.*')
               .preload(agent: :project)
-              .to_a
+
+            query = query.where(agent_id: agent.id) if agent
+            query.to_a
           end
           # rubocop: enable CodeReuse/ActiveRecord
 
diff --git a/app/models/clusters/agent.rb b/app/models/clusters/agent.rb
index f0300f33cf5364bea1cd82e3aa12aad5f3773c71..8b58f31e8ff550e2f4e4a39cdb8435cb6a3c898a 100644
--- a/app/models/clusters/agent.rb
+++ b/app/models/clusters/agent.rb
@@ -93,6 +93,10 @@ def user_access_authorizations
       ).select('config').compact.first
     end
 
+    def resource_management_enabled?
+      false # Overridden in EE
+    end
+
     private
 
     def all_ci_access_authorized_projects_for(user)
diff --git a/app/services/environments/create_for_job_service.rb b/app/services/environments/create_for_job_service.rb
index 02545ce03e0ea6d28b74411ef18ad48ced1a3a4c..74b3aedb2594d29e575b665c72f3940d60678c43 100644
--- a/app/services/environments/create_for_job_service.rb
+++ b/app/services/environments/create_for_job_service.rb
@@ -27,6 +27,13 @@ def to_resource(job)
         environment.auto_stop_in = expanded_auto_stop_in(job)
         environment.tier = job.environment_tier_from_options
         environment.merge_request = job.pipeline.merge_request
+
+        if resource_management_feature_enabled?(job)
+          authorization = matching_authorization(job)
+          if authorization && authorization.agent.resource_management_enabled?
+            environment.cluster_agent = authorization.agent
+          end
+        end
       end
     end
     # rubocop: enable Performance/ActiveRecordSubtransactionMethods
@@ -36,5 +43,39 @@ def expanded_auto_stop_in(job)
 
       ExpandVariables.expand(job.environment_auto_stop_in, -> { job.simple_variables.sort_and_expand_all })
     end
+
+    def cluster_agent_path(job)
+      environment_options(job).dig(:kubernetes, :agent)
+    end
+
+    def environment_options(job)
+      job.options&.dig(:environment) || {}
+    end
+
+    def matching_authorization(job)
+      return false unless cluster_agent_path(job)
+
+      requested_project_path, requested_agent_name = expanded_cluster_agent_path(job).split(':')
+
+      ci_access_authorizations_for_project(job).find do |authorization|
+        requested_project_path == authorization.config_project.full_path &&
+          requested_agent_name == authorization.agent.name &&
+          authorization.config.dig('resource_management', 'enabled') == true
+      end
+    end
+
+    def ci_access_authorizations_for_project(job)
+      Clusters::Agents::Authorizations::CiAccess::Finder.new(job.project).execute
+    end
+
+    def expanded_cluster_agent_path(job)
+      return unless cluster_agent_path(job)
+
+      ExpandVariables.expand(cluster_agent_path(job), -> { job.simple_variables.sort_and_expand_all })
+    end
+
+    def resource_management_feature_enabled?(job)
+      ::Feature.enabled?(:gitlab_managed_cluster_resources, job.project)
+    end
   end
 end
diff --git a/config/feature_flags/beta/gitlab_managed_cluster_resources.yml b/config/feature_flags/beta/gitlab_managed_cluster_resources.yml
new file mode 100644
index 0000000000000000000000000000000000000000..dc11310983bbae5724490269afab3b60517c6a31
--- /dev/null
+++ b/config/feature_flags/beta/gitlab_managed_cluster_resources.yml
@@ -0,0 +1,9 @@
+---
+name: gitlab_managed_cluster_resources
+feature_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/507268
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/178824
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/514901
+milestone: '17.9'
+group: group::environments
+type: beta
+default_enabled: false
diff --git a/ee/app/models/ee/clusters/agent.rb b/ee/app/models/ee/clusters/agent.rb
index 89f57bb4dd51cf8940e57426628ef154697c98d7..55c9dd87b19b107a0bfc5e0a863bd2725b91e28b 100644
--- a/ee/app/models/ee/clusters/agent.rb
+++ b/ee/app/models/ee/clusters/agent.rb
@@ -55,6 +55,11 @@ module Agent
             unversioned_latest_workspaces_agent_config: { enabled: true }
           )
         end
+
+        def resource_management_enabled?
+          ::Feature.enabled?(:gitlab_managed_cluster_resources, project) &&
+            project.licensed_feature_available?(:agent_managed_resources)
+        end
       end
     end
   end
diff --git a/ee/app/models/gitlab_subscriptions/features.rb b/ee/app/models/gitlab_subscriptions/features.rb
index 2b063996a0e00435b23526f0f7d185a6e0ef9af1..be74bfcf2746301aa02d25fe4ec484ee2d5c0b9f 100644
--- a/ee/app/models/gitlab_subscriptions/features.rb
+++ b/ee/app/models/gitlab_subscriptions/features.rb
@@ -85,6 +85,7 @@ class Features
       ai_chat
       adjourned_deletion_for_projects_and_groups
       admin_audit_log
+      agent_managed_resources
       auditor_user
       blocking_merge_requests
       board_assignee_lists
diff --git a/ee/spec/models/ee/clusters/agent_spec.rb b/ee/spec/models/ee/clusters/agent_spec.rb
index ed642ba2ea0e91bc9829f6c837e028b99f46c0f7..d1c002c7bcac47e0d15d50270949a76c21b79b3d 100644
--- a/ee/spec/models/ee/clusters/agent_spec.rb
+++ b/ee/spec/models/ee/clusters/agent_spec.rb
@@ -61,5 +61,47 @@
           agent_1, agent_2, agent_3, agent_with_remote_development_config_disabled)
       end
     end
+
+    describe '#resource_management_enabled?' do
+      subject { agent_1.resource_management_enabled? }
+
+      context 'when feature flag is disabled' do
+        before do
+          stub_feature_flags(gitlab_managed_cluster_resources: false)
+        end
+
+        context 'when licensed feature is not available' do
+          before do
+            stub_licensed_features(agent_managed_resources: false)
+          end
+
+          it { is_expected.to be_falsey }
+        end
+
+        context 'when licensed feature is available' do
+          before do
+            stub_licensed_features(agent_managed_resources: true)
+          end
+
+          it { is_expected.to be_falsey }
+        end
+      end
+
+      context 'when licensed feature is not available' do
+        before do
+          stub_licensed_features(agent_managed_resources: false)
+        end
+
+        it { is_expected.to be_falsey }
+      end
+
+      context 'when licensed feature is available' do
+        before do
+          stub_licensed_features(agent_managed_resources: true)
+        end
+
+        it { is_expected.to be_truthy }
+      end
+    end
   end
 end
diff --git a/lib/gitlab/ci/build/prerequisite/managed_resource.rb b/lib/gitlab/ci/build/prerequisite/managed_resource.rb
index 6cfd05397daec871226c0b3786c991821b7d859c..777b995ce0fe07a1d9251e643e88212d59d3f987 100644
--- a/lib/gitlab/ci/build/prerequisite/managed_resource.rb
+++ b/lib/gitlab/ci/build/prerequisite/managed_resource.rb
@@ -10,9 +10,8 @@ class ManagedResource < Base
           DEFAULT_TEMPLATE_NAME = "default"
 
           def unmet?
-            return false unless resource_management_enabled?
-
             return false unless valid_for_managed_resources?(environment:, build:)
+            return false unless resource_management_enabled?
 
             !managed_resource&.completed?
           end
@@ -33,9 +32,13 @@ def complete!
 
           private
 
-          # TODO: Check "resource_management.enabled" flag in the follow-up MR.
           def resource_management_enabled?
-            false
+            return false unless environment.cluster_agent.resource_management_enabled?
+
+            authorization = ::Clusters::Agents::Authorizations::CiAccess::Finder
+                              .new(build.project, agent: environment.cluster_agent).execute.first
+
+            authorization.present? && authorization.config.dig('resource_management', 'enabled') == true
           end
 
           def ensure_environment
diff --git a/spec/factories/clusters/agents/authorizations/ci_access/project_authorizations.rb b/spec/factories/clusters/agents/authorizations/ci_access/project_authorizations.rb
index f23c7d8bf279277f126bd0c11c8202c521e64320..6961881f9100b48dc6df78178c514f87499c3a12 100644
--- a/spec/factories/clusters/agents/authorizations/ci_access/project_authorizations.rb
+++ b/spec/factories/clusters/agents/authorizations/ci_access/project_authorizations.rb
@@ -8,12 +8,14 @@
     transient do
       environments { nil }
       protected_branches_only { false }
+      resource_management_enabled { false }
     end
 
     config do
       { default_namespace: 'production' }.tap do |c|
         c[:environments] = environments if environments
         c[:protected_branches_only] = protected_branches_only
+        c[:resource_management] = { enabled: true } if resource_management_enabled
       end
     end
   end
diff --git a/spec/finders/clusters/agents/authorizations/ci_access/finder_spec.rb b/spec/finders/clusters/agents/authorizations/ci_access/finder_spec.rb
index 0d010729d5c220f9958f7ad313d146aeaf4310f4..5db321b416139a8bc49871e93c7830c8d5e5e615 100644
--- a/spec/finders/clusters/agents/authorizations/ci_access/finder_spec.rb
+++ b/spec/finders/clusters/agents/authorizations/ci_access/finder_spec.rb
@@ -19,7 +19,9 @@
     let_it_be(:staging_agent) { create(:cluster_agent, project: agent_configuration_project) }
     let_it_be(:production_agent) { create(:cluster_agent, project: agent_configuration_project) }
 
-    subject { described_class.new(requesting_project).execute }
+    let(:finder) { described_class.new(requesting_project) }
+
+    subject { finder.execute }
 
     shared_examples_for 'access_as' do
       let(:config) { { access_as: { access_as => {} } } }
@@ -50,34 +52,54 @@
     end
 
     describe 'project authorizations' do
-      context 'agent configuration project does not share a root namespace with the given project' do
-        let(:unrelated_agent) { create(:cluster_agent) }
+      context 'when initialized without an agent' do
+        context 'agent configuration project does not share a root namespace with the given project' do
+          let(:unrelated_agent) { create(:cluster_agent) }
+
+          before do
+            create(:agent_ci_access_project_authorization, agent: unrelated_agent, project: requesting_project)
+          end
 
-        before do
-          create(:agent_ci_access_project_authorization, agent: unrelated_agent, project: requesting_project)
+          it { is_expected.to be_empty }
         end
 
-        it { is_expected.to be_empty }
-      end
+        context 'agent configuration project shares a root namespace, but does not belong to an ancestor of the given project' do
+          let!(:project_authorization) { create(:agent_ci_access_project_authorization, agent: non_ancestor_agent, project: requesting_project) }
 
-      context 'agent configuration project shares a root namespace, but does not belong to an ancestor of the given project' do
-        let!(:project_authorization) { create(:agent_ci_access_project_authorization, agent: non_ancestor_agent, project: requesting_project) }
+          it { is_expected.to match_array([project_authorization]) }
+        end
 
-        it { is_expected.to match_array([project_authorization]) }
-      end
+        context 'with project authorizations present' do
+          let!(:authorization) { create(:agent_ci_access_project_authorization, agent: production_agent, project: requesting_project) }
 
-      context 'with project authorizations present' do
-        let!(:authorization) { create(:agent_ci_access_project_authorization, agent: production_agent, project: requesting_project) }
+          it { is_expected.to match_array [authorization] }
+        end
 
-        it { is_expected.to match_array [authorization] }
+        context 'with overlapping authorizations' do
+          let!(:agent) { create(:cluster_agent, project: requesting_project) }
+          let!(:project_authorization) { create(:agent_ci_access_project_authorization, agent: agent, project: requesting_project) }
+          let!(:group_authorization) { create(:agent_ci_access_group_authorization, agent: agent, group: bottom_level_group) }
+
+          it { is_expected.to match_array [project_authorization] }
+        end
+
+        context 'with multiple authorizations' do
+          let!(:authorization1) { create(:agent_ci_access_project_authorization, agent: production_agent, project: requesting_project) }
+          let!(:authorization2) { create(:agent_ci_access_project_authorization, agent: staging_agent, project: requesting_project) }
+
+          it { is_expected.to contain_exactly(authorization1, authorization2) }
+        end
       end
 
-      context 'with overlapping authorizations' do
-        let!(:agent) { create(:cluster_agent, project: requesting_project) }
-        let!(:project_authorization) { create(:agent_ci_access_project_authorization, agent: agent, project: requesting_project) }
-        let!(:group_authorization) { create(:agent_ci_access_group_authorization, agent: agent, group: bottom_level_group) }
+      context 'when initialized with an agent' do
+        let!(:authorization1) { create(:agent_ci_access_project_authorization, agent: production_agent, project: requesting_project) }
+        let!(:authorization2) { create(:agent_ci_access_project_authorization, agent: staging_agent, project: requesting_project) }
+
+        let!(:finder) { described_class.new(requesting_project, agent: production_agent) }
 
-        it { is_expected.to match_array [project_authorization] }
+        it 'returns authorizations for the given agent' do
+          expect(subject).to contain_exactly(authorization1)
+        end
       end
 
       it_behaves_like 'access_as' do
@@ -86,54 +108,82 @@
     end
 
     describe 'implicit authorizations' do
-      let!(:associated_agent) { create(:cluster_agent, project: requesting_project) }
+      let!(:associated_agent_1) { create(:cluster_agent, project: requesting_project) }
+      let!(:associated_agent_2) { create(:cluster_agent, project: requesting_project) }
+
+      context 'when initialized without an agent' do
+        it 'returns all authorizations for agents directly associated with the project' do
+          expect(subject.count).to eq(2)
+          expect(subject).to all(be_a(Clusters::Agents::Authorizations::CiAccess::ImplicitAuthorization))
+          expect(subject.map(&:agent)).to contain_exactly(associated_agent_1, associated_agent_2)
+        end
+      end
 
-      it 'returns authorizations for agents directly associated with the project' do
-        expect(subject.count).to eq(1)
+      context 'when initialized with an agent' do
+        let!(:finder) { described_class.new(requesting_project, agent: associated_agent_1) }
 
-        authorization = subject.first
-        expect(authorization).to be_a(Clusters::Agents::Authorizations::CiAccess::ImplicitAuthorization)
-        expect(authorization.agent).to eq(associated_agent)
+        it 'returns authorizations for the given agent' do
+          expect(subject.count).to eq(1)
+
+          authorization = subject.first
+          expect(authorization).to be_a(Clusters::Agents::Authorizations::CiAccess::ImplicitAuthorization)
+          expect(authorization.agent).to eq(associated_agent_1)
+        end
       end
     end
 
     describe 'authorized groups' do
-      context 'agent configuration project is outside the requesting project hierarchy' do
-        let(:unrelated_agent) { create(:cluster_agent) }
+      context 'when initialized without an agent' do
+        context 'agent configuration project is outside the requesting project hierarchy' do
+          let(:unrelated_agent) { create(:cluster_agent) }
 
-        before do
-          create(:agent_ci_access_group_authorization, agent: unrelated_agent, group: top_level_group)
+          before do
+            create(:agent_ci_access_group_authorization, agent: unrelated_agent, group: top_level_group)
+          end
+
+          it { is_expected.to be_empty }
         end
 
-        it { is_expected.to be_empty }
-      end
+        context 'multiple agents are authorized for the same group' do
+          let!(:staging_auth) { create(:agent_ci_access_group_authorization, agent: staging_agent, group: bottom_level_group) }
+          let!(:production_auth) { create(:agent_ci_access_group_authorization, agent: production_agent, group: bottom_level_group) }
 
-      context 'multiple agents are authorized for the same group' do
-        let!(:staging_auth) { create(:agent_ci_access_group_authorization, agent: staging_agent, group: bottom_level_group) }
-        let!(:production_auth) { create(:agent_ci_access_group_authorization, agent: production_agent, group: bottom_level_group) }
+          it 'returns authorizations for all agents' do
+            expect(subject).to contain_exactly(staging_auth, production_auth)
+          end
+        end
 
-        it 'returns authorizations for all agents' do
-          expect(subject).to contain_exactly(staging_auth, production_auth)
+        context 'a single agent is authorized to more than one matching group' do
+          let!(:bottom_level_auth) { create(:agent_ci_access_group_authorization, agent: production_agent, group: bottom_level_group) }
+          let!(:top_level_auth) { create(:agent_ci_access_group_authorization, agent: production_agent, group: top_level_group) }
+
+          it 'picks the authorization for the closest group to the requesting project' do
+            expect(subject).to contain_exactly(bottom_level_auth)
+          end
         end
-      end
 
-      context 'a single agent is authorized to more than one matching group' do
-        let!(:bottom_level_auth) { create(:agent_ci_access_group_authorization, agent: production_agent, group: bottom_level_group) }
-        let!(:top_level_auth) { create(:agent_ci_access_group_authorization, agent: production_agent, group: top_level_group) }
+        context 'agent configuration project does not belong to an ancestor of the authorized group' do
+          let!(:group_authorization) { create(:agent_ci_access_group_authorization, agent: non_ancestor_agent, group: bottom_level_group) }
+
+          it { is_expected.to match_array([group_authorization]) }
+        end
 
-        it 'picks the authorization for the closest group to the requesting project' do
-          expect(subject).to contain_exactly(bottom_level_auth)
+        it_behaves_like 'access_as' do
+          let!(:authorization) { create(:agent_ci_access_group_authorization, agent: production_agent, group: top_level_group, config: config) }
         end
       end
 
-      context 'agent configuration project does not belong to an ancestor of the authorized group' do
-        let!(:group_authorization) { create(:agent_ci_access_group_authorization, agent: non_ancestor_agent, group: bottom_level_group) }
+      context 'when initialized with an agent' do
+        let(:finder) { described_class.new(requesting_project, agent: production_agent) }
 
-        it { is_expected.to match_array([group_authorization]) }
-      end
+        context 'multiple agents are authorized for the same group' do
+          let!(:staging_auth) { create(:agent_ci_access_group_authorization, agent: staging_agent, group: bottom_level_group) }
+          let!(:production_auth) { create(:agent_ci_access_group_authorization, agent: production_agent, group: bottom_level_group) }
 
-      it_behaves_like 'access_as' do
-        let!(:authorization) { create(:agent_ci_access_group_authorization, agent: production_agent, group: top_level_group, config: config) }
+          it 'returns authorizations for the given agent' do
+            expect(subject).to contain_exactly(production_auth)
+          end
+        end
       end
     end
   end
diff --git a/spec/lib/gitlab/ci/build/prerequisite/managed_resource_spec.rb b/spec/lib/gitlab/ci/build/prerequisite/managed_resource_spec.rb
index 6e7a2c4426d0445a08c838b5444e5b0b55af1dff..30bd99b1bb221ae912d9977946c90b6c0dcefcac 100644
--- a/spec/lib/gitlab/ci/build/prerequisite/managed_resource_spec.rb
+++ b/spec/lib/gitlab/ci/build/prerequisite/managed_resource_spec.rb
@@ -4,19 +4,23 @@
 
 RSpec.describe Gitlab::Ci::Build::Prerequisite::ManagedResource, feature_category: :continuous_integration do
   describe '#unmet?' do
-    let_it_be(:agent_management_project) { create(:project, :private, :repository) }
+    let_it_be(:organization) { create(:group) }
+    let_it_be(:agent_management_project) { create(:project, :private, :repository, group: organization) }
     let_it_be(:cluster_agent) { create(:cluster_agent, project: agent_management_project) }
 
-    let_it_be(:deployment_project) { create(:project, :private, :repository) }
-    let_it_be_with_reload(:environment) do
+    let_it_be(:deployment_project) { create(:project, :private, :repository, group: organization) }
+    let_it_be(:environment) do
       create(:environment, project: deployment_project, cluster_agent: cluster_agent)
     end
 
-    let_it_be(:user) { create(:user) }
+    let_it_be(:user) { create(:user, developer_of: deployment_project) }
     let_it_be(:deployment) { create(:deployment, environment: environment, user: user) }
-    let_it_be_with_reload(:build) { create(:ci_build, environment: environment, user: user, deployment: deployment) }
+    let_it_be(:build) do
+      create(:ci_build, project: deployment_project, environment: environment, user: user, deployment: deployment)
+    end
+
     let(:status) { :processing }
-    let!(:managed_resource) do
+    let(:managed_resource) do
       create(:managed_resource,
         build: build,
         project: deployment_project,
@@ -29,42 +33,7 @@
 
     subject(:execute_unmet) { instance.unmet? }
 
-    context 'when resource_management is not enabled' do
-      it 'returns false' do
-        expect(execute_unmet).to be_falsey
-      end
-    end
-
-    context 'when resource_management is enabled' do
-      before do
-        allow(instance).to receive(:resource_management_enabled?).and_return(true)
-      end
-
-      context 'when the build is valid for managed resources' do
-        context 'when the managed resource record does not exist' do
-          let!(:managed_resource) { nil }
-
-          it { is_expected.to be_truthy }
-        end
-
-        context 'when managed resources completed successfully' do
-          let!(:status) { :completed }
-
-          it 'returns false`' do
-            expect(execute_unmet).to be_falsey
-          end
-        end
-
-        context 'when managed resources failed' do
-          let!(:status) { :failed }
-
-          it 'returns true' do
-            managed_resource.reload
-            expect(execute_unmet).to be_truthy
-          end
-        end
-      end
-
+    context 'when not valid for managed resources' do
       context 'when the build does not have a deployment' do
         let_it_be(:build) { create(:ci_build, deployment: nil) }
 
@@ -85,6 +54,73 @@
         it { is_expected.to be_falsey }
       end
     end
+
+    context 'when valid for managed resources' do
+      context 'when agent\'s resource management is disabled' do
+        before do
+          allow_next_instance_of(Clusters::Agent) do |instance|
+            allow(instance).to receive(:resource_management_enabled?).and_return(false)
+          end
+        end
+
+        it 'returns false' do
+          expect(execute_unmet).to be_falsey
+        end
+      end
+
+      context 'when agent\'s resource management is enabled' do
+        before do
+          environment.reload
+          allow(environment.cluster_agent).to receive(:resource_management_enabled?).and_return(true)
+        end
+
+        context 'when authorization exists' do
+          context 'when the resource_management is not enabled' do
+            let_it_be(:agent_ci_access_group_authorization) do
+              create(:agent_ci_access_group_authorization, agent: cluster_agent, group: organization)
+            end
+
+            context 'when the managed resource record has failed status' do
+              let!(:status) { :failed }
+
+              it 'returns false' do
+                expect(execute_unmet).to be_falsey
+              end
+            end
+          end
+
+          context 'when authorization exists with resource_management enabled' do
+            let_it_be(:agent_ci_access_group_authorization) do
+              create(:agent_ci_access_group_authorization, agent: cluster_agent, group: organization,
+                config: { resource_management: { enabled: true } })
+            end
+
+            context 'when the managed resource record does not exist' do
+              let(:managed_resource) { nil }
+
+              it { is_expected.to be_truthy }
+            end
+
+            context 'when the managed resource record has completed status' do
+              let(:status) { :completed }
+
+              it 'returns false`' do
+                managed_resource.reload
+                expect(execute_unmet).to be_falsey
+              end
+            end
+
+            context 'when the managed resource record has failed status' do
+              let(:status) { :failed }
+
+              it 'returns true' do
+                expect(execute_unmet).to be_truthy
+              end
+            end
+          end
+        end
+      end
+    end
   end
 
   describe '#complete!' do
diff --git a/spec/models/clusters/agent_spec.rb b/spec/models/clusters/agent_spec.rb
index 062d5062658a28cae1c153a712b4fc1cadd4c7e7..9f9d6b9544096b1d43988509120266c08d8a9ab5 100644
--- a/spec/models/clusters/agent_spec.rb
+++ b/spec/models/clusters/agent_spec.rb
@@ -317,4 +317,8 @@
       end
     end
   end
+
+  describe '#resource_management_enabled?' do
+    it { expect(subject.resource_management_enabled?).to be_falsey }
+  end
 end
diff --git a/spec/support/shared_examples/environments/create_for_job_shared_examples.rb b/spec/support/shared_examples/environments/create_for_job_shared_examples.rb
index 3acdc8c142fdc8c7cff8d55d3ebcef4775dfe70f..aec4bde2673f9da90898777685f6ec33828d2829 100644
--- a/spec/support/shared_examples/environments/create_for_job_shared_examples.rb
+++ b/spec/support/shared_examples/environments/create_for_job_shared_examples.rb
@@ -149,6 +149,115 @@
       end
     end
 
+    context 'when job has a cluster agent attribute' do
+      let_it_be(:agent) { create(:cluster_agent, project: project) }
+
+      let(:environment_name) { 'production' }
+      let(:agent_path) { "#{project.full_path}:#{agent.name}" }
+      let(:attributes) do
+        {
+          environment: environment_name,
+          options: {
+            environment: {
+              name: environment_name,
+              kubernetes: {
+                agent: agent_path
+              }
+            }
+          }
+        }
+      end
+
+      let(:authorizations) { [ci_access_authorization] }
+      let(:ci_access_authorization) do
+        create(:agent_ci_access_project_authorization,
+          resource_management_enabled: true,
+          project: project,
+          agent: agent
+        )
+      end
+
+      before do
+        allow_next_instance_of(Clusters::Agents::Authorizations::CiAccess::Finder, project) do |finder|
+          allow(finder).to receive(:execute).and_return(authorizations)
+        end
+      end
+
+      context 'when the agent has resource management enabled' do
+        before do
+          allow(agent).to receive(:resource_management_enabled?).and_return(true)
+        end
+
+        it 'creates an environment with the specified cluster agent' do
+          expect { subject }.to change { Environment.count }.by(1)
+
+          expect(subject).to be_a(Environment)
+          expect(subject).to be_persisted
+          expect(subject.cluster_agent).to eq(agent)
+        end
+
+        context 'when the gitlab_managed_cluster_resources feature flag is disabled' do
+          before do
+            stub_feature_flags(gitlab_managed_cluster_resources: false)
+          end
+
+          it 'creates an environment without the specified cluster agent' do
+            expect(Clusters::Agents::Authorizations::CiAccess::Finder).not_to receive(:new)
+
+            expect { subject }.to change { Environment.count }.by(1)
+
+            expect(subject).to be_a(Environment)
+            expect(subject).to be_persisted
+            expect(subject.cluster_agent).to be_nil
+          end
+        end
+
+        context 'when the agent is not configured for resource_management' do
+          let(:ci_access_authorization) do
+            create(:agent_ci_access_project_authorization,
+              project: project,
+              agent: agent,
+              config: {} # resource_management is not enabled
+            )
+          end
+
+          it 'creates an environment without the specified cluster agent' do
+            expect { subject }.to change { Environment.count }.by(1)
+
+            expect(subject).to be_a(Environment)
+            expect(subject).to be_persisted
+            expect(subject.cluster_agent).to be_nil
+          end
+        end
+
+        context 'when the agent is not authorized for this project' do
+          let(:authorizations) { [] }
+
+          it 'creates an environment without the specified cluster agent' do
+            expect { subject }.to change { Environment.count }.by(1)
+
+            expect(subject).to be_a(Environment)
+            expect(subject).to be_persisted
+            expect(subject.cluster_agent).to be_nil
+          end
+        end
+      end
+
+      context 'when the agent does not have resource management enabled' do
+        before do
+          allow(agent).to receive(:resource_management_enabled).and_return(false)
+        end
+
+        it 'creates an environment without the specified cluster agent' do
+          expect { subject }.to change { Environment.count }.by(1)
+
+          expect(subject).to be_a(Environment)
+          expect(subject).to be_persisted
+          expect(subject.cluster_agent).to be_nil
+        end
+      end
+    end
+
     context 'when job starts a review app' do
       let(:environment_name) { 'review/$CI_COMMIT_REF_NAME' }
       let(:expected_environment_name) { "review/#{job.ref}" }