diff --git a/CHANGELOG.md b/CHANGELOG.md
index 46cf05e..4c34a76 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,29 @@
 
 ## Unreleased
 
+- Retry a dead effect or broadcast. `SolidObjects.dead_letters` keeps its
+  message meaning and answers `effects` and `broadcasts`, so the kind rides on
+  the receiver. `retry` returns a dead row to pending with a zero attempt count
+  and no claim, reuses the stable id so a deduplicating handler sees the same
+  key, and acts only on a dead row, so a second press cannot double-enqueue. A
+  dead transmit effect replays rather than stay lost.
+- Redrive a whole scope. `redrive` opens a durable task, returns at once, and is
+  idempotent over its scope and filters, which a dashboard button needs. A
+  unique index on the active scope enforces that in the database, so two
+  processes that start the same redrive share one task. The supervisor advances
+  one bounded batch per pass, so a redrive never holds a transaction longer than
+  one batch. `SolidObjects.redrives` reads tasks back, and `task.cancel` stops
+  one and leaves the rows it already moved. A redrive moves what was dead when
+  it started, so a still-broken handler cannot make it run forever.
+- Record who pressed what. Every retry and every redrive transition writes one
+  row to `solid_objects_administration_events`. The identity comes from the
+  authorization context through a new `administration_identity` hook.
+- Add `redrive_batch_size`, which defaults to 100, and `redrive_batch_pause`,
+  which defaults to 0.05 seconds.
+- Add two tables, `solid_objects_administration_events` and
+  `solid_objects_redrives`. Run `bin/rails solid_objects:install:migrations` and
+  migrate.
+
 - Select a wake-up adapter automatically. `config.wake_up_adapter` now takes a
   name or an adapter, as `config.cache_store` and
   `config.active_job.queue_adapter` do, and defaults to `:automatic`. Selection
diff --git a/app/models/solid_objects/administration_event.rb b/app/models/solid_objects/administration_event.rb
new file mode 100644
index 0000000..fbfdd20
--- /dev/null
+++ b/app/models/solid_objects/administration_event.rb
@@ -0,0 +1,7 @@
+# rbs_inline: enabled
+
+module SolidObjects
+  class AdministrationEvent < Record
+    self.table_name = SolidObjects.table_name(:administration_events)
+  end
+end
diff --git a/app/models/solid_objects/redrive.rb b/app/models/solid_objects/redrive.rb
new file mode 100644
index 0000000..3687698
--- /dev/null
+++ b/app/models/solid_objects/redrive.rb
@@ -0,0 +1,7 @@
+# rbs_inline: enabled
+
+module SolidObjects
+  class Redrive < Record
+    self.table_name = SolidObjects.table_name(:redrives)
+  end
+end
diff --git a/db/migrate/20260922000000_add_solid_objects_administration_events.rb b/db/migrate/20260922000000_add_solid_objects_administration_events.rb
new file mode 100644
index 0000000..95e69c4
--- /dev/null
+++ b/db/migrate/20260922000000_add_solid_objects_administration_events.rb
@@ -0,0 +1,26 @@
+# rbs_inline: enabled
+
+class AddSolidObjectsAdministrationEvents < ActiveRecord::Migration[7.1]
+  # @rbs () -> void
+  def change
+    create_table SolidObjects.table_name(:administration_events) do |definition|
+      definition.string :action, null: false, limit: 64
+      definition.string :kind, null: false, limit: 32
+      definition.string :subject_id, limit: 191
+      definition.public_send(json_type, :filters)
+      definition.string :actor, limit: 255
+      definition.datetime :occurred_at, null: false, precision: 6
+      definition.timestamps precision: 6, null: false
+
+      definition.index [ :occurred_at, :id ], name: "idx_so_admin_events_occurred"
+      definition.index [ :kind, :subject_id ], name: "idx_so_admin_events_subject"
+    end
+  end
+
+  private
+
+  # @rbs () -> Symbol
+  def json_type
+    connection.adapter_name.match?(/postgres/i) ? :jsonb : :json
+  end
+end
diff --git a/db/migrate/20260922000001_add_solid_objects_redrives.rb b/db/migrate/20260922000001_add_solid_objects_redrives.rb
new file mode 100644
index 0000000..4da7471
--- /dev/null
+++ b/db/migrate/20260922000001_add_solid_objects_redrives.rb
@@ -0,0 +1,32 @@
+# rbs_inline: enabled
+
+class AddSolidObjectsRedrives < ActiveRecord::Migration[7.1]
+  # @rbs () -> void
+  def change
+    create_table SolidObjects.table_name(:redrives), id: :string, limit: 64 do |definition|
+      definition.string :kind, null: false, limit: 32
+      definition.public_send(json_type, :filters, null: false)
+      definition.string :status, null: false, default: "running", limit: 32
+      definition.string :active_scope, limit: 191
+      definition.integer :moved, null: false, default: 0
+      definition.integer :move_limit
+      definition.string :actor, limit: 255
+      definition.datetime :started_at, null: false, precision: 6
+      definition.datetime :finished_at, precision: 6
+      definition.timestamps precision: 6, null: false
+
+      definition.index :active_scope, unique: true, name: "idx_so_redrives_active_scope"
+      definition.index [ :status, :started_at, :id ], name: "idx_so_redrives_poll"
+      definition.check_constraint "moved >= 0", name: "chk_so_redrives_moved"
+      definition.check_constraint "move_limit IS NULL OR move_limit > 0", name: "chk_so_redrives_limit"
+      definition.check_constraint "status IN ('running', 'completed', 'cancelled')", name: "chk_so_redrives_status"
+    end
+  end
+
+  private
+
+  # @rbs () -> Symbol
+  def json_type
+    connection.adapter_name.match?(/postgres/i) ? :jsonb : :json
+  end
+end
diff --git a/docs/dashboard.md b/docs/dashboard.md
index 395c9a5..6a408a2 100644
--- a/docs/dashboard.md
+++ b/docs/dashboard.md
@@ -149,6 +149,16 @@ actor class that no longer exists, a full mailbox, a payload over the cap. The
 dashboard renders the dead letter again with the reason and a 422 status,
 rather than failing the request.
 
+Dead effects and broadcasts have the same API, which the dashboard does not yet
+surface. `SolidObjects.dead_letters.effects` and
+`SolidObjects.dead_letters.broadcasts` read and retry their own kind, and
+`redrive` moves a whole scope as a durable task. See
+[Operations](operations.md) for both.
+
+Every retry and every redrive transition writes one row to
+`solid_objects_administration_events`, holding the action, the kind, the
+subject, and the identity that asked for it.
+
 **Pause an instance** sets `paused_at`, and the activation manager stops
 claiming that identity. Two consequences matter:
 
diff --git a/docs/operations.md b/docs/operations.md
index 38b219f..f11563f 100644
--- a/docs/operations.md
+++ b/docs/operations.md
@@ -282,6 +282,95 @@ interval, and current interval. The polling-only warning is also emitted as
 adapters should return `true` for a notification and `false` for a timeout; an
 older adapter that returns `nil` remains compatible and keeps the fast cadence.
 
+## Dead letters, retry, and redrive
+
+A message that exhausts its attempts becomes a dead letter. An effect or a
+broadcast that exhausts its attempts stays in its own table with
+`status = 'dead'`. All three are read and retried through one receiver, which
+carries the kind:
+
+```ruby
+SolidObjects.dead_letters.all(authorization_context: current_admin)
+SolidObjects.dead_letters.retry(dead_letter_id, authorization_context: current_admin)
+
+SolidObjects.dead_letters.effects.all(authorization_context: current_admin)
+SolidObjects.dead_letters.effects.retry(effect_id, authorization_context: current_admin)
+SolidObjects.dead_letters.broadcasts.retry(broadcast_id, authorization_context: current_admin)
+```
+
+An effect or broadcast retry returns the row to pending with a zero attempt
+count, no claim, and immediate availability. It keeps the stable id, so a
+handler that deduplicates on `effect_id` still sees the same key. An effect is
+at-least-once by contract, so a retried effect can run twice.
+
+Retry acts only on a dead row. A row that is pending, processing, or completed
+comes back unchanged, so pressing a button twice cannot double-enqueue and
+cannot take a row away from a worker that holds it.
+
+An incident produces dead rows in the hundreds, so a scope also answers
+`redrive`:
+
+```ruby
+task = SolidObjects.dead_letters.effects.redrive(
+  actor_type: "payments",
+  failed_after: 6.hours.ago,
+  limit: 5_000,
+  authorization_context: current_admin
+)
+
+task.id         # => "redrive_..."
+task.status     # => "running"
+task.moved      # => 412
+task.remaining  # => 4_588
+
+task.cancel(authorization_context: current_admin)
+```
+
+`redrive` returns at once. The task is durable, and the supervisor advances one
+bounded batch per pass, so a redrive of thousands of rows never holds a
+transaction longer than one batch. `redrive_batch_size` defaults to 100 and
+`redrive_batch_pause` to 0.05 seconds.
+
+A redrive moves the rows that were already dead when it started. A row that
+fails again lands back in the same scope, and without that bound a task whose
+handler is still broken would move it forever.
+
+A redrive is idempotent over its scope and its filters. Starting the same one
+while it runs returns the running task rather than a second one, which a
+dashboard button an operator can press twice needs. A different scope or a
+different filter starts its own task, and the same scope can be redriven again
+once the first task finishes.
+
+Read tasks back with `SolidObjects.redrives`:
+
+```ruby
+SolidObjects.redrives.find(task.id, authorization_context: current_admin)
+SolidObjects.redrives.all(status: :running, authorization_context: current_admin)
+```
+
+A running task reports what is left to move rather than a stored estimate,
+because rows die and are retried while it runs.
+
+Retry, redrive, and cancel each go through `authorize_administration` under
+their own resource name: `dead_letters`, `effect_dead_letters`,
+`broadcast_dead_letters`, and `redrives`. Every retry and every task transition
+writes one row to `solid_objects_administration_events`, holding the action, the
+kind, the subject, the identity, and when it happened. The identity comes from
+`administration_identity`, which receives the authorization context the caller
+passed and defaults to its `to_s`.
+
+An event records an authorized press, not a state transition. Pressing retry
+twice writes two rows, because an operator did two things and a log that shows
+one cannot answer who pressed what. The row the event names carries the outcome.
+A refused caller writes nothing, and a retry that raises after the lookup writes
+nothing, because the event shares the transaction with the work. The redrive
+transitions are different: `redrive.start`, `redrive.finish`, and
+`redrive.cancel` are written only when the task actually changes.
+
+Automatic redrive on a schedule is deliberately absent. A dead row means a
+person decided something, and these APIs give that person an alternative to an
+`UPDATE` against a runtime table.
+
 ## Graceful shutdown
 
 The supervisor requests shutdown, stops new claims, lets active loops return,
