diff --git a/app/actions/app_assign_service_account.rb b/app/actions/app_assign_service_account.rb
new file mode 100644
index 00000000000..5a848b47d2b
--- /dev/null
+++ b/app/actions/app_assign_service_account.rb
@@ -0,0 +1,64 @@
+require 'repositories/service_account_event_repository'
+
+module VCAP::CloudController
+  class AppAssignServiceAccount
+    class Error < StandardError; end
+    class Unauthorized < Error; end
+    class Conflict < Error; end
+
+    def initialize(permissions, actor: nil)
+      @permissions = permissions
+      @actor = actor
+    end
+
+    def assign(app, account, provision: false)
+      job = nil
+      app.db.transaction do
+        app.lock!
+        raise Unauthorized.new('not authorized to update the app') unless @permissions.can_write_to_active_space?(app.space.id)
+
+        if account
+          account.lock!
+          raise Conflict.new('account must belong to the app owning space') unless account.space_guid == app.space_guid
+
+          validate_ready!(account, provision)
+          raise Conflict.new('another service account is already assigned; unbind first') if app.service_account_guid && app.service_account_guid != account.guid
+
+          job = provision_account(account) unless account.status == 'ready'
+        end
+
+        unless app.service_account_guid == account&.guid
+          previous_guid = app.service_account_guid
+          app.update(service_account: account)
+          record_assignment(app, account, previous_guid) if @actor
+        end
+      end
+      provision ? job : app
+    end
+
+    private
+
+    def record_assignment(app, account, previous_guid)
+      Repositories::ServiceAccountEventRepository.record(
+        app, account ? 'assign' : 'unassign', @actor,
+        service_account_guid: account&.guid || previous_guid
+      )
+    end
+
+    def validate_ready!(account, provision)
+      raise Conflict.new('service account is not ready or enabled') unless account.enabled && (account.status == 'ready' || provision)
+
+      if PollableJobModel.where(resource_guid: account.guid, resource_type: 'service_account', state: %w[PROCESSING POLLING]).exclude(operation: 'service_account.provision').any?
+        raise Conflict.new('service account lifecycle operation is in progress')
+      end
+    end
+
+    def provision_account(account)
+      job = PollableJobModel.first(resource_guid: account.guid, operation: 'service_account.provision', state: %w[PROCESSING POLLING])
+      return job if job
+
+      account.update(status: 'reconciling')
+      Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue_pollable(Jobs::V3::ServiceAccountProvision.new(account.guid))
+    end
+  end
+end
diff --git a/app/actions/process_restart.rb b/app/actions/process_restart.rb
index 6a405e2647a..ba2f213bc07 100644
--- a/app/actions/process_restart.rb
+++ b/app/actions/process_restart.rb
@@ -14,7 +14,7 @@ def restart(process:, config:, stop_in_runtime:, revision: nil)
             runners(config).runner_for_process(process).stop
           end
 
-          process.update(state: ProcessModel::STARTED, revision: revision_to_set)
+          process.update(state: ProcessModel::STARTED, revision: revision_to_set, service_account_guid: process.app.service_account_guid, service_account_snapshot: true)
           runners(config).runner_for_process(process).start
         end
       end
diff --git a/app/actions/service_account_creation_budget.rb b/app/actions/service_account_creation_budget.rb
new file mode 100644
index 00000000000..48fd6ee9e68
--- /dev/null
+++ b/app/actions/service_account_creation_budget.rb
@@ -0,0 +1,25 @@
+module VCAP::CloudController
+  class ServiceAccountCreationBudget
+    WINDOW = 7.days
+
+    def self.consume(principal, exempt:)
+      return yield if exempt
+
+      # Lock the authenticated principal, not a space: requests on different API
+      # instances and in different spaces must share one atomic budget.
+      principal.lock!
+      now = Time.now.utc
+      creations = ServiceAccountModel.db[:service_account_creations]
+      limit = Config.config.get(:service_account_creation_limit) || -1
+      used = creations.where(principal_guid: principal.guid).where { created_at > now - WINDOW }.count
+      raise CloudController::Errors::V3::ApiError.new_from_details('ServiceAccountCreationLimitExceeded') if limit.between?(0, used)
+
+      result = yield
+      # Independent of accounts, tombstones and audit-event retention. Failed
+      # creations roll this back; successful deletions never refund the budget.
+      creations.insert(principal_guid: principal.guid, created_at: now) # rubocop:disable Rails/SkipsModelValidations -- Transactional creation ledger, not an ActiveRecord model.
+      creations.where(principal_guid: principal.guid).where { created_at <= now - WINDOW }.delete
+      result
+    end
+  end
+end
diff --git a/app/actions/service_account_provision.rb b/app/actions/service_account_provision.rb
new file mode 100644
index 00000000000..47b6166743e
--- /dev/null
+++ b/app/actions/service_account_provision.rb
@@ -0,0 +1,112 @@
+module VCAP::CloudController
+  class ServiceAccountProvision
+    class Conflict < StandardError; end
+
+    def initialize(clients, identity_ca:)
+      @clients = clients
+      @identity_ca = identity_ca
+    end
+
+    def provision(account)
+      failure = nil
+      account.db.transaction(savepoint: true) do
+        account.lock!
+        raise Conflict.new('service account is disabled') unless account.enabled
+
+        begin
+          account.db.transaction(savepoint: true) do
+            principal = User.first(guid: account.client_id)
+            raise Conflict.new('principal identity collision') if principal && !principal.is_oauth_client
+
+            desired = registration(account)
+            existing = existing_client(account.client_id)
+            if existing
+              raise Conflict.new('client identity collision') unless matching_registration?(existing, desired)
+            else
+              @clients.add(:client, desired)
+            end
+
+            User.create(guid: account.client_id, is_oauth_client: true, active: true) unless principal
+            account.update(status: 'ready')
+          end
+        rescue StandardError => e
+          failure = e
+          account.reload.update(status: 'failed')
+        end
+      end
+      raise failure if failure
+
+      account
+    end
+
+    def deprovision(account, delete: false)
+      failure = nil
+      account.db.transaction(savepoint: true) do
+        account.lock!
+        raise Conflict.new('service account was enabled again') if account.enabled
+        raise Conflict.new('service account is still assigned or in use') if delete && account.in_use?
+
+        begin
+          account.db.transaction(savepoint: true) do
+            principal = User.first(guid: account.client_id)
+            raise Conflict.new('principal identity collision') if principal && !principal.is_oauth_client
+
+            existing = existing_client(account.client_id)
+            if existing
+              raise Conflict.new('client identity collision') unless matching_registration?(existing, registration(account))
+
+              @clients.delete(:client, account.client_id)
+            end
+            if delete
+              principal&.destroy
+              account.destroy
+            else
+              account.update(status: 'disabled')
+            end
+          end
+        rescue StandardError => e
+          failure = e
+          account.reload.update(status: 'failed')
+        end
+      end
+      raise failure if failure
+    end
+
+    private
+
+    def matching_registration?(existing, desired)
+      tls_keys = existing.keys.select { |key| key.start_with?('tls-client-auth-', 'tls_client_auth_') }
+      return false unless tls_keys.sort == desired.keys.grep(/\Atls[-_]/).sort
+
+      desired.all? do |key, value|
+        actual = existing[key]
+        actual = [] if key == 'scope' && !existing.key?(key)
+        actual = [] if key == 'scope' && actual == ['uaa.none']
+        if value.is_a?(Array)
+          actual.is_a?(Array) && actual.all? { |entry| entry.is_a?(String) } && actual.sort == value.sort
+        else
+          actual == value
+        end
+      end
+    end
+
+    def existing_client(client_id)
+      @clients.get(:client, client_id)
+    rescue CF::UAA::NotFound
+      nil
+    end
+
+    def registration(account)
+      {
+        'client_id' => account.client_id,
+        'authorized_grant_types' => ['client_credentials'],
+        'authorities' => %w[cloud_controller.read cloud_controller.write],
+        'scope' => [],
+        'access_token_validity' => 300,
+        'tls-client-auth-ca' => @identity_ca,
+        'tls_client_auth_san_dns' => account.certificate_dns_san,
+        'cf_service_account_guid' => account.guid
+      }
+    end
+  end
+end
diff --git a/app/actions/space_delete.rb b/app/actions/space_delete.rb
index 6d73dd46b4c..707d0aa683a 100644
--- a/app/actions/space_delete.rb
+++ b/app/actions/space_delete.rb
@@ -1,5 +1,6 @@
 require 'actions/v3/service_instance_delete'
 require 'jobs/v3/delete_service_instance_job'
+require 'repositories/service_account_event_repository'
 
 module VCAP::CloudController
   class SpaceDelete
@@ -24,6 +25,11 @@ def delete(dataset)
 
         next unless instance_delete_errors.empty? && instance_unshare_errors.empty?
 
+        account_delete_errors = delete_service_accounts(space_model)
+        err = accumulate_space_deletion_error(account_delete_errors, space_model.name)
+        errors << err unless err.nil?
+        next unless account_delete_errors.empty?
+
         Space.db.transaction do
           delete_apps(space_model)
           space_model.destroy
@@ -76,7 +82,36 @@ def unshare_service_instances(space_model)
     end
 
     def delete_apps(space_model)
-      AppDelete.new(@user_audit_info).delete(space_model.app_models)
+      AppDelete.new(@user_audit_info).delete(space_model.app_models_dataset.all)
+    end
+
+    def delete_service_accounts(space_model)
+      accounts = ServiceAccountModel.where(space_guid: space_model.guid).order(:guid).all
+      return [] if accounts.empty?
+
+      ServiceAccountModel.db.transaction do
+        accounts.each do |account|
+          account.lock!
+          if PollableJobModel.where(resource_guid: account.guid, resource_type: 'service_account', state: %w[PROCESSING POLLING]).any?
+            raise ServiceAccountProvision::Conflict.new('service account operation is already in progress')
+          end
+
+          account.update(enabled: false, status: 'deleting')
+        end
+      end
+
+      # AppDelete removes service bindings, tasks and process snapshots. Do not
+      # roll this back if later UAA cleanup fails: retries must see real progress.
+      delete_apps(space_model)
+      provisioner = CloudController::DependencyLocator.instance.service_account_provisioner
+      accounts.each_with_object([]) do |account, errors|
+        Repositories::ServiceAccountEventRepository.record(account, 'delete', @user_audit_info, { recursive: true })
+        provisioner.deprovision(account, delete: true)
+      rescue StandardError => e
+        errors << e
+      end
+    rescue StandardError => e
+      [e]
     end
 
     def delete_service_brokers(space_model)
