diff --git a/app/services/ci/workloads/workload_branch_service.rb b/app/services/ci/workloads/workload_branch_service.rb
index b070b45ebe3aace02f7bcd7d04d62c01866b8b78..dda96790b421d9bf29b8b12702200d03255fa208 100644
--- a/app/services/ci/workloads/workload_branch_service.rb
+++ b/app/services/ci/workloads/workload_branch_service.rb
@@ -10,28 +10,51 @@ def initialize(project:, source_branch:, current_user:)
       end
 
       def execute
+        workload_ref = "workloads/#{SecureRandom.hex[0..10]}"
+        source_ref = @project.repository.branch_exists?(@source_branch) ? @source_branch : default_branch
+
+        if Feature.enabled?(:use_internal_refs_for_workload_pipelines, @project)
+          create_internal_refs(source_ref, workload_ref)
+        else
+          create_workload_branch(source_ref, workload_ref)
+        end
+      rescue Gitlab::Git::CommandError => e
+        Gitlab::ErrorTracking.track_exception(e)
+        ServiceResponse.error(message: 'Failed to create workload ref')
+      end
+
+      private
+
+      def create_internal_refs(source_ref, workload_ref)
+        source_sha = @project.repository.commit(source_ref)&.sha
+        return ServiceResponse.error(message: 'Source ref not found') unless source_sha
+
+        workload_ref_path = "refs/#{workload_ref}"
+        if @project.repository.ref_exists?(workload_ref_path)
+          return ServiceResponse.error(message: 'Ref already exists')
+        end
+
+        result = @project.repository.create_ref(source_sha, workload_ref_path)
+        return ServiceResponse.error(message: 'Error in git ref creation') unless result
+
+        ServiceResponse.success(payload: { ref: workload_ref_path })
+      end
+
+      def create_workload_branch(source_ref, workload_ref)
         unless @current_user.can?(:push_code, @project)
           return ServiceResponse.error(message: 'You are not allowed to create branches in this project')
         end
 
-        branch_name = "workloads/#{SecureRandom.hex[0..10]}"
-        if @project.repository.branch_exists?(branch_name)
+        if @project.repository.branch_exists?(workload_ref)
           return ServiceResponse.error(message: 'Branch already exists')
         end
 
-        ref = @project.repository.branch_exists?(@source_branch) ? @source_branch : default_branch
-
-        repo_branch = @project.repository.add_branch(@current_user, branch_name, ref, skip_ci: true)
-        return ServiceResponse.error(message: 'Error in git branch creation') unless repo_branch
+        result = @project.repository.add_branch(@current_user, workload_ref, source_ref, skip_ci: true)
+        return ServiceResponse.error(message: 'Failed to create branch') unless result
 
-        ServiceResponse.success(payload: { branch_name: branch_name })
-      rescue Gitlab::Git::CommandError => e
-        Gitlab::ErrorTracking.track_exception(e)
-        ServiceResponse.error(message: 'Failed to create branch')
+        ServiceResponse.success(payload: { ref: workload_ref })
       end
 
-      private
-
       def default_branch
         @project.default_branch_or_main
       end
diff --git a/config/feature_flags/wip/use_internal_refs_for_workload_pipelines.yml b/config/feature_flags/wip/use_internal_refs_for_workload_pipelines.yml
new file mode 100644
index 0000000000000000000000000000000000000000..d6f884824a988a723d7b8c964f5f98fb287b4841
--- /dev/null
+++ b/config/feature_flags/wip/use_internal_refs_for_workload_pipelines.yml
@@ -0,0 +1,10 @@
+---
+name: use_internal_refs_for_workload_pipelines
+description: Feature flag to use internal git refs instead of branches for ci workloads
+feature_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/577239
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/215525
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/583179
+milestone: '18.7'
+group: group::agent foundations
+type: wip
+default_enabled: false
diff --git a/ee/app/services/ai/duo_workflows/start_workflow_service.rb b/ee/app/services/ai/duo_workflows/start_workflow_service.rb
index 9a6ec69b81f100351a540e5336d94be3d9c3cdd8..bd4d6127aa35d1e27bacfc9a0385b70d420f74cb 100644
--- a/ee/app/services/ai/duo_workflows/start_workflow_service.rb
+++ b/ee/app/services/ai/duo_workflows/start_workflow_service.rb
@@ -46,7 +46,7 @@ def execute
         branch_response = create_workload_branch
         return branch_response unless branch_response.success?
 
