diff --git a/app/jobs/runtime/service_operations_binding_delete_stuck_in_progress_retry.rb b/app/jobs/runtime/service_operations_binding_delete_stuck_in_progress_retry.rb
new file mode 100644
index 00000000000..7b602219656
--- /dev/null
+++ b/app/jobs/runtime/service_operations_binding_delete_stuck_in_progress_retry.rb
@@ -0,0 +1,109 @@
+module VCAP::CloudController
+  module Jobs
+    module Runtime
+      class ServiceOperationsBindingDeleteStuckInProgressRetry < VCAP::CloudController::Jobs::CCJob
+        BATCH_SIZE = 10
+
+        def perform
+          logger.info("Retrying stuck binding 'delete' operations")
+          retry_stuck(ServiceBindingOperation, ServiceBinding, :service_binding_id, 'service_bindings.delete')
+          retry_stuck(ServiceKeyOperation,     ServiceKey,     :service_key_id,     'service_keys.delete')
+        end
+
+        def max_attempts
+          1
+        end
+
+        private
+
+        def retry_stuck(operation_model, instance_model, foreign_key, jobs_operation)
+          # Find stuck binding 'delete' operations where the broker may still be working
+          # but CC's polling job has permanently failed due to a transient error (e.g. brief db connection flip).
+          #
+          # Unlike create we do not mark the operation failed and do not mitigate orphans: for a delete we
+          # re-enqueue the original polling job so the unbind is driven to completion. The original delayed_job's
+          # serialized handler is reused, preserving @start_time so the ReoccurringJob max-duration expiry
+          # (which marks the operation failed via handle_timeout) still fires against the original polling window.
+          operation_table = operation_model.table_name
+          instance_table = instance_model.table_name
+
+          stuck = operation_model.
+                  join(instance_table, id: Sequel[operation_table][foreign_key]).
+                  join(:jobs, resource_guid: Sequel[instance_table][:guid]).
+                  join(:delayed_jobs, guid: Sequel[:jobs][:delayed_job_guid]).
+                  where(Sequel[operation_table][:state] => 'in progress').
+                  where(Sequel[operation_table][:type] => 'delete').
+                  where(Sequel.lit("#{operation_table}.created_at > CURRENT_TIMESTAMP - INTERVAL '?' SECOND", default_maximum_duration_seconds.to_i)).
+                  where(Sequel[:jobs][:state] => [PollableJobModel::POLLING_STATE, PollableJobModel::FAILED_STATE]).
+                  where(Sequel[:jobs][:operation] => jobs_operation).
+                  exclude(Sequel[:delayed_jobs][:failed_at] => nil).
+                  select(
+                    Sequel[:jobs][:guid].as(:pollable_guid),
+                    Sequel[operation_table][:id].as(:op_id),
+                    Sequel[operation_table][foreign_key].as(:resource_id)
+                  ).
+                  order(Sequel[operation_table][:created_at]).
+                  limit(BATCH_SIZE)
+
+          stuck.each do |row|
+            resolve_stuck(operation_model, instance_model, row[:op_id], row[:resource_id], row[:pollable_guid])
+          end
+        end
+
+        def resolve_stuck(operation_model, instance_model, op_id, resource_id, pollable_guid)
+          operation_model.db.transaction do
+            operation = operation_model.where(id: op_id, state: 'in progress').for_update.skip_locked.first
+            return unless operation
+
+            binding = instance_model.first(id: resource_id)
+            return unless binding
+
+            pollable = PollableJobModel.first(guid: pollable_guid)
+            return unless pollable
+
+            handler = deserialize_handler(pollable)
+            return unless handler
+
+            binding_type = instance_model.to_s.split('::').last
+
+            logger.info(
+              "#{binding_type} #{binding.guid} delete operation is stuck in 'in progress'. Re-enqueuing the polling job.",
+              binding_type: binding_type,
+              binding_guid: binding.guid,
+              operation_id: op_id,
+              pollable_job_guid: pollable_guid
+            )
+
+            pollable.update(state: PollableJobModel::POLLING_STATE, cf_api_error: nil)
+            Jobs::GenericEnqueuer.shared.enqueue_pollable(handler, existing_guid: pollable.guid, preserve_priority: true)
+          end
+        end
+
+        # Reuse the original delete polling job by deserializing the failed delayed_job's handler and unwrapping
+        # the wrapper chain (LoggingContextJob → TimeoutJob → PollableJobWrapper → DeleteBindingJob).
+        # This preserves the original @user_audit_info, @start_time and the binding @type.
+        def deserialize_handler(pollable)
+          delayed_job = Delayed::Job[guid: pollable.delayed_job_guid]
+          return unless delayed_job
+
+          Jobs::Enqueuer.unwrap_job(delayed_job.payload_object)
+        rescue StandardError => e
+          logger.error("Could not deserialize delayed job '#{pollable.delayed_job_guid}' for pollable '#{pollable.guid}': #{e.class}: #{e.message}")
+          nil
+        end
+
+        def default_maximum_duration_seconds
+          Config.config.get(:broker_client_max_async_poll_duration_minutes).minutes
+        end
+
+        def logger
+          @logger ||= Steno.logger('cc.background.service-operations-binding-delete-stuck-in-progress-retry')
+        end
+
+        def job_name_in_configuration
+          :service_operations_binding_delete_stuck_in_progress_retry
+        end
+      end
+    end
+  end
+end
diff --git a/app/jobs/runtime/service_operations_create_in_progress_cleanup.rb b/app/jobs/runtime/service_operations_create_in_progress_cleanup.rb
index afc9f4bb4a3..e7ac4f15c19 100644
--- a/app/jobs/runtime/service_operations_create_in_progress_cleanup.rb
+++ b/app/jobs/runtime/service_operations_create_in_progress_cleanup.rb
@@ -40,6 +40,10 @@ def cleanup_operations(operation_model, instance_model, foreign_key, jobs_operat
           #     service instance that happen to share the same resource_guid
           #   - delayed_jobs.failed_at IS NOT NULL: the delayed job permanently failed (exhausted max_attempts);
           #     jobs still alive or locked have failed_at=NULL and must not be touched
+          #   - service_instances.guid NOT IN (live pollables for this operation): skip resources that still
+          #     have a POLLING/PROCESSING pollable driving this operation. A prior operation on the same
+          #     resource can leave a stale, permanently-failed pollable behind; without this guard that dead
+          #     row would match by resource_guid and cause the current healthy operation to be marked failed
           operation_table = operation_model.table_name
           instance_table = instance_model.table_name
 
@@ -53,6 +57,7 @@ def cleanup_operations(operation_model, instance_model, foreign_key, jobs_operat
                   where(Sequel[:jobs][:state] => [PollableJobModel::POLLING_STATE, PollableJobModel::FAILED_STATE]).
                   where(Sequel[:jobs][:operation] => jobs_operation).
                   exclude(Sequel[:delayed_jobs][:failed_at] => nil).
+                  exclude(live_pollable_exists(operation_model, instance_table, jobs_operation)).
                   select(
                     Sequel[:jobs][:guid].as(:pollable_guid),
                     Sequel[operation_table][:id].as(:op_id),
@@ -104,6 +109,25 @@ def default_maximum_duration_seconds
           Config.config.get(:broker_client_max_async_poll_duration_minutes).minutes
         end
 
+        # NOT EXISTS guard: skip a resource if it still has a pollable job actively driving
+        # THIS operation — state POLLING or PROCESSING AND backed by a delayed_job that has
+        # NOT permanently failed (failed_at IS NULL, or no delayed_job row yet). A stale,
+        # permanently-failed pollable left behind by a previous operation on the same
+        # resource must NOT cause the current healthy operation to be marked failed. A
+        # POLLING pollable whose delayed_job IS failed is itself stuck (the DB flip happened
+        # before the failure hook could write FAILED) and must NOT count as live. Correlated
+        # (resource_guid = instance.guid) so a NULL jobs.resource_guid elsewhere cannot
+        # poison the result the way a NOT IN subquery would.
+        def live_pollable_exists(operation_model, instance_table, jobs_operation)
+          operation_model.db[:jobs].
+            left_join(:delayed_jobs, guid: Sequel[:jobs][:delayed_job_guid]).
+            where(Sequel[:jobs][:operation] => jobs_operation).
+            where(Sequel[:jobs][:state] => [PollableJobModel::POLLING_STATE, PollableJobModel::PROCESSING_STATE]).
+            where(Sequel[:delayed_jobs][:failed_at] => nil).
+            where(Sequel[:jobs][:resource_guid] => Sequel[instance_table][:guid]).
+            exists
+        end
+
         def logger
           @logger ||= Steno.logger('cc.background.service-operations-create-in-progress-cleanup')
         end
diff --git a/app/jobs/runtime/service_operations_delete_stuck_in_progress_retry.rb b/app/jobs/runtime/service_operations_delete_stuck_in_progress_retry.rb
new file mode 100644
index 00000000000..a08b1f8bd40
--- /dev/null
+++ b/app/jobs/runtime/service_operations_delete_stuck_in_progress_retry.rb
@@ -0,0 +1,108 @@
+module VCAP::CloudController
+  module Jobs
+    module Runtime
+      class ServiceOperationsDeleteStuckInProgressRetry < VCAP::CloudController::Jobs::CCJob
+        BATCH_SIZE = 10
+
+        def perform
+          logger.info("Retrying stuck service 'delete' operations")
+          retry_stuck(ServiceInstanceOperation, ServiceInstance, :service_instance_id, 'service_instance.delete')
+        end
+
+        def max_attempts
+          1
+        end
+
+        private
+
+        def retry_stuck(operation_model, instance_model, foreign_key, jobs_operation)
+          # Find stuck service instance 'delete' operations where the broker may still be working
+          # but CC's polling job has permanently failed due to a transient error (e.g. brief db connection flip).
+          #
+          # Unlike create/update we do not mark the operation failed: for a delete we re-enqueue the original
+          # polling job so the deprovision is driven to completion. The original delayed_job's serialized handler
+          # is reused, preserving @start_time so the ReoccurringJob max-duration expiry (which marks the operation
+          # failed via handle_timeout) still fires against the original polling window.
+          operation_table = operation_model.table_name
+          instance_table = instance_model.table_name
+
+          stuck = operation_model.
+                  join(instance_table, id: Sequel[operation_table][foreign_key]).
+                  join(:jobs, resource_guid: Sequel[instance_table][:guid]).
+                  join(:delayed_jobs, guid: Sequel[:jobs][:delayed_job_guid]).
+                  where(Sequel[operation_table][:state] => 'in progress').
+                  where(Sequel[operation_table][:type] => 'delete').
+                  where(Sequel.lit("#{operation_table}.created_at > CURRENT_TIMESTAMP - INTERVAL '?' SECOND", default_maximum_duration_seconds.to_i)).
+                  where(Sequel[:jobs][:state] => [PollableJobModel::POLLING_STATE, PollableJobModel::FAILED_STATE]).
+                  where(Sequel[:jobs][:operation] => jobs_operation).
+                  exclude(Sequel[:delayed_jobs][:failed_at] => nil).
+                  select(
+                    Sequel[:jobs][:guid].as(:pollable_guid),
+                    Sequel[operation_table][:id].as(:op_id),
+                    Sequel[operation_table][foreign_key].as(:resource_id)
+                  ).
+                  order(Sequel[operation_table][:created_at]).
+                  limit(BATCH_SIZE)
+
+          stuck.each do |row|
+            resolve_stuck(operation_model, instance_model, row[:op_id], row[:resource_id], row[:pollable_guid])
+          end
+        end
+
+        def resolve_stuck(operation_model, instance_model, op_id, resource_id, pollable_guid)
+          operation_model.db.transaction do
+            operation = operation_model.where(id: op_id, state: 'in progress').for_update.skip_locked.first
+            return unless operation
+
+            instance = instance_model.first(id: resource_id)
+            return unless instance
+
+            pollable = PollableJobModel.first(guid: pollable_guid)
+            return unless pollable
+
+            handler = deserialize_handler(pollable)
+            return unless handler
+
+            instance_type = instance_model.to_s.split('::').last
+
+            logger.info(
+              "#{instance_type} #{instance.guid} delete operation is stuck in 'in progress'. Re-enqueuing the polling job.",
+              instance_type: instance_type,
+              instance_guid: instance.guid,
+              operation_id: op_id,
+              pollable_job_guid: pollable_guid
+            )
+
+            pollable.update(state: PollableJobModel::POLLING_STATE, cf_api_error: nil)
+            Jobs::GenericEnqueuer.shared.enqueue_pollable(handler, existing_guid: pollable.guid, preserve_priority: true)
+          end
+        end
+
+        # Reuse the original delete polling job by deserializing the failed delayed_job's handler and unwrapping
+        # the wrapper chain (LoggingContextJob → TimeoutJob → PollableJobWrapper → DeleteServiceInstanceJob).
+        # This preserves the original @user_audit_info, @start_time and the recursive-vs-plain delete variant.
+        def deserialize_handler(pollable)
+          delayed_job = Delayed::Job[guid: pollable.delayed_job_guid]
+          return unless delayed_job
+
+          Jobs::Enqueuer.unwrap_job(delayed_job.payload_object)
+        rescue StandardError => e
+          logger.error("Could not deserialize delayed job '#{pollable.delayed_job_guid}' for pollable '#{pollable.guid}': #{e.class}: #{e.message}")
+          nil
+        end
+
+        def default_maximum_duration_seconds
+          Config.config.get(:broker_client_max_async_poll_duration_minutes).minutes
+        end
+
+        def logger
+          @logger ||= Steno.logger('cc.background.service-operations-delete-stuck-in-progress-retry')
+        end
+
+        def job_name_in_configuration
+          :service_operations_delete_stuck_in_progress_retry
+        end
+      end
+    end
+  end
+end
diff --git a/app/jobs/runtime/service_operations_update_stuck_in_progress_failed.rb b/app/jobs/runtime/service_operations_update_stuck_in_progress_failed.rb
new file mode 100644
index 00000000000..5cadd90cf20
--- /dev/null
+++ b/app/jobs/runtime/service_operations_update_stuck_in_progress_failed.rb
@@ -0,0 +1,104 @@
+module VCAP::CloudController
+  module Jobs
+    module Runtime
+      class ServiceOperationsUpdateStuckInProgressFailed < VCAP::CloudController::Jobs::CCJob
+        BATCH_SIZE = 10
+
+        def perform
+          logger.info("Marking stuck service 'update' operations as 'failed'")
+          mark_stuck_in_progress_failed(ServiceInstanceOperation, ServiceInstance, :service_instance_id, 'service_instance.update')
+        end
+
+        def max_attempts
+          1
+        end
+
+        private
+
+        def mark_stuck_in_progress_failed(operation_model, instance_model, foreign_key, jobs_operation)
+          operation_table = operation_model.table_name
+          instance_table = instance_model.table_name
+
+          stuck = operation_model.
+                  join(instance_table, id: Sequel[operation_table][foreign_key]).
+                  join(:jobs, resource_guid: Sequel[instance_table][:guid]).
+                  join(:delayed_jobs, guid: Sequel[:jobs][:delayed_job_guid]).
+                  where(Sequel[operation_table][:state] => 'in progress').
+                  where(Sequel[operation_table][:type] => 'update').
+                  where(Sequel.lit("#{operation_table}.created_at > CURRENT_TIMESTAMP - INTERVAL '?' SECOND", default_maximum_duration_seconds.to_i)).
+                  where(Sequel[:jobs][:state] => [PollableJobModel::POLLING_STATE, PollableJobModel::FAILED_STATE]).
+                  where(Sequel[:jobs][:operation] => jobs_operation).
+                  exclude(Sequel[:delayed_jobs][:failed_at] => nil).
+                  exclude(live_pollable_exists(operation_model, instance_table, jobs_operation)).
+                  select(
+                    Sequel[:jobs][:guid].as(:pollable_guid),
+                    Sequel[operation_table][:id].as(:op_id),
+                    Sequel[operation_table][foreign_key].as(:resource_id)
+                  ).
+                  order(Sequel[operation_table][:created_at]).
+                  limit(BATCH_SIZE)
+
+          stuck.each do |row|
+            resolve_stuck(operation_model, instance_model, row[:op_id], row[:resource_id], row[:pollable_guid])
+          end
+        end
+
+        def resolve_stuck(operation_model, instance_model, op_id, resource_id, pollable_guid)
+          operation_model.db.transaction do
+            operation = operation_model.where(id: op_id, state: 'in progress').for_update.skip_locked.first
+            return unless operation
+
+            instance = instance_model.first(id: resource_id)
+            return unless instance
+
+            instance_type = instance_model.to_s.split('::').last
+
+            logger.info(
+              "#{instance_type} #{instance.guid} update operation is stuck in 'in progress'. " \
+              "Setting operation's state to 'failed' and pollable job's state to 'FAILED'.",
+              instance_type: instance_type,
+              instance_guid: instance.guid,
+              operation_id: op_id,
+              pollable_job_guid: pollable_guid
+            )
+
+            operation.update(state: 'failed',
+                             description: "Operation was stuck in 'in progress' state. Set to 'failed' by cleanup job.")
+            PollableJobModel.where(guid: pollable_guid).update(state: PollableJobModel::FAILED_STATE)
+          end
+        end
+
+        def default_maximum_duration_seconds
+          Config.config.get(:broker_client_max_async_poll_duration_minutes).minutes
+        end
+
+        # NOT EXISTS guard: skip a resource if it still has a pollable job actively driving
+        # THIS operation — state POLLING or PROCESSING AND backed by a delayed_job that has
+        # NOT permanently failed (failed_at IS NULL, or no delayed_job row yet). A stale,
+        # permanently-failed pollable left behind by a previous operation on the same
+        # resource must NOT cause the current healthy operation to be marked failed. A
+        # POLLING pollable whose delayed_job IS failed is itself stuck (the DB flip happened
+        # before the failure hook could write FAILED) and must NOT count as live.
+        # Correlated (resource_guid = instance.guid) so a NULL jobs.resource_guid elsewhere
+        # cannot poison the result the way a NOT IN subquery would.
+        def live_pollable_exists(operation_model, instance_table, jobs_operation)
+          operation_model.db[:jobs].
+            left_join(:delayed_jobs, guid: Sequel[:jobs][:delayed_job_guid]).
+            where(Sequel[:jobs][:operation] => jobs_operation).
+            where(Sequel[:jobs][:state] => [PollableJobModel::POLLING_STATE, PollableJobModel::PROCESSING_STATE]).
+            where(Sequel[:delayed_jobs][:failed_at] => nil).
+            where(Sequel[:jobs][:resource_guid] => Sequel[instance_table][:guid]).
+            exists
+        end
+
+        def logger
+          @logger ||= Steno.logger('cc.background.service-operations-update-stuck-in-progress-failed')
+        end
+
+        def job_name_in_configuration
+          :service_operations_update_stuck_in_progress_failed
+        end
+      end
+    end
+  end
+end
diff --git a/config/cloud_controller.yml b/config/cloud_controller.yml
index 30edab7f28b..58596af8255 100644
--- a/config/cloud_controller.yml
+++ b/config/cloud_controller.yml
@@ -58,6 +58,15 @@ service_operations_initial_cleanup:
 service_operations_create_in_progress_cleanup:
   frequency_in_seconds: 3600 #1h
 
+service_operations_update_stuck_in_progress_failed:
+  frequency_in_seconds: 3600 #1h
+
+service_operations_delete_stuck_in_progress_retry:
+  frequency_in_seconds: 3600 #1h
+
+service_operations_binding_delete_stuck_in_progress_retry:
+  frequency_in_seconds: 3600 #1h
+
 # One-off backfill - to be removed in a future version.
 lifecycle_type_backfill:
   frequency_in_seconds: 3600 #1h
diff --git a/lib/cloud_controller/clock/scheduler.rb b/lib/cloud_controller/clock/scheduler.rb
index e128e4db906..2de24c96816 100644
--- a/lib/cloud_controller/clock/scheduler.rb
+++ b/lib/cloud_controller/clock/scheduler.rb
@@ -26,6 +26,9 @@ class Scheduler
       { name: 'failed_jobs', class: Jobs::Runtime::FailedJobsCleanup },
       { name: 'service_operations_initial_cleanup', class: Jobs::Runtime::ServiceOperationsInitialCleanup },
       { name: 'service_operations_create_in_progress_cleanup', class: Jobs::Runtime::ServiceOperationsCreateInProgressCleanup },
+      { name: 'service_operations_update_stuck_in_progress_failed', class: Jobs::Runtime::ServiceOperationsUpdateStuckInProgressFailed },
+      { name: 'service_operations_delete_stuck_in_progress_retry', class: Jobs::Runtime::ServiceOperationsDeleteStuckInProgressRetry },
+      { name: 'service_operations_binding_delete_stuck_in_progress_retry', class: Jobs::Runtime::ServiceOperationsBindingDeleteStuckInProgressRetry },
       # One-off backfill - to be removed in a future version.
       { name: 'lifecycle_type_backfill', class: Jobs::Runtime::LifecycleTypeBackfill }
     ].freeze
diff --git a/lib/cloud_controller/config_schemas/clock_schema.rb b/lib/cloud_controller/config_schemas/clock_schema.rb
index 79b4461b66c..1b385af7526 100644
--- a/lib/cloud_controller/config_schemas/clock_schema.rb
+++ b/lib/cloud_controller/config_schemas/clock_schema.rb
@@ -37,6 +37,15 @@ class ClockSchema < VCAP::Config
           service_operations_create_in_progress_cleanup: {
             frequency_in_seconds: Integer
           },
+          service_operations_update_stuck_in_progress_failed: {
+            frequency_in_seconds: Integer
+          },
+          service_operations_delete_stuck_in_progress_retry: {
+            frequency_in_seconds: Integer
+          },
+          service_operations_binding_delete_stuck_in_progress_retry: {
+            frequency_in_seconds: Integer
+          },
           # One-off backfill - to be removed in a future version.
           lifecycle_type_backfill: {
             frequency_in_seconds: Integer
diff --git a/lib/cloud_controller/jobs.rb b/lib/cloud_controller/jobs.rb
index eaeae0dc0d9..b55264cec31 100644
--- a/lib/cloud_controller/jobs.rb
+++ b/lib/cloud_controller/jobs.rb
@@ -26,6 +26,9 @@
 require 'jobs/runtime/expired_orphaned_blob_cleanup'
 require 'jobs/runtime/expired_resource_cleanup'
 require 'jobs/runtime/service_operations_create_in_progress_cleanup'
+require 'jobs/runtime/service_operations_update_stuck_in_progress_failed'
+require 'jobs/runtime/service_operations_delete_stuck_in_progress_retry'
+require 'jobs/runtime/service_operations_binding_delete_stuck_in_progress_retry'
 require 'jobs/runtime/failed_jobs_cleanup'
 require 'jobs/runtime/service_operations_initial_cleanup'
 require 'jobs/runtime/legacy_jobs'
diff --git a/spec/unit/jobs/runtime/service_operations_binding_delete_stuck_in_progress_retry_spec.rb b/spec/unit/jobs/runtime/service_operations_binding_delete_stuck_in_progress_retry_spec.rb
new file mode 100644
index 00000000000..a24f543c701
--- /dev/null
+++ b/spec/unit/jobs/runtime/service_operations_binding_delete_stuck_in_progress_retry_spec.rb
@@ -0,0 +1,221 @@
+require 'spec_helper'
+
+module VCAP::CloudController
+  module Jobs::Runtime
+    RSpec.describe ServiceOperationsBindingDeleteStuckInProgressRetry, job_context: :worker do
+      subject(:job) { ServiceOperationsBindingDeleteStuckInProgressRetry.new }
+
+      let(:fake_logger) { instance_double(Steno::Logger, info: nil, warn: nil, error: nil) }
+      let(:max_poll_duration_minutes) { 60 }
+      let(:user_audit_info) { UserAuditInfo.new(user_guid: create(:user).guid, user_email: 'foo@example.com') }
+      let(:enqueuer) { instance_double(Jobs::GenericEnqueuer, enqueue_pollable: nil) }
+
+      before do
+        allow(Steno).to receive(:logger).and_return(fake_logger)
+        TestConfig.override(broker_client_max_async_poll_duration_minutes: max_poll_duration_minutes)
+        allow(Jobs::GenericEnqueuer).to receive(:shared).and_return(enqueuer)
+      end
+
+      # Enqueue a real DeleteBindingJob so the delayed_job carries a genuine serialized handler,
+      # then simulate the permanent failure (failed_at set) that leaves the operation stuck in progress.
+      def prepare_stuck_binding(
+        binding_type:,
+        operation_state: 'in progress',
+        operation_type: 'delete',
+        operation_created_at: Time.now,
+        pollable_job_state: PollableJobModel::FAILED_STATE,
+        pollable_job_operation: nil,
+        delayed_job_failed_at: Time.now
+      )
+        if binding_type == :credential
+          binding = create(:service_binding)
+          create(:service_binding_operation, service_binding_id: binding.id, type: operation_type, state: operation_state, created_at: operation_created_at)
+          default_operation = 'service_bindings.delete'
+          resource_type = 'service_bindings'
+        else
+          binding = create(:service_key)
+          create(:service_key_operation, service_key_id: binding.id, type: operation_type, state: operation_state, created_at: operation_created_at)
+          default_operation = 'service_keys.delete'
+          resource_type = 'service_keys'
+        end
+
+        delete_job = V3::DeleteBindingJob.new(binding_type, binding.guid, user_audit_info: user_audit_info)
+        pjob = Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue_pollable(delete_job)
+        pjob.update(state: pollable_job_state, operation: pollable_job_operation || default_operation, resource_type: resource_type)
+
+        dj = Delayed::Job[guid: pjob.delayed_job_guid]
+        dj.update(failed_at: delayed_job_failed_at)
+
+        { binding: binding, pjob: pjob, delayed_job: dj }
+      end
+
+      it { is_expected.to be_a_valid_job }
+
+      %i[credential key].each do |binding_type|
+        describe "#perform for #{binding_type} bindings" do
+          shared_examples 'does not retry the operation' do
+            it 'leaves the operation in progress, the pollable job untouched, and does not re-enqueue' do
+              scenario = subject_scenario
+              original_pollable_state = scenario[:pjob].state
+              job.perform
+              expect(scenario[:binding].last_operation.reload.state).to eq('in progress')
+              expect(scenario[:pjob].reload.state).to eq(original_pollable_state)
+              expect(enqueuer).not_to have_received(:enqueue_pollable)
+            end
+          end
+
+          context 'when operation state is not in progress' do
+            it 'does not retry when state is succeeded' do
+              scenario = prepare_stuck_binding(binding_type: binding_type, operation_state: 'succeeded')
+              job.perform
+              expect(scenario[:binding].last_operation.reload.state).to eq('succeeded')
+              expect(enqueuer).not_to have_received(:enqueue_pollable)
+            end
+
+            it 'does not retry when state is failed' do
+              scenario = prepare_stuck_binding(binding_type: binding_type, operation_state: 'failed')
+              job.perform
+              expect(scenario[:binding].last_operation.reload.state).to eq('failed')
+              expect(enqueuer).not_to have_received(:enqueue_pollable)
+            end
+          end
+
+          context 'when operation type is not delete' do
+            let(:subject_scenario) do
+              prepare_stuck_binding(binding_type: binding_type, operation_type: 'create',
+                                    pollable_job_operation: binding_type == :credential ? 'service_bindings.create' : 'service_keys.create')
+            end
+
+            it_behaves_like 'does not retry the operation'
+          end
+
+          context 'when operation created_at is beyond the max polling window' do
+            let(:subject_scenario) { prepare_stuck_binding(binding_type: binding_type, operation_created_at: Time.now - (max_poll_duration_minutes + 1).minutes) }
+
+            it_behaves_like 'does not retry the operation'
+          end
+
+          context 'when delayed_job.failed_at is nil (job still running or locked)' do
+            let(:subject_scenario) { prepare_stuck_binding(binding_type: binding_type, delayed_job_failed_at: nil) }
+
+            it_behaves_like 'does not retry the operation'
+          end
+
+          context 'when pollable job state is COMPLETE' do
+            let(:subject_scenario) { prepare_stuck_binding(binding_type: binding_type, pollable_job_state: PollableJobModel::COMPLETE_STATE) }
+
+            it_behaves_like 'does not retry the operation'
+          end
+
+          context 'when pollable job state is PROCESSING' do
+            let(:subject_scenario) { prepare_stuck_binding(binding_type: binding_type, pollable_job_state: PollableJobModel::PROCESSING_STATE) }
+
+            it_behaves_like 'does not retry the operation'
+          end
+
+          context 'when pollable job operation does not match the delete operation' do
+            let(:subject_scenario) do
+              prepare_stuck_binding(binding_type: binding_type,
+                                    pollable_job_operation: binding_type == :credential ? 'service_bindings.create' : 'service_keys.create')
+            end
+
+            it_behaves_like 'does not retry the operation'
+          end
+
+          context 'when a binding delete job is stuck with state FAILED' do
+            it 'resets the pollable job to POLLING and re-enqueues the original delete job' do
+              scenario = prepare_stuck_binding(binding_type: binding_type)
+              job.perform
+
+              expect(scenario[:binding].last_operation.reload.state).to eq('in progress')
+              expect(scenario[:pjob].reload.state).to eq(PollableJobModel::POLLING_STATE)
+              expect(enqueuer).to have_received(:enqueue_pollable).with(
+                an_instance_of(V3::DeleteBindingJob),
+                hash_including(existing_guid: scenario[:pjob].guid, preserve_priority: true)
+              )
+            end
+          end
+
+          context 'when a binding delete job is stuck with state POLLING (DB flip before failure hook)' do
+            it 'resets the pollable job to POLLING and re-enqueues the original delete job' do
+              scenario = prepare_stuck_binding(binding_type: binding_type, pollable_job_state: PollableJobModel::POLLING_STATE)
+              job.perform
+
+              expect(scenario[:pjob].reload.state).to eq(PollableJobModel::POLLING_STATE)
+              expect(enqueuer).to have_received(:enqueue_pollable).with(
+                an_instance_of(V3::DeleteBindingJob),
+                hash_including(existing_guid: scenario[:pjob].guid)
+              )
+            end
+          end
+
+          context 'when there are multiple stuck jobs within the batch size' do
+            it 'retries each one' do
+              3.times { prepare_stuck_binding(binding_type: binding_type) }
+              job.perform
+              expect(enqueuer).to have_received(:enqueue_pollable).exactly(3).times
+            end
+          end
+
+          context 'when there are more stuck jobs than the batch size' do
+            it 'processes only up to BATCH_SIZE jobs per run' do
+              (ServiceOperationsBindingDeleteStuckInProgressRetry::BATCH_SIZE + 1).times { prepare_stuck_binding(binding_type: binding_type) }
+              job.perform
+              expect(enqueuer).to have_received(:enqueue_pollable).exactly(ServiceOperationsBindingDeleteStuckInProgressRetry::BATCH_SIZE).times
+            end
+          end
+        end
+      end
+
+      describe '#perform cross-type isolation' do
+        it 'retries both a stuck credential-binding delete and a stuck key delete' do
+          prepare_stuck_binding(binding_type: :credential)
+          prepare_stuck_binding(binding_type: :key)
+          job.perform
+          expect(enqueuer).to have_received(:enqueue_pollable).exactly(2).times
+        end
+      end
+
+      describe '#resolve_stuck' do
+        context 'when another process already resolved it (skip_locked returns nil)' do
+          it 'does nothing and does not re-enqueue' do
+            scenario = prepare_stuck_binding(binding_type: :credential)
+
+            expect do
+              job.send(:resolve_stuck, ServiceBindingOperation, ServiceBinding,
+                       -1, scenario[:binding].id, scenario[:pjob].guid)
+            end.not_to raise_error
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::FAILED_STATE)
+            expect(enqueuer).not_to have_received(:enqueue_pollable)
+          end
+        end
+
+        context 'when the delayed job handler cannot be deserialized' do
+          it 'does not re-enqueue and leaves the pollable job untouched' do
+            scenario = prepare_stuck_binding(binding_type: :credential)
+            Delayed::Job[guid: scenario[:pjob].delayed_job_guid].update(handler: 'not-valid-yaml: ]')
+            op = scenario[:binding].last_operation
+
+            job.send(:resolve_stuck, ServiceBindingOperation, ServiceBinding,
+                     op.id, scenario[:binding].id, scenario[:pjob].guid)
+
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::FAILED_STATE)
+            expect(enqueuer).not_to have_received(:enqueue_pollable)
+          end
+        end
+
+        context 'when the operation is stuck in progress' do
+          it 'resets the pollable job from its failed state to POLLING' do
+            scenario = prepare_stuck_binding(binding_type: :credential)
+            op = scenario[:binding].last_operation
+
+            expect do
+              job.send(:resolve_stuck, ServiceBindingOperation, ServiceBinding,
+                       op.id, scenario[:binding].id, scenario[:pjob].guid)
+            end.to change { scenario[:pjob].reload.state }.from(PollableJobModel::FAILED_STATE).to(PollableJobModel::POLLING_STATE)
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/spec/unit/jobs/runtime/service_operations_create_in_progress_cleanup_spec.rb b/spec/unit/jobs/runtime/service_operations_create_in_progress_cleanup_spec.rb
index 5980373b90b..d76b701b3cb 100644
--- a/spec/unit/jobs/runtime/service_operations_create_in_progress_cleanup_spec.rb
+++ b/spec/unit/jobs/runtime/service_operations_create_in_progress_cleanup_spec.rb
@@ -55,6 +55,24 @@ def prepare_stuck_service_instance(
         { service_instance: service_instance, pjob: pjob, delayed_job: dj }
       end
 
+      # Attach an additional live pollable job (POLLING/PROCESSING, delayed_job NOT failed)
+      # for the same instance + operation. Mirrors a second create actively polling while a
+      # stale, permanently-failed pollable from a previous attempt lingers.
+      def add_live_pollable(service_instance, operation: 'service_instance.create', state: PollableJobModel::POLLING_STATE)
+        dj = Delayed::Job.create!(
+          guid: SecureRandom.uuid,
+          handler: 'fake',
+          run_at: Time.now,
+          queue: 'cc-generic'
+        )
+        create(:pollable_job_model,
+               state: state,
+               operation: operation,
+               resource_guid: service_instance.guid,
+               resource_type: 'service_instances',
+               delayed_job_guid: dj.guid)
+      end
+
       shared_examples 'does not trigger orphan mitigation' do
         before { job.perform }
 
@@ -119,6 +137,34 @@ def prepare_stuck_service_instance(
           it_behaves_like 'does not trigger orphan mitigation'
         end
 
+        context 'when a live pollable job is still driving the same operation' do
+          # A previous create attempt left a stale, permanently-failed pollable behind; a
+          # second create on the same instance is now actively polling. The stale row must
+          # not cause the healthy current operation to be marked failed / mitigated.
+          it 'does not mitigate and leaves both pollables untouched' do
+            scenario = prepare_stuck_service_instance
+            live_pjob = add_live_pollable(scenario[:service_instance])
+
+            job.perform
+
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('in progress')
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::FAILED_STATE)
+            expect(live_pjob.reload.state).to eq(PollableJobModel::POLLING_STATE)
+            expect(fake_mitigator).not_to have_received(:cleanup_failed_provision)
+          end
+
+          it 'still mitigates once the live pollable is gone' do
+            scenario = prepare_stuck_service_instance
+            live_pjob = add_live_pollable(scenario[:service_instance])
+            live_pjob.destroy
+
+            job.perform
+
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('failed')
+            expect(fake_mitigator).to have_received(:cleanup_failed_provision).with(scenario[:service_instance])
+          end
+        end
+
         context 'when a service instance create job is stuck with state FAILED' do
           it 'sets operation to failed, pollable job to FAILED, and triggers orphan mitigation' do
             scenario = prepare_stuck_service_instance
diff --git a/spec/unit/jobs/runtime/service_operations_delete_stuck_in_progress_retry_spec.rb b/spec/unit/jobs/runtime/service_operations_delete_stuck_in_progress_retry_spec.rb
new file mode 100644
index 00000000000..79f70339519
--- /dev/null
+++ b/spec/unit/jobs/runtime/service_operations_delete_stuck_in_progress_retry_spec.rb
@@ -0,0 +1,199 @@
+require 'spec_helper'
+
+module VCAP::CloudController
+  module Jobs::Runtime
+    RSpec.describe ServiceOperationsDeleteStuckInProgressRetry, job_context: :worker do
+      subject(:job) { ServiceOperationsDeleteStuckInProgressRetry.new }
+
+      let(:fake_logger) { instance_double(Steno::Logger, info: nil, warn: nil, error: nil) }
+      let(:max_poll_duration_minutes) { 60 }
+      let(:user_audit_info) { UserAuditInfo.new(user_guid: create(:user).guid, user_email: 'foo@example.com') }
+      let(:enqueuer) { instance_double(Jobs::GenericEnqueuer, enqueue_pollable: nil) }
+
+      before do
+        allow(Steno).to receive(:logger).and_return(fake_logger)
+        TestConfig.override(broker_client_max_async_poll_duration_minutes: max_poll_duration_minutes)
+        allow(Jobs::GenericEnqueuer).to receive(:shared).and_return(enqueuer)
+      end
+
+      # Enqueue a real DeleteServiceInstanceJob so the delayed_job carries a genuine serialized handler,
+      # then simulate the permanent failure (failed_at set) that leaves the operation stuck in progress.
+      def prepare_stuck_service_instance(
+        service_instance_state: 'in progress',
+        service_instance_type: 'delete',
+        service_instance_created_at: Time.now,
+        pollable_job_state: PollableJobModel::FAILED_STATE,
+        pollable_job_operation: 'service_instance.delete',
+        delayed_job_failed_at: Time.now
+      )
+        service_instance = create(:managed_service_instance)
+
+        create(:service_instance_operation,
+               service_instance_id: service_instance.id,
+               type: service_instance_type,
+               state: service_instance_state,
+               created_at: service_instance_created_at)
+
+        delete_job = V3::DeleteServiceInstanceJob.new(service_instance.guid, user_audit_info)
+        pjob = Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue_pollable(delete_job)
+        pjob.update(state: pollable_job_state, operation: pollable_job_operation)
+
+        dj = Delayed::Job[guid: pjob.delayed_job_guid]
+        dj.update(failed_at: delayed_job_failed_at)
+
+        { service_instance: service_instance, pjob: pjob, delayed_job: dj }
+      end
+
+      shared_examples 'does not retry the operation' do
+        it 'leaves the operation in progress, the pollable job untouched, and does not re-enqueue' do
+          scenario = subject_scenario
+          original_pollable_state = scenario[:pjob].state
+          job.perform
+          expect(scenario[:service_instance].last_operation.reload.state).to eq('in progress')
+          expect(scenario[:pjob].reload.state).to eq(original_pollable_state)
+          expect(enqueuer).not_to have_received(:enqueue_pollable)
+        end
+      end
+
+      it { is_expected.to be_a_valid_job }
+
+      describe '#perform' do
+        context 'when sio state is not in progress' do
+          it 'does not retry when state is succeeded' do
+            scenario = prepare_stuck_service_instance(service_instance_state: 'succeeded')
+            job.perform
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('succeeded')
+            expect(enqueuer).not_to have_received(:enqueue_pollable)
+          end
+
+          it 'does not retry when state is failed' do
+            scenario = prepare_stuck_service_instance(service_instance_state: 'failed')
+            job.perform
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('failed')
+            expect(enqueuer).not_to have_received(:enqueue_pollable)
+          end
+        end
+
+        context 'when sio type is not delete' do
+          let(:subject_scenario) { prepare_stuck_service_instance(service_instance_type: 'update', pollable_job_operation: 'service_instance.update') }
+
+          it_behaves_like 'does not retry the operation'
+        end
+
+        context 'when sio created_at is beyond the max polling window' do
+          let(:subject_scenario) { prepare_stuck_service_instance(service_instance_created_at: Time.now - (max_poll_duration_minutes + 1).minutes) }
+
+          it_behaves_like 'does not retry the operation'
+        end
+
+        context 'when delayed_job.failed_at is nil (job still running or locked)' do
+          let(:subject_scenario) { prepare_stuck_service_instance(delayed_job_failed_at: nil) }
+
+          it_behaves_like 'does not retry the operation'
+        end
+
+        context 'when pollable job state is COMPLETE' do
+          let(:subject_scenario) { prepare_stuck_service_instance(pollable_job_state: PollableJobModel::COMPLETE_STATE) }
+
+          it_behaves_like 'does not retry the operation'
+        end
+
+        context 'when pollable job state is PROCESSING' do
+          let(:subject_scenario) { prepare_stuck_service_instance(pollable_job_state: PollableJobModel::PROCESSING_STATE) }
+
+          it_behaves_like 'does not retry the operation'
+        end
+
+        context 'when pollable job operation is not service_instance.delete' do
+          let(:subject_scenario) { prepare_stuck_service_instance(pollable_job_operation: 'service_instance.create') }
+
+          it_behaves_like 'does not retry the operation'
+        end
+
+        context 'when a service instance delete job is stuck with state FAILED' do
+          it 'resets the pollable job to POLLING and re-enqueues the original delete job' do
+            scenario = prepare_stuck_service_instance
+            job.perform
+
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('in progress')
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::POLLING_STATE)
+            expect(enqueuer).to have_received(:enqueue_pollable).with(
+              an_instance_of(V3::DeleteServiceInstanceJob),
+              hash_including(existing_guid: scenario[:pjob].guid, preserve_priority: true)
+            )
+          end
+        end
+
+        context 'when a service instance delete job is stuck with state POLLING (DB flip before failure hook)' do
+          it 'resets the pollable job to POLLING and re-enqueues the original delete job' do
+            scenario = prepare_stuck_service_instance(pollable_job_state: PollableJobModel::POLLING_STATE)
+            job.perform
+
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::POLLING_STATE)
+            expect(enqueuer).to have_received(:enqueue_pollable).with(
+              an_instance_of(V3::DeleteServiceInstanceJob),
+              hash_including(existing_guid: scenario[:pjob].guid)
+            )
+          end
+        end
+
+        context 'when there are multiple stuck jobs within the batch size' do
+          it 'retries each one' do
+            3.times { prepare_stuck_service_instance }
+            job.perform
+            expect(enqueuer).to have_received(:enqueue_pollable).exactly(3).times
+          end
+        end
+
+        context 'when there are more stuck jobs than the batch size' do
+          it 'processes only up to BATCH_SIZE jobs per run' do
+            (ServiceOperationsDeleteStuckInProgressRetry::BATCH_SIZE + 1).times { prepare_stuck_service_instance }
+            job.perform
+            expect(enqueuer).to have_received(:enqueue_pollable).exactly(ServiceOperationsDeleteStuckInProgressRetry::BATCH_SIZE).times
+          end
+        end
+      end
+
+      describe '#resolve_stuck' do
+        context 'when another process already resolved it (skip_locked returns nil)' do
+          it 'does nothing and does not re-enqueue' do
+            scenario = prepare_stuck_service_instance
+
+            expect do
+              job.send(:resolve_stuck, ServiceInstanceOperation, ServiceInstance,
+                       -1, scenario[:service_instance].id, scenario[:pjob].guid)
+            end.not_to raise_error
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::FAILED_STATE)
+            expect(enqueuer).not_to have_received(:enqueue_pollable)
+          end
+        end
+
+        context 'when the delayed job handler cannot be deserialized' do
+          it 'does not re-enqueue and leaves the pollable job untouched' do
+            scenario = prepare_stuck_service_instance
+            Delayed::Job[guid: scenario[:pjob].delayed_job_guid].update(handler: 'not-valid-yaml: ]')
+            op = scenario[:service_instance].last_operation
+
+            job.send(:resolve_stuck, ServiceInstanceOperation, ServiceInstance,
+                     op.id, scenario[:service_instance].id, scenario[:pjob].guid)
+
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::FAILED_STATE)
+            expect(enqueuer).not_to have_received(:enqueue_pollable)
+          end
+        end
+
+        context 'when the operation is stuck in progress' do
+          it 'resets the pollable job from its failed state to POLLING' do
+            scenario = prepare_stuck_service_instance
+            op = scenario[:service_instance].last_operation
+
+            expect do
+              job.send(:resolve_stuck, ServiceInstanceOperation, ServiceInstance,
+                       op.id, scenario[:service_instance].id, scenario[:pjob].guid)
+            end.to change { scenario[:pjob].reload.state }.from(PollableJobModel::FAILED_STATE).to(PollableJobModel::POLLING_STATE)
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/spec/unit/jobs/runtime/service_operations_update_stuck_in_progress_failed_spec.rb b/spec/unit/jobs/runtime/service_operations_update_stuck_in_progress_failed_spec.rb
new file mode 100644
index 00000000000..6489397ae47
--- /dev/null
+++ b/spec/unit/jobs/runtime/service_operations_update_stuck_in_progress_failed_spec.rb
@@ -0,0 +1,247 @@
+require 'spec_helper'
+
+module VCAP::CloudController
+  module Jobs::Runtime
+    RSpec.describe ServiceOperationsUpdateStuckInProgressFailed, job_context: :worker do
+      subject(:job) { ServiceOperationsUpdateStuckInProgressFailed.new }
+
+      let(:fake_logger) { instance_double(Steno::Logger, info: nil, warn: nil) }
+      let(:max_poll_duration_minutes) { 60 }
+
+      before do
+        allow(Steno).to receive(:logger).and_return(fake_logger)
+        TestConfig.override(broker_client_max_async_poll_duration_minutes: max_poll_duration_minutes)
+      end
+
+      def prepare_stuck_service_instance(
+        service_instance_state: 'in progress',
+        service_instance_type: 'update',
+        service_instance_created_at: Time.now,
+        pollable_job_state: PollableJobModel::FAILED_STATE,
+        pollable_job_operation: 'service_instance.update',
+        delayed_job_failed_at: Time.now
+      )
+        service_instance = create(:managed_service_instance)
+
+        create(:service_instance_operation,
+               service_instance_id: service_instance.id,
+               type: service_instance_type,
+               state: service_instance_state,
+               created_at: service_instance_created_at)
+
+        dj = Delayed::Job.create!(
+          guid: SecureRandom.uuid,
+          handler: 'fake',
+          run_at: Time.now,
+          failed_at: delayed_job_failed_at,
+          queue: 'cc-generic'
+        )
+
+        pjob = create(:pollable_job_model,
+                      state: pollable_job_state,
+                      operation: pollable_job_operation,
+                      resource_guid: service_instance.guid,
+                      resource_type: 'service_instances',
+                      delayed_job_guid: dj.guid)
+
+        { service_instance: service_instance, pjob: pjob, delayed_job: dj }
+      end
+
+      # Attach an additional live pollable job (POLLING/PROCESSING, no failed delayed_job)
+      # for the same resource + operation. This mirrors a second update that is actively
+      # polling while a stale, permanently-failed pollable from a previous operation lingers.
+      def add_live_pollable(service_instance, operation: 'service_instance.update', state: PollableJobModel::POLLING_STATE)
+        dj = Delayed::Job.create!(
+          guid: SecureRandom.uuid,
+          handler: 'fake',
+          run_at: Time.now,
+          queue: 'cc-generic'
+        )
+        create(:pollable_job_model,
+               state: state,
+               operation: operation,
+               resource_guid: service_instance.guid,
+               resource_type: 'service_instances',
+               delayed_job_guid: dj.guid)
+      end
+
+      shared_examples 'does not resolve the operation' do
+        it 'leaves the operation in progress and the pollable job untouched' do
+          scenario = subject_scenario
+          original_pollable_state = scenario[:pjob].state
+          job.perform
+          expect(scenario[:service_instance].last_operation.reload.state).to eq('in progress')
+          expect(scenario[:pjob].reload.state).to eq(original_pollable_state)
+        end
+      end
+
+      it { is_expected.to be_a_valid_job }
+
+      describe '#perform' do
+        context 'when sio state is not in progress' do
+          it 'does not resolve when state is succeeded' do
+            scenario = prepare_stuck_service_instance(service_instance_state: 'succeeded')
+            job.perform
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('succeeded')
+          end
+
+          it 'does not resolve when state is failed' do
+            scenario = prepare_stuck_service_instance(service_instance_state: 'failed')
+            job.perform
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('failed')
+          end
+        end
+
+        context 'when sio type is not update' do
+          let(:subject_scenario) { prepare_stuck_service_instance(service_instance_type: 'create') }
+
+          it_behaves_like 'does not resolve the operation'
+        end
+
+        context 'when sio created_at is beyond the max polling window' do
+          let(:subject_scenario) { prepare_stuck_service_instance(service_instance_created_at: Time.now - (max_poll_duration_minutes + 1).minutes) }
+
+          it_behaves_like 'does not resolve the operation'
+        end
+
+        context 'when delayed_job.failed_at is nil (job still running or locked)' do
+          let(:subject_scenario) { prepare_stuck_service_instance(delayed_job_failed_at: nil) }
+
+          it_behaves_like 'does not resolve the operation'
+        end
+
+        context 'when pollable job state is COMPLETE' do
+          let(:subject_scenario) { prepare_stuck_service_instance(pollable_job_state: PollableJobModel::COMPLETE_STATE) }
+
+          it_behaves_like 'does not resolve the operation'
+        end
+
+        context 'when pollable job state is PROCESSING' do
+          let(:subject_scenario) { prepare_stuck_service_instance(pollable_job_state: PollableJobModel::PROCESSING_STATE) }
+
+          it_behaves_like 'does not resolve the operation'
+        end
+
+        context 'when pollable job operation is not service_instance.update' do
+          let(:subject_scenario) { prepare_stuck_service_instance(pollable_job_operation: 'service_instance.create') }
+
+          it_behaves_like 'does not resolve the operation'
+        end
+
+        context 'when a live pollable job is still driving the same operation' do
+          # A previous update left a stale, permanently-failed pollable behind; a second
+          # update on the same instance is now actively polling. The stale row must not
+          # cause the healthy current operation to be marked failed.
+          it 'does not resolve the operation and leaves the live pollable untouched' do
+            scenario = prepare_stuck_service_instance
+            live_pjob = add_live_pollable(scenario[:service_instance])
+
+            job.perform
+
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('in progress')
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::FAILED_STATE)
+            expect(live_pjob.reload.state).to eq(PollableJobModel::POLLING_STATE)
+          end
+
+          it 'still resolves once the live pollable is gone' do
+            scenario = prepare_stuck_service_instance
+            live_pjob = add_live_pollable(scenario[:service_instance])
+            live_pjob.destroy
+
+            job.perform
+
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('failed')
+          end
+
+          it 'is unaffected by an unrelated live pollable with a NULL resource_guid (NOT EXISTS, not NOT IN)' do
+            # A NOT IN subquery selecting jobs.resource_guid would evaluate to UNKNOWN for
+            # every row if any candidate row has a NULL resource_guid, silently disabling the
+            # whole job. The correlated NOT EXISTS form is immune. Guard against regressing.
+            scenario = prepare_stuck_service_instance
+            dj = Delayed::Job.create!(guid: SecureRandom.uuid, handler: 'fake', run_at: Time.now, queue: 'cc-generic')
+            create(:pollable_job_model,
+                   state: PollableJobModel::POLLING_STATE,
+                   operation: 'service_instance.update',
+                   resource_guid: nil,
+                   resource_type: 'service_instances',
+                   delayed_job_guid: dj.guid)
+
+            job.perform
+
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('failed')
+          end
+        end
+
+        context 'when a service instance update job is stuck with state FAILED' do
+          it 'sets operation to failed and pollable job to FAILED' do
+            scenario = prepare_stuck_service_instance
+            job.perform
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('failed')
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::FAILED_STATE)
+          end
+        end
+
+        context 'when a service instance update job is stuck with state POLLING (DB flip before failure hook)' do
+          it 'sets operation to failed and pollable job to FAILED' do
+            scenario = prepare_stuck_service_instance(pollable_job_state: PollableJobModel::POLLING_STATE)
+            job.perform
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('failed')
+            expect(scenario[:pjob].reload.state).to eq(PollableJobModel::FAILED_STATE)
+          end
+        end
+
+        context 'when there are multiple stuck jobs within the batch size' do
+          it 'resolves each one' do
+            3.times { prepare_stuck_service_instance }
+            job.perform
+            expect(ServiceInstanceOperation.where(state: 'failed').count).to eq(3)
+          end
+        end
+
+        context 'when there are more stuck jobs than the batch size' do
+          it 'processes only up to BATCH_SIZE jobs per run' do
+            (ServiceOperationsUpdateStuckInProgressFailed::BATCH_SIZE + 1).times { prepare_stuck_service_instance }
+            job.perform
+            expect(ServiceInstanceOperation.where(state: 'failed').count).to eq(ServiceOperationsUpdateStuckInProgressFailed::BATCH_SIZE)
+          end
+        end
+      end
+
+      describe '#resolve_stuck' do
+        context 'when another process already resolved it (skip_locked returns nil)' do
+          it 'does nothing' do
+            scenario = prepare_stuck_service_instance
+
+            expect do
+              job.send(:resolve_stuck, ServiceInstanceOperation, ServiceInstance,
+                       -1, scenario[:service_instance].id, scenario[:pjob].guid)
+            end.not_to raise_error
+            expect(scenario[:service_instance].last_operation.reload.state).to eq('in progress')
+          end
+        end
+
+        context 'when the operation is stuck in progress' do
+          it 'sets the operation state from in progress to failed' do
+            scenario = prepare_stuck_service_instance
+            op = scenario[:service_instance].last_operation
+
+            expect do
+              job.send(:resolve_stuck, ServiceInstanceOperation, ServiceInstance,
+                       op.id, scenario[:service_instance].id, scenario[:pjob].guid)
+            end.to change { op.reload.state }.from('in progress').to('failed')
+          end
+
+          it 'sets the pollable job state to FAILED' do
+            scenario = prepare_stuck_service_instance(pollable_job_state: PollableJobModel::POLLING_STATE)
+            op = scenario[:service_instance].last_operation
+
+            expect do
+              job.send(:resolve_stuck, ServiceInstanceOperation, ServiceInstance,
+                       op.id, scenario[:service_instance].id, scenario[:pjob].guid)
+            end.to change { scenario[:pjob].reload.state }.from(PollableJobModel::POLLING_STATE).to(PollableJobModel::FAILED_STATE)
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/spec/unit/lib/cloud_controller/clock/scheduler_spec.rb b/spec/unit/lib/cloud_controller/clock/scheduler_spec.rb
index 21c5b94d459..5e9fa8013a0 100644
--- a/spec/unit/lib/cloud_controller/clock/scheduler_spec.rb
+++ b/spec/unit/lib/cloud_controller/clock/scheduler_spec.rb
@@ -22,6 +22,9 @@ module VCAP::CloudController
           pollable_jobs: { cutoff_age_in_days: 2 },
           service_operations_initial_cleanup: { frequency_in_seconds: 600 },
           service_operations_create_in_progress_cleanup: { frequency_in_seconds: 600 },