diff --git a/docs/roadmap.md b/docs/roadmap.md
index b53f245..8fe2c27 100644
--- a/docs/roadmap.md
+++ b/docs/roadmap.md
@@ -178,15 +178,18 @@
   or turn off. Every route declares its
   own administration policy and a route declared without one raises at load
   time, so the deny-by-default posture is enforced by construction rather than
-  by remembering to add a check. It changes only two things: an idempotent dead
-  letter retry and instance pause/resume. What does not exist is audit records
-  of who pressed what, and bulk-safe tools: retry is one dead letter at a time,
-  because `DeadLetterManager` exposes no bulk operation. Pause is an operator
-  brake and not a stop, since a pass already in flight finishes its turn and a
-  synchronous caller waiting on a paused instance times out. Retry also only
-  exists for message dead letters: a dead effect or broadcast has no retry
-  API, which matters for transmit effects because a dead one is a lost
-  replay until an operator returns its row to pending. The page cost was
+  by remembering to add a check. It changes only three things: an idempotent dead
+  letter retry, a redrive, and instance pause/resume. Retry covers all three
+  kinds. `SolidObjects.dead_letters` keeps its message meaning and answers
+  `effects` and `broadcasts`, so a dead effect or broadcast returns to pending
+  through an API rather than an operator's `UPDATE`, and a dead transmit effect
+  is no longer a lost replay. `redrive` moves a whole scope as a durable task
+  that is idempotent over its filters, cancellable, and advanced in bounded
+  batches by the supervisor. Every retry and task transition writes one row to
+  `solid_objects_administration_events`, so who pressed what is recorded. Pause
+  is an operator brake and not a stop, since a pass already in flight finishes
+  its turn and a synchronous caller waiting on a paused instance times out. The
+  dashboard does not yet surface the scopes or redrive; the API does. The page cost was
   reasoned about rather than measured: the summary bar issues a fixed set of
   indexed aggregate queries per page, which is why `HEAD /` exists for uptime
   monitors, but no dashboard latency has been benchmarked against a large
diff --git a/lib/solid_objects.rb b/lib/solid_objects.rb
index 4cbbf03..faf9e9c 100644
--- a/lib/solid_objects.rb
+++ b/lib/solid_objects.rb
@@ -27,6 +27,11 @@
 require "solid_objects/actor"
 require "solid_objects/reference"
 require "solid_objects/message_reference"
+require "solid_objects/administration_audit"
+require "solid_objects/redrive_task"
+require "solid_objects/redrive_manager"
+require "solid_objects/redrive_runner"
+require "solid_objects/dead_letter_scope"
 require "solid_objects/dead_letter_manager"
 require "solid_objects/message_pruner"
 require "solid_objects/instance_pruner"
@@ -160,6 +165,11 @@ def dead_letters
       @dead_letters ||= DeadLetterManager.new
     end
 
+    # @rbs () -> RedriveManager
+    def redrives
+      @redrives ||= RedriveManager.new
+    end
+
     # @rbs () -> Administration
     def administration
       @administration ||= Administration.new
@@ -192,6 +202,7 @@ def reset!
       @effect_registry = EffectRegistry.new
       @commit_action_registry = CommitActionRegistry.new
       @dead_letters = nil
+      @redrives = nil
       @administration = nil
     end
 
diff --git a/lib/solid_objects/administration_audit.rb b/lib/solid_objects/administration_audit.rb
new file mode 100644
index 0000000..840902b
--- /dev/null
+++ b/lib/solid_objects/administration_audit.rb
@@ -0,0 +1,30 @@
+# rbs_inline: enabled
+
+module SolidObjects
+  module AdministrationAudit
+    module_function
+
+    # @rbs (action: String, kind: String, ?subject_id: untyped, ?filters: Hash[Symbol | String, untyped]?, ?actor: String?) -> void
+    def record(action:, kind:, subject_id: nil, filters: nil, actor: nil)
+      AdministrationEvent.create!(
+        action:,
+        kind:,
+        subject_id: subject_id&.to_s,
+        filters:,
+        actor:,
+        occurred_at: SolidObjects.database_adapter.database_now
+      )
+      nil
+    end
+
+    # @rbs (untyped) -> String?
+    def identity(authorization_context)
+      return nil if authorization_context.nil?
+
+      SolidObjects.configuration.administration_identity
+        .call(authorization_context)
+        &.to_s
+        &.byteslice(0, 255)
+    end
+  end
+end
diff --git a/lib/solid_objects/configuration.rb b/lib/solid_objects/configuration.rb
index 87b2523..7a33ec9 100644
--- a/lib/solid_objects/configuration.rb
+++ b/lib/solid_objects/configuration.rb
@@ -31,6 +31,8 @@ class Configuration
     # @rbs @instance_retention_by_actor_type: Hash[String, Numeric]
     # @rbs @process_retention: Numeric
     # @rbs @prune_batch_size: Integer
+    # @rbs @redrive_batch_size: Integer
+    # @rbs @redrive_batch_pause: Float
     # @rbs @worker_count: Integer
     # @rbs @effect_worker_count: Integer
     # @rbs @broadcast_worker_count: Integer
@@ -49,6 +51,7 @@ class Configuration
     # @rbs @authorize_subscription: Proc
     # @rbs @authorize_administration: Proc
     # @rbs @authorize_transmission: Proc
+    # @rbs @administration_identity: Proc
     # @rbs @transmission_actor_type_resolver: Proc
 
     attr_accessor :table_name_prefix,
@@ -80,6 +83,8 @@ class Configuration
       :instance_retention_by_actor_type,
       :process_retention,
       :prune_batch_size,
+      :redrive_batch_size,
+      :redrive_batch_pause,
       :worker_count,
       :effect_worker_count,
       :broadcast_worker_count,
@@ -98,6 +103,7 @@ class Configuration
       :authorize_subscription,
       :authorize_administration,
       :authorize_transmission,
+      :administration_identity,
       :transmission_actor_type_resolver
 
     # @rbs @additional_components: Array[untyped]
@@ -134,6 +140,8 @@ def initialize
       @instance_retention_by_actor_type = {}
       @process_retention = 7.days
       @prune_batch_size = 1_000
+      @redrive_batch_size = 100
+      @redrive_batch_pause = 0.05
       @worker_count = 1
       @effect_worker_count = 1
       @broadcast_worker_count = 1
@@ -156,6 +164,7 @@ def initialize
       @authorize_subscription = ->(**) { false }
       @authorize_administration = ->(**) { false }
       @authorize_transmission = ->(**) { false }
+      @administration_identity = ->(authorization_context) { authorization_context.to_s }
       @transmission_actor_type_resolver = ->(actor_type) { actor_type }
       @additional_components = []
     end
@@ -226,6 +235,9 @@ def validate!
       positive_values.each do |name, value|
         raise ArgumentError, "#{name} must be positive" unless value.positive?
       end
+      if redrive_batch_pause.negative?
+        raise ArgumentError, "redrive_batch_pause must not be negative"
+      end
       if warn_state_bytes > max_state_bytes
         raise ArgumentError, "warn_state_bytes must not exceed max_state_bytes"
       end
@@ -247,6 +259,9 @@ def validate!
       unless payload_authorization_context.respond_to?(:call)
         raise ArgumentError, "payload_authorization_context must respond to call"
       end
+      unless administration_identity.respond_to?(:call)
+        raise ArgumentError, "administration_identity must respond to call"
+      end
 
       self
     end
@@ -295,7 +310,8 @@ def positive_values
         shutdown_timeout:,
         message_retention:,
         process_retention:,
-        prune_batch_size:
+        prune_batch_size:,
+        redrive_batch_size:
       }
     end
 
diff --git a/lib/solid_objects/dead_letter_manager.rb b/lib/solid_objects/dead_letter_manager.rb
index 65cb457..832645b 100644
--- a/lib/solid_objects/dead_letter_manager.rb
+++ b/lib/solid_objects/dead_letter_manager.rb
@@ -11,10 +11,48 @@ def all(authorization_context: nil)
     # @rbs (Integer, ?authorization_context: untyped) -> MessageReference
     def retry(dead_letter_id, authorization_context: nil)
       authorize!(:retry, authorization_context:, dead_letter_id:)
-      dead_letter = DeadLetter.find(dead_letter_id)
-      return MessageReference.from_message(Message.find(dead_letter.retried_message_id)) if dead_letter.retried_message_id
+      actor = AdministrationAudit.identity(authorization_context)
+      SolidObjects.database_adapter.transaction do
+        dead_letter = DeadLetter.find(dead_letter_id)
+        reference = if dead_letter.retried_message_id
+          MessageReference.from_message(Message.find(dead_letter.retried_message_id))
+        else
+          enqueue_retry(dead_letter)
+        end
+        AdministrationAudit.record(
+          action: "dead_letter.retry",
+          kind: "message",
+          subject_id: dead_letter.id,
+          actor:
+        )
+        reference
+      end
+    end
+
+    # @rbs () -> DeadLetterScope
+    def effects
+      @effects ||= DeadLetterScope.new(
+        model: Effect,
+        resource: "effect_dead_letters",
+        identifier: :effect_id,
+        kind: "effect"
+      )
+    end
 