-        @ref = branch_response.payload[:branch_name]
+        @ref = branch_response.payload[:ref]
         service = ::Ci::Workloads::RunWorkloadService.new(
           project: project,
           current_user: @workload_user,
@@ -174,12 +174,17 @@ def main_workflow_commands
           %(echo $DUO_WORKFLOW_DEFINITION),
           %(echo $DUO_WORKFLOW_GOAL),
           %(echo $DUO_WORKFLOW_SOURCE_BRANCH),
-          %(git checkout $CI_WORKLOAD_REF),
           %(echo $DUO_WORKFLOW_FLOW_CONFIG),
           %(echo $DUO_WORKFLOW_FLOW_CONFIG_SCHEMA_VERSION),
           %(echo $DUO_WORKFLOW_ADDITIONAL_CONTEXT_CONTENT),
           %(echo Starting Workflow #{@workflow.id})
-        ] + set_up_executor_commands
+        ] + branch_checkout_commands + set_up_executor_commands
+      end
+
+      def branch_checkout_commands
+        return [] if Feature.enabled?(:use_internal_refs_for_workload_pipelines, project)
+
+        [%(git checkout $CI_WORKLOAD_REF)]
       end
 
       def set_up_executor_commands
diff --git a/ee/app/services/ai/flow_triggers/run_service.rb b/ee/app/services/ai/flow_triggers/run_service.rb
index f88b7def76277a0d1b716b1a78e8bd781ba914c4..733dff8b19e5f1b3e6790441e50479dad91d4fbe 100644
--- a/ee/app/services/ai/flow_triggers/run_service.rb
+++ b/ee/app/services/ai/flow_triggers/run_service.rb
@@ -182,7 +182,7 @@ def branch_args
         branch_response = workload_branch_service.execute
         return branch_response unless branch_response.success?
 
-        ServiceResponse.success(payload: { ref: branch_response.payload[:branch_name] })
+        ServiceResponse.success(payload: { ref: branch_response.payload[:ref] })
       end
 
       def composite_identity_token
diff --git a/ee/spec/services/ai/duo_workflows/start_workflow_service_spec.rb b/ee/spec/services/ai/duo_workflows/start_workflow_service_spec.rb
index 9e8e9d58cb2394254dcab086c0ba90abb5760f43..36b1cf27f5833a6404d7c615f56247aee59e9c4e 100644
--- a/ee/spec/services/ai/duo_workflows/start_workflow_service_spec.rb
+++ b/ee/spec/services/ai/duo_workflows/start_workflow_service_spec.rb
@@ -51,8 +51,7 @@
         workload_id: workload_id)
 
       workload = Ci::Workloads::Workload.find_by_id([workload_id])
-      expect(workload.branch_name).to start_with('workloads/')
-      expect(workload.branch_name).to start_with('workloads/')
+      expect(workload.branch_name).to start_with('refs/workloads/')
     end
   end
 
@@ -860,7 +859,6 @@ def standard_context_content(parsed_context)
         %(echo $DUO_WORKFLOW_DEFINITION),
         %(echo $DUO_WORKFLOW_GOAL),
         %(echo $DUO_WORKFLOW_SOURCE_BRANCH),
-        %(git checkout $CI_WORKLOAD_REF),
         %(echo $DUO_WORKFLOW_FLOW_CONFIG),
         %(echo $DUO_WORKFLOW_FLOW_CONFIG_SCHEMA_VERSION),
         %(echo $DUO_WORKFLOW_ADDITIONAL_CONTEXT_CONTENT),