diff --git a/app/actions/task_create.rb b/app/actions/task_create.rb
index 2eddace7bad..36a7fc49cac 100644
--- a/app/actions/task_create.rb
+++ b/app/actions/task_create.rb
@@ -24,6 +24,8 @@ def create(app, message, user_audit_info, droplet: nil)
         task = TaskModel.create(
           name: use_requested_name_or_generate_name(message),
           app: app,
+          service_account_guid: app.service_account_guid,
+          service_account_snapshot: true,
           state: TaskModel::PENDING_STATE,
           droplet: droplet,
           command: command(message, template_process),
diff --git a/app/collection_transformers/username_populator.rb b/app/collection_transformers/username_populator.rb
index 866baf3e145..99892d72a54 100644
--- a/app/collection_transformers/username_populator.rb
+++ b/app/collection_transformers/username_populator.rb
@@ -8,9 +8,11 @@ def initialize(uaa_client)
 
     def transform(users, _opts={})
       users = Array(users)
-      user_ids = users.collect(&:guid)
-      username_mapping = uaa_client.usernames_for_ids(user_ids)
-      users.each { |user| user.username = username_mapping[user.guid] }
+      accounts = ServiceAccountModel.where(name: users.select(&:is_oauth_client).map { |user| user.guid.delete_prefix('cf:service-account:') }).all
+      account_names = accounts.to_h { |account| [account.client_id, account.name] }
+      human_users = users.reject { |user| user.is_oauth_client && account_names.key?(user.guid) }
+      username_mapping = human_users.empty? ? {} : uaa_client.usernames_for_ids(human_users.collect(&:guid))
+      users.each { |user| user.username = user.is_oauth_client && account_names.key?(user.guid) ? account_names[user.guid] : username_mapping[user.guid] }
     end
   end
 end
diff --git a/app/controllers/v3/app_service_accounts_controller.rb b/app/controllers/v3/app_service_accounts_controller.rb
new file mode 100644
index 00000000000..b0690fa1e37
--- /dev/null
+++ b/app/controllers/v3/app_service_accounts_controller.rb
@@ -0,0 +1,43 @@
+require 'actions/app_assign_service_account'
+require 'messages/app_service_account_update_message'
+require 'fetchers/app_fetcher'
+require 'jobs/v3/service_account_provision'
+
+class AppServiceAccountsController < ApplicationController
+  def show
+    app, = fetch_readable_app
+    render status: :ok, json: relationship(app)
+  end
+
+  def update
+    app, space = fetch_readable_app
+    unauthorized! unless permission_queryer.can_write_to_active_space?(space.id)
+    require_writable_space!(space)
+    message = AppServiceAccountUpdateMessage.new(hashed_params[:body])
+    unprocessable!(message.errors.full_messages) unless message.valid?
+
+    account = ServiceAccountModel.where(guid: message.account_guid).first if message.account_guid
+    resource_not_found!(:service_account) if message.account_guid && !account
+    job = AppAssignServiceAccount.new(permission_queryer, actor: user_audit_info).assign(app, account, provision: Config.config.get(:service_account_provisioning_enabled) == true)
+    add_warning_headers(["Restart #{app.name} for the service-account assignment change to take effect."])
+    return head :accepted, 'Location' => url_builder.build_url(path: "/v3/jobs/#{job.guid}") if job.is_a?(PollableJobModel)
+
+    render status: :ok, json: relationship(app)
+  rescue AppAssignServiceAccount::Unauthorized
+    unauthorized!
+  rescue AppAssignServiceAccount::Conflict => e
+    raise CloudController::Errors::V3::ApiError.new_from_details('ServiceAccountAssignmentConflict', e.message)
+  end
+
+  private
+
+  def fetch_readable_app
+    app, space = AppFetcher.new.fetch(hashed_params[:app_guid])
+    resource_not_found!(:app) unless app && permission_queryer.can_read_from_space?(space.id, space.organization_id)
+    [app, space]
+  end
+
+  def relationship(app)
+    { data: app.service_account_guid ? { guid: app.service_account_guid } : nil }
+  end
+end
diff --git a/app/controllers/v3/service_accounts_controller.rb b/app/controllers/v3/service_accounts_controller.rb
new file mode 100644
index 00000000000..fb978e9205d
--- /dev/null
+++ b/app/controllers/v3/service_accounts_controller.rb
@@ -0,0 +1,141 @@
+require 'messages/service_account_create_message'
+require 'presenters/v3/service_account_presenter'
+require 'messages/service_account_update_message'
+require 'messages/service_accounts_list_message'
+require 'messages/apps_list_message'
+require 'fetchers/app_list_fetcher'
+require 'presenters/v3/app_presenter'
+require 'repositories/service_account_event_repository'
+require 'actions/service_account_creation_budget'
+
+class ServiceAccountsController < ApplicationController
+  def index
+    message = ServiceAccountsListMessage.from_params(query_params)
+    invalid_param!(message.errors.full_messages) unless message.valid?
+    dataset = ServiceAccountModel.dataset
+    dataset = dataset.where(space_guid: permission_queryer.readable_space_guids_query) unless permission_queryer.can_read_globally?
+    dataset = dataset.where(name: message.names) if message.requested?(:names)
+    dataset = dataset.where(space_guid: message.space_guids) if message.requested?(:space_guids)
+    render_list(dataset, message, Presenters::V3::ServiceAccountPresenter, '/v3/service_accounts')
+  end
+
+  def apps
+    account = readable_account
+    message = AppsListMessage.from_params(query_params)
+    invalid_param!(message.errors.full_messages) unless message.valid?
+    dataset = AppListFetcher.fetch(message, [account.space_guid]).where(service_account_guid: account.guid)
+    render_list(dataset, message, Presenters::V3::AppPresenter, "/v3/service_accounts/#{account.guid}/apps")
+  end
+
+  def show
+    account = readable_account
+
+    render status: :ok, json: Presenters::V3::ServiceAccountPresenter.new(account)
+  end
+
+  def create
+    message = ServiceAccountCreateMessage.new(hashed_params[:body])
+    unprocessable!(message.errors.full_messages) unless message.valid?
+
+    space = Space.where(guid: message.space_guid).first
+    unprocessable!('Space not found') unless space && permission_queryer.can_read_from_space?(space.id, space.organization_id)
+    authorize_management!(space)
+
+    account = nil
+    ServiceAccountModel.db.transaction do
+      ServiceAccountCreationBudget.consume(current_user, exempt: permission_queryer.can_write_globally?) do
+        account = ServiceAccountModel.create(name: message.name, space: space)
+        apply_metadata(account, message)
+        Repositories::ServiceAccountEventRepository.record(account, 'create', user_audit_info)
+      end
+    end
+    render status: :created, json: Presenters::V3::ServiceAccountPresenter.new(account)
+  rescue Sequel::ValidationFailed => e
+    raise CloudController::Errors::V3::ApiError.new_from_details('ServiceAccountNameReserved') if e.message.include?('is already reserved')
+
+    unprocessable!(e.message)
+  end
+
+  def update
+    account = readable_account
+    authorize_management!(account.space)
+    message = ServiceAccountUpdateMessage.new(hashed_params[:body])
+    unprocessable!(message.errors.full_messages) unless message.valid?
+    job = nil
+    account.db.transaction do
+      account.lock!
+      apply_metadata(account, message)
+      if message.requested?(:enabled)
+        require_provisioning!
+        reject_active_operation!(account)
+        account.update(enabled: message.enabled, status: 'reconciling')
+        job = enqueue_lifecycle(account, message.enabled ? 'provision' : 'disable')
+      end
+      Repositories::ServiceAccountEventRepository.record(account, 'update', user_audit_info, message.audit_hash)
+    end
+    return head :accepted, 'Location' => url_builder.build_url(path: "/v3/jobs/#{job.guid}") if job
+
+    render status: :ok, json: Presenters::V3::ServiceAccountPresenter.new(account.reload)
+  end
+
+  def destroy
+    account = readable_account
+    authorize_management!(account.space)
+    require_provisioning!
+    job = nil
+    account.db.transaction do
+      account.lock!
+      lifecycle_conflict!('service account is still assigned or in use') if account.in_use?
+      reject_active_operation!(account)
+      account.update(enabled: false, status: 'deleting')
+      job = enqueue_lifecycle(account, 'delete')
+      Repositories::ServiceAccountEventRepository.record(account, 'delete', user_audit_info)
+    end
+    head :accepted, 'Location' => url_builder.build_url(path: "/v3/jobs/#{job.guid}")
+  end
+
+  private
+
+  def require_provisioning!
+    lifecycle_conflict!('service account provisioning is not enabled') unless Config.config.get(:service_account_provisioning_enabled) == true
+  end
+
+  def reject_active_operation!(account)
+    return unless PollableJobModel.where(resource_guid: account.guid, resource_type: 'service_account', state: %w[PROCESSING POLLING]).any?
+
+    lifecycle_conflict!('service account operation is already in progress')
+  end
+
+  def lifecycle_conflict!(detail)
+    raise CloudController::Errors::V3::ApiError.new_from_details('ServiceAccountAssignmentConflict', detail)
+  end
+
+  def enqueue_lifecycle(account, operation)
+    Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue_pollable(Jobs::V3::ServiceAccountProvision.new(account.guid, operation: operation))
+  end
+
+  def readable_account
+    account = ServiceAccountModel.first(guid: hashed_params[:guid])
+    resource_not_found!(:service_account) unless account && permission_queryer.can_read_from_space?(account.space.id, account.space.organization_id)
+    account
+  end
+
+  def authorize_management!(space)
+    unauthorized! unless permission_queryer.can_write_globally? || space.managers_dataset.where(id: current_user.id).any?
+    require_writable_space!(space)
+  end
+
+  def apply_metadata(account, message)
+    account.update(description: message.description || '') if message.requested?(:description)
+    LabelsUpdate.update(account, message.labels, ServiceAccountLabelModel)
+    AnnotationsUpdate.update(account, message.annotations, ServiceAccountAnnotationModel)
+  end
+
+  def render_list(dataset, message, presenter, path)
+    render status: :ok, json: Presenters::V3::PaginatedListPresenter.new(
+      presenter: presenter,
+      paginated_result: SequelPaginator.new.get_page(dataset, message.pagination_options),
+      path: path, message: message
+    )
+  end
+end
diff --git a/app/jobs/v3/service_account_provision.rb b/app/jobs/v3/service_account_provision.rb
new file mode 100644
index 00000000000..756ad02fad8
--- /dev/null
+++ b/app/jobs/v3/service_account_provision.rb
@@ -0,0 +1,48 @@
+require 'jobs/cc_job'
+require 'actions/service_account_provision'
+
+module VCAP::CloudController
+  module Jobs
+    module V3
+      class ServiceAccountProvision < CCJob
+        attr_reader :resource_guid
+
+        def initialize(account_guid, operation: 'provision')
+          @resource_guid = account_guid
+          @operation = operation
+        end
+
+        def perform
+          account = ServiceAccountModel.first(guid: resource_guid)
+          raise CloudController::Errors::ApiError.new_from_details('ResourceNotFound', 'The service account could not be found') unless account
+
+          provisioner = CloudController::DependencyLocator.instance.service_account_provisioner
+          case @operation
+          when 'provision'
+            provisioner.provision(account)
+          when 'disable', 'delete'
+            provisioner.deprovision(account, delete: @operation == 'delete')
+          else
+            raise ArgumentError.new('unsupported service account operation')
+          end
+        end
+
+        def max_attempts
+          3
+        end
+
+        def resource_type
+          'service_account'
+        end
+
+        def display_name
+          "service_account.#{@operation}"
+        end
+
+        def job_name_in_configuration
+          :service_account_provision
+        end
+      end
+    end
+  end
+end
diff --git a/app/messages/app_service_account_update_message.rb b/app/messages/app_service_account_update_message.rb
new file mode 100644
index 00000000000..03b952c5660
--- /dev/null
+++ b/app/messages/app_service_account_update_message.rb
@@ -0,0 +1,24 @@
+require 'messages/base_message'
+
+module VCAP::CloudController
+  class AppServiceAccountUpdateMessage < BaseMessage
+    register_allowed_keys [:data]
+    validates_with NoAdditionalKeysValidator
+    validate :relationship_data
+
+    def account_guid
+      HashUtils.dig(data, :guid)
+    end
+
+    def relationship_data
+      unless requested?(:data)
+        errors.add(:data, 'must be provided')
+        return
+      end
+      return if data.nil?
+      return if data.is_a?(Hash) && data.keys == [:guid] && account_guid.is_a?(String) && account_guid.present?
+
+      errors.add(:data, 'must be null or an object containing only a nonempty guid string')
+    end
+  end
+end
diff --git a/app/messages/service_account_create_message.rb b/app/messages/service_account_create_message.rb
new file mode 100644
index 00000000000..4c9b0e82a65
--- /dev/null
+++ b/app/messages/service_account_create_message.rb
@@ -0,0 +1,27 @@
+require 'messages/metadata_base_message'
+
+module VCAP::CloudController
+  class ServiceAccountCreateMessage < MetadataBaseMessage
+    register_allowed_keys %i[name relationships description]
+    validates :description, string: true, allow_nil: true, length: { maximum: 250 }
+    validates_with NoAdditionalKeysValidator, RelationshipValidator
+    validates :name, presence: true, string: true,
+                     format: { with: ->(_) { ServiceAccountModel::NAME_PATTERN } }, length: { in: 3..63 }
+
+    delegate :space_guid, to: :relationships_message
+
+    def relationships_message
+      @relationships_message ||= Relationships.new(relationships.deep_symbolize_keys)
+    end
+
+    class Relationships < BaseMessage
+      register_allowed_keys [:space]
+      validates_with NoAdditionalKeysValidator
+      validates :space, presence: true, to_one_relationship: true
+
+      def space_guid
+        HashUtils.dig(space, :data, :guid)
+      end
+    end
+  end
+end
diff --git a/app/messages/service_account_update_message.rb b/app/messages/service_account_update_message.rb
new file mode 100644
index 00000000000..6bc199effe0
--- /dev/null
+++ b/app/messages/service_account_update_message.rb
@@ -0,0 +1,10 @@
+require 'messages/metadata_base_message'
+
+module VCAP::CloudController
+  class ServiceAccountUpdateMessage < MetadataBaseMessage
+    register_allowed_keys %i[description enabled]
+    validates :enabled, boolean: true, if: -> { requested?(:enabled) }
+    validates_with NoAdditionalKeysValidator
+    validates :description, string: true, allow_nil: true, length: { maximum: 250 }
+  end
+end
diff --git a/app/messages/service_accounts_list_message.rb b/app/messages/service_accounts_list_message.rb
new file mode 100644
index 00000000000..bfa960a7f22
--- /dev/null
+++ b/app/messages/service_accounts_list_message.rb
@@ -0,0 +1,17 @@
+require 'messages/list_message'
+
+module VCAP::CloudController
+  class ServiceAccountsListMessage < ListMessage
+    register_allowed_keys %i[names space_guids]
+    validates_with NoAdditionalParamsValidator
+    validates :names, :space_guids, array: true, allow_nil: true
+
+    def self.from_params(params)
+      super(params, %w[names space_guids])
+    end
+
+    def valid_order_by_values
+      super + [:name]
+    end
+  end
+end
diff --git a/app/models.rb b/app/models.rb
index 7944a73f04b..f839d668447 100644
--- a/app/models.rb
+++ b/app/models.rb
@@ -13,6 +13,7 @@
 require 'models/runtime/docker_lifecycle_data_model'
 require 'models/runtime/task_model'
 require 'models/runtime/isolation_segment_model'
+require 'models/runtime/service_account_model'
 require 'models/runtime/pollable_job_model'
 require 'models/runtime/job_warning_model'
 require 'models/runtime/security_group'
@@ -89,6 +90,8 @@
 require 'models/runtime/domain_label_model'
 require 'models/runtime/droplet_label_model'
 require 'models/runtime/isolation_segment_label_model'
+require 'models/runtime/service_account_label_model'
+require 'models/runtime/service_account_annotation_model'
 require 'models/runtime/organization_label_model'
 require 'models/runtime/package_label_model'
 require 'models/runtime/process_label_model'
diff --git a/app/models/runtime/app_model.rb b/app/models/runtime/app_model.rb
index 95de344a631..37d6dd63a2f 100644
--- a/app/models/runtime/app_model.rb
+++ b/app/models/runtime/app_model.rb
@@ -17,6 +17,7 @@ class AppModel < Sequel::Model(:apps)
     one_to_many :tasks, class: 'VCAP::CloudController::TaskModel', key: :app_guid, primary_key: :guid
 
     many_to_one :space, class: 'VCAP::CloudController::Space', key: :space_guid, primary_key: :guid, without_guid_generation: true
+    many_to_one :service_account, class: 'VCAP::CloudController::ServiceAccountModel', key: :service_account_guid, primary_key: :guid, without_guid_generation: true
     one_through_one :organization, join_table: Space.table_name, left_key: :guid, left_primary_key: :space_guid, right_primary_key: :id, right_key: :organization_id
 
     one_to_many :processes, class: 'VCAP::CloudController::ProcessModel', key: :app_guid, primary_key: :guid do |dataset|
@@ -94,6 +95,8 @@ def validate
       validate_environment_variables
       validate_droplet_is_staged
 
+      errors.add(:service_account, 'must belong to the app owning space') if service_account && service_account.space_guid != space_guid
+
       validates_includes Lifecycles::TYPES, :lifecycle_type
     end
 
diff --git a/app/models/runtime/process_model.rb b/app/models/runtime/process_model.rb
index eb031eda821..d376ba07695 100644
--- a/app/models/runtime/process_model.rb
+++ b/app/models/runtime/process_model.rb
@@ -304,10 +304,18 @@ def before_validation
     end
 
     def before_save
+      if being_started?
+        self.service_account_guid = app.service_account_guid
+        self.service_account_snapshot = true
+      end
       set_new_version if version_needs_to_be_updated?
       super
     end
 
+    def runtime_service_account_guid
+      service_account_snapshot ? service_account_guid : nil
+    end
+
     # rubocop:disable Metrics/CyclomaticComplexity
     def version_needs_to_be_updated?
       # change version if:
diff --git a/app/models/runtime/service_account_annotation_model.rb b/app/models/runtime/service_account_annotation_model.rb
new file mode 100644
index 00000000000..234db2f44bf
--- /dev/null
+++ b/app/models/runtime/service_account_annotation_model.rb
@@ -0,0 +1,6 @@
+module VCAP::CloudController
+  class ServiceAccountAnnotationModel < Sequel::Model(:service_account_annotations)
+    many_to_one :service_account, class: 'VCAP::CloudController::ServiceAccountModel', primary_key: :guid, key: :resource_guid, without_guid_generation: true
+    include MetadataModelMixin
+  end
+end
diff --git a/app/models/runtime/service_account_label_model.rb b/app/models/runtime/service_account_label_model.rb
new file mode 100644
index 00000000000..15b2e7270e6
--- /dev/null
+++ b/app/models/runtime/service_account_label_model.rb
@@ -0,0 +1,6 @@
+module VCAP::CloudController
+  class ServiceAccountLabelModel < Sequel::Model(:service_account_labels)
+    many_to_one :service_account, class: 'VCAP::CloudController::ServiceAccountModel', primary_key: :guid, key: :resource_guid, without_guid_generation: true
+    include MetadataModelMixin
+  end
+end
diff --git a/app/models/runtime/service_account_model.rb b/app/models/runtime/service_account_model.rb
new file mode 100644
index 00000000000..59054afd2d2
--- /dev/null
+++ b/app/models/runtime/service_account_model.rb
@@ -0,0 +1,45 @@
+module VCAP::CloudController
+  class ServiceAccountModel < Sequel::Model(:service_accounts)
+    NAME_PATTERN = /\A[a-z0-9][a-z0-9-]{1,61}[a-z0-9]\z/
+
+    many_to_one :space, key: :space_guid, primary_key: :guid, without_guid_generation: true
+    one_to_many :apps, class: 'VCAP::CloudController::AppModel', key: :service_account_guid, primary_key: :guid
+    one_to_many :labels, class: 'VCAP::CloudController::ServiceAccountLabelModel', key: :resource_guid, primary_key: :guid
+    one_to_many :annotations, class: 'VCAP::CloudController::ServiceAccountAnnotationModel', key: :resource_guid, primary_key: :guid
+    add_association_dependencies labels: :destroy, annotations: :destroy
+
+    def validate
+      super
+      validates_presence :space
+      validates_format NAME_PATTERN, :name
+      errors.add(:name, 'is immutable') if !new? && column_changed?(:name)
+      errors.add(:space, 'is immutable') if !new? && column_changed?(:space_guid)
+    end
+
+    def around_create
+      db.transaction(savepoint: true) do
+        # Keep this reservation permanently, even when the account is destroyed.
+        # The unique constraint, rather than a preflight query, serializes claims.
+        db[:service_account_names].insert(name: name) # rubocop:disable Rails/SkipsModelValidations -- Atomic namespace reservation, not an ActiveRecord model.
+        yield
+      end
+    rescue Sequel::UniqueConstraintViolation
+      errors.add(:name, 'is already reserved')
+      raise validation_failed_error
+    end
+
+    def client_id
+      "cf:service-account:#{name}"
+    end
+
+    def in_use?
+      apps_dataset.any? ||
+        ProcessModel.where(service_account_guid: guid, state: ProcessModel::STARTED).any? ||
+        TaskModel.where(service_account_guid: guid, state: [TaskModel::PENDING_STATE, TaskModel::RUNNING_STATE, TaskModel::CANCELING_STATE]).any?
+    end
+
+    def certificate_dns_san
+      "#{name}.svc.identity"
+    end
+  end
+end
diff --git a/app/models/runtime/task_model.rb b/app/models/runtime/task_model.rb
index bad22c00eef..7cf82b686bc 100644
--- a/app/models/runtime/task_model.rb
+++ b/app/models/runtime/task_model.rb
@@ -28,6 +28,10 @@ class TaskModel < Sequel::Model(:tasks)
     set_field_as_encrypted :environment_variables, column: :encrypted_environment_variables
     serializes_via_json :environment_variables
 
+    def runtime_service_account_guid
+      service_account_snapshot ? service_account_guid : nil
+    end
+
     def after_update
       super
 
diff --git a/app/presenters/v3/service_account_presenter.rb b/app/presenters/v3/service_account_presenter.rb
new file mode 100644
index 00000000000..4e410709713
--- /dev/null
+++ b/app/presenters/v3/service_account_presenter.rb
@@ -0,0 +1,29 @@
+require 'presenters/v3/base_presenter'
+require 'presenters/mixins/metadata_presentation_helpers'
+
+module VCAP::CloudController
+  module Presenters
+    module V3
+      class ServiceAccountPresenter < BasePresenter
+        include Mixins::MetadataPresentationHelpers
+
+        def to_hash
+          {
+            guid: @resource.guid,
+            name: @resource.name,
+            description: @resource.description,
+            metadata: { labels: hashified_labels(@resource.labels), annotations: hashified_annotations(@resource.annotations) },
+            created_at: @resource.created_at,
+            updated_at: @resource.updated_at,
+            client_id: @resource.client_id,
+            certificate_dns_san: @resource.certificate_dns_san,
+            enabled: @resource.enabled,
+            status: @resource.status,
+            relationships: { space: { data: { guid: @resource.space_guid } } },
+            links: { self: { href: url_builder.build_url(path: "/v3/service_accounts/#{@resource.guid}") } }
+          }
+        end
+      end
+    end
+  end
+end
diff --git a/app/repositories/service_account_event_repository.rb b/app/repositories/service_account_event_repository.rb
new file mode 100644
index 00000000000..4e3077d2af0
--- /dev/null
+++ b/app/repositories/service_account_event_repository.rb
@@ -0,0 +1,21 @@
+module VCAP::CloudController
+  module Repositories
+    class ServiceAccountEventRepository
+      def self.record(resource, operation, actor, metadata={})
+        Event.create(
+          type: "audit.#{resource.is_a?(AppModel) ? 'app.service_account' : 'service_account'}.#{operation}",
+          space: resource.space,
+          actee: resource.guid,
+          actee_type: resource.is_a?(AppModel) ? 'app' : 'service_account',
+          actee_name: resource.name,
+          actor: actor.user_guid,
+          actor_type: 'user',
+          actor_name: actor.user_email,
+          actor_username: actor.user_name,
+          timestamp: Sequel::CURRENT_TIMESTAMP,
+          metadata: metadata
+        )
+      end
+    end
+  end
+end
diff --git a/config/routes.rb b/config/routes.rb
index 28526281d86..612c691e9fd 100644
--- a/config/routes.rb
+++ b/config/routes.rb
@@ -5,6 +5,15 @@
   post '/admin/actions/clear_buildpack_cache', to: 'admin_actions#clear_buildpack_cache'
 
   # apps
+  post '/service_accounts', to: 'service_accounts#create'
+  get '/service_accounts', to: 'service_accounts#index'
+  patch '/service_accounts/:guid', to: 'service_accounts#update'
+  delete '/service_accounts/:guid', to: 'service_accounts#destroy'
+  get '/service_accounts/:guid/apps', to: 'service_accounts#apps'
+  get '/service_accounts/:guid', to: 'service_accounts#show'
+  get '/apps/:app_guid/relationships/service_account', to: 'app_service_accounts#show'
+  patch '/apps/:app_guid/relationships/service_account', to: 'app_service_accounts#update'
+
   get '/apps', to: 'apps_v3#index'
   post '/apps', to: 'apps_v3#create'
   get '/apps/:guid', to: 'apps_v3#show'
diff --git a/db/migrations/20261005120000_create_service_accounts.rb b/db/migrations/20261005120000_create_service_accounts.rb
new file mode 100644
index 00000000000..44bd93543d8
--- /dev/null
+++ b/db/migrations/20261005120000_create_service_accounts.rb
@@ -0,0 +1,20 @@
+Sequel.migration do
+  change do
+    create_table :service_account_names do
+      primary_key :id, name: :service_account_names_pkey
+      String :name, size: 63, null: false
+      index :name, unique: true, name: :service_account_names_name_unique
+    end
+
+    create_table :service_accounts do
+      VCAP::Migration.common(self)
+      String :name, size: 63, null: false
+      foreign_key [:name], :service_account_names, key: :name, name: :fk_service_accounts_name
+      index :name, unique: true, name: :service_accounts_name_unique
+      String :space_guid, size: 255, null: false
+      foreign_key [:space_guid], :spaces, key: :guid, name: :fk_service_accounts_space
+      TrueClass :enabled, null: false, default: true
+      String :status, size: 32, null: false, default: 'reserved'
+    end
+  end
+end
diff --git a/db/migrations/20261005120100_add_app_service_account.rb b/db/migrations/20261005120100_add_app_service_account.rb
new file mode 100644
index 00000000000..6bc61c54722
--- /dev/null
+++ b/db/migrations/20261005120100_add_app_service_account.rb
@@ -0,0 +1,33 @@
+Sequel.migration do
+  no_transaction
+
+  up do
+    options = { size: 255 }
+    if database_type == :mysql
+      parent = fetch("SHOW FULL COLUMNS FROM service_accounts LIKE 'guid'").first
+      options[:collate] = parent[:Collation] || parent[:collation]
+    end
+    existing_column = schema(:apps).any? { |name, _| name == :service_account_guid }
+    alter_table :apps do
+      if existing_column
+        set_column_type :service_account_guid, String, **options, size: 255
+      else
+        add_column :service_account_guid, String, **options, size: 255
+      end
+      add_foreign_key [:service_account_guid], :service_accounts, key: :guid, name: :fk_apps_service_account_guid
+    end
+    VCAP::Migration.with_concurrent_timeout(self) do
+      add_index :apps, :service_account_guid, name: :apps_service_account_guid_index, concurrently: database_type == :postgres
+    end
+  end
+
+  down do
+    VCAP::Migration.with_concurrent_timeout(self) do
+      drop_index :apps, :service_account_guid, name: :apps_service_account_guid_index, concurrently: database_type == :postgres
+    end
+    alter_table :apps do
+      drop_foreign_key [:service_account_guid], name: :fk_apps_service_account_guid
+      drop_column :service_account_guid
+    end
+  end
+end
diff --git a/db/migrations/20261005120200_add_service_account_metadata.rb b/db/migrations/20261005120200_add_service_account_metadata.rb
new file mode 100644
index 00000000000..31f06b1f531
--- /dev/null
+++ b/db/migrations/20261005120200_add_service_account_metadata.rb
@@ -0,0 +1,16 @@
+Sequel.migration do
+  change do
+    alter_table(:service_accounts) do
+      add_column :description, String, size: 250, null: false, default: ''
+    end
+    create_table(:service_account_labels) do
+      VCAP::Migration.common(self)
+      VCAP::Migration.labels_common(self, :service_account_labels, :service_accounts)
+    end
+    create_table(:service_account_annotations) do
+      VCAP::Migration.common(self)
+      VCAP::Migration.annotations_common(self, :service_account_annotations, :service_accounts)
+    end
+    rename_column :service_account_annotations, :key, :key_name
+  end
+end
diff --git a/db/migrations/20261005120300_add_task_service_account_snapshot.rb b/db/migrations/20261005120300_add_task_service_account_snapshot.rb
new file mode 100644
index 00000000000..cdcdf66ba30
--- /dev/null
+++ b/db/migrations/20261005120300_add_task_service_account_snapshot.rb
@@ -0,0 +1,8 @@
+Sequel.migration do
+  change do
+    alter_table(:tasks) do
+      add_column :service_account_guid, String, size: 255
+      add_column :service_account_snapshot, TrueClass, null: false, default: false
+    end
+  end
+end
diff --git a/db/migrations/20261005120400_add_process_service_account_snapshot.rb b/db/migrations/20261005120400_add_process_service_account_snapshot.rb
new file mode 100644
index 00000000000..65f3a6c124d
--- /dev/null
+++ b/db/migrations/20261005120400_add_process_service_account_snapshot.rb
@@ -0,0 +1,8 @@
+Sequel.migration do
+  change do
+    alter_table(:processes) do
+      add_column :service_account_guid, String, size: 255
+      add_column :service_account_snapshot, TrueClass, null: false, default: false
+    end
+  end
+end
diff --git a/db/migrations/20261007120000_create_service_account_creations.rb b/db/migrations/20261007120000_create_service_account_creations.rb
new file mode 100644
index 00000000000..5367cc8a40d
--- /dev/null
+++ b/db/migrations/20261007120000_create_service_account_creations.rb
@@ -0,0 +1,10 @@
+Sequel.migration do
+  change do
+    create_table :service_account_creations do
+      primary_key :id, name: :service_account_creations_pkey
+      String :principal_guid, size: 255, null: false
+      DateTime :created_at, null: false
+      index %i[principal_guid created_at], name: :service_account_creations_principal_time
+    end
+  end
+end
diff --git a/docs/v3/source/includes/api_resources/_service_accounts.md b/docs/v3/source/includes/api_resources/_service_accounts.md
new file mode 100644
index 00000000000..ef822789f43
--- /dev/null
+++ b/docs/v3/source/includes/api_resources/_service_accounts.md
@@ -0,0 +1,148 @@
+# Service accounts (experimental)
+
+Service accounts are immutable, space-owned identities shared by apps in that
+space. Each app has zero or one desired account. Names are foundation-unique
+lowercase DNS labels of 3–63 characters and remain reserved after deletion.
+Cross-space assignment is forbidden, including within the same organization.
+
+## Resource and permissions
+
+Account representations include `guid`, `name`, `description`, `metadata`,
+`enabled`, `status`, `client_id`, `certificate_dns_san`, timestamps and the owning
+`relationships.space`. Identity fields and owning space cannot be updated.
+The client ID is `cf:service-account:<name>` and DNS SAN is `<name>.svc.identity`.
+
+Owning-space readers can show/list accounts and their apps. Space managers and
+platform administrators manage accounts in writable spaces. App writers may assign
+an enabled same-space account. Account management does not grant app-write access.
+Account assignment does not grant any resource roles to the account principal.
+
+| Method | Path | Behavior |
+| --- | --- | --- |
+| POST | `/v3/service_accounts` | Reserve an account; returns 201, initially `reserved` |
+| GET | `/v3/service_accounts` | Paginated readable accounts; `names`, `space_guids`, standard pagination/order parameters |
+| GET | `/v3/service_accounts/:guid` | Show readable account |
+| GET | `/v3/service_accounts/:guid/apps` | Paginated apps with desired assignment |
+| PATCH | `/v3/service_accounts/:guid` | Update description/metadata (200) or enabled state (202) |
+| DELETE | `/v3/service_accounts/:guid` | Queue unused-account deletion; returns 202 |
+| GET | `/v3/apps/:guid/relationships/service_account` | Desired relationship `{ "data": null }` or account GUID |
+| PATCH | `/v3/apps/:guid/relationships/service_account` | Assign GUID or explicitly unbind with `{ "data": null }` |
+
+Minimal creation body (permitted roles: owning-space manager or platform admin):
+
+```json
+{
+  "name": "payments-worker",
+  "relationships": { "space": { "data": { "guid": "SPACE_GUID" } } }
+}
+```
+
+Optional creation fields are `description` and standard `metadata.labels` and
+`metadata.annotations`. Metadata updates use the normal merge/removal semantics.
+
+Operators can set `service_account_creation_limit` in API configuration to a
+nonnegative maximum successful creations per authenticated principal over a rolling
+seven-day window. Omitted or `-1` means unlimited; `0` disables non-admin creation.
+The budget spans all spaces and includes automation clients, but platform admins
+and admin automation are exempt. Exhaustion returns 429
+`CF-ServiceAccountCreationLimitExceeded`. Failed creation consumes nothing;
+deletion does not refund the budget. Accounting is independent of account deletion
+and audit-event retention, and concurrent requests share one atomic budget.
+
+Example account object:
+
+```json
+{
+  "guid": "ACCOUNT_GUID",
+  "name": "payments-worker",
+  "description": "",
+  "enabled": true,
+  "status": "reserved",
+  "client_id": "cf:service-account:payments-worker",
+  "certificate_dns_san": "payments-worker.svc.identity",
+  "created_at": "2026-10-05T12:00:00Z",
+  "updated_at": "2026-10-05T12:00:00Z",
+  "metadata": { "labels": {}, "annotations": {} },
+  "relationships": { "space": { "data": { "guid": "SPACE_GUID" } } },
+  "links": {
+    "self": { "href": "https://api.example.org/v3/service_accounts/ACCOUNT_GUID" }
+  }
+}
+```
+
+Permitted roles for GET endpoints are owning-space readers and global readers.
+PATCH/DELETE account endpoints require an owning-space manager or platform admin;
+PATCH app relationship requires app-write permission in the owning space.
+
+Assign an account with `{ "data": { "guid": "ACCOUNT_GUID" } }` on the app
+relationship endpoint. Clear it with `{ "data": null }`. PATCH an account with
+`{ "enabled": false }` or `{ "enabled": true }` to disable or enable authentication.
+
+## Provisioning and lifecycle
+
+First authorized bind is asynchronous when operator provisioning is enabled.
+It persists the desired assignment, marks the account `reconciling`, and returns
+202 with a `/v3/jobs/:guid` Location. Poll until completion before starting or
+restarting the app. Concurrent binds share the active provisioning job. A ready
+account bind returns 200. Replacing an account requires explicit unbind first.
+Bind/unbind return restart guidance and never restart the app automatically.
+
+Provisioning registers one canonical certificate-authenticated, secretless UAA
+client and creates a roleless OAuth principal with GUID equal to the client ID.
+Client collisions are rejected rather than overwritten/adopted. Transient worker
+failures expose `failed` state and retry up to three attempts. A later authorized
+bind can retry after the previous job has terminated. Unbind while provisioning
+does not cause the worker to restore an assignment.
+
+Use `/v3/roles` with `relationships.user.data.guid` equal to the canonical client
+ID. Explicit organization membership is required before granting space roles.
+Roles belong to the shared account principal and are retained with zero apps or
+while authentication is disabled. An account token cannot identify which app used it.
+
+PATCH `enabled: false` immediately prevents new assignment/launch identity and
+queues UAA registration deletion. Enabling recreates the canonical registration
+and preserves resource roles. The local UAA lifecycle strategy uses client deletion
+because the selected UAA model has no enabled-client property. This must be verified
+against the deployed UAA configuration before claiming token-issuance denial.
+
+Deleting requires no desired app assignments, started process snapshots or active
+task snapshots. The worker rechecks usage and registration trust before removing
+the client, principal and account. The name reservation is retained permanently.
+Concurrent lifecycle operations return 409. All asynchronous operations use the
+standard job polling endpoint.
+Space/org deletion cascades owned accounts: service bindings and workloads are
+deleted before managed clients, principals/roles and accounts; name reservations
+remain. Active account operations and cleanup failures prevent final parent
+deletion and are reported by the deletion job. Retry the parent deletion after
+resolving the failure; completed cleanup is not rolled back.
+
+## Runtime and rollout
+
+Runtime identity is enabled separately from provisioning. New process launches and
+tasks capture the account at launch/creation. Later bind/unbind changes require
+restart; scaling/redriving retains the captured identity. Pre-feature launches have
+no account identity until explicit launch/restart. Staging excludes account identity
+and discovery. Non-ready, disabled, missing or wrong-space accounts fail closed.
+
+Runtime `VCAP_SERVICE_ACCOUNT` contains account GUID/name, canonical client ID/SAN
+and the configured HTTPS mTLS token endpoint, without secrets. Workloads use their
+existing instance credentials; platform-derived typed certificate properties
+preserve app/space/org organizational units. Account SANs are not route names.
+
+BOSH properties under `cc.service_accounts`:
+
+- `provisioning_enabled` (default false): enable API lifecycle/first-bind enqueueing
+  and dedicated worker provisioning configuration.
+- `runtime_enabled` (default false): enable account identity only after compatible
+  BBS, rep and cell versions are present throughout the foundation.
+- `token_endpoint`: HTTPS UAA mTLS endpoint without credentials/query/fragment.
+- Worker-only `management_client_id`, `management_client_secret`, `identity_ca`:
+  dedicated UAA management credentials and the instance identity CA PEM, separate
+  from the UAA server TLS CA. Configure namespace protection independently in UAA.
+
+App-hosted local workers may receive equivalent `service_account_provisioning`
+configuration separately; the API BOSH template does not render management secrets.
+UAA audience/issuer, protected namespace, trusted proxy profile, leaf-expiry token
+lifetime cap and the selected UAA `cnf` claim deviation require integration evidence.
+Disabling/deleting clients, unbinding and restarting do not revoke already-issued
+JWTs or certificates immediately. Certificate-only route grants are separate.
diff --git a/docs/v3/source/index.html.md b/docs/v3/source/index.html.md
index efb673f35ec..9af499384d7 100644
--- a/docs/v3/source/index.html.md
+++ b/docs/v3/source/index.html.md
@@ -29,6 +29,7 @@ includes:
   - api_resources/routes
   - api_resources/security_groups
   - api_resources/service_brokers
+  - api_resources/service_accounts
   - api_resources/service_offerings
   - api_resources/service_plans
   - api_resources/service_plan_visibility
diff --git a/errors/v3.yml b/errors/v3.yml
index 35a7cc6dd35..703a21d3b55 100644
--- a/errors/v3.yml
+++ b/errors/v3.yml
@@ -3,6 +3,21 @@
   http_code: 422
   message: "%s"
 
+10017:
+  name: ServiceAccountNameReserved
+  http_code: 409
+  message: "Service account name is already reserved"
+
+10018:
+  name: ServiceAccountAssignmentConflict
+  http_code: 409
+  message: "%s"
+
+10019:
+  name: ServiceAccountCreationLimitExceeded
+  http_code: 429
+  message: "Service account creation budget exhausted for this principal; retry when creations leave the rolling seven-day window"
+
 270010:
   name: ServiceBrokerNotRemovable
   http_code: 422
diff --git a/lib/cloud_controller/backends/staging_environment_builder.rb b/lib/cloud_controller/backends/staging_environment_builder.rb
index 9395c5d3c5c..e22c393d790 100644
--- a/lib/cloud_controller/backends/staging_environment_builder.rb
+++ b/lib/cloud_controller/backends/staging_environment_builder.rb
@@ -27,7 +27,8 @@ def build(app, space, lifecycle, memory_limit, staging_disk_in_mb, vars_from_mes
             'MEMORY_LIMIT' => "#{memory_limit}m"
           }
         ).