-      original_message = dead_letter.message
+    # @rbs () -> DeadLetterScope
+    def broadcasts
+      @broadcasts ||= DeadLetterScope.new(
+        model: Broadcast,
+        resource: "broadcast_dead_letters",
+        identifier: :broadcast_id,
+        kind: "broadcast"
+      )
+    end
+
+    private
+
+    # @rbs (DeadLetter) -> MessageReference
+    def enqueue_retry(dead_letter)
       message_reference = Mailbox.new.enqueue(
         reference: Reference.new(
           actor_type: dead_letter.actor_type,
@@ -22,15 +60,13 @@ def retry(dead_letter_id, authorization_context: nil)
         ),
         operation: dead_letter.operation,
         arguments: dead_letter.arguments,
-        delivery_mode: original_message.delivery_mode,
+        delivery_mode: dead_letter.message.delivery_mode,
         idempotency_key: "dead-letter:#{dead_letter.id}"
       )
       dead_letter.update!(retried_message_id: message_reference.id)
       message_reference
     end
 
-    private
-
     # @rbs (Symbol, authorization_context: untyped, ?dead_letter_id: Integer?) -> void
     def authorize!(action, authorization_context:, dead_letter_id: nil)
       authorized = SolidObjects.configuration.authorize_administration.call(
diff --git a/lib/solid_objects/dead_letter_scope.rb b/lib/solid_objects/dead_letter_scope.rb
new file mode 100644
index 0000000..1cb9b5f
--- /dev/null
+++ b/lib/solid_objects/dead_letter_scope.rb
@@ -0,0 +1,131 @@
+# rbs_inline: enabled
+
+module SolidObjects
+  class DeadLetterScope
+    DEAD = "dead"
+    PENDING = "pending"
+
+    # @rbs @model: untyped
+    # @rbs @resource: String
+    # @rbs @identifier: Symbol
+
+    attr_reader :resource, :kind
+
+    # @rbs (model: untyped, resource: String, identifier: Symbol, kind: String) -> void
+    def initialize(model:, resource:, identifier:, kind:)
+      @model = model
+      @resource = resource
+      @identifier = identifier
+      @kind = kind
+    end
+
+    # @rbs (String) -> DeadLetterScope
+    def self.for_kind(kind)
+      return SolidObjects.dead_letters.effects if kind == "effect"
+      return SolidObjects.dead_letters.broadcasts if kind == "broadcast"
+
+      raise ArgumentError, "unknown dead letter kind #{kind.inspect}"
+    end
+
+    # @rbs (?authorization_context: untyped) -> ActiveRecord::Relation[untyped]
+    def all(authorization_context: nil)
+      authorize!(:inspect, authorization_context:)
+      dead.order(updated_at: :desc, id: :desc)
+    end
+
+    # @rbs (String, ?authorization_context: untyped) -> untyped
+    def retry(identifier_value, authorization_context: nil)
+      authorize!(:retry, authorization_context:, resource_id: identifier_value)
+      actor = AdministrationAudit.identity(authorization_context)
+      SolidObjects.database_adapter.transaction do
+        row = model.find_by!(identifier => identifier_value)
+        revive(row) if row.status == DEAD
+        AdministrationAudit.record(
+          action: "dead_letter.retry",
+          kind: kind,
+          subject_id: identifier_value,
+          actor:
+        )
+        row
+      end
+    end
+
+    # @rbs (?actor_type: String?, ?failed_after: untyped, ?limit: Integer?, ?authorization_context: untyped) -> RedriveTask
+    def redrive(actor_type: nil, failed_after: nil, limit: nil, authorization_context: nil)
+      if failed_after && !failed_after.respond_to?(:utc)
+        raise ArgumentError, "failed_after must be a time"
+      end
+      if limit && !(limit.is_a?(Integer) && limit.positive?)
+        raise ArgumentError, "limit must be a positive integer"
+      end
+
+      SolidObjects.redrives.start(
+        scope: self,
+        filters: {
+          "actor_type" => actor_type,
+          "failed_after" => failed_after&.utc&.iso8601(6),
+          "limit" => limit
+        },
+        authorization_context:
+      )
+    end
+
+    # @rbs () -> ActiveRecord::Relation[untyped]
+    def dead
+      model.where(status: DEAD)
+    end
+
+    # @rbs (Hash[String, untyped], ?dead_before: untyped) -> ActiveRecord::Relation[untyped]
+    def matching(filters, dead_before: nil)
+      relation = dead
+      actor_type = filters["actor_type"]
+      failed_after = filters["failed_after"]
+      relation = relation.joins(:instance).where(Instance.table_name => { actor_type: }) if actor_type
+      relation = relation.where(updated_at: Time.parse(failed_after)..) if failed_after
+      relation = relation.where(updated_at: ..dead_before) if dead_before
+      relation
+    end
+
+    # @rbs (Array[untyped]) -> Integer
+    def revive_all(identifiers)
+      model.where(id: identifiers, status: DEAD).update_all(revival_attributes)
+    end
+
+    # @rbs (untyped) -> Integer
+    def revive(row)
+      model.where(id: row.id, status: DEAD).update_all(revival_attributes).tap do
+        row.reload
+      end
+    end
+
+    # @rbs (Symbol, authorization_context: untyped, ?resource_id: String?) -> void
+    def authorize!(action, authorization_context:, resource_id: nil)
+      authorized = SolidObjects.configuration.authorize_administration.call(
+        action: action.to_s,
+        resource: resource,
+        resource_id: resource_id,
+        authorization_context:
+      )
+      return if authorized
+
+      raise Unauthorized, "actor administration is not authorized"
+    end
+
+    private
+
+    attr_reader :model, :identifier
+
+    # @rbs () -> Hash[Symbol, untyped]
+    def revival_attributes
+      now = SolidObjects.database_adapter.database_now
+      {
+        status: PENDING,
+        attempt_count: 0,
+        available_at: now,
+        claimed_by: nil,
+        claimed_at: nil,
+        updated_at: now
+      }
+    end
+  end
+end
diff --git a/lib/solid_objects/doctor.rb b/lib/solid_objects/doctor.rb
index 4ed4a92..0a26561 100644
--- a/lib/solid_objects/doctor.rb
+++ b/lib/solid_objects/doctor.rb
@@ -80,7 +80,9 @@ def to_s
       effects: %w[id message_id instance_id effect_id status available_at],
       effect_recoveries: %w[effect_id instance_id recovery_operation status_operation recovery_timeout retired_at],
       broadcasts: %w[id message_id instance_id broadcast_id status available_at],
-      dead_letters: %w[id message_id instance_id actor_type actor_id attempts]
+      dead_letters: %w[id message_id instance_id actor_type actor_id attempts],
+      administration_events: %w[id action kind subject_id actor occurred_at],
+      redrives: %w[id kind filters status active_scope moved move_limit started_at finished_at]
     }.freeze
 
     class ProbeActor < Actor