@@ -1065,9 +1063,9 @@ def standard_context_content(parsed_context)
           expect(Ci::Workloads::RunWorkloadService).to have_received(:new) do |workload_definition:, **_kwargs|
             commands = workload_definition.commands
             expect(commands[0]).to eq('npm install')
-            expect(commands[1..8]).to eq(shared_main_commands)
-            expect(commands[9..12]).to eq(cli_install_commands)
-            expect(commands[13..]).to eq(wrapped_commands)
+            expect(commands[1..7]).to eq(shared_main_commands)
+            expect(commands[8..11]).to eq(cli_install_commands)
+            expect(commands[12..]).to eq(wrapped_commands)
           end
         end
       end
@@ -1125,6 +1123,23 @@ def standard_context_content(parsed_context)
         it_behaves_like 'sandbox is disabled'
       end
     end
+
+    context 'when feature flag use_internal_refs_for_workload_pipelines is disabled' do
+      before do
+        stub_feature_flags(use_internal_refs_for_workload_pipelines: false)
+        allow(duo_config).to receive(:setup_script).and_return([])
+      end
+
+      it 'prepends the checkout command' do
+        expect(Ci::Workloads::RunWorkloadService).to receive(:new) do |workload_definition:, **_kwargs|
+          commands = workload_definition.commands
+          expect(commands).to include('git checkout $CI_WORKLOAD_REF')
+          expect(commands.size).to eq(8)
+        end.and_call_original
+
+        expect(execute).to be_success
+      end
+    end
   end
 
   context 'with cache configuration' do
diff --git a/ee/spec/services/ai/flow_triggers/run_service_spec.rb b/ee/spec/services/ai/flow_triggers/run_service_spec.rb
index bfea52d8dff78825bb7332be994729f622ffe22b..77dfe4bcebe2de8536cdf2ae83803e2918d4df4a 100644
--- a/ee/spec/services/ai/flow_triggers/run_service_spec.rb
+++ b/ee/spec/services/ai/flow_triggers/run_service_spec.rb
@@ -630,52 +630,72 @@ def expected_gitlab_hostname
 
       let_it_be(:resource) { merge_request }
 
-      it 'creates workload branch from merge request source branch and passes ref to RunWorkloadService' do
-        workload_branch_service = instance_double(Ci::Workloads::WorkloadBranchService)
-        expect(Ci::Workloads::WorkloadBranchService).to receive(:new).with(
-          current_user: service_account,
-          project: project,
-          source_branch: 'feature-branch'
-        ).and_return(workload_branch_service)
-        expect(workload_branch_service).to receive(:execute).and_return(
-          ServiceResponse.success(payload: { branch_name: 'workloads/abc123' })
-        )
+      shared_examples 'passes workload ref to RunWorkloadService' do |ref_value|
+        it "creates workload ref from merge request source branch and passes ref #{ref_value} to RunWorkloadService" do
+          workload_branch_service = instance_double(Ci::Workloads::WorkloadBranchService)
+          expect(Ci::Workloads::WorkloadBranchService).to receive(:new).with(
+            current_user: service_account,
+            project: project,
+            source_branch: 'feature-branch'
+          ).and_return(workload_branch_service)
+          expect(workload_branch_service).to receive(:execute).and_return(
+            ServiceResponse.success(payload: { ref: ref_value })
+          )
 
-        expect(Ci::Workloads::RunWorkloadService).to receive(:new).with(
-          project: project,
-          current_user: service_account,
-          source: :duo_workflow,
-          workload_definition: an_instance_of(Ci::Workloads::WorkloadDefinition),
-          ci_variables_included: %w[API_KEY DATABASE_URL],
-          ref: 'workloads/abc123'
-        ).and_call_original
+          expect(Ci::Workloads::RunWorkloadService).to receive(:new).with(
+            project: project,
+            current_user: service_account,
+            source: :duo_workflow,
+            workload_definition: an_instance_of(Ci::Workloads::WorkloadDefinition),
+            ci_variables_included: %w[API_KEY DATABASE_URL],
+            ref: ref_value
+          ).and_call_original
 