-        merge(SystemEnvPresenter.new(app).system_env.stringify_keys)
+        merge(SystemEnvPresenter.new(app).system_env.stringify_keys).
+        except('VCAP_SERVICE_ACCOUNT', :VCAP_SERVICE_ACCOUNT)
     end
   end
 end
diff --git a/lib/cloud_controller/config_schemas/api_schema.rb b/lib/cloud_controller/config_schemas/api_schema.rb
index 511febfebf7..a8f0849b953 100644
--- a/lib/cloud_controller/config_schemas/api_schema.rb
+++ b/lib/cloud_controller/config_schemas/api_schema.rb
@@ -30,6 +30,15 @@ class ApiSchema < VCAP::Config
             optional(:custom) => Hash
           },
           optional(:custom_root_links) => Array,
+          optional(:service_account_provisioning_enabled) => bool,
+          optional(:service_account_runtime_enabled) => bool,
+          optional(:service_account_creation_limit) => Integer,
+          optional(:service_account_token_endpoint) => String,
+          optional(:service_account_provisioning) => {
+            client_id: String,
+            client_secret: String,
+            identity_ca: String
+          },
 
           system_domain: String,
           optional(:system_domain_organization) => enum(String, NilClass),
diff --git a/lib/cloud_controller/config_schemas/clock_schema.rb b/lib/cloud_controller/config_schemas/clock_schema.rb
index 4aa96c33bc4..d3d6c3dd968 100644
--- a/lib/cloud_controller/config_schemas/clock_schema.rb
+++ b/lib/cloud_controller/config_schemas/clock_schema.rb
@@ -6,6 +6,8 @@ class ClockSchema < VCAP::Config
       # rubocop:disable Metrics/BlockLength
       define_schema do
         {
+          optional(:service_account_runtime_enabled) => bool,
+          optional(:service_account_token_endpoint) => String,
           external_port: Integer,
           external_domain: String,
           tls_port: Integer,
diff --git a/lib/cloud_controller/config_schemas/worker_schema.rb b/lib/cloud_controller/config_schemas/worker_schema.rb
index c6b99c27a67..c8803229ba8 100644
--- a/lib/cloud_controller/config_schemas/worker_schema.rb
+++ b/lib/cloud_controller/config_schemas/worker_schema.rb
@@ -29,6 +29,14 @@ class WorkerSchema < VCAP::Config
             client_timeout: Integer
           },
 