diff --git a/lib/solid_objects/redrive_manager.rb b/lib/solid_objects/redrive_manager.rb
new file mode 100644
index 0000000..dd4cbe9
--- /dev/null
+++ b/lib/solid_objects/redrive_manager.rb
@@ -0,0 +1,130 @@
+# rbs_inline: enabled
+
+require "digest"
+
+module SolidObjects
+  class RedriveManager
+    RUNNING = "running"
+    COMPLETED = "completed"
+    CANCELLED = "cancelled"
+
+    # @rbs (scope: DeadLetterScope, filters: Hash[String, untyped], authorization_context: untyped) -> RedriveTask
+    def start(scope:, filters:, authorization_context:)
+      scope.authorize!(:redrive, authorization_context:)
+      active_scope = "#{scope.kind}:#{Digest::SHA256.hexdigest(filters.to_json)}"
+      running = Redrive.find_by(active_scope:)
+      return task_for(running) if running
+
+      task_for(open_task(scope:, filters:, active_scope:, authorization_context:))
+    rescue ActiveRecord::RecordNotUnique
+      task_for(Redrive.find_by!(active_scope:))
+    end
+
+    # @rbs (String, ?authorization_context: untyped) -> RedriveTask
+    def find(id, authorization_context: nil)
+      authorize!(:inspect, authorization_context:, resource_id: id)
+      task_for(Redrive.find(id))
+    end
+
+    # @rbs (?status: Symbol | String | nil, ?authorization_context: untyped) -> Array[RedriveTask]
+    def all(status: nil, authorization_context: nil)
+      authorize!(:inspect, authorization_context:)
+      relation = Redrive.order(started_at: :desc, id: :desc)
+      relation = relation.where(status: status.to_s) if status
+      relation.map { |record| task_for(record) }
+    end
+
+    # @rbs (String, ?authorization_context: untyped) -> RedriveTask
+    def cancel(id, authorization_context: nil)
+      authorize!(:cancel, authorization_context:, resource_id: id)
+      record = Redrive.find(id)
+      return task_for(record) unless record.status == RUNNING
+
+      audit(record, action: "redrive.cancel") if close(record, status: CANCELLED)
+      task_for(record)
+    end
+
+    # @rbs (Redrive, status: String) -> bool
+    def close(record, status:)
+      changed = Redrive.where(id: record.id, status: RUNNING).update_all(
+        status:,
+        active_scope: nil,
+        finished_at: SolidObjects.database_adapter.database_now,
+        updated_at: SolidObjects.database_adapter.database_now
+      )
+      record.reload if changed.positive?
+      changed.positive?
+    end
+
+    # @rbs (Redrive, action: String) -> void
+    def audit(record, action:)
+      AdministrationAudit.record(
+        action:,
+        kind: record.kind,
+        subject_id: record.id,
+        filters: record.filters,
+        actor: record.actor
+      )
+    end
+
+    # @rbs (Redrive) -> RedriveTask
+    def task_for(record)
+      RedriveTask.new(
+        id: record.id,
+        kind: record.kind,
+        filters: Serialization.readonly_copy(record.filters),
+        status: record.status,
+        moved: record.moved,
+        remaining: remaining_for(record),
+        started_at: record.started_at,
+        finished_at: record.finished_at
+      )
+    end
+
+    private
+
+    # @rbs (scope: DeadLetterScope, filters: Hash[String, untyped], active_scope: String, authorization_context: untyped) -> Redrive
+    def open_task(scope:, filters:, active_scope:, authorization_context:)
+      record = Redrive.create!(
+        id: "redrive_#{SecureRandom.uuid}",
+        kind: scope.kind,
+        filters:,
+        status: RUNNING,
+        active_scope:,
+        moved: 0,
+        move_limit: filters["limit"],
+        actor: AdministrationAudit.identity(authorization_context),
+        started_at: SolidObjects.database_adapter.database_now
+      )
+      audit(record, action: "redrive.start")
+      record
+    end
+
+    # @rbs (Redrive) -> Integer
+    def remaining_for(record)
+      return 0 unless record.status == RUNNING
+
+      matching = DeadLetterScope
+        .for_kind(record.kind)
+        .matching(record.filters, dead_before: record.started_at)
+        .count
+      limit = record.move_limit
+      return matching unless limit
+
+      [ matching, limit - record.moved ].min
+    end
+
+    # @rbs (Symbol, authorization_context: untyped, ?resource_id: String?) -> void
+    def authorize!(action, authorization_context:, resource_id: nil)
+      authorized = SolidObjects.configuration.authorize_administration.call(
+        action: action.to_s,
+        resource: "redrives",
+        resource_id:,
+        authorization_context:
+      )
+      return if authorized
+
+      raise Unauthorized, "actor administration is not authorized"
+    end
+  end
+end
diff --git a/lib/solid_objects/redrive_runner.rb b/lib/solid_objects/redrive_runner.rb
new file mode 100644
index 0000000..22f6977
--- /dev/null
+++ b/lib/solid_objects/redrive_runner.rb
@@ -0,0 +1,60 @@
+# rbs_inline: enabled
+
+module SolidObjects
+  class RedriveRunner
+    # @rbs () -> bool
+    def run_once
+      SolidObjects.database_adapter.transaction do
+        record = claim
+        next false unless record
+
+        advance(record)
+      end
+    end
+
+    private
+
+    # @rbs () -> Redrive?
+    def claim
+      relation = Redrive.where(status: RedriveManager::RUNNING).order(:started_at, :id)
+      SolidObjects.database_adapter.lock_candidates(relation).first
+    end
+
+    # @rbs (Redrive) -> bool
+    def advance(record)
+      moved = move_batch(record)
+      return true if moved.positive?
+
+      finish(record)
+      false
+    end
+
+    # @rbs (Redrive) -> Integer
+    def move_batch(record)
+      scope = DeadLetterScope.for_kind(record.kind)
+      configured = SolidObjects.configuration.redrive_batch_size
+      limit = record.move_limit
+      size = limit ? [ configured, limit - record.moved ].min : configured
+      return 0 unless size.positive?
+
+      identifiers = scope
+        .matching(record.filters, dead_before: record.started_at)
+        .order(:id)
+        .limit(size)
+        .pluck(:id)
+      return 0 if identifiers.empty?
+
+      revived = scope.revive_all(identifiers)
+      record.update!(moved: record.moved + revived)
+      revived
+    end
+
+    # @rbs (Redrive) -> void
+    def finish(record)
+      manager = SolidObjects.redrives
+      return unless manager.close(record, status: RedriveManager::COMPLETED)
+
+      manager.audit(record, action: "redrive.finish")
+    end
+  end
+end
diff --git a/lib/solid_objects/redrive_task.rb b/lib/solid_objects/redrive_task.rb
new file mode 100644
index 0000000..1172352
--- /dev/null
+++ b/lib/solid_objects/redrive_task.rb
@@ -0,0 +1,14 @@
+# rbs_inline: enabled
+
+module SolidObjects
+  RedriveTask = Data.define(
+    :id, :kind, :filters, :status, :moved, :remaining, :started_at, :finished_at
+  )
+
+  class RedriveTask
+    # @rbs (?authorization_context: untyped) -> RedriveTask
+    def cancel(authorization_context: nil)
+      SolidObjects.redrives.cancel(id, authorization_context:)
+    end
+  end
+end
diff --git a/lib/solid_objects/supervisor.rb b/lib/solid_objects/supervisor.rb
index 3d7ff91..f3f27e3 100644
--- a/lib/solid_objects/supervisor.rb
+++ b/lib/solid_objects/supervisor.rb
@@ -31,6 +31,7 @@ def initialize(
       @started = false
       @cleaned_up_at = nil
       @retention = nil
+      @redrive = nil
       @lifecycle = Thread::Mutex.new
     end
 
@@ -50,6 +51,7 @@ def start
       @threads = components.map { |component| supervise(component) }
       @monitor = Thread.new { monitor_loop }
       @retention = Thread.new { retention_loop }
+      @redrive = Thread.new { redrive_loop }
       SolidObjects.instrument(:"supervisor.started", component_count: components.length)
     end
 
@@ -64,6 +66,7 @@ def stop
         @lifecycle.synchronize { @started = false }
         stop_monitor
         stop_retention
+        stop_redrive
         components.each(&:request_shutdown)
         join_until_timeout
         components.reject(&:stopped?).each(&:stop)
@@ -204,6 +207,46 @@ def retention_pause(failures)
       [ backoff, interval ].min
     end
 
+    # @rbs () -> void
+    def redrive_loop
+      while @started
+        advance_redrive ? pause_between_batches : wait_for_next_redrive
+      end
+    end
+
+    # @rbs () -> bool
+    def advance_redrive
+      RedriveRunner.new.run_once
+    rescue => error
+      SolidObjects.instrument(
+        :"supervisor.redrive_failed",
+        error_class: error.class.name,
+        error_message: error.message
+      )
+      false
+    end
+
+    # @rbs () -> void
+    def pause_between_batches
+      pause = SolidObjects.configuration.redrive_batch_pause
+      sleep pause if pause.positive?
+    end
+
+    # @rbs () -> void
+    def wait_for_next_redrive
+      sleep SolidObjects.configuration.supervisor_monitor_interval
+    end
+
+    # @rbs () -> void
+    def stop_redrive
+      redrive = @redrive
+      @redrive = nil
+      return unless redrive
+
+      redrive.join(SolidObjects.configuration.shutdown_timeout)
+      redrive.kill if redrive.alive?
+    end
+
     # @rbs () -> void
     def stop_retention
       retention = @retention
diff --git a/lib/solid_objects/test_helper.rb b/lib/solid_objects/test_helper.rb
index 804469c..5cd466e 100644
--- a/lib/solid_objects/test_helper.rb
+++ b/lib/solid_objects/test_helper.rb
@@ -31,6 +31,8 @@ def reset_actors!
       # @rbs () -> Array[Class]
       def actor_owned_models
         [
+          AdministrationEvent,
+          Redrive,
           DeadLetter,
           ClaimedMessage,
           ReadyMessage,
diff --git a/sig/generated/lib/solid_objects.rbs b/sig/generated/lib/solid_objects.rbs
index 7c56789..0d7563d 100644
--- a/sig/generated/lib/solid_objects.rbs
+++ b/sig/generated/lib/solid_objects.rbs
@@ -39,6 +39,9 @@ module SolidObjects
   # @rbs () -> DeadLetterManager
   def self.dead_letters: () -> DeadLetterManager
 
+  # @rbs () -> RedriveManager
+  def self.redrives: () -> RedriveManager
+
   # @rbs () -> Administration
   def self.administration: () -> Administration
 
diff --git a/sig/generated/lib/solid_objects/administration_audit.rbs b/sig/generated/lib/solid_objects/administration_audit.rbs
new file mode 100644
index 0000000..9a3c096
--- /dev/null
+++ b/sig/generated/lib/solid_objects/administration_audit.rbs
@@ -0,0 +1,11 @@
+# Generated from lib/solid_objects/administration_audit.rb with RBS::Inline
+
+module SolidObjects
+  module AdministrationAudit
+    # @rbs (action: String, kind: String, ?subject_id: untyped, ?filters: Hash[Symbol | String, untyped]?, ?actor: String?) -> void
+    def self?.record: (action: String, kind: String, ?subject_id: untyped, ?filters: Hash[Symbol | String, untyped]?, ?actor: String?) -> void
+
+    # @rbs (untyped) -> String?
+    def self?.identity: (untyped) -> String?
+  end
+end
diff --git a/sig/generated/lib/solid_objects/configuration.rbs b/sig/generated/lib/solid_objects/configuration.rbs
index 0cecf1f..af49e26 100644
--- a/sig/generated/lib/solid_objects/configuration.rbs
+++ b/sig/generated/lib/solid_objects/configuration.rbs
@@ -2,9 +2,7 @@
 
 module SolidObjects
   class Configuration
-    @process_heartbeat_interval: Float
-
-    @process_alive_threshold: Float
+    @table_name_prefix: String
 
     @shutdown_timeout: Float
 
@@ -24,6 +22,10 @@ module SolidObjects
 
     @prune_batch_size: Integer
 
+    @redrive_batch_size: Integer
+
+    @redrive_batch_pause: Float
+
     @worker_count: Integer
 
     @effect_worker_count: Integer
@@ -60,9 +62,9 @@ module SolidObjects
 
     @authorize_transmission: Proc
 
-    @transmission_actor_type_resolver: Proc
+    @administration_identity: Proc
 
-    @table_name_prefix: String
+    @transmission_actor_type_resolver: Proc
 
     @polling_interval: Float
 
@@ -98,6 +100,10 @@ module SolidObjects
 
     @lock_retry_attempts: Integer
 
+    @process_heartbeat_interval: Float
+
+    @process_alive_threshold: Float
+
     attr_accessor table_name_prefix: untyped
 
     attr_accessor polling_interval: untyped
@@ -156,6 +162,10 @@ module SolidObjects
 
     attr_accessor prune_batch_size: untyped
 
+    attr_accessor redrive_batch_size: untyped
+
+    attr_accessor redrive_batch_pause: untyped
+
     attr_accessor worker_count: untyped
 
     attr_accessor effect_worker_count: untyped
@@ -192,6 +202,8 @@ module SolidObjects
 
     attr_accessor authorize_transmission: untyped
 
+    attr_accessor administration_identity: untyped
+
     attr_accessor transmission_actor_type_resolver: untyped
 
     # @rbs @additional_components: Array[untyped]
diff --git a/sig/generated/lib/solid_objects/dead_letter_manager.rbs b/sig/generated/lib/solid_objects/dead_letter_manager.rbs
index 373e5f5..856fe79 100644
--- a/sig/generated/lib/solid_objects/dead_letter_manager.rbs
+++ b/sig/generated/lib/solid_objects/dead_letter_manager.rbs
@@ -8,8 +8,17 @@ module SolidObjects
     # @rbs (Integer, ?authorization_context: untyped) -> MessageReference
     def retry: (Integer, ?authorization_context: untyped) -> MessageReference
 
+    # @rbs () -> DeadLetterScope
+    def effects: () -> DeadLetterScope
+
+    # @rbs () -> DeadLetterScope
+    def broadcasts: () -> DeadLetterScope
+
     private
 
+    # @rbs (DeadLetter) -> MessageReference
+    def enqueue_retry: (DeadLetter) -> MessageReference
+
     # @rbs (Symbol, authorization_context: untyped, ?dead_letter_id: Integer?) -> void
     def authorize!: (Symbol, authorization_context: untyped, ?dead_letter_id: Integer?) -> void
   end
diff --git a/sig/generated/lib/solid_objects/dead_letter_scope.rbs b/sig/generated/lib/solid_objects/dead_letter_scope.rbs
new file mode 100644
index 0000000..3f65fd2
--- /dev/null
+++ b/sig/generated/lib/solid_objects/dead_letter_scope.rbs
@@ -0,0 +1,58 @@
+# Generated from lib/solid_objects/dead_letter_scope.rb with RBS::Inline
+
+module SolidObjects
+  class DeadLetterScope
+    DEAD: ::String
+
+    PENDING: ::String
+
+    @identifier: Symbol
+
+    @resource: String
+
+    @model: untyped
+
+    attr_reader resource: untyped
+
+    attr_reader kind: untyped
+
+    # @rbs (model: untyped, resource: String, identifier: Symbol, kind: String) -> void
+    def initialize: (model: untyped, resource: String, identifier: Symbol, kind: String) -> void
+
+    # @rbs (String) -> DeadLetterScope
+    def self.for_kind: (String) -> DeadLetterScope
+
+    # @rbs (?authorization_context: untyped) -> ActiveRecord::Relation[untyped]
+    def all: (?authorization_context: untyped) -> ActiveRecord::Relation[untyped]
+
+    # @rbs (String, ?authorization_context: untyped) -> untyped
+    def retry: (String, ?authorization_context: untyped) -> untyped
+
+    # @rbs (?actor_type: String?, ?failed_after: untyped, ?limit: Integer?, ?authorization_context: untyped) -> RedriveTask
+    def redrive: (?actor_type: String?, ?failed_after: untyped, ?limit: Integer?, ?authorization_context: untyped) -> RedriveTask
+
+    # @rbs () -> ActiveRecord::Relation[untyped]
+    def dead: () -> ActiveRecord::Relation[untyped]
+
+    # @rbs (Hash[String, untyped], ?dead_before: untyped) -> ActiveRecord::Relation[untyped]
+    def matching: (Hash[String, untyped], ?dead_before: untyped) -> ActiveRecord::Relation[untyped]
+
+    # @rbs (Array[untyped]) -> Integer
+    def revive_all: (Array[untyped]) -> Integer
+
+    # @rbs (untyped) -> Integer
+    def revive: (untyped) -> Integer
+
+    # @rbs (Symbol, authorization_context: untyped, ?resource_id: String?) -> void
+    def authorize!: (Symbol, authorization_context: untyped, ?resource_id: String?) -> void
+
+    private
+
+    attr_reader model: untyped
+
+    attr_reader identifier: untyped
+
+    # @rbs () -> Hash[Symbol, untyped]
+    def revival_attributes: () -> Hash[Symbol, untyped]
+  end
+end
diff --git a/sig/generated/lib/solid_objects/redrive_manager.rbs b/sig/generated/lib/solid_objects/redrive_manager.rbs
new file mode 100644
index 0000000..dc68c8a
--- /dev/null
+++ b/sig/generated/lib/solid_objects/redrive_manager.rbs
@@ -0,0 +1,43 @@
+# Generated from lib/solid_objects/redrive_manager.rb with RBS::Inline
+
+module SolidObjects
+  class RedriveManager
+    RUNNING: ::String
+
+    COMPLETED: ::String
+
+    CANCELLED: ::String
+
+    # @rbs (scope: DeadLetterScope, filters: Hash[String, untyped], authorization_context: untyped) -> RedriveTask
+    def start: (scope: DeadLetterScope, filters: Hash[String, untyped], authorization_context: untyped) -> RedriveTask
+
+    # @rbs (String, ?authorization_context: untyped) -> RedriveTask
+    def find: (String, ?authorization_context: untyped) -> RedriveTask
+
+    # @rbs (?status: Symbol | String | nil, ?authorization_context: untyped) -> Array[RedriveTask]
+    def all: (?status: Symbol | String | nil, ?authorization_context: untyped) -> Array[RedriveTask]
+
+    # @rbs (String, ?authorization_context: untyped) -> RedriveTask
+    def cancel: (String, ?authorization_context: untyped) -> RedriveTask
+
+    # @rbs (Redrive, status: String) -> bool
+    def close: (Redrive, status: String) -> bool
+
+    # @rbs (Redrive, action: String) -> void
+    def audit: (Redrive, action: String) -> void
+
+    # @rbs (Redrive) -> RedriveTask
+    def task_for: (Redrive) -> RedriveTask
+
+    private
+
+    # @rbs (scope: DeadLetterScope, filters: Hash[String, untyped], active_scope: String, authorization_context: untyped) -> Redrive
+    def open_task: (scope: DeadLetterScope, filters: Hash[String, untyped], active_scope: String, authorization_context: untyped) -> Redrive
+
+    # @rbs (Redrive) -> Integer
+    def remaining_for: (Redrive) -> Integer
+
+    # @rbs (Symbol, authorization_context: untyped, ?resource_id: String?) -> void
+    def authorize!: (Symbol, authorization_context: untyped, ?resource_id: String?) -> void
+  end
+end
diff --git a/sig/generated/lib/solid_objects/redrive_runner.rbs b/sig/generated/lib/solid_objects/redrive_runner.rbs
new file mode 100644
index 0000000..d48e357
--- /dev/null
+++ b/sig/generated/lib/solid_objects/redrive_runner.rbs
@@ -0,0 +1,22 @@
+# Generated from lib/solid_objects/redrive_runner.rb with RBS::Inline
+
+module SolidObjects
+  class RedriveRunner
+    # @rbs () -> bool
+    def run_once: () -> bool
+
+    private
+
+    # @rbs () -> Redrive?
+    def claim: () -> Redrive?
+
+    # @rbs (Redrive) -> bool
+    def advance: (Redrive) -> bool
+
+    # @rbs (Redrive) -> Integer
+    def move_batch: (Redrive) -> Integer
+
+    # @rbs (Redrive) -> void
+    def finish: (Redrive) -> void
+  end
+end
diff --git a/sig/generated/lib/solid_objects/redrive_task.rbs b/sig/generated/lib/solid_objects/redrive_task.rbs
new file mode 100644
index 0000000..301e732
--- /dev/null
+++ b/sig/generated/lib/solid_objects/redrive_task.rbs
@@ -0,0 +1,33 @@
+# Generated from lib/solid_objects/redrive_task.rb with RBS::Inline
+
+module SolidObjects
+  class RedriveTask < Data
+    attr_reader id(): untyped
+
+    attr_reader kind(): untyped
+
+    attr_reader filters(): untyped
+
+    attr_reader status(): untyped
+
+    attr_reader moved(): untyped
+
+    attr_reader remaining(): untyped
+
+    attr_reader started_at(): untyped
+
+    attr_reader finished_at(): untyped
+
+    def self.new: (untyped id, untyped kind, untyped filters, untyped status, untyped moved, untyped remaining, untyped started_at, untyped finished_at) -> instance
+                | (id: untyped, kind: untyped, filters: untyped, status: untyped, moved: untyped, remaining: untyped, started_at: untyped, finished_at: untyped) -> instance
+
+    def self.members: () -> [ :id, :kind, :filters, :status, :moved, :remaining, :started_at, :finished_at ]
+
+    def members: () -> [ :id, :kind, :filters, :status, :moved, :remaining, :started_at, :finished_at ]
+  end
+
+  class RedriveTask
+    # @rbs (?authorization_context: untyped) -> RedriveTask
+    def cancel: (?authorization_context: untyped) -> RedriveTask
+  end
+end
diff --git a/sig/generated/lib/solid_objects/supervisor.rbs b/sig/generated/lib/solid_objects/supervisor.rbs
index 0572fda..bf179f1 100644
--- a/sig/generated/lib/solid_objects/supervisor.rbs
+++ b/sig/generated/lib/solid_objects/supervisor.rbs
@@ -83,6 +83,21 @@ module SolidObjects
     # @rbs (Integer) -> Float
     def retention_pause: (Integer) -> Float
 
+    # @rbs () -> void
+    def redrive_loop: () -> void
+
+    # @rbs () -> bool
+    def advance_redrive: () -> bool
+
+    # @rbs () -> void
+    def pause_between_batches: () -> void
+
+    # @rbs () -> void
+    def wait_for_next_redrive: () -> void
+
+    # @rbs () -> void
+    def stop_redrive: () -> void
+
     # @rbs () -> void
     def stop_retention: () -> void
 
diff --git a/sig/generated/models/solid_objects/administration_event.rbs b/sig/generated/models/solid_objects/administration_event.rbs
new file mode 100644
index 0000000..7787bff
--- /dev/null
+++ b/sig/generated/models/solid_objects/administration_event.rbs
@@ -0,0 +1,6 @@
+# Generated from app/models/solid_objects/administration_event.rb with RBS::Inline
+
+module SolidObjects
+  class AdministrationEvent < Record
+  end
+end
diff --git a/sig/generated/models/solid_objects/redrive.rbs b/sig/generated/models/solid_objects/redrive.rbs
new file mode 100644
index 0000000..0b51664
--- /dev/null
+++ b/sig/generated/models/solid_objects/redrive.rbs
@@ -0,0 +1,6 @@
+# Generated from app/models/solid_objects/redrive.rb with RBS::Inline
+
+module SolidObjects
+  class Redrive < Record
+  end
+end
diff --git a/test/database_test_helper.rb b/test/database_test_helper.rb
index 56b42b5..0ff0f22 100644
--- a/test/database_test_helper.rb
+++ b/test/database_test_helper.rb
@@ -25,11 +25,15 @@
 require_relative "../db/migrate/20260806000000_add_state_revision_to_solid_objects_instances"
 require_relative "../db/migrate/20260813000000_rename_message_dispatch_columns"
 require_relative "../db/migrate/20260915000000_add_solid_objects_effect_recoveries"
+require_relative "../db/migrate/20260922000000_add_solid_objects_administration_events"
+require_relative "../db/migrate/20260922000001_add_solid_objects_redrives"
 
 CreateSolidObjectsTables.new.migrate(:up)
 AddStateRevisionToSolidObjectsInstances.new.migrate(:up)
 RenameMessageDispatchColumns.new.migrate(:up)
 AddSolidObjectsEffectRecoveries.new.migrate(:up)
+AddSolidObjectsAdministrationEvents.new.migrate(:up)
+AddSolidObjectsRedrives.new.migrate(:up)
 
 ActiveRecord::Base.connection.create_table(:solid_objects_test_domain_records) do |table|
   table.string :name, null: false
@@ -37,6 +41,8 @@
 
 require "solid_objects/database_adapter"
 require_relative "../app/models/solid_objects/record"
+require_relative "../app/models/solid_objects/administration_event"
+require_relative "../app/models/solid_objects/redrive"
 require_relative "../app/models/solid_objects/process"
 require_relative "../app/models/solid_objects/instance"
 require_relative "../app/models/solid_objects/message"
@@ -59,6 +65,8 @@ class ActiveSupport::TestCase
   teardown do
     SolidObjects::Instance.delete_all
     SolidObjects::Process.delete_all
+    SolidObjects::AdministrationEvent.delete_all
+    SolidObjects::Redrive.delete_all
     SolidObjectsTestDomainRecord.delete_all
   end
 
diff --git a/test/integration/administration_audit_test.rb b/test/integration/administration_audit_test.rb
new file mode 100644
index 0000000..58bdebf
--- /dev/null
+++ b/test/integration/administration_audit_test.rb
@@ -0,0 +1,194 @@
+# frozen_string_literal: true
+
+require "database_test_helper"
+
+class AdministrationAuditTest < ActiveSupport::TestCase
+  class LedgerActor < SolidObjects::Actor
+    actor_type "audited-ledger"
+
+    attribute :count, default: 0
+    attribute :entries, default: 0
+
+    observable :count
+
+    def post
+      self.entries += 1
+      emit :settle, entry: "one"
+    end
+
+    def touch
+      self.count += 1
+    end
+  end
+
+  class PoisonActor < SolidObjects::Actor
+    actor_type "audited-poison"
+
+    class << self
+      attr_accessor :fail
+    end
+
+    def run
+      raise "poison message" if self.class.fail
+    end
+  end
+
+  setup do
+    SolidObjects.configuration.retry_delay = ->(_attempt) { 0 }
+    SolidObjects.configuration.max_attempts = 1
+    SolidObjects.configuration.authorize_administration = ->(**) { true }
+    PoisonActor.fail = true
+  end
+
+  test "writes one audit row for an effect retry" do
+    effect = dead_effect
+
+    SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator")
+
+    event = SolidObjects::AdministrationEvent.sole
+    assert_equal "dead_letter.retry", event.action
+    assert_equal "effect", event.kind
+    assert_equal effect.effect_id, event.subject_id
+    assert_equal "operator", event.actor
+    assert event.occurred_at
+  end
+
+  test "writes one audit row for a broadcast retry" do
+    broadcast = dead_broadcast
+
+    SolidObjects.dead_letters.broadcasts.retry(
+      broadcast.broadcast_id,
+      authorization_context: "operator"
+    )
+
+    event = SolidObjects::AdministrationEvent.sole
+    assert_equal "dead_letter.retry", event.action
+    assert_equal "broadcast", event.kind
+    assert_equal broadcast.broadcast_id, event.subject_id
+  end
+
+  test "writes one audit row for a message retry" do
+    PoisonActor.ref("one").async.run
+    run_actors
+    dead_letter = SolidObjects::DeadLetter.sole
+    PoisonActor.fail = false
+
+    SolidObjects.dead_letters.retry(dead_letter.id, authorization_context: "operator")
+
+    event = SolidObjects::AdministrationEvent.sole
+    assert_equal "dead_letter.retry", event.action
+    assert_equal "message", event.kind
+    assert_equal dead_letter.id.to_s, event.subject_id
+  end
+
+  test "writes one audit row for each press, including a repeat" do
+    effect = dead_effect
+
+    2.times do
+      SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator")
+    end
+
+    assert_equal 2, SolidObjects::AdministrationEvent.count
+  end
+
+  test "records the identity the application names" do
+    effect = dead_effect
+    SolidObjects.configuration.administration_identity = ->(context) { "user:#{context.fetch(:id)}" }
+
+    SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: { id: 42 })
+
+    assert_equal "user:42", SolidObjects::AdministrationEvent.sole.actor
+  end
+
+  test "writes no audit row when the retry itself fails" do
+    effect = dead_effect
+
+    with_failing_method(SolidObjects::DeadLetterScope, :revive) do
+      assert_raises(RuntimeError) do
+        SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator")
+      end
+    end
+
+    assert_equal 0, SolidObjects::AdministrationEvent.count
+  end
+
+  test "writes no audit row when a message retry fails to enqueue" do
+    PoisonActor.ref("one").async.run
+    run_actors
+    dead_letter = SolidObjects::DeadLetter.sole
+
+    with_failing_method(SolidObjects::Mailbox, :enqueue) do
+      assert_raises(RuntimeError) do
+        SolidObjects.dead_letters.retry(dead_letter.id, authorization_context: "operator")
+      end
+    end
+
+    assert_equal 0, SolidObjects::AdministrationEvent.count
+  end
+
+  test "writes no audit row when the caller is refused" do
+    effect = dead_effect
+    SolidObjects.configuration.authorize_administration = ->(**) { false }
+
+    assert_raises(SolidObjects::Unauthorized) do
+      SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator")
+    end
+
+    assert_equal 0, SolidObjects::AdministrationEvent.count
+  end
+
+  test "reading dead letters writes no audit row" do
+    dead_effect
+
+    SolidObjects.dead_letters.effects.all(authorization_context: "operator").to_a
+
+    assert_equal 0, SolidObjects::AdministrationEvent.count
+  end
+
+  private
+
+  def with_failing_method(target, name)
+    original = target.instance_method(name)
+    target.define_method(name) { |*, **| raise "injected failure" }
+    yield
+  ensure
+    target.define_method(name, original)
+  end
+
+  def dead_effect
+    LedgerActor.ref("one").async.post
+    run_actors
+    SolidObjects.register_effect(:settle) { raise "settlement declined" }
+    run_effects
+    SolidObjects::Effect.find_by!(name: "settle")
+  end
+
+  def dead_broadcast
+    SolidObjects.configuration.broadcast_adapter = ->(_payload) { raise "transport down" }
+    LedgerActor.ref("broadcast").async.touch
+    run_actors
+    run_broadcasts
+    SolidObjects::Broadcast.where(status: "dead").sole
+  end
+
+  def run_actors
+    worker = SolidObjects::Worker.new
+    worker.run_until_idle
+  ensure
+    worker&.stop
+  end
+
+  def run_effects
+    executor = SolidObjects::EffectExecutor.new
+    executor.run_once while SolidObjects::Effect.exists?(status: "pending")
+  ensure
+    executor&.stop
+  end
+
+  def run_broadcasts
+    executor = SolidObjects::BroadcastExecutor.new
+    executor.run_once while SolidObjects::Broadcast.exists?(status: "pending")
+  ensure
+    executor&.stop
+  end
+end
diff --git a/test/integration/dead_letter_scopes_test.rb b/test/integration/dead_letter_scopes_test.rb
new file mode 100644
index 0000000..2169879
--- /dev/null
+++ b/test/integration/dead_letter_scopes_test.rb
@@ -0,0 +1,235 @@
+# frozen_string_literal: true
+
+require "database_test_helper"
+
+class DeadLetterScopesTest < ActiveSupport::TestCase
+  class OrderActor < SolidObjects::Actor
+    actor_type "scoped-dead-letter-orders"
+
+    attribute :count, default: 0
+    attribute :orders, default: 0
+
+    observable :count
+
+    def place
+      self.orders += 1
+      emit :charge_order, order_id: "order-1"
+    end
+
+    def touch
+      self.count += 1
+    end
+
+    def send_elsewhere
+      transmit.touch
+    end
+  end
+
+  class PoisonActor < SolidObjects::Actor
+    actor_type "scoped-dead-letter-poison"
+
+    class << self
+      attr_accessor :fail
+    end
+
+    def run
+      raise "poison message" if self.class.fail
+    end
+  end
+
+  setup do
+    SolidObjects.configuration.retry_delay = ->(_attempt) { 0 }
+    SolidObjects.configuration.max_attempts = 1
+    SolidObjects.configuration.authorize_administration = ->(**) { true }
+    PoisonActor.fail = true
+    @charges = []
+  end
+
+  test "returns a dead effect to pending and runs it again" do
+    effect = dead_effect
+
+    SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator")
+
+    assert_equal "pending", effect.reload.status
+    assert_equal 0, effect.attempt_count
+    assert_nil effect.claimed_by
+    SolidObjects.register_effect(:charge_order) { |arguments, _context| @charges << arguments }
+    run_effects
+
+    assert_equal "completed", effect.reload.status
+    assert_equal [ { "order_id" => "order-1" } ], @charges
+  end
+
+  test "reuses the stable effect id when it retries" do
+    effect = dead_effect
+    original_id = effect.effect_id
+
+    SolidObjects.dead_letters.effects.retry(original_id, authorization_context: "operator")
+
+    assert_equal original_id, effect.reload.effect_id
+    assert_equal 1, SolidObjects::Effect.count
+  end
+
+  test "leaves an effect that is already pending alone" do
+    effect = dead_effect
+    SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator")
+    available_at = effect.reload.available_at
+
+    SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator")
+
+    assert_equal "pending", effect.reload.status
+    assert_equal available_at, effect.available_at
+    assert_equal 1, SolidObjects::Effect.count
+  end
+
+  test "returns a dead broadcast to pending and delivers it" do
+    broadcast = dead_broadcast
+
+    SolidObjects.dead_letters.broadcasts.retry(
+      broadcast.broadcast_id,
+      authorization_context: "operator"
+    )
+
+    assert_equal "pending", broadcast.reload.status
+    assert_equal 0, broadcast.attempt_count
+    delivered = []
+    SolidObjects.configuration.broadcast_adapter = ->(payload) { delivered << payload }
+    run_broadcasts
+
+    assert_equal "delivered", broadcast.reload.status
+    assert_equal 1, delivered.size
+  end
+
+  test "replays a dead transmit effect" do
+    OrderActor.ref("one").async.send_elsewhere
+    run_actors
+    transmitted = []
+    SolidObjects.register_effect(SolidObjects::Transmission::EFFECT_NAME) { raise "carrier down" }
+    run_effects
+    effect = SolidObjects::Effect.find_by!(name: SolidObjects::Transmission::EFFECT_NAME)
+    assert_equal "dead", effect.status
+
+    SolidObjects.register_effect(SolidObjects::Transmission::EFFECT_NAME) do |arguments, _context|
+      transmitted << arguments
+    end
+    SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator")
+    run_effects
+
+    assert_equal "completed", effect.reload.status
+    assert_equal [ "touch" ], transmitted.map { |arguments| arguments.fetch("operation") }
+  end
+
+  test "reads and retries message dead letters as it always has" do
+    PoisonActor.ref("one").async.run
+    run_actors
+    dead_letter = SolidObjects::DeadLetter.sole
+    PoisonActor.fail = false
+
+    listed = SolidObjects.dead_letters.all(authorization_context: "operator")
+    reference = SolidObjects.dead_letters.retry(dead_letter.id, authorization_context: "operator")
+    run_actors
+
+    assert_equal [ dead_letter.id ], listed.map(&:id)
+    assert_equal reference.id, dead_letter.reload.retried_message_id
+    assert_equal "completed", reference.status
+  end
+
+  test "each scope reads only its own kind" do
+    effect = dead_effect
+    broadcast = dead_broadcast
+    PoisonActor.ref("one").async.run
+    run_actors
+
+    effects = SolidObjects.dead_letters.effects.all(authorization_context: "operator")
+    broadcasts = SolidObjects.dead_letters.broadcasts.all(authorization_context: "operator")
+    messages = SolidObjects.dead_letters.all(authorization_context: "operator")
+
+    assert_equal [ effect.effect_id ], effects.map(&:effect_id)
+    assert_equal [ broadcast.broadcast_id ], broadcasts.map(&:broadcast_id)
+    assert_equal 1, messages.count
+  end
+
+  test "reads only dead rows, not pending ones" do
+    dead = dead_effect
+    OrderActor.ref("two").async.place
+    run_actors
+
+    effects = SolidObjects.dead_letters.effects.all(authorization_context: "operator")
+
+    assert_equal 2, SolidObjects::Effect.count
+    assert_equal [ dead.effect_id ], effects.map(&:effect_id)
+  end
+
+  test "refuses an unauthorized caller" do
+    effect = dead_effect
+    broadcast = dead_broadcast
+    SolidObjects.configuration.authorize_administration = ->(**) { false }
+
+    assert_raises(SolidObjects::Unauthorized) do
+      SolidObjects.dead_letters.effects.all(authorization_context: "operator")
+    end
+    assert_raises(SolidObjects::Unauthorized) do
+      SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator")
+    end
+    assert_raises(SolidObjects::Unauthorized) do
+      SolidObjects.dead_letters.broadcasts.retry(
+        broadcast.broadcast_id,
+        authorization_context: "operator"
+      )
+    end
+  end
+
+  test "names the scope it authorizes" do
+    effect = dead_effect
+    seen = []
+    SolidObjects.configuration.authorize_administration = lambda do |action:, resource:, **|
+      seen << [ action, resource ]
+      true
+    end
+
+    SolidObjects.dead_letters.effects.retry(effect.effect_id, authorization_context: "operator")
+
+    assert_equal [ [ "retry", "effect_dead_letters" ] ], seen
+  end
+
+  private
+
+  def dead_effect
+    OrderActor.ref("one").async.place
+    run_actors
+    SolidObjects.register_effect(:charge_order) { raise "charge declined" }
+    run_effects
+    SolidObjects::Effect.find_by!(name: "charge_order").tap do |effect|
+      assert_equal "dead", effect.status
+    end
+  end
+
+  def dead_broadcast
+    SolidObjects.configuration.broadcast_adapter = ->(_payload) { raise "transport down" }
+    OrderActor.ref("broadcast").async.touch
+    run_actors
+    run_broadcasts
+    SolidObjects::Broadcast.where(status: "dead").sole
+  end
+
+  def run_actors
+    worker = SolidObjects::Worker.new
+    worker.run_until_idle
+  ensure
+    worker&.stop
+  end
+
+  def run_effects
+    executor = SolidObjects::EffectExecutor.new
+    executor.run_once while SolidObjects::Effect.exists?(status: "pending")
+  ensure
+    executor&.stop
+  end
+
+  def run_broadcasts
+    executor = SolidObjects::BroadcastExecutor.new
+    executor.run_once while SolidObjects::Broadcast.exists?(status: "pending")
+  ensure
+    executor&.stop
+  end
+end
diff --git a/test/integration/redrive_test.rb b/test/integration/redrive_test.rb
new file mode 100644
index 0000000..64a76cb
--- /dev/null
+++ b/test/integration/redrive_test.rb
@@ -0,0 +1,360 @@
+# frozen_string_literal: true
+
+require "database_test_helper"
+
+class RedriveTest < ActiveSupport::TestCase
+  class PaymentActor < SolidObjects::Actor
+    actor_type "redrive-payments"
+
+    attribute :placed, default: 0
+
+    def place(order:)
+      self.placed += 1
+      emit :settle, order: order
+    end
+  end
+
+  class ShipmentActor < SolidObjects::Actor
+    actor_type "redrive-shipments"
+
+    attribute :count, default: 0
+
+    observable :count
+
+    def touch
+      self.count += 1
+    end
+  end
+
+  setup do
+    SolidObjects.configuration.retry_delay = ->(_attempt) { 0 }
+    SolidObjects.configuration.authorize_administration = ->(**) { true }
+    SolidObjects.configuration.redrive_batch_size = 10
+    SolidObjects.configuration.redrive_batch_pause = 0
+  end
+
+  test "moves every matching row in bounded batches" do
+    dead_effects(25)
+
+    task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+    runner = SolidObjects::RedriveRunner.new
+
+    assert_equal "running", task.status
+    assert runner.run_once
+    assert_equal 10, SolidObjects::Effect.where(status: "pending").count
+    assert runner.run_once
+    assert_equal 20, SolidObjects::Effect.where(status: "pending").count
+    assert runner.run_once
+    assert_equal 25, SolidObjects::Effect.where(status: "pending").count
+
+    refute runner.run_once
+    finished = SolidObjects.redrives.find(task.id, authorization_context: "operator")
+    assert_equal "completed", finished.status
+    assert_equal 25, finished.moved
+    assert_equal 0, finished.remaining
+  end
+
+  test "stops at the limit and leaves the rest dead" do
+    dead_effects(25)
+
+    task = SolidObjects.dead_letters.effects.redrive(limit: 15, authorization_context: "operator")
+    drain
+
+    finished = SolidObjects.redrives.find(task.id, authorization_context: "operator")
+    assert_equal 15, finished.moved
+    assert_equal "completed", finished.status
+    assert_equal 10, SolidObjects::Effect.where(status: "dead").count
+  end
+
+  test "returns the running task when the same scope is redriven again" do
+    dead_effects(25)
+
+    first = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+    second = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+
+    assert_equal first.id, second.id
+    assert_equal 1, SolidObjects::Redrive.count
+  end
+
+  test "starts a separate task for another scope while one runs" do
+    dead_effects(5)
+    dead_broadcast
+
+    effects_task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+    broadcasts_task = SolidObjects.dead_letters.broadcasts.redrive(authorization_context: "operator")
+
+    refute_equal effects_task.id, broadcasts_task.id
+    assert_equal [ "broadcast", "effect" ], SolidObjects::Redrive.pluck(:kind).sort
+  end
+
+  test "starts a separate task for different filters" do
+    dead_effects(5)
+
+    first = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+    second = SolidObjects.dead_letters.effects.redrive(
+      actor_type: "redrive-payments",
+      authorization_context: "operator"
+    )
+
+    refute_equal first.id, second.id
+  end
+
+  test "starts a new task once the first finishes" do
+    dead_effects(5)
+    first = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+    drain
+    SolidObjects::Effect.where.not(status: "dead").update_all(
+      status: "dead", updated_at: SolidObjects.database_adapter.database_now
+    )
+
+    second = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+
+    refute_equal first.id, second.id
+    assert_equal "running", second.status
+  end
+
+  test "cancels a running task and keeps the rows it already moved" do
+    dead_effects(25)
+    task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+    runner = SolidObjects::RedriveRunner.new
+    runner.run_once
+
+    task.cancel(authorization_context: "operator")
+
+    refute runner.run_once
+    cancelled = SolidObjects.redrives.find(task.id, authorization_context: "operator")
+    assert_equal "cancelled", cancelled.status
+    assert_equal 10, cancelled.moved
+    assert_equal 10, SolidObjects::Effect.where(status: "pending").count
+    assert_equal 15, SolidObjects::Effect.where(status: "dead").count
+  end
+
+  test "does not move a row that died after the task started" do
+    dead_effects(2)
+    task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+    sleep 0.01
+    PaymentActor.ref("late").async.place(order: "late")
+    run_actors
+    SolidObjects::Effect.where.not(status: "dead").update_all(
+      status: "dead", updated_at: SolidObjects.database_adapter.database_now
+    )
+
+    drain
+
+    assert_equal 2, SolidObjects.redrives.find(task.id, authorization_context: "operator").moved
+    assert_equal 1, SolidObjects::Effect.where(status: "dead").count
+  end
+
+  test "filters by actor type" do
+    dead_effects(3)
+    ShipmentActor.ref("one").async.touch
+    run_actors
+
+    task = SolidObjects.dead_letters.effects.redrive(
+      actor_type: "redrive-shipments",
+      authorization_context: "operator"
+    )
+    drain
+
+    finished = SolidObjects.redrives.find(task.id, authorization_context: "operator")
+    assert_equal 0, finished.moved
+    assert_equal 3, SolidObjects::Effect.where(status: "dead").count
+  end
+
+  test "filters by failure time" do
+    dead_effects(3)
+    future = SolidObjects.database_adapter.database_now + 60
+
+    task = SolidObjects.dead_letters.effects.redrive(
+      failed_after: future,
+      authorization_context: "operator"
+    )
+    drain
+
+    assert_equal 0, SolidObjects.redrives.find(task.id, authorization_context: "operator").moved
+  end
+
+  test "reports what a running task has left to move" do
+    dead_effects(25)
+
+    task = SolidObjects.dead_letters.effects.redrive(limit: 15, authorization_context: "operator")
+    assert_equal 15, task.remaining
+
+    SolidObjects::RedriveRunner.new.run_once
+
+    assert_equal 5, SolidObjects.redrives.find(task.id, authorization_context: "operator").remaining
+  end
+
+  test "reads tasks back by id and by status" do
+    dead_effects(5)
+    task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+
+    running = SolidObjects.redrives.all(status: :running, authorization_context: "operator")
+    assert_equal [ task.id ], running.map(&:id)
+
+    drain
+
+    assert_empty SolidObjects.redrives.all(status: :running, authorization_context: "operator")
+    completed = SolidObjects.redrives.all(status: :completed, authorization_context: "operator")
+    assert_equal [ task.id ], completed.map(&:id)
+    assert_equal [ task.id ], SolidObjects.redrives.all(authorization_context: "operator").map(&:id)
+  end
+
+  test "writes one audit row for each task transition" do
+    dead_effects(5)
+
+    task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+    drain
+
+    events = SolidObjects::AdministrationEvent.order(:id).map(&:action)
+    assert_equal [ "redrive.start", "redrive.finish" ], events
+    assert_equal [ task.id, task.id ], SolidObjects::AdministrationEvent.order(:id).map(&:subject_id)
+    assert_equal({ "actor_type" => nil, "failed_after" => nil, "limit" => nil },
+      SolidObjects::AdministrationEvent.order(:id).first.filters)
+  end
+
+  test "writes one audit row when a task is cancelled" do
+    dead_effects(5)
+    task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+
+    task.cancel(authorization_context: "operator")
+
+    assert_equal [ "redrive.start", "redrive.cancel" ],
+      SolidObjects::AdministrationEvent.order(:id).map(&:action)
+  end
+
+  test "refuses an invalid filter rather than redrive everything" do
+    dead_effects(2)
+
+    assert_raises(ArgumentError) do
+      SolidObjects.dead_letters.effects.redrive(limit: 0, authorization_context: "operator")
+    end
+    assert_raises(ArgumentError) do
+      SolidObjects.dead_letters.effects.redrive(limit: 1.5, authorization_context: "operator")
+    end
+
+    assert_equal 0, SolidObjects::Redrive.count
+    assert_equal 0, SolidObjects::AdministrationEvent.count
+  end
+
+  test "writes no audit row when a retry names a row that does not exist" do
+    assert_raises(ActiveRecord::RecordNotFound) do
+      SolidObjects.dead_letters.effects.retry("missing", authorization_context: "operator")
+    end
+
+    assert_equal 0, SolidObjects::AdministrationEvent.count
+  end
+
+  test "a cancel cannot overwrite a task the runner already finished" do
+    dead_effects(1)
+    task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+    drain
+
+    task.cancel(authorization_context: "operator")
+
+    assert_equal "completed",
+      SolidObjects.redrives.find(task.id, authorization_context: "operator").status
+    assert_equal [ "redrive.start", "redrive.finish" ],
+      SolidObjects::AdministrationEvent.order(:id).map(&:action)
+  end
+
+  test "refuses an unauthorized caller that reaches the manager directly" do
+    SolidObjects.configuration.authorize_administration = ->(**) { false }
+
+    assert_raises(SolidObjects::Unauthorized) do
+      SolidObjects.redrives.start(
+        scope: SolidObjects.dead_letters.effects,
+        filters: { "actor_type" => nil, "failed_after" => nil, "limit" => nil },
+        authorization_context: "operator"
+      )
+    end
+  end
+
+  test "refuses an unauthorized caller" do
+    dead_effects(5)
+    task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+    SolidObjects.configuration.authorize_administration = ->(**) { false }
+
+    assert_raises(SolidObjects::Unauthorized) do
+      SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+    end
+    assert_raises(SolidObjects::Unauthorized) do
+      task.cancel(authorization_context: "operator")
+    end
+    assert_raises(SolidObjects::Unauthorized) do
+      SolidObjects.redrives.all(authorization_context: "operator")
+    end
+    assert_raises(SolidObjects::Unauthorized) do
+      SolidObjects.redrives.find(task.id, authorization_context: "operator")
+    end
+  end
+
+  test "a supervised runtime advances a redrive without a caller driving it" do
+    dead_effects(12)
+    SolidObjects.configuration.redrive_batch_size = 4
+    supervisor = SolidObjects::Supervisor.new(
+      worker_count: 0,
+      effect_worker_count: 0,
+      broadcast_worker_count: 0,
+      reminder_scheduler_count: 1
+    )
+    task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+
+    supervisor.start
+    deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 10
+    until SolidObjects.redrives.find(task.id, authorization_context: "operator").status == "completed"
+      flunk "the supervisor did not finish the redrive" if
+        Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline
+      sleep 0.05
+    end
+
+    assert_equal 12, SolidObjects::Effect.where(status: "pending").count
+  ensure
+    supervisor&.stop
+  end
+
+  test "reports a task as a frozen value" do
+    dead_effects(1)
+
+    task = SolidObjects.dead_letters.effects.redrive(authorization_context: "operator")
+
+    assert_kind_of SolidObjects::RedriveTask, task
+    assert task.frozen?
+    assert_equal "effect", task.kind
+    assert task.started_at
+    assert_nil task.finished_at
+  end
+
+  private
+
+  def dead_effects(count)
+    SolidObjects.register_effect(:settle) { raise "declined" }
+    count.times { |index| PaymentActor.ref("order-#{index}").async.place(order: index) }
+    run_actors
+    SolidObjects::Effect.where.not(status: "dead").update_all(
+      status: "dead", updated_at: SolidObjects.database_adapter.database_now
+    )
+    assert_equal count, SolidObjects::Effect.where(status: "dead").count
+  end
+
+  def dead_broadcast
+    SolidObjects.configuration.broadcast_adapter = ->(_payload) { raise "transport down" }
+    ShipmentActor.ref("one").async.touch
+    run_actors
+    SolidObjects::Broadcast.where.not(status: "dead").update_all(
+      status: "dead", updated_at: SolidObjects.database_adapter.database_now
+    )
+  end
+
+  def drain
+    runner = SolidObjects::RedriveRunner.new
+    nil while runner.run_once
+  end
+
+  def run_actors
+    worker = SolidObjects::Worker.new
+    worker.run_until_idle
+  ensure
+    worker&.stop
+  end
+end