-        service.execute(params)
+          service.execute(params)
+        end
+      end
+
+      context 'with workload branch' do
+        it_behaves_like 'passes workload ref to RunWorkloadService', 'workloads/xyz789'
+      end
+
+      context 'with workload internal_refs' do
+        it_behaves_like 'passes workload ref to RunWorkloadService', 'refs/workloads/123'
       end
     end
 
     context 'when resource is not a MergeRequest' do
-      it 'creates workload branch without source branch and passes ref to RunWorkloadService' do
-        workload_branch_service = instance_double(Ci::Workloads::WorkloadBranchService)
-        expect(Ci::Workloads::WorkloadBranchService).to receive(:new).with(
-          current_user: service_account,
-          project: project,
-          source_branch: nil
-        ).and_return(workload_branch_service)
-        expect(workload_branch_service).to receive(:execute).and_return(
-          ServiceResponse.success(payload: { branch_name: 'workloads/xyz789' })
-        )
+      shared_examples 'creates workload ref and passes ref to RunWorkloadService' do |ref_value|
+        it "creates workload ref without source branch and passes ref #{ref_value} to RunWorkloadService" do
+          workload_branch_service = instance_double(Ci::Workloads::WorkloadBranchService)
+          expect(Ci::Workloads::WorkloadBranchService).to receive(:new).with(
+            current_user: service_account,
+            project: project,
+            source_branch: nil
+          ).and_return(workload_branch_service)
+          expect(workload_branch_service).to receive(:execute).and_return(
+            ServiceResponse.success(payload: { ref: ref_value })
+          )
 
-        expect(Ci::Workloads::RunWorkloadService).to receive(:new).with(
-          project: project,
-          current_user: service_account,
-          source: :duo_workflow,
-          workload_definition: an_instance_of(Ci::Workloads::WorkloadDefinition),
-          ci_variables_included: %w[API_KEY DATABASE_URL],
-          ref: 'workloads/xyz789'
-        ).and_call_original
+          expect(Ci::Workloads::RunWorkloadService).to receive(:new).with(
+            project: project,
+            current_user: service_account,
+            source: :duo_workflow,
+            workload_definition: an_instance_of(Ci::Workloads::WorkloadDefinition),
+            ci_variables_included: %w[API_KEY DATABASE_URL],
+            ref: ref_value
+          ).and_call_original
 
-        service.execute(params)
+          service.execute(params)
+        end
+      end
+
+      context 'with workload branch' do
+        it_behaves_like 'creates workload ref and passes ref to RunWorkloadService', 'workloads/xyz789'
+      end
+
+      context 'with workload internal_refs' do
+        it_behaves_like 'creates workload ref and passes ref to RunWorkloadService', 'refs/workloads/123'
       end
     end
 
diff --git a/spec/services/ci/workloads/run_workload_service_spec.rb b/spec/services/ci/workloads/run_workload_service_spec.rb
index 9507224fc016d524686187a51018c0b9f5ff0bdc..0490395a04675f91aa33f9aad00aaeadb3114149 100644
--- a/spec/services/ci/workloads/run_workload_service_spec.rb
+++ b/spec/services/ci/workloads/run_workload_service_spec.rb
@@ -103,6 +103,30 @@
       end
     end
 
+    context 'with internal refs for workloads' do
+      let(:ref) { 'refs/workloads/123' }
+
+      before do
+        source_ref = project.default_branch_or_main
+        source_sha = project.repository.commit(source_ref)&.sha
+        project.repository.create_ref(source_sha, ref)
+      end
+
+      it 'starts a pipeline to execute workload' do
+        expect_next_instance_of(Ci::CreatePipelineService, project, user,
+          hash_including(ref: ref)) do |pipeline_service|
+          expect(pipeline_service).to receive(:execute)
+                                        .and_call_original
+        end
+        result = execute
+        expect(result).to be_success
+
+        workload = result.payload
+        expect(workload).to be_a(Ci::Workloads::Workload)
+        expect(workload.pipeline).to be_present
+      end
+    end
+
     context 'with unsupported source' do
       let(:source) { :foo }
 