+          optional(:service_account_provisioning) => {
+            client_id: String,
+            client_secret: String,
+            identity_ca: String
+          },
+          optional(:service_account_runtime_enabled) => bool,
+          optional(:service_account_token_endpoint) => String,
+
           logging: {
             level: String, # debug, info, etc.
             file: String # Log file to use
diff --git a/lib/cloud_controller/dependency_locator.rb b/lib/cloud_controller/dependency_locator.rb
index 1f1ebf5aaaf..37c5216f3f4 100644
--- a/lib/cloud_controller/dependency_locator.rb
+++ b/lib/cloud_controller/dependency_locator.rb
@@ -26,6 +26,8 @@
 require 'cloud_controller/metrics/prometheus_updater'
 require 'statsd/instrument'
 require 'cloud_controller/execution_context'
+require 'cloud_controller/uaa/service_account_client'
+require 'actions/service_account_provision'
 
 module CloudController
   class DependencyLocator
@@ -305,6 +307,23 @@ def uaa_username_lookup_client
       )
     end
 
+    def service_account_provisioner
+      return @dependencies[:service_account_provisioner] if @dependencies[:service_account_provisioner]
+
+      settings = config.get(:service_account_provisioning)
+      unless settings.is_a?(Hash) && %i[client_id client_secret identity_ca].all? { |key| settings[key].is_a?(String) && settings[key].present? }
+        raise 'service account provisioning is not configured'
+      end
+
+      clients = ServiceAccountClient.new(
+        uaa_target: config.get(:uaa, :internal_url),
+        client_id: settings[:client_id],
+        secret: settings[:client_secret],
+        ca_file: config.get(:uaa, :ca_file)
+      )
+      register(:service_account_provisioner, ServiceAccountProvision.new(clients, identity_ca: settings[:identity_ca]))
+    end
+
     def uaa_shadow_user_creation_client
       client = config.get(:uaa, :clients)&.find { |client_config| client_config['name'] == 'cloud_controller_shadow_user_creation' }
 
diff --git a/lib/cloud_controller/diego/app_recipe_builder.rb b/lib/cloud_controller/diego/app_recipe_builder.rb
index 25a4c10866f..7688a2ae7df 100644
--- a/lib/cloud_controller/diego/app_recipe_builder.rb
+++ b/lib/cloud_controller/diego/app_recipe_builder.rb
@@ -10,6 +10,7 @@
 require 'credhub/config_helpers'
 require 'models/helpers/health_check_types'
 require 'cloud_controller/diego/main_lrp_action_builder'
+require 'cloud_controller/diego/service_account_identity'
 
 module VCAP::CloudController
   module Diego
@@ -98,9 +99,7 @@ def app_lrp_arguments
           check_definition: generate_healthcheck_definition(desired_lrp_builder),
           routes: ::Diego::Bbs::Models::ProtoRoutes.new(routes:),
           max_pids: @config.get(:diego, :pid_limit),
-          certificate_properties: ::Diego::Bbs::Models::CertificateProperties.new(
-            organizational_unit: ["organization:#{process.organization.guid}", "space:#{process.space.guid}", "app:#{process.app_guid}"]
-          ),
+          certificate_properties: ServiceAccountIdentity.new(process.app, config, account_guid: process.runtime_service_account_guid).certificate_properties,
           image_username: process.desired_droplet.docker_receipt_username,
           image_password: process.desired_droplet.docker_receipt_password,
           volume_mounted_files: ServiceBindingFilesBuilder.build(process)
diff --git a/lib/cloud_controller/diego/environment.rb b/lib/cloud_controller/diego/environment.rb
index d5789782432..d30ff19a5af 100644
--- a/lib/cloud_controller/diego/environment.rb
+++ b/lib/cloud_controller/diego/environment.rb
@@ -1,6 +1,7 @@
 require 'presenters/system_environment/system_env_presenter'
 require 'cloud_controller/diego/normal_env_hash_to_diego_env_array_philosopher'
 require_relative '../../vcap/vars_builder'
+require 'cloud_controller/diego/service_account_identity'
 
 module VCAP::CloudController
   module Diego
@@ -43,6 +44,9 @@ def common_json_and_merge(&blk)
           merge(blk.call).
           merge(SystemEnvPresenter.new(process).system_env)
 
+        diego_env = diego_env.except('VCAP_SERVICE_ACCOUNT', :VCAP_SERVICE_ACCOUNT).
+                    merge(ServiceAccountIdentity.new(process.app, Config.config, account_guid: process.runtime_service_account_guid).environment)
+
         diego_env = diego_env.merge(DATABASE_URL: process.database_uri) if process.database_uri
 
         NormalEnvHashToDiegoEnvArrayPhilosopher.muse(diego_env)
diff --git a/lib/cloud_controller/diego/service_account_identity.rb b/lib/cloud_controller/diego/service_account_identity.rb
new file mode 100644
index 00000000000..324584f52a4
--- /dev/null
+++ b/lib/cloud_controller/diego/service_account_identity.rb
@@ -0,0 +1,59 @@
+require 'uri'
+
+module VCAP::CloudController
+  module Diego
+    class ServiceAccountIdentity
+      def initialize(app, config, account_guid: app.service_account_guid)
+        @app = app
+        @config = config
+        @account_guid = account_guid
+      end
+
+      def certificate_properties
+        account = ready_account
+        attributes = { organizational_unit: ["organization:#{@app.organization_guid}", "space:#{@app.space_guid}", "app:#{@app.guid}"] }
+        attributes[:service_account] = ::Diego::Bbs::Models::ServiceAccount.new(name: account.name) if account
+        ::Diego::Bbs::Models::CertificateProperties.new(attributes)
+      end
+
+      def environment
+        account = ready_account
+        return {} unless account
+
+        endpoint = @config.get(:service_account_token_endpoint)
+        raise CloudController::Errors::ApiError.new_from_details('UnprocessableEntity', 'Service account token endpoint is not configured') unless valid_endpoint?(endpoint)
+
+        { 'VCAP_SERVICE_ACCOUNT' => {
+          guid: account.guid, name: account.name, client_id: account.client_id,
+          certificate_dns_san: account.certificate_dns_san, token_endpoint: endpoint
+        } }
+      end
+
+      private
+
+      def valid_endpoint?(endpoint)
+        return false unless endpoint.is_a?(String)
+
+        uri = URI.parse(endpoint)
+        uri.is_a?(URI::HTTPS) && uri.host.present? && uri.userinfo.nil? && uri.query.nil? && uri.fragment.nil?
+      rescue URI::InvalidURIError
+        false
+      end
+
+      def ready_account
+        return unless @account_guid
+
+        unless @config.get(:service_account_runtime_enabled) == true
+          raise CloudController::Errors::ApiError.new_from_details('UnprocessableEntity', 'Service account runtime is not enabled')
+        end
+
+        account = ServiceAccountModel.first(guid: @account_guid)
+        unless account && account.enabled && account.status == 'ready' && account.space_guid == @app.space_guid
+          raise CloudController::Errors::ApiError.new_from_details('UnprocessableEntity', 'Service account is not ready for runtime credentials')
+        end
+
+        account
+      end
+    end
+  end
+end
diff --git a/lib/cloud_controller/diego/task_environment.rb b/lib/cloud_controller/diego/task_environment.rb
index dad26816613..129f7ec0055 100644
--- a/lib/cloud_controller/diego/task_environment.rb
+++ b/lib/cloud_controller/diego/task_environment.rb
@@ -1,4 +1,5 @@
 require 'credhub/config_helpers'
+require 'cloud_controller/diego/service_account_identity'
 
 module VCAP::CloudController
   module Diego
@@ -21,6 +22,9 @@ def build
           merge('VCAP_APPLICATION' => vcap_application, 'MEMORY_LIMIT' => "#{task.memory_in_mb}m").
           merge(SystemEnvPresenter.new(app).system_env.stringify_keys)
 
+        task_env = task_env.except('VCAP_SERVICE_ACCOUNT', :VCAP_SERVICE_ACCOUNT).
+                   merge(ServiceAccountIdentity.new(app, Config.config, account_guid: task.runtime_service_account_guid).environment)
+
         task_env = task_env.merge('VCAP_PLATFORM_OPTIONS' => credhub_url) if credhub_url.present? && cred_interpolation_enabled?
 
         task_env = task_env.merge('LANG' => DEFAULT_LANG) if [BuildpackLifecycleDataModel::LIFECYCLE_TYPE, CNBLifecycleDataModel::LIFECYCLE_TYPE].include?(app.lifecycle_type)
diff --git a/lib/cloud_controller/diego/task_recipe_builder.rb b/lib/cloud_controller/diego/task_recipe_builder.rb
index f1a4962095d..f5b92869752 100644
--- a/lib/cloud_controller/diego/task_recipe_builder.rb
+++ b/lib/cloud_controller/diego/task_recipe_builder.rb
@@ -6,6 +6,7 @@
 require 'cloud_controller/diego/task_completion_callback_generator'
 require 'cloud_controller/diego/task_cpu_weight_calculator'
 require 'cloud_controller/diego/service_binding_files_builder'
+require 'cloud_controller/diego/service_account_identity'
 
 module VCAP::CloudController
   module Diego
@@ -45,13 +46,7 @@ def build_app_task(config, task)
           root_fs: task_action_builder.stack,
           environment_variables: task_action_builder.task_environment_variables,
           placement_tags: [VCAP::CloudController::IsolationSegmentSelector.for_space(task.space)].compact,
-          certificate_properties: ::Diego::Bbs::Models::CertificateProperties.new(
-            organizational_unit: [
-              "organization:#{task.app.organization.guid}",
-              "space:#{task.app.space_guid}",
-              "app:#{task.app_guid}"
-            ]
-          ),
+          certificate_properties: ServiceAccountIdentity.new(task.app, config, account_guid: task.runtime_service_account_guid).certificate_properties,
           image_username: task.droplet.docker_receipt_username,
           image_password: task.droplet.docker_receipt_password,
           volume_mounted_files: ServiceBindingFilesBuilder.build(task.app)
diff --git a/lib/cloud_controller/uaa/service_account_client.rb b/lib/cloud_controller/uaa/service_account_client.rb
new file mode 100644
index 00000000000..81646f6cc0a
--- /dev/null
+++ b/lib/cloud_controller/uaa/service_account_client.rb
@@ -0,0 +1,23 @@
+require 'cloud_controller/uaa/uaa_client'
+
+module VCAP::CloudController
+  class ServiceAccountClient < UaaClient
+    def get(type, id)
+      raise ArgumentError.new('only client resources are supported') unless type == :client
+
+      super
+    end
+
+    def add(type, registration)
+      raise ArgumentError.new('only client resources are supported') unless type == :client
+
+      with_cache_retry { scim.add(type, registration) }
+    end
+
+    def delete(type, id)
+      raise ArgumentError.new('only client resources are supported') unless type == :client
+
+      with_cache_retry { scim.delete(type, id) }
+    end
+  end
+end
diff --git a/lib/diego/bbs/models/certificate_properties_pb.rb b/lib/diego/bbs/models/certificate_properties_pb.rb
index 003d022b902..16355dfbcd3 100644
--- a/lib/diego/bbs/models/certificate_properties_pb.rb
+++ b/lib/diego/bbs/models/certificate_properties_pb.rb
@@ -5,15 +5,16 @@
 require 'google/protobuf'
 
 
-descriptor_data = "\n\x1c\x63\x65rtificate_properties.proto\x12\x10\x64iego.bbs.models\"4\n\x15\x43\x65rtificateProperties\x12\x1b\n\x13organizational_unit\x18\x01 \x03(\tb\x06proto3"
+descriptor_data = "\n\x1c\x63\x65rtificate_properties.proto\x12\x10\x64iego.bbs.models\"o\n\x15\x43\x65rtificateProperties\x12\x1b\n\x13organizational_unit\x18\x01 \x03(\t\x12\x39\n\x0fservice_account\x18\x02 \x01(\x0b\x32 .diego.bbs.models.ServiceAccount\"\x1e\n\x0eServiceAccount\x12\x0c\n\x04name\x18\x01 \x01(\tb\x06proto3"
 
-pool = Google::Protobuf::DescriptorPool.generated_pool
+pool = ::Google::Protobuf::DescriptorPool.generated_pool
 pool.add_serialized_file(descriptor_data)
 
 module Diego
   module Bbs
     module Models
       CertificateProperties = ::Google::Protobuf::DescriptorPool.generated_pool.lookup("diego.bbs.models.CertificateProperties").msgclass
+      ServiceAccount = ::Google::Protobuf::DescriptorPool.generated_pool.lookup("diego.bbs.models.ServiceAccount").msgclass
     end
   end
 end