+          service_operations_update_stuck_in_progress_failed: { frequency_in_seconds: 600 },
+          service_operations_delete_stuck_in_progress_retry: { frequency_in_seconds: 600 },
+          service_operations_binding_delete_stuck_in_progress_retry: { frequency_in_seconds: 600 },
           lifecycle_type_backfill: { frequency_in_seconds: 500 },
           service_usage_events: { cutoff_age_in_days: 5 },
           completed_tasks: { cutoff_age_in_days: 6 },
@@ -169,6 +172,24 @@ module VCAP::CloudController
           expect(block.call).to be_instance_of(Jobs::Runtime::ServiceOperationsCreateInProgressCleanup)
         end
 
+        expect(clock).to receive(:schedule_frequent_worker_job) do |args, &block|
+          expect(args).to eql(name: 'service_operations_update_stuck_in_progress_failed', interval: 600)
+          expect(Jobs::Runtime::ServiceOperationsUpdateStuckInProgressFailed).to receive(:new).and_call_original
+          expect(block.call).to be_instance_of(Jobs::Runtime::ServiceOperationsUpdateStuckInProgressFailed)
+        end
+
+        expect(clock).to receive(:schedule_frequent_worker_job) do |args, &block|
+          expect(args).to eql(name: 'service_operations_delete_stuck_in_progress_retry', interval: 600)
+          expect(Jobs::Runtime::ServiceOperationsDeleteStuckInProgressRetry).to receive(:new).and_call_original
+          expect(block.call).to be_instance_of(Jobs::Runtime::ServiceOperationsDeleteStuckInProgressRetry)
+        end
+
+        expect(clock).to receive(:schedule_frequent_worker_job) do |args, &block|
+          expect(args).to eql(name: 'service_operations_binding_delete_stuck_in_progress_retry', interval: 600)
+          expect(Jobs::Runtime::ServiceOperationsBindingDeleteStuckInProgressRetry).to receive(:new).and_call_original
+          expect(block.call).to be_instance_of(Jobs::Runtime::ServiceOperationsBindingDeleteStuckInProgressRetry)
+        end
+
         expect(clock).to receive(:schedule_frequent_worker_job) do |args, &block|
           expect(args).to eql(name: 'lifecycle_type_backfill', interval: 500)
           expect(Jobs::Runtime::LifecycleTypeBackfill).to receive(:new).and_call_original