diff --git a/spec/services/ci/workloads/workload_branch_service_spec.rb b/spec/services/ci/workloads/workload_branch_service_spec.rb
index 4f0235b06003c0f7e6d7744596394de59ddd53f8..b58d1a5a21019884501f220e04c5788be3207531 100644
--- a/spec/services/ci/workloads/workload_branch_service_spec.rb
+++ b/spec/services/ci/workloads/workload_branch_service_spec.rb
@@ -16,111 +16,179 @@
       allow(SecureRandom).to receive(:hex).and_return('abcdef12345')
     end
 
-    context 'when user does not have access to push code to the project' do
-      let(:user) { create(:user) }
-
-      it 'returns an error response' do
-        result = execute
-        expect(result).to be_error
-        expect(result.message).to eq('You are not allowed to create branches in this project')
-      end
-    end
-
     context 'when source branch exists' do
       before do
         project.repository.create_branch(source_branch, project.default_branch)
       end
 
-      it 'creates a new workload branch from the source branch' do
-        expected_branch_name = 'workloads/abcdef12345'
+      it 'creates a new workload ref from the source branch' do
+        workload_ref = 'refs/workloads/abcdef12345'
+        source_sha = project.repository.commit(source_branch).sha
 
-        expect(project.repository).to receive(:add_branch)
-                                        .with(user, expected_branch_name, source_branch, skip_ci: true)
-                                        .and_return(double)
+        expect(project.repository).to receive(:create_ref)
+                                        .with(source_sha, workload_ref)
+                                        .and_return(true)
 
         result = execute
 
         expect(result).to be_success
-        expect(result.payload[:branch_name]).to eq(expected_branch_name)
+        expect(result.payload[:ref]).to eq(workload_ref)
       end
     end
 
     context 'when source branch does not exist' do
       let(:source_branch) { 'non-existent-branch' }
 
-      it 'creates a new workload branch from the default branch' do
-        expected_branch_name = 'workloads/abcdef12345'
+      it 'creates a new workload ref from the default branch' do
+        workload_ref = 'refs/workloads/abcdef12345'
+        default_sha = project.repository.commit(project.default_branch_or_main).sha
 
-        expect(project.repository).to receive(:add_branch)
-                                        .with(user, expected_branch_name, project.default_branch_or_main, skip_ci: true)
-                                        .and_return(double)
+        expect(project.repository).to receive(:create_ref)
+                                        .with(default_sha, workload_ref)
+                                        .and_return(true)
 
         result = execute
 
         expect(result).to be_success
-        expect(result.payload[:branch_name]).to eq(expected_branch_name)
+        expect(result.payload[:ref]).to eq(workload_ref)
       end
     end
 
     context 'when source branch is nil' do
       let(:source_branch) { nil }
 
-      it 'creates a new workload branch from the default branch' do
-        expected_branch_name = 'workloads/abcdef12345'
+      it 'creates a new workload ref from the default branch' do
+        workload_ref = 'refs/workloads/abcdef12345'
+        default_sha = project.repository.commit(project.default_branch_or_main).sha
 
-        expect(project.repository).to receive(:add_branch)
-                                        .with(user, expected_branch_name, project.default_branch_or_main, skip_ci: true)
-                                        .and_return(double)
+        expect(project.repository).to receive(:create_ref)
+                                        .with(default_sha, workload_ref)
+                                        .and_return(true)
 
         result = execute
 
         expect(result).to be_success
-        expect(result.payload[:branch_name]).to eq(expected_branch_name)
+        expect(result.payload[:ref]).to eq(workload_ref)
       end
     end
 