diff --git a/spec/request/app_service_accounts_spec.rb b/spec/request/app_service_accounts_spec.rb
new file mode 100644
index 00000000000..0f9ce6fdd61
--- /dev/null
+++ b/spec/request/app_service_accounts_spec.rb
@@ -0,0 +1,202 @@
+require 'spec_helper'
+
+RSpec.describe 'App service account relationship' do
+  let(:space) { create(:space) }
+  let(:user) { create(:user) }
+  let(:app_model) { create(:app_model, space: space, desired_state: 'STARTED') }
+  let(:account) { VCAP::CloudController::ServiceAccountModel.create(name: 'payments-worker', space: space, status: 'ready') }
+  let(:path) { "/v3/apps/#{app_model.guid}/relationships/service_account" }
+
+  def headers(role)
+    set_user_with_header_as_role(role: role, org: space.organization, space: space, user: user)
+  end
+
+  it 'lets a developer bind a ready same-space account without restarting the app' do
+    patch path, { data: { guid: account.guid } }.to_json, headers('space_developer')
+    expect(last_response.status).to eq(200)
+    expect(Oj.load(last_response.body)['data']).to eq('guid' => account.guid)
+    expect(last_response.headers['X-Cf-Warnings']).to include('Restart')
+    expect(app_model.reload.service_account).to eq(account)
+    expect(app_model.desired_state).to eq('STARTED')
+    event = VCAP::CloudController::Event.first(type: 'audit.app.service_account.assign', actee: app_model.guid)
+    expect(event).not_to be_nil
+    expect(event.metadata['service_account_guid']).to eq(account.guid)
+  end
+
+  it 'shows an empty relationship for an unbound app' do
+    get path, nil, headers('space_auditor')
+    expect(last_response.status).to eq(200)
+    expect(Oj.load(last_response.body)['data']).to be_nil
+  end
+
+  it 'denies an auditor assignment permission' do
+    patch path, { data: { guid: account.guid } }.to_json, headers('space_auditor')
+    expect(last_response.status).to eq(403)
+    expect(app_model.reload.service_account).to be_nil
+  end
+
+  it 'denies a cross-space account even for an administrator' do
+    other = VCAP::CloudController::ServiceAccountModel.create(name: 'reporting-reader', space: create(:space, organization: space.organization), status: 'ready')
+    patch path, { data: { guid: other.guid } }.to_json, headers('admin')
+    expect(last_response.status).to eq(409)
+    expect(app_model.reload.service_account).to be_nil
+  end
+
+  it 'requires unbinding before replacing an account' do
+    app_model.update(service_account: account)
+    other = VCAP::CloudController::ServiceAccountModel.create(name: 'reporting-reader', space: space, status: 'ready')
+    patch path, { data: { guid: other.guid } }.to_json, headers('space_developer')
+    expect(last_response.status).to eq(409)
+    expect(app_model.reload.service_account).to eq(account)
+  end
+
+  it 'supports explicit unbinding and returns restart guidance' do
+    app_model.update(service_account: account)
+    patch path, { data: nil }.to_json, headers('space_developer')
+    expect(last_response.status).to eq(200)
+    expect(app_model.reload.service_account).to be_nil
+    expect(last_response.headers['X-Cf-Warnings']).to include('Restart')
+    expect(VCAP::CloudController::Event.where(type: 'audit.app.service_account.unassign', actee: app_model.guid).count).to eq(1)
+  end
+
+  it 'rejects assignment while the space is suspended' do
+    space.update(status: VCAP::CloudController::Space::SUSPENDED)
+    patch path, { data: { guid: account.guid } }.to_json, headers('space_developer')
+    expect(last_response.status).to eq(403)
+  end
+
+  it 'rejects unprovisioned accounts rather than claiming successful binding' do
+    account.update(status: 'reserved')
+    patch path, { data: { guid: account.guid } }.to_json, headers('space_developer')
+    expect(last_response.status).to eq(409)
+  end
+
+  context 'when asynchronous provisioning is enabled' do
+    let(:clients) { instance_double(CF::UAA::Scim) }
+
+    before do
+      TestConfig.override(service_account_provisioning_enabled: true)
+      CloudController::DependencyLocator.instance.register(
+        :service_account_provisioner,
+        VCAP::CloudController::ServiceAccountProvision.new(clients, identity_ca: 'identity-ca')
+      )
+      account.update(status: 'reserved')
+      allow(clients).to receive(:get).and_raise(CF::UAA::NotFound)
+      allow(clients).to receive(:add)
+    end
+
+    it 'accepts desired binding and exposes one pollable provisioning job without restarting' do
+      patch path, { data: { guid: account.guid } }.to_json, headers('space_developer')
+      expect(last_response.status).to eq(202)
+      location = last_response.headers['Location']
+      expect(location).to include('/v3/jobs/')
+      expect(app_model.reload.service_account_guid).to eq(account.guid)
+      expect(app_model.desired_state).to eq('STARTED')
+      expect(account.reload.status).to eq('reconciling')
+      expect(VCAP::CloudController::User.first(guid: account.client_id)).to be_nil
+      expect(clients).not_to have_received(:add)
+
+      get URI(location).path, nil, headers('space_developer')
+      expect(last_response.status).to eq(200)
+      expect(Oj.load(last_response.body)).to include('state' => 'PROCESSING', 'operation' => 'service_account.provision')
+      expect(Delayed::Worker.new.work_off).to eq([1, 0])
+      expect(account.reload.status).to eq('ready')
+      get URI(location).path, nil, headers('space_developer')
+      expect(Oj.load(last_response.body)['state']).to eq('COMPLETE')
+      expect(VCAP::CloudController::User.first(guid: account.client_id).spaces).to be_empty
+    end
+
+    it 'reuses the active job for repeated binding and another app sharing the account' do
+      auth = headers('space_developer')
+      patch path, { data: { guid: account.guid } }.to_json, auth
+      location = last_response.headers['Location']
+      patch path, { data: { guid: account.guid } }.to_json, auth
+      expect(last_response.status).to eq(202)
+      expect(last_response.headers['Location']).to eq(location)
+      other_app = create(:app_model, space: space)
+      patch "/v3/apps/#{other_app.guid}/relationships/service_account", { data: { guid: account.guid } }.to_json, auth
+      expect(last_response.headers['Location']).to eq(location)
+      expect(Delayed::Job.count).to eq(1)
+    end
+
+    it 'allows unbinding while queued and does not restore the assignment when provisioning finishes' do
+      auth = headers('space_developer')
+      patch path, { data: { guid: account.guid } }.to_json, auth
+      patch path, { data: nil }.to_json, auth
+      expect(last_response.status).to eq(200)
+      Delayed::Worker.new.work_off
+      expect(app_model.reload.service_account_guid).to be_nil
+      expect(account.reload.status).to eq('ready')
+    end
+
+    %w[space_auditor space_manager].each do |role|
+      it "does not queue provisioning for #{role}" do
+        patch path, { data: { guid: account.guid } }.to_json, headers(role)
+        expect(last_response.status).to eq(403)
+        expect(Delayed::Job.count).to eq(0)
+        expect(account.reload.status).to eq('reserved')
+      end
+    end
+
+    it 'does not queue provisioning for a disabled account' do
+      account.update(enabled: false)
+      patch path, { data: { guid: account.guid } }.to_json, headers('space_developer')
+      expect(last_response.status).to eq(409)
+      expect(Delayed::Job.count).to eq(0)
+    end
+
+    it 'does not queue a cross-space account even for an administrator' do
+      other = VCAP::CloudController::ServiceAccountModel.create(name: 'other-worker', space: create(:space))
+      patch path, { data: { guid: other.guid } }.to_json, headers('admin')
+      expect(last_response.status).to eq(409)
+      expect(Delayed::Job.count).to eq(0)
+    end
+
+    it 'does not bind an account while a delete operation is retrying' do
+      account.update(status: 'failed')
+      VCAP::CloudController::PollableJobModel.create(
+        delayed_job_guid: SecureRandom.uuid, operation: 'service_account.delete',
+        resource_guid: account.guid, resource_type: 'service_account', state: 'PROCESSING'
+      )
+      patch path, { data: { guid: account.guid } }.to_json, headers('space_developer')
+      expect(last_response.status).to eq(409)
+      expect(app_model.reload.service_account_guid).to be_nil
+    end
+
+    it 'rolls back desired binding and state if enqueueing fails' do
+      error = CloudController::Errors::ApiError.new_from_details('ServerError')
+      allow_any_instance_of(VCAP::CloudController::Jobs::Enqueuer).to receive(:enqueue_pollable).and_raise(error)
+      patch path, { data: { guid: account.guid } }.to_json, headers('space_developer')
+      expect(last_response.status).to eq(500)
+      expect(app_model.reload.service_account_guid).to be_nil
+      expect(account.reload.status).to eq('reserved')
+    end
+
+    it 'retries a failed bind through the same pollable job and then reports completion' do
+      allow(clients).to receive(:add).and_raise(CF::UAA::BadTarget, 'unavailable')
+      patch path, { data: { guid: account.guid } }.to_json, headers('space_developer')
+      expect(last_response.status).to eq(202)
+      location = last_response.headers['Location']
+      expect(Delayed::Worker.new.work_off).to eq([0, 1])
+      expect(account.reload.status).to eq('failed')
+      allow(clients).to receive(:add)
+      Delayed::Job.first.update(run_at: Time.now.utc - 1)
+      expect(Delayed::Worker.new.work_off).to eq([1, 0])
+      get URI(location).path, nil, headers('space_developer')
+      expect(Oj.load(last_response.body)['state']).to eq('COMPLETE')
+    end
+  end
+
+  [{}, { data: {} }, { data: { guid: nil } }, { data: { guid: 'x', name: 'injected' } }, { data: [] }].each do |body|
+    it "rejects malformed relationship body #{body.inspect}" do
+      patch path, body.to_json, headers('space_developer')
+      expect(last_response.status).to eq(422)
+    end
+  end
+
+  it 'hides apps outside the caller space' do
+    other_app = create(:app_model, space: create(:space, organization: space.organization))
+    get "/v3/apps/#{other_app.guid}/relationships/service_account", nil, headers('space_developer')
+    expect(last_response.status).to eq(404)
+  end
+end
diff --git a/spec/request/service_account_roles_spec.rb b/spec/request/service_account_roles_spec.rb
new file mode 100644
index 00000000000..37485b203a3
--- /dev/null
+++ b/spec/request/service_account_roles_spec.rb
@@ -0,0 +1,52 @@
+require 'spec_helper'
+
+RSpec.describe 'Explicit service account resource roles' do
+  let(:space) { create(:space) }
+  let(:admin) { create(:user) }
+  let(:account) { VCAP::CloudController::ServiceAccountModel.create(name: 'payments-worker', space: space) }
+  let(:clients) { instance_double(CF::UAA::Scim) }
+  let(:principal) { VCAP::CloudController::User.first(guid: account.client_id) }
+
+  before do
+    allow(clients).to receive(:get).and_raise(CF::UAA::NotFound)
+    allow(clients).to receive(:add)
+    VCAP::CloudController::ServiceAccountProvision.new(clients, identity_ca: 'identity-ca').provision(account)
+    lookup = instance_double(VCAP::CloudController::UaaClient, usernames_for_ids: {})
+    expect(lookup).not_to receive(:usernames_for_ids).with([principal.guid])
+    allow(CloudController::DependencyLocator.instance).to receive(:uaa_username_lookup_client).and_return(lookup)
+  end
+
+  def grant(type, resource, guid)
+    post '/v3/roles', {
+      type: type,
+      relationships: { user: { data: { guid: principal.guid } }, resource => { data: { guid: guid } } }
+    }.to_json, admin_headers_for(admin)
+  end
+
+  it 'requires explicit organization membership and then a space role before account tokens can write apps' do
+    grant('space_developer', :space, space.guid)
+    expect(last_response.status).to eq(422)
+    grant('organization_user', :organization, space.organization.guid)
+    expect(last_response.status).to eq(201)
+
+    app_body = { name: 'account-created', relationships: { space: { data: { guid: space.guid } } } }
+    post '/v3/apps', app_body.to_json, headers_for(principal, client: true)
+    expect(last_response.status).to eq(422)
+
+    grant('space_developer', :space, space.guid)
+    expect(last_response.status).to eq(201)
+    post '/v3/apps', app_body.to_json, headers_for(principal, client: true)
+    expect(last_response.status).to eq(201)
+
+    post '/v3/service_accounts', { name: 'privilege-escalation', relationships: { space: { data: { guid: space.guid } } } }.to_json, headers_for(principal, client: true)
+    expect(last_response.status).to eq(403)
+  end
+
+  it 'keeps account tokens denied in unrelated spaces after an explicit role grant' do
+    grant('organization_user', :organization, space.organization.guid)
+    grant('space_developer', :space, space.guid)
+    other = create(:space, organization: space.organization)
+    post '/v3/apps', { name: 'unauthorized', relationships: { space: { data: { guid: other.guid } } } }.to_json, headers_for(principal, client: true)
+    expect(last_response.status).to eq(422)
+  end
+end
diff --git a/spec/request/service_accounts_spec.rb b/spec/request/service_accounts_spec.rb
new file mode 100644
index 00000000000..78dc6e041b5
--- /dev/null
+++ b/spec/request/service_accounts_spec.rb
@@ -0,0 +1,346 @@
+require 'spec_helper'
+
+RSpec.describe 'Service accounts' do
+  let(:org) { create(:organization) }
+  let(:space) { create(:space, organization: org) }
+  let(:user) { create(:user) }
+  let(:body) { { name: 'payments-worker', relationships: { space: { data: { guid: space.guid } } } } }
+
+  def headers(role)
+    set_user_with_header_as_role(role: role, org: org, space: space, user: user)
+  end
+
+  it 'allows a space manager to reserve an identity without provisioning a client' do
+    post '/v3/service_accounts', body.to_json, headers('space_manager')
+
+    expect(last_response.status).to eq(201)
+    result = Oj.load(last_response.body)
+    expect(result).to include('name' => 'payments-worker', 'status' => 'reserved', 'enabled' => true,
+                              'client_id' => 'cf:service-account:payments-worker', 'certificate_dns_san' => 'payments-worker.svc.identity')
+    expect(result.dig('relationships', 'space', 'data', 'guid')).to eq(space.guid)
+    expect(VCAP::CloudController::ServiceAccountModel.where(guid: result['guid']).first.space_guid).to eq(space.guid)
+    event = VCAP::CloudController::Event.first(type: 'audit.service_account.create', actee: result['guid'])
+    expect(event).not_to be_nil
+    expect(event.actor).to eq(user.guid)
+    expect(event.space_guid).to eq(space.guid)
+  end
+
+  it 'allows a platform administrator to create an account' do
+    post '/v3/service_accounts', body.to_json, headers('admin')
+    expect(last_response.status).to eq(201)
+  end
+
+  context 'weekly creation budget' do
+    before { TestConfig.override(service_account_creation_limit: 1) }
+
+    it 'limits a principal across spaces without refunding deleted accounts' do
+      auth = headers('space_manager')
+      post '/v3/service_accounts', body.to_json, auth
+      expect(last_response.status).to eq(201)
+      VCAP::CloudController::ServiceAccountModel.first.destroy
+      other_space = create(:space, organization: org)
+      other_space.add_manager(user)
+      other_body = { name: 'other-worker', relationships: { space: { data: { guid: other_space.guid } } } }
+
+      post '/v3/service_accounts', other_body.to_json, auth
+
+      expect(last_response.status).to eq(429)
+      expect(Oj.load(last_response.body).dig('errors', 0, 'title')).to eq('CF-ServiceAccountCreationLimitExceeded')
+      expect(VCAP::CloudController::ServiceAccountModel.count).to eq(0)
+    end
+
+    it 'uses a rolling seven-day window' do
+      auth = headers('space_manager')
+      Timecop.freeze(Time.utc(2026, 10, 1)) do
+        post '/v3/service_accounts', body.to_json, auth
+        expect(last_response.status).to eq(201)
+      end
+      Timecop.freeze(Time.utc(2026, 10, 7, 23, 59, 59)) do
+        post '/v3/service_accounts', body.merge(name: 'second-worker').to_json, auth
+        expect(last_response.status).to eq(429)
+      end
+      Timecop.freeze(Time.utc(2026, 10, 8)) do
+        post '/v3/service_accounts', body.merge(name: 'second-worker').to_json, auth
+        expect(last_response.status).to eq(201)
+      end
+    end
+
+    it 'does not consume budget on a failed creation' do
+      auth = headers('space_manager')
+      VCAP::CloudController::ServiceAccountModel.create(name: body[:name], space: space)
+      post '/v3/service_accounts', body.to_json, auth
+      expect(last_response.status).to eq(409)
+      post '/v3/service_accounts', body.merge(name: 'second-worker').to_json, auth
+      expect(last_response.status).to eq(201)
+      post '/v3/service_accounts', body.merge(name: 'third-worker').to_json, auth
+      expect(last_response.status).to eq(429)
+    end
+
+    it 'does not share budget between principals' do
+      post '/v3/service_accounts', body.to_json, headers('space_manager')
+      expect(last_response.status).to eq(201)
+      other_user = create(:user)
+      auth = set_user_with_header_as_role(role: 'space_manager', org: org, space: space, user: other_user)
+      post '/v3/service_accounts', body.merge(name: 'second-worker').to_json, auth
+      expect(last_response.status).to eq(201)
+    end
+
+    it 'supports explicit unlimited configuration' do
+      TestConfig.override(service_account_creation_limit: -1)
+      auth = headers('space_manager')
+      %w[first-worker second-worker].each do |name|
+        post '/v3/service_accounts', body.merge(name: name).to_json, auth
+        expect(last_response.status).to eq(201)
+      end
+    end
+
+    it 'exempts platform admins even when the configured budget is zero' do
+      TestConfig.override(service_account_creation_limit: 0)
+      auth = headers('admin')
+      %w[first-worker second-worker].each do |name|
+        post '/v3/service_accounts', body.merge(name: name).to_json, auth
+        expect(last_response.status).to eq(201)
+      end
+      post '/v3/service_accounts', body.to_json, headers('space_manager')
+      expect(last_response.status).to eq(429)
+    end
+
+    it 'applies to non-admin automation clients and exempts admin automation' do
+      user.update(is_oauth_client: true)
+      org.add_user(user)
+      space.add_manager(user)
+      coder = CF::UAA::TokenCoder.new(audience_ids: TestConfig.config[:uaa][:resource_id], skey: TestConfig.config[:uaa][:symmetric_secret], pkey: nil)
+      token = { client_id: user.guid, scope: %w[cloud_controller.read cloud_controller.write], jti: 'automation', iss: UAAIssuer::ISSUER }
+      auth = base_json_headers('HTTP_AUTHORIZATION' => "bearer #{coder.encode(token)}")
+      post '/v3/service_accounts', body.to_json, auth
+      expect(last_response.status).to eq(201)
+      post '/v3/service_accounts', body.merge(name: 'second-worker').to_json, auth
+      expect(last_response.status).to eq(429)
+      token[:scope] << 'cloud_controller.admin'
+      admin_auth = base_json_headers('HTTP_AUTHORIZATION' => "bearer #{coder.encode(token)}")
+      post '/v3/service_accounts', body.merge(name: 'second-worker').to_json, admin_auth
+      expect(last_response.status).to eq(201)
+    end
+  end
+
+  it 'denies a space developer account-creation authority' do
+    post '/v3/service_accounts', body.to_json, headers('space_developer')
+    expect(last_response.status).to eq(403)
+    expect(VCAP::CloudController::ServiceAccountModel.count).to eq(0)
+  end
+
+  it 'rejects invalid names without normalizing them' do
+    post '/v3/service_accounts', body.merge(name: 'Payments').to_json, headers('admin')
+    expect(last_response.status).to eq(422)
+  end
+
+  it 'rejects caller-supplied identity and provisioning fields' do
+    post '/v3/service_accounts', body.merge(client_id: 'attacker', status: 'ready').to_json, headers('admin')
+    expect(last_response.status).to eq(422)
+  end
+
+  it 'rejects malformed relationship bodies' do
+    post '/v3/service_accounts', { name: 'payments-worker', relationships: nil }.to_json, headers('admin')
+    expect(last_response.status).to eq(422)
+  end
+
+  it 'returns conflict for a permanently reserved name' do
+    account = VCAP::CloudController::ServiceAccountModel.create(name: 'payments-worker', space: space)
+    account.destroy
+    post '/v3/service_accounts', body.to_json, headers('admin')
+    expect(last_response.status).to eq(409)
+  end
+
+  it 'denies creation in a suspended organization' do
+    org.update(status: 'suspended')
+    post '/v3/service_accounts', body.to_json, headers('space_manager')
+    expect(last_response.status).to eq(403)
+  end
+
+  it 'allows an owning-space member to read the account' do
+    account = VCAP::CloudController::ServiceAccountModel.create(name: 'payments-worker', space: space)
+    get "/v3/service_accounts/#{account.guid}", nil, headers('space_developer')
+
+    expect(last_response.status).to eq(200)
+    expect(Oj.load(last_response.body)['guid']).to eq(account.guid)
+  end
+
+  it 'hides accounts from developers in another space of the same organization' do
+    account = VCAP::CloudController::ServiceAccountModel.create(name: 'payments-worker', space: create(:space, organization: org))
+    get "/v3/service_accounts/#{account.guid}", nil, headers('space_developer')
+    expect(last_response.status).to eq(404)
+  end
+
+  it 'rejects unauthenticated access' do
+    post '/v3/service_accounts', body.to_json, base_json_headers
+    expect(last_response.status).to eq(401)
+  end
+
+  it 'rejects an organization ownership relationship' do
+    post '/v3/service_accounts', { name: 'payments-worker', relationships: { organization: { data: { guid: org.guid } } } }.to_json, headers('admin')
+    expect(last_response.status).to eq(422)
+  end
+
+  it 'denies creation in a suspended space' do
+    space.update(status: VCAP::CloudController::Space::SUSPENDED)
+    post '/v3/service_accounts', body.to_json, headers('space_manager')
+    expect(last_response.status).to eq(403)
+  end
+
+  it 'creates and updates description and metadata using normal merge/remove semantics' do
+    auth = headers('space_manager')
+    post '/v3/service_accounts', body.merge(description: 'Batch payments', metadata: { labels: { team: 'payments' }, annotations: { note: 'original' } }).to_json, auth
+    expect(last_response.status).to eq(201)
+    result = Oj.load(last_response.body)
+    expect(result['description']).to eq('Batch payments')
+    patch "/v3/service_accounts/#{result['guid']}", { description: 'Revised', metadata: { labels: { team: nil, owner: 'finance' } } }.to_json, auth
+    expect(last_response.status).to eq(200)
+    result = Oj.load(last_response.body)
+    expect(result['description']).to eq('Revised')
+    expect(result['metadata']).to eq('labels' => { 'owner' => 'finance' }, 'annotations' => { 'note' => 'original' })
+    expect(VCAP::CloudController::Event.where(type: 'audit.service_account.update', actee: result['guid']).count).to eq(1)
+  end
+
+  it 'lists only readable accounts with pagination and space/name filters' do
+    own = VCAP::CloudController::ServiceAccountModel.create(name: 'payments-worker', space: space)
+    VCAP::CloudController::ServiceAccountModel.create(name: 'hidden-worker', space: create(:space))
+    get '/v3/service_accounts?per_page=1&names=payments-worker', nil, headers('space_developer')
+    expect(last_response.status).to eq(200)
+    result = Oj.load(last_response.body)
+    expect(result.dig('pagination', 'total_results')).to eq(1)
+    expect(result['resources'].pluck('guid')).to eq([own.guid])
+    get "/v3/service_accounts?space_guids=#{space.guid}", nil, headers('admin')
+    expect(Oj.load(last_response.body)['resources'].pluck('guid')).to eq([own.guid])
+  end
+
+  it 'lists assigned apps only for account readers' do
+    own = VCAP::CloudController::ServiceAccountModel.create(name: 'payments-worker', space: space)
+    app = create(:app_model, space: space, service_account: own)
+    get "/v3/service_accounts/#{own.guid}/apps", nil, headers('space_auditor')
+    expect(last_response.status).to eq(200)
+    expect(Oj.load(last_response.body)['resources'].pluck('guid')).to eq([app.guid])
+    get "/v3/service_accounts/#{own.guid}/apps?names=missing-app", nil, headers('space_auditor')
+    expect(Oj.load(last_response.body)['resources']).to eq([])
+  end
+
+  it 'denies developers updates and hides unreadable resources' do
+    own = VCAP::CloudController::ServiceAccountModel.create(name: 'payments-worker', space: space)
+    patch "/v3/service_accounts/#{own.guid}", { description: 'unauthorized' }.to_json, headers('space_developer')
+    expect(last_response.status).to eq(403)
+    hidden = VCAP::CloudController::ServiceAccountModel.create(name: 'hidden-worker', space: create(:space))
+    patch "/v3/service_accounts/#{hidden.guid}", { description: 'unauthorized' }.to_json, headers('space_manager')
+    expect(last_response.status).to eq(404)
+  end
+
+  %i[name relationships status client_id certificate_dns_san].each do |key|
+    it "rejects mutation of platform-owned #{key}" do
+      own = VCAP::CloudController::ServiceAccountModel.create(name: 'payments-worker', space: space)
+      patch "/v3/service_accounts/#{own.guid}", { key => 'injected' }.to_json, headers('space_manager')
+      expect(last_response.status).to eq(422)
+    end
+  end
+
+  it 'rejects invalid metadata and list parameters' do
+    post '/v3/service_accounts', body.merge(metadata: { labels: { 'invalid/key/key' => 'value' } }).to_json, headers('admin')
+    expect(last_response.status).to eq(422)
+    get '/v3/service_accounts?unknown=value', nil, headers('admin')
+    expect(last_response.status).to eq(400)
+  end
+
+  context 'account lifecycle' do
+    let(:account) { VCAP::CloudController::ServiceAccountModel.create(name: 'payments-worker', space: space) }
+    let(:clients) { instance_double(CF::UAA::Scim) }
+    let(:provisioner) { VCAP::CloudController::ServiceAccountProvision.new(clients, identity_ca: 'identity-ca') }
+    let(:account_path) { "/v3/service_accounts/#{account.guid}" }
+
+    before do
+      TestConfig.override(service_account_provisioning_enabled: true)
+      CloudController::DependencyLocator.instance.register(:service_account_provisioner, provisioner)
+      allow(clients).to receive(:get).and_raise(CF::UAA::NotFound)
+      allow(clients).to receive(:add)
+      allow(clients).to receive(:delete)
+    end
+
+    it 'disables new authentication and enables again while retaining explicit principal roles' do
+      provisioner.provision(account)
+      principal = VCAP::CloudController::User.first(guid: account.client_id)
+      org.add_user(principal)
+      space.add_developer(principal)
+      registration = provisioner.send(:registration, account)
+      allow(clients).to receive(:get).and_return(registration)
+      auth = headers('space_manager')
+
+      patch account_path, { enabled: false }.to_json, auth
+      expect(last_response.status).to eq(202)
+      expect(account.reload.enabled).to be(false)
+      expect(Delayed::Worker.new.work_off).to eq([1, 0])
+      expect(clients).to have_received(:delete).with(:client, account.client_id).once
+      expect(account.reload.status).to eq('disabled')
+      expect(principal.reload.spaces).to include(space)
+
+      allow(clients).to receive(:get).and_raise(CF::UAA::NotFound)
+      patch account_path, { enabled: true }.to_json, auth
+      expect(last_response.status).to eq(202)
+      expect(Delayed::Worker.new.work_off).to eq([1, 0])
+      expect(account.reload.status).to eq('ready')
+      expect(principal.reload.spaces).to include(space)
+    end
+
+    it 'deletes an unused account asynchronously and keeps its name permanently reserved' do
+      provisioner.provision(account)
+      allow(clients).to receive(:get).and_return(provisioner.send(:registration, account))
+      guid = account.guid
+      auth = headers('space_manager')
+      delete account_path, nil, auth
+      expect(last_response.status).to eq(202)
+      expect(account.reload.status).to eq('deleting')
+      expect(account.enabled).to be(false)
+      expect(Delayed::Worker.new.work_off).to eq([1, 0])
+      expect(VCAP::CloudController::ServiceAccountModel.first(guid: guid)).to be_nil
+      expect(VCAP::CloudController::User.first(guid: account.client_id)).to be_nil
+      post '/v3/service_accounts', body.to_json, auth
+      expect(last_response.status).to eq(409)
+    end
+
+    it 'rejects deleting an account still assigned to an app' do
+      create(:app_model, space: space, service_account: account)
+      delete account_path, nil, headers('space_manager')
+      expect(last_response.status).to eq(409)
+      expect(account.reload.status).to eq('reserved')
+      expect(Delayed::Job.count).to eq(0)
+    end
+
+    it 'denies developer deletion and rejects non-boolean enabled values' do
+      delete account_path, nil, headers('space_developer')
+      expect(last_response.status).to eq(403)
+      patch account_path, { enabled: 'false' }.to_json, headers('space_manager')
+      expect(last_response.status).to eq(422)
+    end
+
+    it 'refuses to delete an unmanaged colliding client' do
+      allow(clients).to receive(:get).and_return('client_id' => account.client_id)
+      delete account_path, nil, headers('space_manager')
+      expect(last_response.status).to eq(202)
+      expect(Delayed::Worker.new.work_off).to eq([0, 1])
+      expect(clients).not_to have_received(:delete)
+      expect(account.reload.status).to eq('failed')
+    end
+
+    it 'rejects deletion while an unbound app still has the account in a running launch snapshot' do
+      app = create(:app_model, space: space)
+      process = create(:process_model, app: app, state: 'STARTED')
+      process.update(service_account_guid: account.guid, service_account_snapshot: true)
+      delete account_path, nil, headers('space_manager')
+      expect(last_response.status).to eq(409)
+      expect(Delayed::Job.count).to eq(0)
+    end
+
+    it 'rejects deletion while an active task holds the account snapshot' do
+      app = create(:app_model, space: space)
+      create(:task_model, app: app, service_account_guid: account.guid, service_account_snapshot: true, state: 'RUNNING')
+      delete account_path, nil, headers('space_manager')
+      expect(last_response.status).to eq(409)
+      expect(Delayed::Job.count).to eq(0)
+    end
+  end
+end
diff --git a/spec/unit/actions/app_assign_service_account_spec.rb b/spec/unit/actions/app_assign_service_account_spec.rb
new file mode 100644
index 00000000000..e2e91617322
--- /dev/null
+++ b/spec/unit/actions/app_assign_service_account_spec.rb
@@ -0,0 +1,69 @@
+require 'spec_helper'
+require 'actions/app_assign_service_account' if File.exist?('app/actions/app_assign_service_account.rb')
+
+module VCAP::CloudController
+  RSpec.describe 'Same-space service account assignment' do
+    let(:space) { create(:space) }
+    let(:app) { create(:app_model, space: space) }
+    let(:account) { VCAP::CloudController.const_get(:ServiceAccountModel).create(name: 'payments-worker', space: space, status: 'ready') }
+    let(:permissions) { instance_double(Permissions, can_write_to_active_space?: true) }
+    let(:action) { VCAP::CloudController.const_get(:AppAssignServiceAccount).new(permissions) }
+
+    it 'allows an app writer to use an account in the owning space without a use grant' do
+      action.assign(app, account)
+      expect(app.reload.service_account).to eq(account)
+      expect(app.desired_state).to eq('STOPPED')
+    end
+
+    it 'denies callers without app-write permission' do
+      allow(permissions).to receive(:can_write_to_active_space?).and_return(false)
+      expect { action.assign(app, account) }.to raise_error(AppAssignServiceAccount::Unauthorized)
+      expect(app.reload.service_account).to be_nil
+    end
+
+    it 'rejects an account in another space of the same organization' do
+      other = ServiceAccountModel.create(name: 'reporting-reader', space: create(:space, organization: space.organization), status: 'ready')
+      expect { action.assign(app, other) }.to raise_error(AppAssignServiceAccount::Conflict, /owning space/)
+    end
+
+    it 'is idempotent but requires explicit unbinding before replacement' do
+      action.assign(app, account)
+      action.assign(app, account)
+      other = ServiceAccountModel.create(name: 'reporting-reader', space: space, status: 'ready')
+      expect { action.assign(app, other) }.to raise_error(AppAssignServiceAccount::Conflict, /already assigned/)
+      expect(app.reload.service_account).to eq(account)
+    end
+
+    it 'allows an app writer to unbind' do
+      app.update(service_account: account)
+      action.assign(app, nil)
+      expect(app.reload.service_account).to be_nil
+    end
+
+    it 'rejects disabled and unprovisioned accounts' do
+      account.update(enabled: false)
+      expect { action.assign(app, account) }.to raise_error(AppAssignServiceAccount::Conflict, /not ready/)
+      account.update(enabled: true, status: 'reserved')
+      expect { action.assign(app, account) }.to raise_error(AppAssignServiceAccount::Conflict, /not ready/)
+    end
+
+    it 'serializes concurrent first binds to two apps into a single provisioning job', isolation: :truncation do
+      account.update(status: 'reserved')
+      other_app = create(:app_model, space: space)
+      ids = [app.guid, other_app.guid]
+      writer = create(:user)
+      space.organization.add_user(writer)
+      space.add_developer(writer)
+      permission_queryer = Permissions.new(writer)
+      threads = ids.map do |guid|
+        Thread.new do
+          AppAssignServiceAccount.new(permission_queryer).assign(AppModel.first(guid: guid), ServiceAccountModel.first(guid: account.guid), provision: true)
+        end
+      end
+      jobs = threads.map(&:value)
+      expect(jobs.map(&:guid).uniq.size).to eq(1)
+      expect(PollableJobModel.where(resource_guid: account.guid).count).to eq(1)
+      expect(AppModel.where(guid: ids).select_map(:service_account_guid)).to eq([account.guid, account.guid])
+    end
+  end
+end
diff --git a/spec/unit/actions/organization_delete_spec.rb b/spec/unit/actions/organization_delete_spec.rb
index 3d6329a71e8..e70968fc7b4 100644
--- a/spec/unit/actions/organization_delete_spec.rb
+++ b/spec/unit/actions/organization_delete_spec.rb
@@ -73,6 +73,38 @@ module VCAP::CloudController
       end
 
       describe 'recursive deletion' do
+        context 'service accounts' do
+          let!(:account) { ServiceAccountModel.create(name: 'org-worker', space: space) }
+          let(:clients) { instance_double(CF::UAA::Scim) }
+          let(:provisioner) { ServiceAccountProvision.new(clients, identity_ca: 'identity-ca') }
+
+          before do
+            TestConfig.override(service_account_provisioning_enabled: true)
+            CloudController::DependencyLocator.instance.register(:service_account_provisioner, provisioner)
+            allow(clients).to receive(:get).and_raise(CF::UAA::NotFound)
+            allow(clients).to receive(:delete)
+          end
+
+          it 'cascades account deletion through spaces and retains the name tombstone' do
+            app.update(service_account: account)
+            expect(org_delete.delete(org_dataset)).to be_empty
+            expect(ServiceAccountModel.first(guid: account.guid)).to be_nil
+            expect(Organization.first(guid: org_1.guid)).to be_nil
+            expect(ServiceAccountModel.db[:service_account_names].where(name: account.name).count).to eq(1)
+          end
+
+          it 'keeps the organization until account cleanup succeeds' do
+            allow(clients).to receive(:get).and_raise('UAA unavailable')
+            errors = org_delete.delete(org_dataset)
+            expect(errors.map(&:message).join).to include('UAA unavailable')
+            expect(Organization.first(guid: org_1.guid)).not_to be_nil
+            expect(Space.first(guid: space.guid)).not_to be_nil
+            allow(clients).to receive(:get).and_raise(CF::UAA::NotFound)
+            expect(org_delete.delete(org_dataset)).to be_empty
+            expect(Organization.first(guid: org_1.guid)).to be_nil
+          end
+        end
+
         it 'deletes any spaces in the org' do
           expect do
             org_delete.delete(org_dataset)
diff --git a/spec/unit/actions/process_restart_spec.rb b/spec/unit/actions/process_restart_spec.rb
index 2f659ed34ee..dc22232bb7f 100644
--- a/spec/unit/actions/process_restart_spec.rb
+++ b/spec/unit/actions/process_restart_spec.rb
@@ -49,6 +49,24 @@ module VCAP::CloudController
       end
 
       context 'when the process is STARTED' do
+        it 'does not grant a newly bound account to a pre-feature launch before restart' do
+          account = ServiceAccountModel.create(name: 'payments-worker', space: app.space, status: 'ready')
+          app.update(service_account: account)
+          process.this.update(service_account_snapshot: false, service_account_guid: nil)
+          expect(process.reload.runtime_service_account_guid).to be_nil
+        end
+
+        it 'captures account identity only at restart and preserves it after delayed assignment changes' do
+          account = ServiceAccountModel.create(name: 'payments-worker', space: app.space, status: 'ready')
+          app.update(service_account: account)
+          ProcessRestart.restart(process: process, config: config, stop_in_runtime: true)
+          expect(process.reload.runtime_service_account_guid).to eq(account.guid)
+          app.update(service_account: nil)
+          expect(process.reload.runtime_service_account_guid).to eq(account.guid)
+          ProcessRestart.restart(process: process, config: config, stop_in_runtime: true)
+          expect(process.reload.runtime_service_account_guid).to be_nil
+        end
+
         it 'keeps process state as STARTED' do
           ProcessRestart.restart(process: process, config: config, stop_in_runtime: true)
           expect(process.reload.state).to eq('STARTED')