-    context 'when git branch creation fails' do
+    context 'when git ref creation fails' do
       before do
-        allow(project.repository).to receive(:add_branch).and_return(nil)
+        allow(project.repository).to receive(:create_ref).and_return(nil)
       end
 
       it 'returns an error response' do
         result = execute
 
         expect(result).to be_error
-        expect(result.message).to eq('Error in git branch creation')
+        expect(result.message).to eq('Error in git ref creation')
       end
     end
 
-    context 'when branch name already exists' do
+    context 'when ref already exists' do
       before do
-        allow(project.repository).to receive(:branch_exists?)
-                                       .with(match(%r{^workloads/\w+}))
-                                       .and_return(true)
+        allow(project.repository).to receive(:ref_exists?)
+                                      .with(match(%r{^refs/workloads/\w+}))
+                                      .and_return(true)
       end
 
       it 'returns an error response' do
         result = execute
 
         expect(result).to be_error
-        expect(result.message).to eq('Branch already exists')
+        expect(result.message).to eq('Ref already exists')
+      end
+    end
+
+    context 'when source ref is not found' do
+      before do
+        allow(project.repository).to receive(:commit).and_return(nil)
+      end
+
+      it 'returns an error response' do
+        result = execute
+
+        expect(result).to be_error
+        expect(result.message).to eq('Source ref not found')
       end
     end
 
     context 'when git command raises an error' do
       before do
-        allow(project.repository).to receive(:add_branch)
-                                       .and_raise(Gitlab::Git::CommandError.new('git error'))
+        allow(project.repository).to receive(:create_ref)
+                                      .and_raise(Gitlab::Git::CommandError.new('git error'))
       end
 
       it 'tracks the exception and returns an error response' do
         expect(Gitlab::ErrorTracking).to receive(:track_exception)
-                                           .with(instance_of(Gitlab::Git::CommandError))
+                                          .with(instance_of(Gitlab::Git::CommandError))
 
         result = execute
 
         expect(result).to be_error
-        expect(result.message).to eq('Failed to create branch')
+        expect(result.message).to eq('Failed to create workload ref')
+      end
+    end
+
+    context 'when feature flag is disabled' do
+      before do
+        stub_feature_flags(use_internal_refs_for_workload_pipelines: false)
+      end
+
+      context 'when user does not have access to push code to the project' do
+        let(:user) { create(:user) }
+
+        it 'returns an error response' do
+          result = execute
+          expect(result).to be_error
+          expect(result.message).to eq('You are not allowed to create branches in this project')
+        end
+      end
+
+      context 'when git branch creation fails' do
+        before do
+          allow(project.repository).to receive(:add_branch).and_return(nil)
+        end
+
+        it 'returns an error response' do
+          result = execute
+
+          expect(result).to be_error
+          expect(result.message).to eq('Failed to create branch')
+        end
+      end
+
+      context 'when branch name already exists' do
+        before do
+          allow(project.repository).to receive(:branch_exists?).with("feature-branch").and_return(true)
+          allow(project.repository).to receive(:branch_exists?)
+                                        .with(match(%r{^workloads/\w+}))
+                                        .and_return(true)
+        end
+
+        it 'returns an error response' do
+          result = execute
+
+          expect(result).to be_error
+          expect(result.message).to eq('Branch already exists')
+        end
+      end
+
+      context 'when git command raises an error' do
+        before do
+          allow(project.repository).to receive(:add_branch)
+                                        .and_raise(Gitlab::Git::CommandError.new('git error'))
+        end
+
+        it 'tracks the exception and returns an error response' do
+          expect(Gitlab::ErrorTracking).to receive(:track_exception)
+                                            .with(instance_of(Gitlab::Git::CommandError))
+
+          result = execute
+
+          expect(result).to be_error
+          expect(result.message).to eq('Failed to create workload ref')
+        end
       end
     end
   end