diff --git a/spec/unit/actions/service_account_creation_budget_spec.rb b/spec/unit/actions/service_account_creation_budget_spec.rb
new file mode 100644
index 00000000000..32e24508b08
--- /dev/null
+++ b/spec/unit/actions/service_account_creation_budget_spec.rb
@@ -0,0 +1,32 @@
+require 'spec_helper'
+require 'actions/service_account_creation_budget'
+
+module VCAP::CloudController
+  RSpec.describe ServiceAccountCreationBudget do
+    it 'serializes concurrent creations by the same principal across spaces', isolation: :truncation do
+      TestConfig.override(service_account_creation_limit: 1)
+      principal = create(:user)
+      spaces = create_list(:space, 2)
+      gate = Queue.new
+      threads = spaces.each_with_index.map do |space, index|
+        Thread.new do
+          gate.pop
+          ServiceAccountModel.db.transaction do
+            described_class.consume(User.first(guid: principal.guid), exempt: false) do
+              ServiceAccountModel.create(name: "worker-#{index}", space: space)
+            end
+          end
+          :created
+        rescue CloudController::Errors::V3::ApiError => e
+          e.name
+        end
+      end
+      threads.size.times { gate << true }
+
+      expect(threads.map(&:value)).to contain_exactly(:created, 'ServiceAccountCreationLimitExceeded')
+      expect(ServiceAccountModel.count).to eq(1)
+      expect(ServiceAccountModel.db[:service_account_creations].count).to eq(1)
+      expect(ServiceAccountModel.db[:service_account_names].count).to eq(1)
+    end
+  end
+end
diff --git a/spec/unit/actions/service_account_provision_spec.rb b/spec/unit/actions/service_account_provision_spec.rb
new file mode 100644
index 00000000000..d016543e43f
--- /dev/null
+++ b/spec/unit/actions/service_account_provision_spec.rb
@@ -0,0 +1,140 @@
+require 'spec_helper'
+require 'actions/service_account_provision' if File.exist?('app/actions/service_account_provision.rb')
+
+module VCAP::CloudController
+  RSpec.describe 'Service account provisioning' do
+    let(:account) { ServiceAccountModel.create(name: 'payments-worker', space: create(:space)) }
+    let(:clients) { instance_double(CF::UAA::Scim) }
+    let(:action) { VCAP::CloudController.const_get(:ServiceAccountProvision).new(clients, identity_ca: 'trusted-ca') }
+    let(:registration) do
+      {
+        'client_id' => account.client_id,
+        'authorized_grant_types' => ['client_credentials'],
+        'authorities' => %w[cloud_controller.read cloud_controller.write],
+        'scope' => [],
+        'access_token_validity' => 300,
+        'tls-client-auth-ca' => 'trusted-ca',
+        'tls_client_auth_san_dns' => account.certificate_dns_san,
+        'cf_service_account_guid' => account.guid
+      }
+    end
+
+    it 'creates one canonical client and a roleless OAuth principal before marking ready' do
+      allow(clients).to receive(:get).with(:client, account.client_id).and_raise(CF::UAA::NotFound)
+      expect(clients).to receive(:add).with(:client, registration)
+      action.provision(account)
+
+      principal = User.first(guid: account.client_id)
+      expect(principal.is_oauth_client).to be(true)
+      expect(principal.organizations).to be_empty
+      expect(principal.spaces).to be_empty
+      expect(account.reload.status).to eq('ready')
+    end
+
+    it 'reuses a matching managed registration after a partial failure or retry' do
+      allow(clients).to receive(:get).with(:client, account.client_id).and_return(registration)
+      expect(clients).not_to receive(:add)
+      action.provision(account)
+      action.provision(account)
+      expect(User.where(guid: account.client_id).count).to eq(1)
+    end
+
+    it 'accepts UAA normalized inert scope during disable reconciliation' do
+      account.update(enabled: false)
+      allow(clients).to receive(:get).and_return(registration.merge('scope' => ['uaa.none']))
+      expect(clients).to receive(:delete).with(:client, account.client_id)
+      action.deprovision(account)
+      expect(account.reload.status).to eq('disabled')
+    end
+
+    it 'does not adopt an existing unmanaged client even if its SAN matches' do
+      allow(clients).to receive(:get).and_return(registration.except('cf_service_account_guid'))
+      expect(clients).not_to receive(:add)
+      expect { action.provision(account) }.to raise_error(/client identity collision/)
+      expect(account.reload.status).to eq('failed')
+      expect(User.first(guid: account.client_id)).to be_nil
+    end
+
+    it 'refuses a managed registration with altered trust policy' do
+      allow(clients).to receive(:get).and_return(registration.merge('tls_client_auth_san_dns' => 'other.svc.identity'))
+      expect { action.provision(account) }.to raise_error(/client identity collision/)
+      expect(account.reload.status).to eq('failed')
+    end
+
+    it 'leaves a visible failed state on UAA failure and can retry' do
+      allow(clients).to receive(:get).and_raise(CF::UAA::NotFound)
+      allow(clients).to receive(:add).and_raise(StandardError, 'unavailable')
+      expect { action.provision(account) }.to raise_error('unavailable')
+      expect(account.reload.status).to eq('failed')
+      expect(User.first(guid: account.client_id)).to be_nil
+      allow(clients).to receive(:add).and_return(registration)
+      action.provision(account)
+      expect(account.reload.status).to eq('ready')
+    end
+
+    it 'refuses to repurpose an existing human principal' do
+      create(:user, guid: account.client_id, is_oauth_client: false)
+      expect(clients).not_to receive(:get)
+      expect { action.provision(account) }.to raise_error(/principal identity collision/)
+      expect(account.reload.status).to eq('failed')
+    end
+
+    it 'does not provision disabled accounts' do
+      account.update(enabled: false)
+      expect(clients).not_to receive(:get)
+      expect { action.provision(account) }.to raise_error(/disabled/)
+      expect(account.reload.status).to eq('reserved')
+    end
+
+    context 'with the real UAA client library' do
+      let(:uaa_url) { 'https://service-account-uaa.example.test' }
+      let(:clients) { CF::UAA::Scim.new(uaa_url, 'bearer test-token') }
+      let(:client_url) { "#{uaa_url}/oauth/clients/#{account.client_id}" }
+      let(:response_headers) { { 'content-type' => 'application/json' } }
+
+      it 'serializes the canonical secretless registration over HTTP' do
+        WebMock::API.stub_request(:get, client_url).to_return(status: 404)
+        creation = WebMock::API.stub_request(:post, "#{uaa_url}/oauth/clients").
+                   with(body: registration.to_json, headers: { 'Authorization' => 'bearer test-token' }).
+                   to_return(status: 201, headers: response_headers, body: registration.to_json)
+
+        action.provision(account)
+
+        expect(creation).to have_been_requested.once
+        expect(account.reload.status).to eq('ready')
+      end
+
+      it 'reuses a client with omitted empty scopes and reordered authorities on repeated reconciliation' do
+        response = registration.except('scope').merge('authorities' => registration['authorities'].reverse)
+        WebMock::API.stub_request(:get, client_url).
+          to_return(headers: response_headers, body: response.to_json)
+
+        action.provision(account)
+        action.provision(account)
+
+        expect(account.reload.status).to eq('ready')
+        expect(User.where(guid: account.client_id).count).to eq(1)
+        expect(WebMock::API.a_request(:post, "#{uaa_url}/oauth/clients")).not_to have_been_made
+      end
+
+      [
+        { 'scope' => ['clients.admin'] },
+        { 'authorities' => %w[cloud_controller.read cloud_controller.write cloud_controller.admin] },
+        { 'tls-client-auth-sub-template' => 'admin' },
+        { 'tls-client-auth-aud-templates' => ['other-api'] },
+        { 'tls_client_auth_subject_dn' => 'CN=other' }
+      ].each do |altered_policy|
+        it "refuses an existing client with altered #{altered_policy.keys.first}" do
+          WebMock::API.stub_request(:get, client_url).
+            to_return(headers: response_headers, body: registration.merge(altered_policy).to_json)
+
+          expect { action.provision(account) }.to raise_error(ServiceAccountProvision::Conflict, 'client identity collision')
+
+          expect(account.reload.status).to eq('failed')
+          expect(User.first(guid: account.client_id)).to be_nil
+          expect(WebMock::API.a_request(:post, "#{uaa_url}/oauth/clients")).not_to have_been_made
+        end
+      end
+    end
+  end
+end
diff --git a/spec/unit/actions/space_delete_spec.rb b/spec/unit/actions/space_delete_spec.rb
index f09e6c59980..29f04cded0a 100644
--- a/spec/unit/actions/space_delete_spec.rb
+++ b/spec/unit/actions/space_delete_spec.rb
@@ -27,6 +27,90 @@ module VCAP::CloudController
         expect { space.refresh }.to raise_error Sequel::Error, 'Record not found'
       end
 
+      context 'owned service accounts' do
+        let!(:account) { ServiceAccountModel.create(name: 'payments-worker', space: space) }
+        let(:clients) { instance_double(CF::UAA::Scim) }
+        let(:provisioner) { ServiceAccountProvision.new(clients, identity_ca: 'identity-ca') }
+
+        before do
+          TestConfig.override(service_account_provisioning_enabled: true)
+          CloudController::DependencyLocator.instance.register(:service_account_provisioner, provisioner)
+          allow(clients).to receive(:get).and_raise(CF::UAA::NotFound)
+          allow(clients).to receive(:add)
+          allow(clients).to receive(:delete)
+        end
+
+        it 'deletes assigned workloads, managed clients and principals with explicit roles, retaining names' do
+          provisioner.provision(account)
+          principal = User.first(guid: account.client_id)
+          space.organization.add_user(principal)
+          space.add_auditor(principal)
+          app.update(service_account: account)
+          process = create(:process_model, app: app, state: ProcessModel::STARTED, service_account_guid: account.guid, service_account_snapshot: true)
+          task = create(:task_model, app: app, state: TaskModel::RUNNING_STATE, service_account_guid: account.guid, service_account_snapshot: true)
+          allow(clients).to receive(:get).and_return(provisioner.send(:registration, account))
+          allow(clients).to receive(:delete) do
+            expect(AppModel.first(guid: app.guid)).to be_nil
+            expect(ProcessModel.first(guid: process.guid)).to be_nil
+            expect(TaskModel.first(guid: task.guid)).to be_nil
+            expect(Space.first(guid: space.guid)).not_to be_nil
+          end
+
+          expect(space_delete.delete([space])).to be_empty
+
+          expect(clients).to have_received(:delete).with(:client, account.client_id)
+          expect(ServiceAccountModel.first(guid: account.guid)).to be_nil
+          expect(User.first(guid: principal.guid)).to be_nil
+          expect(Space.first(guid: space.guid)).to be_nil
+          expect(ServiceAccountModel.db[:service_account_names].where(name: account.name).count).to eq(1)
+          expect(Event.where(type: 'audit.service_account.delete', actee: account.guid).count).to eq(1)
+        end
+
+        it 'retains the space on UAA failure and allows cleanup to be retried' do
+          provisioner.provision(account)
+          app.update(service_account: account)
+          allow(clients).to receive(:get).and_return(provisioner.send(:registration, account))
+          allow(clients).to receive(:delete).and_raise('UAA unavailable')
+
+          errors = space_delete.delete([space])
+
+          expect(errors.map(&:message).join).to include('UAA unavailable')
+          expect(Space.first(guid: space.guid)).not_to be_nil
+          expect(account.reload.enabled).to be(false)
+          expect(account.status).to eq('failed')
+          allow(clients).to receive(:delete)
+          expect(space_delete.delete([space])).to be_empty
+          expect(Space.first(guid: space.guid)).to be_nil
+        end
+
+        it 'reports active account reconciliation before deleting workloads' do
+          PollableJobModel.create(resource_guid: account.guid, resource_type: 'service_account', operation: 'service_account.provision', state: 'POLLING')
+          errors = space_delete.delete([space])
+          expect(errors.map(&:message).join).to include('operation is already in progress')
+          expect(AppModel.first(guid: app.guid)).not_to be_nil
+          expect(Space.first(guid: space.guid)).not_to be_nil
+          expect(clients).not_to have_received(:delete)
+        end
+
+        it 'does not adopt or remove an unrelated UAA client' do
+          allow(clients).to receive(:get).and_return('client_id' => account.client_id, 'cf_service_account_guid' => 'someone-else')
+          errors = space_delete.delete([space])
+          expect(errors.map(&:message).join).to include('client identity collision')
+          expect(ServiceAccountModel.first(guid: account.guid)).not_to be_nil
+          expect(Space.first(guid: space.guid)).not_to be_nil
+          expect(clients).not_to have_received(:delete)
+        end
+
+        it 'retains the account and space if workload deletion fails' do
+          allow_any_instance_of(AppDelete).to receive(:delete).and_raise('workload cleanup failed')
+          errors = space_delete.delete([space])
+          expect(errors.map(&:message).join).to include('workload cleanup failed')
+          expect(ServiceAccountModel.first(guid: account.guid)).not_to be_nil
+          expect(Space.first(guid: space.guid)).not_to be_nil
+          expect(clients).not_to have_received(:delete)
+        end
+      end
+
       it 'creates audit events for recursive app deletion and space deletion' do
         space_delete.delete([space])
         expect(VCAP::CloudController::Event.count).to eq(2)
diff --git a/spec/unit/actions/task_create_spec.rb b/spec/unit/actions/task_create_spec.rb
index aa6e4542aa6..e6c46ac8265 100644
--- a/spec/unit/actions/task_create_spec.rb
+++ b/spec/unit/actions/task_create_spec.rb
@@ -72,6 +72,14 @@ module VCAP::CloudController
         expect(TaskModel.count).to eq(1)
       end
 
+      it 'snapshots the app account on task creation so later unbind cannot change task identity' do
+        account = ServiceAccountModel.create(name: 'payments-worker', space: app.space, status: 'ready')
+        app.update(service_account: account)
+        task = task_create_action.create(app, message, user_audit_info)
+        app.update(service_account: nil)
+        expect(task.reload.service_account_guid).to eq(account.guid)
+      end
+
       it "sets the task state to 'RUNNING'" do
         task = task_create_action.create(app, message, user_audit_info)
 
diff --git a/spec/unit/jobs/v3/service_account_provision_spec.rb b/spec/unit/jobs/v3/service_account_provision_spec.rb
new file mode 100644
index 00000000000..16f580798a8
--- /dev/null
+++ b/spec/unit/jobs/v3/service_account_provision_spec.rb
@@ -0,0 +1,71 @@
+require 'spec_helper'
+require 'actions/service_account_provision'
+require 'jobs/v3/service_account_provision' if File.exist?('app/jobs/v3/service_account_provision.rb')
+
+module VCAP::CloudController
+  RSpec.describe 'Service account provisioning job', job_context: :worker do
+    let(:account) { ServiceAccountModel.create(name: 'payments-worker', space: create(:space)) }
+    let(:clients) { instance_double(CF::UAA::Scim) }
+    let(:provisioner) { ServiceAccountProvision.new(clients, identity_ca: 'worker-only-ca') }
+    let(:locator) { CloudController::DependencyLocator.instance }
+    let(:job) { Jobs::V3.const_get(:ServiceAccountProvision).new(account.guid) }
+
+    before do
+      locator.register(:service_account_provisioner, provisioner)
+      allow(clients).to receive(:get).and_raise(CF::UAA::NotFound)
+      allow(clients).to receive(:add)
+    end
+
+    it 'is serializable without credentials and resolves provisioning dependencies in the worker' do
+      serialized = YAML.dump(job)
+      expect(serialized).not_to include('worker-only-ca')
+      expect(serialized).not_to include('CF::UAA::Scim')
+      restored = YAML.unsafe_load(serialized)
+
+      restored.perform
+
+      expect(account.reload.status).to eq('ready')
+      expect(User.first(guid: account.client_id).is_oauth_client).to be(true)
+      expect(restored.resource_guid).to eq(account.guid)
+      expect(restored.resource_type).to eq('service_account')
+      expect(restored.display_name).to eq('service_account.provision')
+      expect(restored).to be_a_valid_job
+      expect(restored.max_attempts).to eq(3)
+    end
+
+    it 'propagates transient errors for retry and reconciles the persisted failed account on the next attempt' do
+      allow(clients).to receive(:add).and_raise(CF::UAA::BadTarget, 'unavailable')
+      expect { job.perform }.to raise_error(CF::UAA::BadTarget)
+      expect(account.reload.status).to eq('failed')
+      expect(User.first(guid: account.client_id)).to be_nil
+
+      allow(clients).to receive(:add)
+      job.perform
+
+      expect(account.reload.status).to eq('ready')
+    end
+
+    it 'does not contact UAA or create a principal if the account was disabled while queued' do
+      queued = job
+      account.update(enabled: false)
+      expect(clients).not_to receive(:get)
+      expect { queued.perform }.to raise_error(ServiceAccountProvision::Conflict, /disabled/)
+      expect(User.first(guid: account.client_id)).to be_nil
+    end
+
+    it 'does not contact UAA if the account was deleted while queued' do
+      queued = job
+      account.destroy
+      expect(clients).not_to receive(:get)
+      expect { queued.perform }.to raise_error(CloudController::Errors::ApiError, /could not be found/)
+    end
+
+    it 'fails closed when worker provisioning has not been configured' do
+      locator.register(:service_account_provisioner, nil)
+      expect(clients).not_to receive(:get)
+      expect { job.perform }.to raise_error(/service account provisioning is not configured/)
+      expect(account.reload.status).to eq('reserved')
+      expect(User.first(guid: account.client_id)).to be_nil
+    end
+  end
+end
diff --git a/spec/unit/lib/cloud_controller/backends/staging_environment_builder_spec.rb b/spec/unit/lib/cloud_controller/backends/staging_environment_builder_spec.rb
index 421d6867c6f..34183d6ed06 100644
--- a/spec/unit/lib/cloud_controller/backends/staging_environment_builder_spec.rb
+++ b/spec/unit/lib/cloud_controller/backends/staging_environment_builder_spec.rb
@@ -60,6 +60,13 @@ module VCAP::CloudController
                                                })
       end
 
+      it 'never exposes account discovery in staging even when supplied by a staging request' do
+        account = ServiceAccountModel.create(name: 'payments-worker', space: app.space, status: 'ready')
+        app.update(service_account: account)
+        result = builder.build(app, space, lifecycle, memory_limit, staging_disk_in_mb, { 'VCAP_SERVICE_ACCOUNT' => 'injected' })
+        expect(result).not_to have_key('VCAP_SERVICE_ACCOUNT')
+      end
+
       context 'when the app has a route associated with it' do
         it 'includes the uris as part of vcap_application' do
           route1 = create(:route, space:)
diff --git a/spec/unit/lib/cloud_controller/diego/app_recipe_builder_spec.rb b/spec/unit/lib/cloud_controller/diego/app_recipe_builder_spec.rb
index e75ebf58b10..14a19509452 100644
--- a/spec/unit/lib/cloud_controller/diego/app_recipe_builder_spec.rb
+++ b/spec/unit/lib/cloud_controller/diego/app_recipe_builder_spec.rb
@@ -356,6 +356,20 @@ module Diego
 
           it_behaves_like 'creating a desired lrp'
 
+          context 'with a launched service account' do
+            let(:config) do
+              Config.new({ service_account_runtime_enabled: true,
+                           diego: { use_privileged_containers_for_running: false, lifecycle_bundles: { 'potato-stack' => 'some-uri' }, pid_limit: 100 } })
+            end
+
+            it 'keeps the launch identity in rebuilt LRPs after desired unbind' do
+              account = ServiceAccountModel.create(name: 'payments-worker', space: space, status: 'ready')
+              process.update(service_account_guid: account.guid, service_account_snapshot: true)
+              app_model.update(service_account: nil)
+              expect(builder.build_app_lrp.certificate_properties.service_account.name).to eq(account.name)
+            end
+          end
+
           it 'creates a desired lrp with buildpack specific properties' do
             lrp = builder.build_app_lrp
             expect(lrp.environment_variables).to contain_exactly(::Diego::Bbs::Models::EnvironmentVariable.new(name: 'foo', value: 'bar'))
diff --git a/spec/unit/lib/cloud_controller/diego/environment_spec.rb b/spec/unit/lib/cloud_controller/diego/environment_spec.rb
index 93642fabd9b..fb2e350f8c2 100644
--- a/spec/unit/lib/cloud_controller/diego/environment_spec.rb
+++ b/spec/unit/lib/cloud_controller/diego/environment_spec.rb
@@ -39,6 +39,21 @@ module VCAP::CloudController::Diego
       ])
     end
 
+    it 'overrides injected account discovery with the ready platform account' do
+      TestConfig.override(service_account_runtime_enabled: true, service_account_token_endpoint: 'https://uaa.example.test/oauth/token/mtls')
+      account = VCAP::CloudController::ServiceAccountModel.create(name: 'payments-worker', space: process.space, status: 'ready')
+      process.app.update(service_account: account)
+      process.update(service_account_guid: account.guid, service_account_snapshot: true)
+      environment['VCAP_SERVICE_ACCOUNT'] = 'injected'
+      value = Environment.new(process).as_json.find { |entry| entry['name'] == 'VCAP_SERVICE_ACCOUNT' }['value']
+      expect(Oj.load(value)).to include('name' => account.name, 'client_id' => account.client_id)
+    end
+
+    it 'removes injected account discovery from unbound apps' do
+      environment['VCAP_SERVICE_ACCOUNT'] = 'injected'
+      expect(Environment.new(process).as_json.pluck('name')).not_to include('VCAP_SERVICE_ACCOUNT')
+    end
+
     context 'when the user specifies their own MEMORY_LIMIT' do
       it 'uses the system provided MEMORY_LIMIT' do
         environment['MEMORY_LIMIT'] = 'i-should-not-be-usedMB'
diff --git a/spec/unit/lib/cloud_controller/diego/service_account_identity_spec.rb b/spec/unit/lib/cloud_controller/diego/service_account_identity_spec.rb
new file mode 100644
index 00000000000..205a3af30ae
--- /dev/null
+++ b/spec/unit/lib/cloud_controller/diego/service_account_identity_spec.rb
@@ -0,0 +1,58 @@
+require 'spec_helper'
+require 'cloud_controller/diego/service_account_identity' if File.exist?('lib/cloud_controller/diego/service_account_identity.rb')
+
+module VCAP::CloudController::Diego
+  RSpec.describe 'Service account runtime identity' do
+    let(:app) { create(:app_model) }
+    let(:account) { VCAP::CloudController::ServiceAccountModel.create(name: 'payments-worker', space: app.space, status: 'ready') }
+    let(:identity) { VCAP::CloudController::Diego.const_get(:ServiceAccountIdentity).new(app, TestConfig.config_instance) }
+
+    before do
+      TestConfig.override(service_account_runtime_enabled: true, service_account_token_endpoint: 'https://uaa.example.test/oauth/token/mtls')
+      app.update(service_account: account)
+    end
+
+    it 'serializes a typed launch identity while preserving organizational units' do
+      properties = identity.certificate_properties
+      decoded = ::Diego::Bbs::Models::CertificateProperties.decode(properties.to_proto)
+      expect(decoded.service_account.name).to eq(account.name)
+      expect(decoded.organizational_unit).to eq(["organization:#{app.organization_guid}", "space:#{app.space_guid}", "app:#{app.guid}"])
+      app.update(service_account: nil)
+      expect(decoded.service_account.name).to eq(account.name)
+    end
+
+    it 'exposes non-secret discovery information' do
+      metadata = {
+        guid: account.guid, name: account.name, client_id: account.client_id, certificate_dns_san: account.certificate_dns_san,
+        token_endpoint: 'https://uaa.example.test/oauth/token/mtls'
+      }
+      expect(identity.environment).to eq('VCAP_SERVICE_ACCOUNT' => metadata)
+    end
+
+    %w[reserved reconciling failed disabled deleting].each do |state|
+      it "rejects new runtime credentials in #{state} state" do
+        account.update(status: state)
+        expect { identity.certificate_properties }.to raise_error(CloudController::Errors::ApiError, /not ready/)
+      end
+    end
+
+    it 'rejects a disabled ready account and a mixed-version disabled runtime' do
+      account.update(enabled: false)
+      expect { identity.certificate_properties }.to raise_error(CloudController::Errors::ApiError, /not ready/)
+      account.update(enabled: true)
+      TestConfig.config[:service_account_runtime_enabled] = false
+      expect { identity.certificate_properties }.to raise_error(CloudController::Errors::ApiError, /runtime is not enabled/)
+    end
+
+    it 'does not give unbound apps account SANs or metadata' do
+      app.update(service_account: nil)
+      expect(identity.certificate_properties.to_h).not_to have_key(:service_account)
+      expect(identity.environment).to eq({})
+    end
+
+    it 'rejects token discovery endpoints containing embedded credentials' do
+      TestConfig.config[:service_account_token_endpoint] = 'https://user:secret@uaa.example.test/oauth/token/mtls'
+      expect { identity.environment }.to raise_error(CloudController::Errors::ApiError, /endpoint/)
+    end
+  end
+end
diff --git a/spec/unit/lib/cloud_controller/diego/task_environment_spec.rb b/spec/unit/lib/cloud_controller/diego/task_environment_spec.rb
index 48bc645d0ba..e7a7c83a649 100644
--- a/spec/unit/lib/cloud_controller/diego/task_environment_spec.rb
+++ b/spec/unit/lib/cloud_controller/diego/task_environment_spec.rb
@@ -42,6 +42,15 @@ module VCAP::CloudController::Diego
         TestConfig.config[:default_app_disk_in_mb]         = staging_disk_in_mb
       end
 
+      it 'inherits the ready platform account discovery and overrides injected values' do
+        TestConfig.override(service_account_runtime_enabled: true, service_account_token_endpoint: 'https://uaa.example.test/oauth/token/mtls')
+        account = VCAP::CloudController::ServiceAccountModel.create(name: 'payments-worker', space: app.space, status: 'ready')
+        app.update(service_account: account)
+        task.update(service_account_guid: account.guid, service_account_snapshot: true)
+        env = TaskEnvironment.new(app, task, space, { 'VCAP_SERVICE_ACCOUNT' => 'injected' }).build
+        expect(env['VCAP_SERVICE_ACCOUNT']).to include(name: account.name, client_id: account.client_id)
+      end
+
       it 'returns the correct environment hash for a v3 app' do
         constructed_envs = TaskEnvironment.new(app, task, space).build
 
diff --git a/spec/unit/lib/cloud_controller/diego/task_recipe_builder_spec.rb b/spec/unit/lib/cloud_controller/diego/task_recipe_builder_spec.rb
index 9daaf547724..5020b0f17cb 100644
--- a/spec/unit/lib/cloud_controller/diego/task_recipe_builder_spec.rb
+++ b/spec/unit/lib/cloud_controller/diego/task_recipe_builder_spec.rb
@@ -524,6 +524,14 @@ module Diego
             allow(TaskCpuWeightCalculator).to receive(:new).with(memory_in_mb: task.memory_in_mb).and_return(calculator)
           end
 
+          it 'propagates a ready service account as typed task certificate properties' do
+            config.config_hash[:service_account_runtime_enabled] = true
+            account = ServiceAccountModel.create(name: 'payments-worker', space: app.space, status: 'ready')
+            app.update(service_account: account)
+            task.update(service_account_guid: account.guid, service_account_snapshot: true)
+            expect(task_recipe_builder.build_app_task(config, task).certificate_properties.service_account.name).to eq(account.name)
+          end
+
           it 'constructs a TaskDefinition with app task instructions' do
             result = task_recipe_builder.build_app_task(config, task)
             expected_callback_url = "https://#{internal_service_hostname}:#{tls_port}/internal/v4/tasks/#{task.guid}/completed"
diff --git a/spec/unit/lib/cloud_controller/service_account_provisioner_configuration_spec.rb b/spec/unit/lib/cloud_controller/service_account_provisioner_configuration_spec.rb
new file mode 100644
index 00000000000..1114b96d1c3
--- /dev/null
+++ b/spec/unit/lib/cloud_controller/service_account_provisioner_configuration_spec.rb
@@ -0,0 +1,86 @@
+require 'spec_helper'
+require 'cloud_controller/dependency_locator'
+
+module VCAP::CloudController
+  RSpec.describe 'Configured service account provisioner', job_context: :worker do
+    let(:locator) { CloudController::DependencyLocator.instance }
+    let(:settings) { { client_id: 'account-manager', client_secret: 'manager-secret', identity_ca: 'instance-identity-ca' } }
+    let(:account) { ServiceAccountModel.create(name: 'payments-worker', space: create(:space)) }
+    let(:uaa_url) { 'https://uaa.service.cf.internal' }
+    let(:headers) { { 'content-type' => 'application/json' } }
+    let(:worker_config) { Config.read_file('config/cloud_controller.yml').merge(service_account_provisioning: settings) }
+
+    before do
+      UaaTokenCache.clear!
+      TestConfig.override(service_account_provisioning: settings)
+    end
+
+    it 'accepts complete worker configuration' do
+      expect { ConfigSchemas::WorkerSchema.validate(worker_config) }.not_to raise_error
+    end
+
+    it 'validates provisioning configuration for API-hosted local workers too' do
+      api_config = Config.read_file('config/cloud_controller.yml').merge(service_account_provisioning: settings.except(:identity_ca))
+      expect { ConfigSchemas::ApiSchema.validate(api_config) }.to raise_error(Membrane::SchemaValidationError, /identity_ca => Missing key/)
+    end
+
+    %i[client_id client_secret identity_ca].each do |key|
+      it "rejects worker configuration missing #{key}" do
+        worker_config[:service_account_provisioning] = settings.except(key)
+        expect { ConfigSchemas::WorkerSchema.validate(worker_config) }.to raise_error(Membrane::SchemaValidationError, /#{key} => Missing key/)
+      end
+
+      it "fails closed before authentication with blank #{key}" do
+        TestConfig.override(service_account_provisioning: settings.merge(key => ' '))
+        expect { locator.service_account_provisioner }.to raise_error(/service account provisioning is not configured/)
+      end
+    end
+
+    it 'uses only the configured management credentials and instance identity CA' do
+      token = WebMock::API.stub_request(:post, "#{uaa_url}/oauth/token").
+              with(basic_auth: %w[account-manager manager-secret], body: { 'grant_type' => 'client_credentials' }).
+              to_return(headers: headers, body: { access_token: 'manager-token', token_type: 'bearer', expires_in: 300 }.to_json)
+      WebMock::API.stub_request(:get, "#{uaa_url}/oauth/clients/#{account.client_id}").
+        with(headers: { 'Authorization' => 'bearer manager-token' }).to_return(status: 404)
+      creation = WebMock::API.stub_request(:post, "#{uaa_url}/oauth/clients").
+                 with(headers: { 'Authorization' => 'bearer manager-token' }) do |request|
+                   payload = Oj.load(request.body)
+                   payload['client_id'] == account.client_id &&
+                     payload['tls-client-auth-ca'] == settings[:identity_ca] &&
+                     payload['tls_client_auth_san_dns'] == account.certificate_dns_san &&
+                     !payload.key?('client_secret')
+                 end.to_return(status: 201, headers: headers, body: { client_id: account.client_id }.to_json)
+
+      provisioner = locator.service_account_provisioner
+      expect(locator.service_account_provisioner).to equal(provisioner)
+      provisioner.provision(account)
+
+      expect(account.reload.status).to eq('ready')
+      expect(token).to have_been_requested.at_least_once
+      expect(creation).to have_been_requested.once
+    end
+
+    it 'refreshes an invalid cached management token before retrying client creation' do
+      UaaTokenCache.set_token(settings[:client_id], 'bearer expired-token')
+      WebMock::API.stub_request(:get, "#{uaa_url}/oauth/clients/#{account.client_id}").to_return(status: 404)
+      WebMock::API.stub_request(:post, "#{uaa_url}/oauth/clients").
+        with(headers: { 'Authorization' => 'bearer expired-token' }).
+        to_return(status: 401, headers: headers, body: { error: 'invalid_token' }.to_json)
+      WebMock::API.stub_request(:post, "#{uaa_url}/oauth/token").
+        to_return(headers: headers, body: { access_token: 'refreshed-token', token_type: 'bearer', expires_in: 300 }.to_json)
+      creation = WebMock::API.stub_request(:post, "#{uaa_url}/oauth/clients").
+                 with(headers: { 'Authorization' => 'bearer refreshed-token' }).
+                 to_return(status: 201, headers: headers, body: { client_id: account.client_id }.to_json)
+
+      locator.service_account_provisioner.provision(account)
+
+      expect(account.reload.status).to eq('ready')
+      expect(creation).to have_been_requested.once
+    end
+
+    it 'fails closed without configuration and does not fall back to another UAA client' do
+      TestConfig.override(service_account_provisioning: nil)
+      expect { locator.service_account_provisioner }.to raise_error(/service account provisioning is not configured/)
+    end
+  end
+end
diff --git a/spec/unit/models/runtime/service_account_assignment_spec.rb b/spec/unit/models/runtime/service_account_assignment_spec.rb
new file mode 100644
index 00000000000..4d1cee6e97e
--- /dev/null
+++ b/spec/unit/models/runtime/service_account_assignment_spec.rb
@@ -0,0 +1,49 @@
+require 'spec_helper'
+
+module VCAP::CloudController
+  RSpec.describe 'Space-owned service account app assignments' do
+    let(:space) { create(:space) }
+    let(:app) { create(:app_model, space: space) }
+    let(:account) { ServiceAccountModel.create(name: 'payments-worker', space: space) }
+
+    it 'leaves existing apps unbound by default' do
+      expect(app.service_account).to be_nil
+    end
+
+    it 'allows multiple apps in the owning space to share one account' do
+      other_app = create(:app_model, space: space)
+      app.update(service_account: account)
+      other_app.update(service_account: account)
+      expect(account.apps).to contain_exactly(app, other_app)
+    end
+
+    it 'supports unbinding' do
+      app.update(service_account: account)
+      app.update(service_account: nil)
+      expect(app.reload.service_account).to be_nil
+    end
+
+    it 'rejects assignment from another space in the same organization' do
+      other_space = create(:space, organization: space.organization)
+      other_account = ServiceAccountModel.create(name: 'reporting-reader', space: other_space)
+      expect { app.update(service_account: other_account) }.to raise_error(Sequel::ValidationFailed, /owning space/)
+      expect(app.reload.service_account).to be_nil
+    end
+
+    it 'rejects moving a bound app to another space' do
+      app.update(service_account: account)
+      expect { app.update(space: create(:space, organization: space.organization)) }.to raise_error(Sequel::ValidationFailed, /owning space/)
+      expect(app.reload.space).to eq(space)
+    end
+
+    it 'does not permit changing an account owner space' do
+      expect { account.update(space: create(:space)) }.to raise_error(Sequel::ValidationFailed, /immutable/)
+      expect(account.reload.space).to eq(space)
+    end
+
+    it 'prevents deleting a referenced account' do
+      app.update(service_account: account)
+      expect { account.db.transaction(savepoint: true) { account.destroy } }.to raise_error(Sequel::ForeignKeyConstraintViolation)
+    end
+  end
+end
diff --git a/spec/unit/models/runtime/service_account_model_spec.rb b/spec/unit/models/runtime/service_account_model_spec.rb
new file mode 100644
index 00000000000..226ba2708f0
--- /dev/null
+++ b/spec/unit/models/runtime/service_account_model_spec.rb
@@ -0,0 +1,63 @@
+require 'spec_helper'
+
+module VCAP::CloudController
+  RSpec.describe 'Service account registry' do
+    let(:space) { create(:space) }
+    let(:model) { VCAP::CloudController.const_get(:ServiceAccountModel) }
+
+    def reserve(name, owner=space)
+      model.create(name: name, space: owner)
+    end
+
+    it 'reserves a space-owned identity without provisioning authentication' do
+      account = reserve('payments-worker')
+
+      expect(account.guid).not_to be_empty
+      expect(account.space).to eq(space)
+      expect(account.client_id).to eq('cf:service-account:payments-worker')
+      expect(account.certificate_dns_san).to eq('payments-worker.svc.identity')
+      expect(account.enabled).to be(true)
+      expect(account.status).to eq('reserved')
+    end
+
+    it 'accepts canonical labels at both length boundaries' do
+      expect(reserve('a-b').name).to eq('a-b')
+      expect(reserve('a' * 63).name).to eq('a' * 63)
+    end
+
+    ['', 'ab', 'a' * 64, '-worker', 'worker-', 'Worker', 'work_er',
+     'work.er', "worker\n", 'wörker', 'work%65r', 'work:er', '*worker'].each do |name|
+      it "rejects noncanonical label #{name.inspect}" do
+        expect { reserve(name) }.to raise_error(Sequel::ValidationFailed, /name/)
+      end
+    end
+
+    it 'enforces foundation-wide uniqueness across owners in the database' do
+      account = reserve('payments-worker')
+      other_owner = create(:space)
+
+      expect { reserve(account.name, other_owner) }.to raise_error(Sequel::ValidationFailed, /name/)
+      expect do
+        model.db.transaction(savepoint: true) do
+          # Deliberately bypass model validation to exercise the database constraint.
+          model.dataset.insert(account.values.except(:id, :guid).merge(guid: SecureRandom.uuid, space_guid: other_owner.guid)) # rubocop:disable Rails/SkipsModelValidations
+        end
+      end.to raise_error(Sequel::UniqueConstraintViolation)
+    end
+
+    it 'does not permit changing the identity name' do
+      account = reserve('payments-worker')
+
+      expect { account.update(name: 'reporting-reader') }.to raise_error(Sequel::ValidationFailed, /immutable/)
+      expect(account.reload.name).to eq('payments-worker')
+    end
+
+    it 'retains the name after deletion so another owner cannot inherit it' do
+      account = reserve('payments-worker')
+      account.destroy
+
+      expect(model.where(guid: account.guid).first).to be_nil
+      expect { reserve('payments-worker', create(:space)) }.to raise_error(Sequel::ValidationFailed, /name/)
+    end
+  end
+end
diff --git a/spec/unit/service_account_migration_spec.rb b/spec/unit/service_account_migration_spec.rb
new file mode 100644
index 00000000000..d390c561a63
--- /dev/null
+++ b/spec/unit/service_account_migration_spec.rb
@@ -0,0 +1,12 @@
+require 'spec_helper'
+
+RSpec.describe 'service account app relationship migration' do
+  it 'matches MySQL parent collation when retrying a partially applied migration' do
+    db = Sequel.mock(host: :mysql, fetch: [{ Field: 'guid', Collation: 'utf8mb3_general_ci' }])
+    allow(db).to receive(:schema).with(:apps).and_return([[:service_account_guid, { type: :string }]])
+    load File.expand_path('../../db/migrations/20261005120100_add_app_service_account.rb', __dir__)
+    migration = Sequel::Migration.descendants.last
+    migration.apply(db, :up)
+    expect(db.sqls.join).to include('COLLATE utf8mb3_general_ci')
+  end
+end