diff --git a/CHANGELOG.md b/CHANGELOG.md
index 133a0e01..f8761d5e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,7 @@
 
 ### Added
 
+- [Deferred] Add `Rage::Deferred.dead_tasks` for listing and inspecting tasks that exhausted their retries.
 - Add low-level `PubSub` API (#399).
 - [Cable] Serialize WS messages per connection (#400).
 - Update fiber scheduler to support Ruby 4.1 (#397).
diff --git a/lib/rage/deferred/backends/disk.rb b/lib/rage/deferred/backends/disk.rb
index 8faba70a..ba6dea6b 100644
--- a/lib/rage/deferred/backends/disk.rb
+++ b/lib/rage/deferred/backends/disk.rb
@@ -10,8 +10,10 @@
 # * `remove_task` - called when a task has to be removed from the storage;
 # * `pending_tasks` - the method should iterate over the underlying storage and return a list of tasks to replay;
 # * `add_dead_task` - called when a task has exhausted its retries or aborted them;
-# * `list_dead_tasks` - return a list of dead tasks, newest first;
-# * `find_dead_task` - return a single dead task;
+# * `each_dead_task` - yields one record for each task ID that exists when traversal starts;
+#   if multiple records have the same task ID, it yields the newest record;
+#   records added during traversal are not included;
+# * `find_dead_task` - returns the newest dead-task record with the exact task ID;
 # * `remove_dead_tasks` - permanently delete dead tasks;
 #
 class Rage::Deferred::Backends::Disk
@@ -53,19 +55,23 @@ def add_dead_task(task_id, context, exception, task_class:, attempts:)
     @dead_tasks_storage.add(task_id, context, exception, task_class:, attempts:)
   end
 
-  # Return a list of dead-lettered tasks, newest first.
-  # @param limit [Integer, nil] the maximum number of records to return
-  # @param offset [Integer] the number of records to skip
-  # @return [Array<Hash>]
-  def list_dead_tasks(limit: nil, offset: 0)
-    @dead_tasks_storage.list(limit:, offset:)
+  # Yield one record for each task ID that exists when traversal starts.
+  # If multiple records have the same task ID, yield the newest record.
+  # Do not include records that Rage adds during traversal.
+  # Yield records from oldest to newest.
+  # @yieldparam record [Hash] a dead-task record
+  # @return [Rage::Deferred::Backends::Disk::DeadTasksStorage]
+  # @private
+  def each_dead_task(&block)
+    @dead_tasks_storage.each(&block)
   end
 
-  # Return a single dead-lettered task.
-  # @param id [String] the id of the dead-task record
-  # @return [Hash, nil]
+  # Return the newest dead-task record with the exact task ID.
+  # @param id [String] the task ID
+  # @return [Hash, nil] the matching record, or nil if there is no matching record
+  # @private
   def find_dead_task(id)
-    @dead_tasks_storage.find(id)
+    @dead_tasks_storage.find_by_id(id)
   end
 
   # Permanently delete dead-lettered tasks.
@@ -421,25 +427,76 @@ def add(task_id, context, exception, task_class:, attempts:)
       end
     end
 
-    # Return dead task records, newest first.
-    # @param limit [Integer, nil] maximum number of records to return, or all records if nil
-    # @param offset [Integer] number of newest records to skip
-    # @return [Array<Hash>] task records that could be read successfully
+    # Yield each selected dead-task record from oldest to newest.
+    #
+    # The method locks the store and opens the current file.
+    # It sets a read boundary after the last complete record.
+    # It then releases the lock.
+    # A record that Rage adds after this step is not part of the traversal.
+    #
+    # The method scans each complete record before it yields the first record.
+    # For each task ID, it selects the newest record that has a valid frame.
+    #
+    # The method decodes one selected record immediately before it yields that record.
+    # It does not decode the next record while the caller processes the current record.
+    # Each call uses independent traversal state.
+    #
+    # @yieldparam record [Hash] a decoded dead-task record
+    # @return [self]
     # @raise [Rage::Deferred::DeadTasksLockTimeout] if the store cannot be locked
-    def list(limit: nil, offset: 0)
-      records = read_records.reverse
-      records = records.drop(offset) if offset > 0
-      records = records.first(limit) if limit
+    # @raise [SystemCallError] when an operation on the selected storage file fails
+    def each
+      with_snapshot do |storage, snapshot_end|
+        records_index(storage, snapshot_end).each do |task_id, (offset, length)|
+          storage.seek(offset, IO::SEEK_SET)
+
+          begin
+            record = Marshal.load(storage.read(length).undump)
+          rescue ArgumentError, NameError, TypeError => e
+            puts "ERROR: Can't deserialize the dead-lettered task with id #{task_id}: (#{e.class}) #{e.message}"
+            next
+          end
+
+          # yield outside the `rescue`, so that an error raised by the caller's block is not swallowed
+          yield record.merge(id: task_id)
+        end
 
-      records
+        self
+      end
     end
 
-    # Find a dead task by its id.
-    # @param id [String] persisted task id
-    # @return [Hash, nil] the task record, or nil when no record matches
+    # Return the newest record with the exact task ID.
+    #
+    # Scan all complete records in the fixed snapshot.
+    # Do not decode a candidate record during the scan.
+    # After the scan, decode only the newest matching record.
+    # @param id [String] the persisted task ID
+    # @return [Hash, nil] the decoded record, or nil if there is no matching record
     # @raise [Rage::Deferred::DeadTasksLockTimeout] if the store cannot be locked
-    def find(id)
-      read_records.find { |record| record[:id] == id }
+    # @raise [SystemCallError] when an operation on the selected storage file fails
+    def find_by_id(id)
+      with_snapshot do |storage, snapshot_end|
+        serialized_record_location = nil
+        each_framed_record(storage, snapshot_end) do |task_id, offset, length|
+          next unless task_id == id
+
+          serialized_record_location = [offset, length]
+        end
+
+        next unless serialized_record_location
+
+        offset, length = serialized_record_location
+        storage.seek(offset, IO::SEEK_SET)
+
+        begin
+          record = Marshal.load(storage.read(length).undump)
+        rescue ArgumentError, NameError, TypeError => e
+          puts "ERROR: Can't deserialize the dead-lettered task with id #{id}: (#{e.class}) #{e.message}"
+          next
+        end
+
+        record.merge(id: id)
+      end
     end
 
     # Permanently delete the records with the given ids.
@@ -459,7 +516,7 @@ def remove(ids)
         File.open(@tmp_storage_path, File::WRONLY | File::CREAT | File::TRUNC | File::BINARY, 0o644) do |tmp|
           File.open(@storage_path, File::RDONLY | File::BINARY) do |storage|
             storage.each_line do |entry|
-              id = entry_id(entry)
+              id, = framed_record(entry.chomp)
 
               if id.nil?
                 next # drop corrupted records; they can neither be listed nor deleted otherwise
@@ -489,95 +546,153 @@ def remove(ids)
 
     private
 
-    # Persist changes to the live file's directory entry, including file creation and replacement.
-    # @return [void]
-    def sync_storage_directory
-      File.open(@storage_path.dirname, File::RDONLY, &:fsync)
+    # Index the newest valid payload location for each task ID.
+    # If the index contains the duplicate task ID, the method removes the ID before it adds the newer record.
+    # This operation moves the ID to the position of its newest record.
+    # Hash insertion order then gives the required order from oldest to newest.
+    # @param storage [File] the descriptor for the fixed snapshot
+    # @param snapshot_end [Integer] the byte boundary after its last complete record
+    # @return [Hash{String => Array<(Integer, Integer)>}] ordered serialized-payload locations
+    def records_index(storage, snapshot_end)
+      records_index = {}
+      each_framed_record(storage, snapshot_end) do |task_id, offset, length|
+        records_index.delete(task_id)
+        records_index[task_id] = [offset, length].freeze
+      end
+
+      records_index
     end
 
-    # Remove an incomplete final entry left by an interrupted append.
-    # @param storage [File] store opened for reading and writing
+    # Scan each complete record up to the fixed snapshot boundary.
+    # Silently skip a malformed frame or an invalid CRC.
+    # Do not decode or inspect a serialized payload.
+    # @param storage [File] the descriptor for the fixed snapshot
+    # @param snapshot_end [Integer] the byte boundary after its last complete record
+    # @yieldparam task_id [String] the task ID from the storage frame
+    # @yieldparam offset [Integer] the serialized payload's byte offset
+    # @yieldparam length [Integer] the serialized payload's byte length
     # @return [void]
-    def repair_torn_tail(storage)
+    # @private
+    def each_framed_record(storage, snapshot_end)
+      position = 0
+      storage.seek(0, IO::SEEK_SET)
+
+      while position < snapshot_end
+        offset = position
+        entry = storage.gets
+        break unless entry
+
+        position += entry.bytesize
+        length = entry.bytesize - 1
+        entry = entry.byteslice(0, length)
+        task_id, serialized_record = framed_record(entry)
+        next unless task_id
+
+        serialized_record_offset = offset + length - serialized_record.bytesize
+        yield task_id, serialized_record_offset, serialized_record.bytesize
+      end
+    end
+
+    # Lock the store, open the live file, and find the last complete record.
+    # Release the lock before the block scans, decodes, or yields a record.
+    # Keep the file descriptor open while the block runs.
+    # The descriptor continues to identify the same file if another operation replaces the live path.
+    # Close the descriptor when the block exits.
+    # If the block and `File#close` both raise exceptions, the method re-raises the block's exception.
+    # @yieldparam storage [File] the descriptor for the fixed snapshot
+    # @yieldparam snapshot_end [Integer] the fixed end position of the last complete record
+    # @yieldreturn [Object] the result from the read operation
+    # @return [Object] the block result
+    def with_snapshot
+      storage = nil
+      active_exception = nil
+
+      begin
+        snapshot_end = with_lock("read tasks from") do
+          storage = File.open(@storage_path, File::RDONLY | File::BINARY)
+          complete_record_end(storage)
+        end
+
+        yield storage, snapshot_end
+      rescue Exception => e
+        active_exception = e
+        raise
+      ensure
+        begin
+          storage.close if storage && !storage.closed?
+        rescue
+          raise unless active_exception
+        end
+      end
+    end
+
+    # Return the byte position after the last complete record.
+    # A complete record ends with a newline.
+    # If the file has an incomplete tail, scan backwards in fixed-size chunks.
+    # Do not parse a record during this scan.
+    # Do not change the file.
+    # Return zero if the file has no complete record.
+    # @param storage [File] an open dead-task file
+    # @return [Integer] the byte boundary after the last complete record
+    def complete_record_end(storage)
       storage.seek(0, IO::SEEK_END)
       end_position = storage.pos
-      return if end_position == 0
+      return 0 if end_position == 0
 
       storage.seek(-1, IO::SEEK_END)
-      return if storage.read(1) == "\n"
+      return end_position if storage.read(1) == "\n"
 
       position = end_position
-      truncate_at = 0
-
       while position > 0
         chunk_start = [position - TAIL_SCAN_CHUNK_SIZE, 0].max
         storage.seek(chunk_start, IO::SEEK_SET)
         chunk = storage.read(position - chunk_start)
 
         if (newline_index = chunk.rindex("\n"))
-          truncate_at = chunk_start + newline_index + 1
-          break
+          return chunk_start + newline_index + 1
         end
 
         position = chunk_start
       end
 
-      storage.truncate(truncate_at)
+      0
     end
 
-    # Read valid records, keeping only the latest entry for each task id.
-    # Corrupted or unreadable records are reported and skipped.
-    # @return [Array<Hash>]
-    # @raise [Rage::Deferred::DeadTasksLockTimeout] if the store cannot be locked
-    def read_records
-      entries = with_lock("read tasks from") do
-        result, corrupted_count = {}, 0
-
-        File.open(@storage_path, File::RDONLY | File::BINARY) do |storage|
-          storage.each_line(chomp: true) do |entry|
-            id = entry_id(entry)
-
-            if id.nil?
-              corrupted_count += 1
-              next
-            end
-
-            # the same task can be dead-lettered more than once if the worker crashed
-            # before the task was removed from the write-ahead log
-            result.delete(id)
-            result[id] = entry
-          end
-        end
-
-        if corrupted_count != 0
-          puts "WARNING: Detected #{corrupted_count} corrupted dead-lettered task(s)"
-        end
+    # Validate the storage frame.
+    # Return its task ID and serialized payload.
+    # Return `nil` for a malformed frame, invalid CRC, unsupported operation, or empty task ID.
+    # Do not decode the payload.
+    # @param entry [String] one complete physical record without the newline
+    # @return [Array<(String, String)>, nil]
+    def framed_record(entry)
+      return unless entry.bytesize > ENTRY_CRC_HEX_WIDTH + 1
+      return unless entry.getbyte(ENTRY_CRC_HEX_WIDTH) == 58 # `:`
+
+      signature = entry.byteslice(0, ENTRY_CRC_HEX_WIDTH)
+      payload = entry.byteslice((ENTRY_CRC_HEX_WIDTH + 1)..)
+      operation_prefix = "#{ENTRY_OP}:"
+      return unless payload.start_with?(operation_prefix)
+      return unless signature == Zlib.crc32(payload).to_s(16).rjust(ENTRY_CRC_HEX_WIDTH, "0")
 
-        result
-      end
+      id_start = operation_prefix.bytesize
+      separator_index = payload.index(":", id_start)
+      return unless separator_index && separator_index > id_start
 
-      entries.filter_map do |id, entry|
-        _, _, _, serialized_record = entry.split(":", 4)
-        Marshal.load(serialized_record.undump)
-      rescue ArgumentError, NameError, TypeError => e
-        puts "ERROR: Can't deserialize the dead-lettered task with id #{id}: (#{e.class}) #{e.message}"
-        nil
-      end
+      [payload.byteslice(id_start...separator_index), payload.byteslice((separator_index + 1)..)]
     end
 
-    # Validate a stored entry and extract its task id.
-    # @param entry [String] serialized entry, optionally ending with a newline
-    # @return [String, nil] task id, or nil if the entry is malformed or corrupted
-    def entry_id(entry)
-      entry = entry.chomp
-
-      signature, payload = entry[0...ENTRY_CRC_HEX_WIDTH], entry[(ENTRY_CRC_HEX_WIDTH + 1)..]
-      return if signature.nil? || payload.nil? || !payload.start_with?("#{ENTRY_OP}:")
-      return unless signature == Zlib.crc32(payload).to_s(16).rjust(ENTRY_CRC_HEX_WIDTH, "0")
+    # Call `fsync` on the storage directory after file creation or replacement.
+    # @return [void]
+    def sync_storage_directory
+      File.open(@storage_path.dirname, File::RDONLY, &:fsync)
+    end
 
-      id_start = ENTRY_CRC_HEX_WIDTH + 1 + ENTRY_OP.length + 1
-      separator_index = entry.index(":", id_start)
-      entry[id_start...separator_index] if separator_index
+    # Remove bytes after the last complete record. An interrupted append can leave these bytes.
+    # @param storage [File] the store opened for reading and writing
+    # @return [void]
+    def repair_torn_tail(storage)
+      truncate_at = complete_record_end(storage)
+      storage.truncate(truncate_at) if truncate_at < storage.size
     end
 
     # Serialize a task record as a checksummed, newline-delimited entry.
@@ -611,12 +726,21 @@ def with_lock(operation)
       end
 
       @locked = true
+      active_exception = nil
 
       begin
         yield
+      rescue Exception => e
+        active_exception = e
+        raise
       ensure
-        @lock_file.flock(File::LOCK_UN)
-        @locked = false
+        begin
+          @lock_file.flock(File::LOCK_UN)
+        rescue
+          raise unless active_exception
+        ensure
+          @locked = false
+        end
       end
     end
   end
diff --git a/lib/rage/deferred/backends/nil.rb b/lib/rage/deferred/backends/nil.rb
index 25829019..83d45e7a 100644
--- a/lib/rage/deferred/backends/nil.rb
+++ b/lib/rage/deferred/backends/nil.rb
@@ -17,8 +17,8 @@ def pending_tasks
   def add_dead_task(_, _, _, **)
   end
 
-  def list_dead_tasks(**)
-    []
+  def each_dead_task
+    self
   end
 
   def find_dead_task(_)
diff --git a/lib/rage/deferred/dead_task.rb b/lib/rage/deferred/dead_task.rb
new file mode 100644
index 00000000..1c997748
--- /dev/null
+++ b/lib/rage/deferred/dead_task.rb
@@ -0,0 +1,96 @@
+# frozen_string_literal: true
+
+##
+# Provides information about one dead task.
+#
+# A dead task is a deferred task that is no longer retried.
+# Creating this object does not load the task class.
+# The first call to {#args} or {#kwargs} deserializes the execution context.
+class Rage::Deferred::DeadTask
+  # @return [String] the ID that Rage assigned when it enqueued the task
+  attr_reader :id
+
+  # @return [String] the name of the task class
+  attr_reader :task_class
+
+  # @return [Integer] the number of processing attempts
+  attr_reader :attempts
+
+  # @return [Time] the time when Rage enqueued the task
+  attr_reader :enqueued_at
+
+  # @return [Time] the time when the last attempt failed
+  attr_reader :failed_at
+
+  # @return [String] the class name of the last exception
+  attr_reader :exception_class
+
+  # @return [String] the message from the last exception
+  attr_reader :exception_message
+
+  # @return [Array<String>] the frozen backtrace from the last exception
+  attr_reader :backtrace
+
+  # @private
+  def initialize(record)
+    @id = record[:id].dup.freeze
+    @task_class = record[:task_class].dup.freeze
+    @attempts = record[:attempts]
+    @enqueued_at = Time.at(record[:enqueued_at])
+    @failed_at = Time.at(record[:failed_at])
+
+    @exception_class = record[:exception_class].dup.freeze
+    @exception_message = record[:exception_message].dup.freeze
+    @backtrace = record[:backtrace].map { |line| line.dup.freeze }.freeze
+    @__context = record[:context].dup.freeze
+  end
+
+  # Return the original positional arguments.
+  #
+  # On the first call, the method deserializes the execution context.
+  # The method freezes all objects in the decoded data.
+  # After a successful call, the method returns the same Array on each later call.
+  # If the stored value is `nil`, the method returns a frozen empty Array.
+  #
+  # @return [Array] frozen positional arguments
+  # @raise [Rage::Deferred::DeadTaskContextDeserializationError] when the context cannot be decoded
+  def args
+    decode_context unless defined?(@__args)
+    @__args
+  end
+
+  # Return the original keyword arguments.
+  #
+  # On the first call, the method deserializes the execution context.
+  # The method freezes all objects in the decoded data.
+  # After a successful call, the method returns the same Hash on each later call.
+  # If the stored value is `nil`, the method returns a frozen empty Hash.
+  #
+  # @return [Hash] frozen keyword arguments
+  # @raise [Rage::Deferred::DeadTaskContextDeserializationError] when the context cannot be decoded
+  def kwargs
+    decode_context unless defined?(@__kwargs)
+    @__kwargs
+  end
+
+  private
+
+  # @private
+  def decode_context
+    context = Marshal.load(@__context, freeze: true)
+    args = Rage::Deferred::Context.get_args(context)
+    kwargs = Rage::Deferred::Context.get_kwargs(context)
+    args = [].freeze if args.nil?
+    kwargs = {}.freeze if kwargs.nil?
+
+    @__args = args
+    @__kwargs = kwargs
+  rescue => e
+    error = Rage::Deferred::DeadTaskContextDeserializationError.new(
+      "Could not deserialize execution context for dead task #{@id}"
+    )
+    raise error, cause: e
+  end
+
+  private_class_method :new
+end
diff --git a/lib/rage/deferred/dead_tasks.rb b/lib/rage/deferred/dead_tasks.rb
new file mode 100644
index 00000000..7ba38d47
--- /dev/null
+++ b/lib/rage/deferred/dead_tasks.rb
@@ -0,0 +1,44 @@
+# frozen_string_literal: true
+
+##
+# Provides access to deferred tasks that Rage no longer retries.
+#
+# {Rage::Deferred.dead_tasks} returns this collection.
+class Rage::Deferred::DeadTasks
+  include Enumerable
+
+  # @private
+  def initialize(backend)
+    @backend = backend
+  end
+
+  # Yield each dead task from oldest to newest.
+  #
+  # Without a block, the method returns a new Enumerator.
+  #
+  # @yieldparam dead_task [Rage::Deferred::DeadTask] information about one dead task
+  # @return [Enumerator, self] a new Enumerator without a block, or this collection with a block
+  def each
+    return enum_for(__method__) unless block_given?
+
+    @backend.each_dead_task do |record|
+      yield Rage::Deferred::DeadTask.send(:new, record)
+    end
+
+    self
+  end
+
+  # Return the dead task with the exact persisted ID.
+  #
+  # The ID must be a String. The method does not convert other values to a String.
+  #
+  # @param id [String] the exact persisted task ID
+  # @return [Rage::Deferred::DeadTask, nil] the matching task, or nil
+  # @raise [TypeError] when the ID is not a String
+  def find_by_id(id)
+    raise TypeError, "dead task id must be a String" unless id.is_a?(String)
+
+    record = @backend.find_dead_task(id)
+    Rage::Deferred::DeadTask.send(:new, record) if record
+  end
+end
diff --git a/lib/rage/deferred/deferred.rb b/lib/rage/deferred/deferred.rb
index 00533fac..642f5a59 100644
--- a/lib/rage/deferred/deferred.rb
+++ b/lib/rage/deferred/deferred.rb
@@ -51,6 +51,24 @@ def self.wrap(instance, delay: nil, delay_until: nil)
     Rage::Deferred::Proxy.new(instance, delay:, delay_until:)
   end
 
+  # Return the collection of tasks that Rage no longer retries.
+  #
+  # The method returns the same collection object on each call.
+  # The collection uses the same backend object as the deferred queue.
+  #
+  # @example List dead tasks
+  #   Rage::Deferred.dead_tasks.each do |task|
+  #     puts "#{task.id}: #{task.exception_message}"
+  #   end
+  #
+  # @example Find one dead task
+  #   task = Rage::Deferred.dead_tasks.find_by_id("1735689600-12345-1")
+  #
+  # @return [Rage::Deferred::DeadTasks]
+  def self.dead_tasks
+    @__dead_tasks ||= Rage::Deferred::DeadTasks.new(__backend)
+  end
+
   # @private
   def self.__backend
     @__backend ||= Rage.config.deferred.backend
@@ -100,11 +118,20 @@ module Backends
   class PushTimeout < StandardError
   end
 
-  # Raised when a dead-tasks store operation cannot acquire the lock within the retry budget.
+  # Raise this error when the dead-task store cannot be locked after all retries.
   class DeadTasksLockTimeout < StandardError
   end
+
+  # Raise this error when a dead task's execution context cannot be deserialized.
+  #
+  # The error message identifies the dead task.
+  # `Exception#cause` contains the original deserialization error.
+  class DeadTaskContextDeserializationError < StandardError
+  end
 end
 
+require_relative "dead_task"
+require_relative "dead_tasks"
 require_relative "task"
 require_relative "scheduler"
 require_relative "queue"
diff --git a/spec/deferred/backends/disk_spec.rb b/spec/deferred/backends/disk_spec.rb
index a47023fa..f41387e6 100644
--- a/spec/deferred/backends/disk_spec.rb
+++ b/spec/deferred/backends/disk_spec.rb
@@ -325,7 +325,7 @@ def stored_records
 
         add_dead_task("2-2-2")
 
-        expect(backend.list_dead_tasks.map { |record| record[:id] }).to eq(["2-2-2", "1-1-1"])
+        expect(backend.enum_for(:each_dead_task).map { |record| record[:id] }).to eq(["1-1-1", "2-2-2"])
       end
 
       it "repairs a file containing only an incomplete entry before appending" do
@@ -334,7 +334,7 @@ def stored_records
 
         add_dead_task("1-1-1")
 
-        expect(backend.list_dead_tasks.map { |record| record[:id] }).to eq(["1-1-1"])
+        expect(backend.enum_for(:each_dead_task).map { |record| record[:id] }).to eq(["1-1-1"])
       end
 
       it "repairs a torn tail longer than one scan chunk" do
@@ -409,13 +409,733 @@ def stored_records
       end
     end
 
+    describe "#each_dead_task" do
+      def each_dead_task
+        backend.enum_for(:each_dead_task).to_a
+      end
+
+      def finish_external_traversal(iterator)
+        records = []
+        while true
+          records << iterator.next
+        end
+      rescue StopIteration
+        records
+      end
+
+      def append_physical_record(outer_id, record)
+        entry = dead_tasks_storage.send(:build_entry, outer_id, record)
+        dead_tasks_path.open("ab") { |storage| storage.write(entry) }
+      end
+
+      def append_serialized_record(outer_id, serialized_record)
+        payload = "dead_task:#{outer_id}:#{serialized_record}"
+        crc = Zlib.crc32(payload).to_s(16).rjust(8, "0")
+        dead_tasks_path.open("ab") { |storage| storage.write("#{crc}:#{payload}\n") }
+      end
+
+      it "keeps backend traversal argument-free" do
+        add_dead_task("1700000001-1-1")
+        add_dead_task("1700000002-1-2")
+        add_dead_task("1700000003-1-3")
+
+        expect(backend.method(:each_dead_task).parameters).to eq([[:block, :block]])
+        expect { backend.each_dead_task(limit: 1) {} }.to raise_error(ArgumentError)
+        expect(each_dead_task.map { |record| record[:id] }).to eq(
+          ["1700000001-1-1", "1700000002-1-2", "1700000003-1-3"]
+        )
+      end
+
+      it "yields the newest frame-valid duplicate" do
+        add_dead_task("1700000001-1-1")
+        older = stored_records.last
+        append_physical_record("1700000001-1-1", older.merge(task_class: "NewTask", failed_at: older[:failed_at] + 1))
+
+        records = each_dead_task
+
+        expect(records.length).to eq(1)
+        expect(records.first[:task_class]).to eq("NewTask")
+      end
+
+      it "orders a duplicate logical task at its winning record's position" do
+        add_dead_task("1700000001-1-1")
+        original = stored_records.last
+        add_dead_task("1700000002-1-2")
+        append_physical_record("1700000001-1-1", original.merge(task_class: "MovedTask"))
+
+        records = each_dead_task
+
+        expect(records.map { |record| record[:id] }).to eq(["1700000002-1-2", "1700000001-1-1"])
+        expect(records.last[:task_class]).to eq("MovedTask")
+      end
+
+      it "falls back to an older duplicate only when newer frames are physically invalid" do
+        add_dead_task("1700000001-1-1")
+        dead_tasks_path.open("ab") { |storage| storage.write("deadbeef:dead_task:1700000001-1-1:garbage\n") }
+
+        records = each_dead_task
+
+        expect(records.length).to eq(1)
+        expect(records.first[:id]).to eq("1700000001-1-1")
+        expect(records.first[:task_class]).to eq("SendWelcomeEmail")
+      end
+
+      it "does not fall back when the newest frame-valid payload cannot be decoded" do
+        add_dead_task("1700000001-1-1")
+        append_serialized_record("1700000001-1-1", "not-a-dumped-string")
+
+        expect { each_dead_task }.to raise_error(RuntimeError, /dumped string/)
+      end
+
+      it "silently skips malformed frames and an incomplete tail" do
+        add_dead_task("1700000001-1-1")
+        dead_tasks_path.open("ab") do |storage|
+          storage.write("bad\n")
+          storage.write("00000000:not_dead_task:id:payload\n")
+          storage.write("deadbeef:dead_task:1700000002-1-2:secret-payload\n")
+          storage.write("incomplete-secret-tail")
+        end
+        snapshot_bytes = dead_tasks_path.binread
+
+        expect {
+          expect(each_dead_task.map { |record| record[:id] }).to eq(["1700000001-1-1"])
+        }.not_to output.to_stdout
+        expect { each_dead_task }.not_to output.to_stderr
+        expect(dead_tasks_path.binread).to eq(snapshot_bytes)
+      end
+
+      it "does not schema-check frame-valid payloads during selection" do
+        add_dead_task("1700000001-1-1")
+        valid = stored_records.last
+        append_physical_record("1700000001-1-1", valid.merge(id: "inner-id", attempts: "three"))
+
+        expect(each_dead_task).to contain_exactly(include(id: "1700000001-1-1", attempts: "three"))
+      end
+
+      it "does not deserialize opaque context bytes" do
+        add_dead_task("1700000001-1-1")
+        record = stored_records.last.merge(context: "not a Marshal payload")
+        append_physical_record("1700000002-1-2", record.merge(id: "1700000002-1-2"))
+
+        expect(each_dead_task.map { |entry| entry[:id] }).to eq(["1700000001-1-1", "1700000002-1-2"])
+      end
+
+      it "reads large physical records during the forward scan" do
+        add_dead_task("1700000001-1-1")
+        record = stored_records.last
+        append_physical_record(
+          "1700000002-1-2",
+          record.merge(id: "1700000002-1-2", exception_message: "x" * 20_000)
+        )
+
+        expect(each_dead_task.map { |entry| entry[:id] }).to eq(["1700000001-1-1", "1700000002-1-2"])
+      end
+
+      it "returns and decodes only the oldest 20 winning records for first and take" do
+        25.times { |index| add_dead_task("170000#{index.to_s.rjust(4, "0")}-1-1") }
+        decode_count = 0
+        allow(Marshal).to receive(:load).and_wrap_original do |original, *args|
+          decode_count += 1
+          original.call(*args)
+        end
+
+        expected_ids = 20.times.map { |index| "170000#{index.to_s.rjust(4, "0")}-1-1" }
+
+        expect(backend.enum_for(:each_dead_task).first(20).map { |record| record[:id] }).to eq(expected_ids)
+        expect(decode_count).to eq(20)
+
+        decode_count = 0
+        expect(backend.enum_for(:each_dead_task).take(20).map { |record| record[:id] }).to eq(expected_ids)
+        expect(decode_count).to eq(20)
+      end
+
+      it "finishes frame-only winner selection before the first yield and decodes only the consumed winner" do
+        5.times { |index| add_dead_task("170000000#{index}-1-1") }
+        scanned_ids = []
+        allow(dead_tasks_storage).to receive(:framed_record).and_wrap_original do |original, entry|
+          original.call(entry).tap { |framed_record| scanned_ids << framed_record&.first }
+        end
+        allow(Marshal).to receive(:load).and_wrap_original do |original, *args|
+          expect(scanned_ids).to eq(5.times.map { |index| "170000000#{index}-1-1" })
+          original.call(*args)
+        end
+
+        first = backend.enum_for(:each_dead_task).first
+
+        expect(first[:id]).to eq("1700000000-1-1")
+        expect(Marshal).to have_received(:load).once
+      end
+
+      it "decodes each winner immediately before yielding it" do
+        3.times { |index| add_dead_task("170000000#{index}-1-1") }
+        decoded_ids = []
+        yielded_ids = []
+        allow(Marshal).to receive(:load).and_wrap_original do |original, *args|
+          original.call(*args).tap { |record| decoded_ids << record[:id] }
+        end
+
+        backend.each_dead_task do |record|
+          yielded_ids << record[:id]
+          expect(decoded_ids).to eq(yielded_ids)
+        end
+      end
+
+      it "does not open or inspect a snapshot until first advancement" do
+        add_dead_task("1700000001-1-1")
+        allow(dead_tasks_storage).to receive(:with_snapshot).and_call_original
+        allow(Marshal).to receive(:load).and_call_original
+
+        iterator = backend.enum_for(:each_dead_task)
+
+        expect(dead_tasks_storage).not_to have_received(:with_snapshot)
+        expect(Marshal).not_to have_received(:load)
+
+        iterator.next
+
+        expect(dead_tasks_storage).to have_received(:with_snapshot).once
+        expect(Marshal).to have_received(:load).once
+      end
+
+      it "excludes an incomplete tail and records appended after first advancement" do
+        add_dead_task("1700000001-1-1")
+        add_dead_task("1700000002-1-2")
+        dead_tasks_path.open("ab") { |storage| storage.write("deadbeef:dead_task:partial") }
+        iterator = backend.enum_for(:each_dead_task)
+
+        expect(iterator.next[:id]).to eq("1700000001-1-1")
+        add_dead_task("1700000003-1-3")
+
+        expect(finish_external_traversal(iterator).map { |record| record[:id] }).to eq(["1700000002-1-2"])
+        expect(each_dead_task.map { |record| record[:id] }).to eq(
+          ["1700000001-1-1", "1700000002-1-2", "1700000003-1-3"]
+        )
+      end
+
+      it "keeps the captured inode stable across rename-based compaction" do
+        add_dead_task("1700000001-1-1")
+        add_dead_task("1700000002-1-2")
+        add_dead_task("1700000003-1-3")
+        iterator = backend.enum_for(:each_dead_task)
+
+        expect(iterator.next[:id]).to eq("1700000001-1-1")
+        backend.remove_dead_tasks("1700000002-1-2")
+
+        expect(finish_external_traversal(iterator).map { |record| record[:id] }).to eq(
+          ["1700000002-1-2", "1700000003-1-3"]
+        )
+        expect(each_dead_task.map { |record| record[:id] }).to eq(["1700000001-1-1", "1700000003-1-3"])
+      end
+
+      it "releases the permanent lock before decoding and yielding" do
+        add_dead_task("1700000001-1-1")
+        add_dead_task("1700000002-1-2")
+        allow(Marshal).to receive(:load).and_wrap_original do |original, *args|
+          expect(dead_tasks_storage.instance_variable_get(:@locked)).to eq(false)
+          original.call(*args)
+        end
+
+        backend.each_dead_task do
+          expect(dead_tasks_storage.instance_variable_get(:@locked)).to eq(false)
+          File.open(lock_path, File::WRONLY) do |lock|
+            expect(lock.flock(File::LOCK_EX | File::LOCK_NB)).to be_truthy
+            lock.flock(File::LOCK_UN)
+          end
+        end
+      end
+
+      it "closes the snapshot after exhaustion, block exit, and terminal Enumerable exits" do
+        add_dead_task("1700000001-1-1")
+        descriptors = []
+        allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation|
+          original.call do |storage, snapshot_end|
+            descriptors << storage
+            operation.call(storage, snapshot_end)
+          end
+        end
+        collection = Rage::Deferred::DeadTasks.new(backend)
+
+        backend.each_dead_task {}
+        backend.each_dead_task { break }
+        catch(:finished) { backend.each_dead_task { throw :finished } }
+        collection.find { true }
+        collection.first
+        collection.take(1)
+
+        expect(descriptors.length).to eq(6)
+        expect(descriptors).to all(be_closed)
+      end
+
+      it "keeps a partial external traversal open until it is exhausted" do
+        add_dead_task("1700000001-1-1")
+        add_dead_task("1700000002-1-2")
+        descriptor = nil
+        allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation|
+          original.call do |storage, snapshot_end|
+            descriptor = storage
+            operation.call(storage, snapshot_end)
+          end
+        end
+        iterator = backend.enum_for(:each_dead_task)
+
+        iterator.next
+        expect(descriptor).not_to be_closed
+
+        finish_external_traversal(iterator)
+        expect(descriptor).to be_closed
+      end
+
+      it "does not let cleanup replace an active user exception" do
+        add_dead_task("1700000001-1-1")
+        descriptor = nil
+        allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation|
+          original.call do |storage, snapshot_end|
+            descriptor = storage
+            allow(descriptor).to receive(:close).and_wrap_original do |close|
+              close.call
+              raise Errno::EIO
+            end
+            operation.call(storage, snapshot_end)
+          end
+        end
+
+        expect {
+          backend.each_dead_task { raise "user failure" }
+        }.to raise_error(RuntimeError, "user failure")
+        expect(descriptor).to be_closed
+      end
+
+      it "propagates a descriptor cleanup error when it is the only failure" do
+        add_dead_task("1700000001-1-1")
+        allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation|
+          original.call do |storage, snapshot_end|
+            allow(storage).to receive(:close).and_wrap_original do |close|
+              close.call
+              raise Errno::EIO
+            end
+            operation.call(storage, snapshot_end)
+          end
+        end
+
+        expect {
+          backend.each_dead_task {}
+        }.to raise_error(Errno::EIO)
+      end
+
+      it "propagates a descriptor cleanup error inside an unrelated rescue" do
+        add_dead_task("1700000001-1-1")
+        allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation|
+          original.call do |storage, snapshot_end|
+            allow(storage).to receive(:close).and_wrap_original do |close|
+              close.call
+              raise Errno::EIO
+            end
+            operation.call(storage, snapshot_end)
+          end
+        end
+
+        begin
+          raise "unrelated failure"
+        rescue RuntimeError
+          expect {
+            backend.each_dead_task {}
+          }.to raise_error(Errno::EIO)
+        end
+      end
+
+      it "propagates a lock cleanup error inside an unrelated rescue" do
+        lock_file = dead_tasks_storage.instance_variable_get(:@lock_file)
+        allow(lock_file).to receive(:flock).and_wrap_original do |flock, operation|
+          flock.call(operation).tap do
+            raise Errno::EIO if operation == File::LOCK_UN
+          end
+        end
+
+        begin
+          raise "unrelated failure"
+        rescue RuntimeError
+          expect {
+            backend.each_dead_task {}
+          }.to raise_error(Errno::EIO)
+        end
+      end
+
+      it "propagates lazy snapshot-acquisition failures and closes the opened descriptor" do
+        descriptor = nil
+        allow(dead_tasks_storage).to receive(:complete_record_end) do |storage|
+          descriptor = storage
+          raise Errno::EACCES
+        end
+        iterator = backend.enum_for(:each_dead_task)
+
+        expect(descriptor).to be_nil
+        expect { iterator.next }.to raise_error(Errno::EACCES)
+        expect(descriptor).to be_closed
+      end
+
+      it "propagates an open failure from first advancement" do
+        backend
+        allow(File).to receive(:open).and_call_original
+        allow(File).to receive(:open).with(dead_tasks_path, File::RDONLY | File::BINARY).and_raise(Errno::EACCES)
+        iterator = backend.enum_for(:each_dead_task)
+
+        expect { iterator.next }.to raise_error(Errno::EACCES)
+      end
+
+      it "propagates forward-scan seek failures and closes the snapshot" do
+        add_dead_task("1700000001-1-1")
+        descriptor = nil
+        allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation|
+          original.call do |storage, snapshot_end|
+            descriptor = storage
+            allow(descriptor).to receive(:seek).and_raise(Errno::EIO)
+            operation.call(storage, snapshot_end)
+          end
+        end
+
+        expect {
+          backend.each_dead_task {}
+        }.to raise_error(Errno::EIO)
+        expect(descriptor).to be_closed
+      end
+
+      it "propagates forward-scan read failures and closes the snapshot" do
+        add_dead_task("1700000001-1-1")
+        descriptor = nil
+        allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation|
+          original.call do |storage, snapshot_end|
+            descriptor = storage
+            allow(descriptor).to receive(:gets).and_raise(Errno::EIO)
+            operation.call(storage, snapshot_end)
+          end
+        end
+
+        expect {
+          backend.each_dead_task {}
+        }.to raise_error(Errno::EIO)
+        expect(descriptor).to be_closed
+      end
+
+      it "propagates snapshot lock timeouts on first advancement" do
+        stub_const("#{described_class}::DeadTasksStorage::LOCK_MAX_ATTEMPTS", 1)
+        dead_tasks_storage.instance_variable_set(:@locked, true)
+        iterator = backend.enum_for(:each_dead_task)
+
+        expect { iterator.next }.to raise_error(Rage::Deferred::DeadTasksLockTimeout, /read tasks from/)
+      ensure
+        dead_tasks_storage.instance_variable_set(:@locked, false)
+      end
+
+      it "keeps overlapping and Fiber-interleaved traversals independent" do
+        add_dead_task("1700000001-1-1")
+        add_dead_task("1700000002-1-2")
+        first = backend.enum_for(:each_dead_task)
+        second = backend.enum_for(:each_dead_task)
+        first_fiber = Fiber.new do
+          ids = [first.next[:id]]
+          Fiber.yield(ids.last)
+          ids << first.next[:id]
+          first.next
+        rescue StopIteration
+          ids
+        end
+        second_fiber = Fiber.new do
+          ids = [second.next[:id]]
+          Fiber.yield(ids.last)
+          ids << second.next[:id]
+          second.next
+        rescue StopIteration
+          ids
+        end
+
+        expect(first_fiber.resume).to eq("1700000001-1-1")
+        expect(second_fiber.resume).to eq("1700000001-1-1")
+        expect(first_fiber.resume).to eq(["1700000001-1-1", "1700000002-1-2"])
+        expect(second_fiber.resume).to eq(["1700000001-1-1", "1700000002-1-2"])
+      end
+
+      it "keeps a paused traversal usable when another traversal raises" do
+        add_dead_task("1700000001-1-1")
+        add_dead_task("1700000002-1-2")
+        paused = backend.enum_for(:each_dead_task)
+
+        expect(paused.next[:id]).to eq("1700000001-1-1")
+        expect { backend.each_dead_task { raise "application failure" } }.to raise_error("application failure")
+        expect(finish_external_traversal(paused).map { |record| record[:id] }).to eq(["1700000002-1-2"])
+      end
+
+      it "establishes separate snapshots on each traversal's first advancement" do
+        add_dead_task("1700000001-1-1")
+        add_dead_task("1700000002-1-2")
+        first = backend.enum_for(:each_dead_task)
+        second = backend.enum_for(:each_dead_task)
+
+        expect(first.next[:id]).to eq("1700000001-1-1")
+        add_dead_task("1700000003-1-3")
+        expect(finish_external_traversal(second).map { |record| record[:id] }).to eq(
+          ["1700000001-1-1", "1700000002-1-2", "1700000003-1-3"]
+        )
+        expect(finish_external_traversal(first).map { |record| record[:id] }).to eq(["1700000002-1-2"])
+      end
+
+      it "keeps nested traversals independent" do
+        add_dead_task("1700000001-1-1")
+        add_dead_task("1700000002-1-2")
+        outer_ids = []
+        nested_ids = nil
+
+        backend.each_dead_task do |record|
+          outer_ids << record[:id]
+          nested_ids ||= each_dead_task.map { |nested| nested[:id] }
+        end
+
+        expect(outer_ids).to eq(["1700000001-1-1", "1700000002-1-2"])
+        expect(nested_ids).to eq(outer_ids)
+      end
+    end
+
+    describe "#find_dead_task" do
+      let(:lookup_id) { "1700000001-1-1" }
+
+      def append_lookup_record(outer_id, record)
+        entry = dead_tasks_storage.send(:build_entry, outer_id, record)
+        dead_tasks_path.open("ab") { |storage| storage.write(entry) }
+      end
+
+      def append_lookup_payload(outer_id, serialized_record)
+        payload = "dead_task:#{outer_id}:#{serialized_record}"
+        crc = Zlib.crc32(payload).to_s(16).rjust(8, "0")
+        dead_tasks_path.open("ab") { |storage| storage.write("#{crc}:#{payload}\n") }
+      end
+
+      def store_lookup_record(mode)
+        add_dead_task(lookup_id)
+        original = stored_records.last
+
+        case mode
+        when :existing
+          original
+        when :duplicate
+          original.merge(task_class: "NewestTask", failed_at: original[:failed_at] + 1).tap do |record|
+            append_lookup_record(lookup_id, record)
+          end
+        when :frame_invalid_duplicate
+          dead_tasks_path.open("ab") { |storage| storage.write("deadbeef:dead_task:#{lookup_id}:secret\n") }
+          original
+        when :selected_winner_failure
+          append_lookup_payload(lookup_id, "not-a-dumped-string")
+          [RuntimeError, /dumped string/]
+        when :schema_incompatible_duplicate
+          original.merge(attempts: "invalid").tap do |record|
+            append_lookup_record(lookup_id, record)
+          end
+        end
+      end
+
+      include_examples "a dead-task exact-lookup backend", empty: false
+
+      it "silently skips malformed physical records without disclosing their payload" do
+        add_dead_task(lookup_id)
+        valid = stored_records.last
+        dead_tasks_path.open("ab") { |storage| storage.write("secret without a valid frame\n") }
+
+        expect { expect(backend.find_dead_task(lookup_id)).to eq(valid) }.not_to output.to_stdout
+        expect { backend.find_dead_task(lookup_id) }.not_to output.to_stderr
+      end
+
+      it "returns a frame-valid schema-incompatible match and excludes an incomplete tail" do
+        backend
+        record = {
+          id: lookup_id, task_class: "Task", attempts: "invalid",
+          enqueued_at: 1, failed_at: 2, exception_class: "RuntimeError",
+          exception_message: "secret", backtrace: [], context: Marshal.dump(["Task", [], {}])
+        }
+        append_lookup_record(lookup_id, record)
+        dead_tasks_path.open("ab") { |storage| storage.write("incomplete-secret-tail") }
+        bytes = dead_tasks_path.binread
+
+        expect(backend.find_dead_task(lookup_id)).to eq(record)
+        expect(dead_tasks_path.binread).to eq(bytes)
+      end
+
+      it "completes the forward frame scan before decoding the newest match" do
+        add_dead_task(lookup_id)
+        add_dead_task("1700000002-1-2")
+        newest = stored_records.first.merge(exception_message: "newest")
+        append_lookup_record(lookup_id, newest)
+        add_dead_task("1700000003-1-3")
+        scanned_ids = []
+        allow(Marshal).to receive(:load).and_call_original
+        allow(dead_tasks_storage).to receive(:framed_record).and_wrap_original do |original, entry|
+          expect(Marshal).not_to have_received(:load)
+          original.call(entry).tap { |framed_record| scanned_ids << framed_record&.first }
+        end
+
+        expect(backend.find_dead_task(lookup_id)[:exception_message]).to eq("newest")
+        expect(scanned_ids).to eq([lookup_id, "1700000002-1-2", lookup_id, "1700000003-1-3"])
+        expect(Marshal).to have_received(:load).once
+      end
+
+      it "uses the outer framed id when the decoded payload id differs" do
+        add_dead_task(lookup_id)
+        append_lookup_record(lookup_id, stored_records.last.merge(id: "different-id"))
+
+        expect(backend.find_dead_task(lookup_id)[:id]).to eq(lookup_id)
+      end
+
+      it "finds records with incompatible opaque contexts without loading them" do
+        add_dead_task(lookup_id)
+        record = stored_records.last.merge(context: "not Marshal")
+        append_lookup_record("1700000002-1-2", record.merge(id: "1700000002-1-2"))
+        allow(Marshal).to receive(:load).and_call_original
+
+        found = backend.find_dead_task("1700000002-1-2")
+
+        expect(found[:id]).to eq("1700000002-1-2")
+        expect(Marshal).to have_received(:load).once
+      end
+
+      it "uses an independent snapshot without disturbing a paused enumeration" do
+        add_dead_task(lookup_id)
+        add_dead_task("1700000002-1-2")
+        dead_tasks_path.open("ab") { |storage| storage.write("malformed-secret\n") }
+        iterator = backend.enum_for(:each_dead_task)
+
+        expect(iterator.next[:id]).to eq(lookup_id)
+        expect(backend.find_dead_task("1700000002-1-2")[:id]).to eq("1700000002-1-2")
+        expect(iterator.next[:id]).to eq("1700000002-1-2")
+        expect { iterator.next }.to raise_error(StopIteration)
+      end
+
+      it "uses the complete-record boundary captured when lookup starts" do
+        add_dead_task(lookup_id)
+        original = stored_records.last
+        appended = false
+        newest = original.merge(exception_message: "added during lookup")
+        allow(dead_tasks_storage).to receive(:framed_record).and_wrap_original do |method, entry|
+          method.call(entry).tap do
+            unless appended
+              appended = true
+              append_lookup_record(lookup_id, newest)
+            end
+          end
+        end
+
+        expect(backend.find_dead_task(lookup_id)).to eq(original)
+        expect(backend.find_dead_task(lookup_id)).to eq(newest)
+      end
+
+      it "propagates snapshot lock timeouts unchanged" do
+        stub_const("#{described_class}::DeadTasksStorage::LOCK_MAX_ATTEMPTS", 1)
+        dead_tasks_storage.instance_variable_set(:@locked, true)
+
+        expect {
+          backend.find_dead_task(lookup_id)
+        }.to raise_error(Rage::Deferred::DeadTasksLockTimeout, /read tasks from/)
+      ensure
+        dead_tasks_storage.instance_variable_set(:@locked, false)
+      end
+
+      it "propagates open failures unchanged" do
+        backend
+        allow(File).to receive(:open).and_call_original
+        allow(File).to receive(:open).with(dead_tasks_path, File::RDONLY | File::BINARY).and_raise(Errno::EACCES)
+
+        expect { backend.find_dead_task(lookup_id) }.to raise_error(Errno::EACCES)
+      end
+
+      it "propagates forward-scan seek failures and closes the snapshot" do
+        add_dead_task(lookup_id)
+        descriptor = nil
+        allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation|
+          original.call do |storage, snapshot_end|
+            descriptor = storage
+            allow(descriptor).to receive(:seek).and_raise(Errno::EIO)
+            operation.call(storage, snapshot_end)
+          end
+        end
+
+        expect { backend.find_dead_task(lookup_id) }.to raise_error(Errno::EIO)
+        expect(descriptor).to be_closed
+      end
+
+      it "propagates forward-scan read failures and closes the snapshot" do
+        add_dead_task(lookup_id)
+        descriptor = nil
+        allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation|
+          original.call do |storage, snapshot_end|
+            descriptor = storage
+            allow(descriptor).to receive(:gets).and_raise(Errno::EIO)
+            operation.call(storage, snapshot_end)
+          end
+        end
+
+        expect { backend.find_dead_task(lookup_id) }.to raise_error(Errno::EIO)
+        expect(descriptor).to be_closed
+      end
+
+      it "does not let cleanup replace an active lookup failure" do
+        add_dead_task(lookup_id)
+        descriptor = nil
+        allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation|
+          original.call do |storage, snapshot_end|
+            descriptor = storage
+            allow(descriptor).to receive(:seek).and_raise(Errno::EACCES)
+            allow(descriptor).to receive(:close).and_wrap_original do |close|
+              close.call
+              raise Errno::EIO
+            end
+            operation.call(storage, snapshot_end)
+          end
+        end
+
+        expect { backend.find_dead_task(lookup_id) }.to raise_error(Errno::EACCES)
+        expect(descriptor).to be_closed
+      end
+
+      it "propagates a descriptor cleanup error when it is the only failure" do
+        add_dead_task(lookup_id)
+        allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation|
+          original.call do |storage, snapshot_end|
+            allow(storage).to receive(:close).and_wrap_original do |close|
+              close.call
+              raise Errno::EIO
+            end
+            operation.call(storage, snapshot_end)
+          end
+        end
+
+        expect { backend.find_dead_task(lookup_id) }.to raise_error(Errno::EIO)
+      end
+
+      it "propagates an operational failure after corruption and a provisional match" do
+        backend
+        dead_tasks_path.open("ab") { |storage| storage.write("deadbeef:dead_task:#{lookup_id}:secret\n") }
+        add_dead_task(lookup_id)
+        add_dead_task("1700000002-1-2")
+        allow(dead_tasks_storage).to receive(:with_snapshot).and_wrap_original do |original, &operation|
+          original.call do |storage, snapshot_end|
+            reads = 0
+            allow(storage).to receive(:gets).and_wrap_original do |gets|
+              reads += 1
+              raise Errno::EIO if reads == 3
+
+              gets.call
+            end
+            operation.call(storage, snapshot_end)
+          end
+        end
+
+        expect { backend.find_dead_task(lookup_id) }.to raise_error(Errno::EIO)
+      end
+    end
+
     describe "#remove_dead_tasks" do
       it "removes the given dead tasks and keeps the others" do
         add_dead_task("1-1-1")
         add_dead_task("2-2-2")
 
         expect(backend.remove_dead_tasks("1-1-1")).to eq(1)
-        expect(backend.list_dead_tasks.map { |record| record[:id] }).to eq(["2-2-2"])
+        expect(backend.enum_for(:each_dead_task).map { |record| record[:id] }).to eq(["2-2-2"])
         expect(tmp_storage_path).not_to exist
       end
 
@@ -543,7 +1263,7 @@ def stored_records
         add_dead_task("1-1-1")
 
         expect(backend.remove_dead_tasks("nope")).to eq(0)
-        expect(backend.list_dead_tasks.map { |record| record[:id] }).to eq(["1-1-1"])
+        expect(backend.enum_for(:each_dead_task).map { |record| record[:id] }).to eq(["1-1-1"])
         expect(tmp_storage_path).not_to exist
       end
 
diff --git a/spec/deferred/backends/nil_spec.rb b/spec/deferred/backends/nil_spec.rb
new file mode 100644
index 00000000..07cb3919
--- /dev/null
+++ b/spec/deferred/backends/nil_spec.rb
@@ -0,0 +1,13 @@
+# frozen_string_literal: true
+
+RSpec.describe Rage::Deferred::Backends::Nil do
+  subject(:backend) { described_class.new }
+
+  it "retains traversal and removal behavior" do
+    expect(backend).to respond_to(:each_dead_task, :find_dead_task, :remove_dead_tasks)
+    expect(backend.enum_for(:each_dead_task).to_a).to eq([])
+    expect(backend.remove_dead_tasks("missing-id")).to eq(0)
+  end
+
+  include_examples "a dead-task exact-lookup backend", empty: true
+end
diff --git a/spec/deferred/dead_tasks_spec.rb b/spec/deferred/dead_tasks_spec.rb
new file mode 100644
index 00000000..907d63bc
--- /dev/null
+++ b/spec/deferred/dead_tasks_spec.rb
@@ -0,0 +1,389 @@
+# frozen_string_literal: true
+
+RSpec.describe Rage::Deferred::DeadTasks do
+  let(:records) do
+    [
+      {
+        id: "1700000001-1-1", task_class: "OlderTask", attempts: 2,
+        enqueued_at: 1_700_000_001, failed_at: 1_700_000_011,
+        exception_class: "ArgumentError", exception_message: "bad argument",
+        backtrace: [], context: Marshal.dump(["OlderTask", [], {}])
+      },
+      {
+        id: "1700000002-1-2", task_class: "MissingTask", attempts: 4,
+        enqueued_at: 1_700_000_002, failed_at: 1_700_000_012,
+        exception_class: "RuntimeError", exception_message: "secret failure",
+        backtrace: ["app/task.rb:1"], context: Marshal.dump(["MissingTask", [], {}])
+      }
+    ]
+  end
+  let(:backend) do
+    stored_records = records
+    Class.new do
+      attr_reader :traversal_count, :lookup_count
+
+      define_method(:initialize) do
+        @traversal_count = 0
+        @lookup_count = 0
+      end
+      define_method(:each_dead_task) do |&block|
+        @traversal_count += 1
+        stored_records.each(&block)
+        self
+      end
+      define_method(:find_dead_task) do |id|
+        @lookup_count += 1
+        stored_records.reverse_each.find { |record| record[:id] == id }
+      end
+    end.new
+  end
+  subject(:dead_tasks) { described_class.new(backend) }
+
+  describe "#each" do
+    it "returns distinct standard Enumerators without starting backend traversal" do
+      expect(backend).not_to receive(:each_dead_task)
+
+      first = dead_tasks.each
+      second = dead_tasks.each
+
+      expect(first).to be_a(Enumerator)
+      expect(second).to be_a(Enumerator)
+      expect(first).not_to equal(second)
+      expect(first).not_to respond_to(:close)
+    end
+
+    it "keeps both the public and private backend traversal APIs argument-free" do
+      expect(dead_tasks.method(:each).parameters).to eq([])
+      expect(backend.method(:each_dead_task).parameters).to eq([[:block, :block]])
+      expect { dead_tasks.each(limit: 1) }.to raise_error(ArgumentError)
+
+      dead_tasks.each {}
+
+      expect(backend.traversal_count).to eq(1)
+    end
+
+    it "yields passive defensive summaries without resolving task classes or contexts" do
+      expect(Object).not_to receive(:const_get)
+      expect(Marshal).not_to receive(:load)
+
+      entries = dead_tasks.to_a
+
+      expect(entries.map(&:id)).to eq(["1700000001-1-1", "1700000002-1-2"])
+      expect(entries.last.task_class).to eq("MissingTask")
+      expect(entries.last.attempts).to eq(4)
+      expect(entries.last.enqueued_at).to eq(Time.at(1_700_000_002))
+      expect(entries.last.failed_at).to eq(Time.at(1_700_000_012))
+      expect(entries.last.id).to be_frozen
+      expect(entries.last.task_class).to be_frozen
+      expect(entries.last.enqueued_at.utc).to equal(entries.last.enqueued_at)
+      expect(entries.last.failed_at.gmtime).to equal(entries.last.failed_at)
+      entries.last.enqueued_at.localtime("+05:00")
+      entries.last.failed_at.localtime("-04:00")
+      fresh_entry = dead_tasks.to_a.last
+
+      expect(entries.last.enqueued_at.utc_offset).to eq(5 * 60 * 60)
+      expect(entries.last.failed_at.utc_offset).to eq(-4 * 60 * 60)
+      expect(fresh_entry.enqueued_at).to eq(Time.at(1_700_000_002))
+      expect(fresh_entry.failed_at).to eq(Time.at(1_700_000_012))
+      expect(fresh_entry.enqueued_at).not_to equal(entries.last.enqueued_at)
+      expect(fresh_entry.failed_at).not_to equal(entries.last.failed_at)
+      expect(records.last.values_at(:enqueued_at, :failed_at)).to eq([1_700_000_002, 1_700_000_012])
+      expect(entries.last).to respond_to(:args, :kwargs, :exception_class, :exception_message, :backtrace)
+    end
+
+    it "retains standard Enumerable find and detect behavior" do
+      fallback = -> { :missing }
+      unused_fallback = -> { raise "must not run" }
+
+      expect(dead_tasks.find { |entry| entry.attempts == 2 }.task_class).to eq("OlderTask")
+      expect(dead_tasks.find(unused_fallback) { |entry| entry.attempts == 2 }.task_class).to eq("OlderTask")
+      expect(dead_tasks.detect(fallback) { |entry| entry.attempts == 99 }).to eq(:missing)
+      expect(dead_tasks.find).to be_a(Enumerator)
+      expect(dead_tasks.detect).to be_a(Enumerator)
+    end
+
+    it "uses independent traversals for separate Enumerable operations" do
+      dead_tasks.first
+      dead_tasks.count
+
+      expect(backend.traversal_count).to eq(2)
+      expect(dead_tasks.instance_variables).to eq([:@backend])
+      expect(dead_tasks.instance_variable_get(:@backend)).to equal(backend)
+    end
+  end
+
+  describe "#find_by_id" do
+    it "validates an exact String before backend access without coercion" do
+      expect(backend).not_to receive(:find_dead_task)
+
+      expect { dead_tasks.find_by_id(:"1700000001-1-1") }.to raise_error(TypeError, /String/)
+      expect { dead_tasks.find_by_id(1_700_000_001) }.to raise_error(TypeError, /String/)
+      expect(backend.lookup_count).to eq(0)
+    end
+
+    it "delegates polymorphically and wraps a matching record" do
+      entry = dead_tasks.find_by_id("1700000002-1-2")
+
+      expect(entry).to be_a(Rage::Deferred::DeadTask)
+      expect(entry.id).to eq("1700000002-1-2")
+      expect(backend.lookup_count).to eq(1)
+      expect(dead_tasks.find_by_id("missing")).to be_nil
+    end
+
+    it "uses the newest backend record without schema validation or fallback" do
+      id = "schema-drift"
+      records << records.first.merge(id:, attempts: 1, context: Marshal.dump(["Task", ["older"], {}]))
+      records << records.first.merge(id:, task_class: nil, attempts: "unvalidated", context: "not Marshal")
+
+      entry = dead_tasks.find_by_id(id)
+
+      expect(entry.task_class).to be_nil
+      expect(entry.attempts).to eq("unvalidated")
+      expect { entry.args }.to raise_error(Rage::Deferred::DeadTaskContextDeserializationError)
+    end
+
+    it "propagates natural entry-construction failures from the selected backend record" do
+      invalid_records = [
+        [[], TypeError],
+        [records.first.merge(backtrace: "not an array"), NoMethodError]
+      ]
+
+      invalid_records.each do |record, error_class|
+        selected_backend = instance_double("Rage::Deferred::Backend", find_dead_task: record)
+        collection = described_class.new(selected_backend)
+
+        expect { collection.find_by_id("selected") }.to raise_error(error_class)
+      end
+    end
+
+    it "exposes frozen exception metadata without decoding context or resolving the task class" do
+      expect(Object).not_to receive(:const_get)
+      expect(Marshal).not_to receive(:load)
+
+      entry = dead_tasks.find_by_id("1700000002-1-2")
+
+      expect(entry.exception_class).to eq("RuntimeError")
+      expect(entry.exception_message).to eq("secret failure")
+      expect(entry.backtrace).to eq(["app/task.rb:1"])
+      expect(entry.exception_class).to be_frozen
+      expect(entry.exception_message).to be_frozen
+      expect(entry.backtrace).to be_frozen
+      expect(entry.backtrace.first).to be_frozen
+      expect { entry.backtrace << "changed" }.to raise_error(FrozenError)
+    end
+
+    it "lazily decodes empty, positional, keyword, and mixed arguments with stable frozen identities" do
+      contexts = [
+        ["Task", nil, nil],
+        ["Task", [["nested"]], nil],
+        ["Task", nil, { key: { nested: "value" } }],
+        ["Task", [1], { key: 2 }]
+      ]
+      load_count = 0
+      allow(Marshal).to receive(:load).and_wrap_original do |original, *args, **kwargs|
+        load_count += 1
+        original.call(*args, **kwargs)
+      end
+
+      contexts.each_with_index do |context, index|
+        record = records.first.merge(id: "context-#{index}", context: Marshal.dump(context))
+        records << record
+        entry = dead_tasks.find_by_id(record[:id])
+
+        expect(entry.args).to eq(context[1] || [])
+        expect(entry.kwargs).to eq(context[2] || {})
+        expect(entry.args).to equal(entry.args)
+        expect(entry.kwargs).to equal(entry.kwargs)
+        expect(entry.args).to be_frozen
+        expect(entry.kwargs).to be_frozen
+      end
+
+      expect(load_count).to eq(contexts.length)
+      expect(dead_tasks.find_by_id("context-1").args.dig(0)).to eq(["nested"])
+      expect(dead_tasks.find_by_id("context-2").kwargs.dig(:key, :nested)).to eq("value")
+    end
+
+    it "uses the deferred context accessors to extract positional and keyword arguments" do
+      context = ["Task", [1], { key: 2 }]
+      records << records.first.merge(id: "context-accessors", context: Marshal.dump(context))
+      entry = dead_tasks.find_by_id("context-accessors")
+
+      expect(Marshal).to receive(:load).with(instance_of(String), freeze: true).and_call_original
+      expect(Rage::Deferred::Context).to receive(:get_args).with(instance_of(Array)).and_call_original
+      expect(Rage::Deferred::Context).to receive(:get_kwargs).with(instance_of(Array)).and_call_original
+
+      expect(entry.args).to eq([1])
+      expect(entry.kwargs).to eq({ key: 2 })
+    end
+
+    it "normalizes missing argument slots through the context accessors" do
+      records << records.first.merge(id: "missing-slots", context: Marshal.dump(["Task"]))
+      entry = dead_tasks.find_by_id("missing-slots")
+
+      expect(entry.args).to eq([])
+      expect(entry.kwargs).to eq({})
+      expect(entry.args).to be_frozen
+      expect(entry.kwargs).to be_frozen
+      expect(entry.args).to equal(entry.args)
+      expect(entry.kwargs).to equal(entry.kwargs)
+    end
+
+    it "freezes nested argument graphs defensively" do
+      shared = ["secret"]
+      shared << shared
+      records << records.first.merge(
+        id: "nested", context: Marshal.dump(["Task", [shared], { shared: }])
+      )
+      entry = dead_tasks.find_by_id("nested")
+
+      expect(entry.args.first).to equal(entry.kwargs[:shared])
+      expect(entry.args.first).to equal(entry.args.first.last)
+      expect(entry.args.first).to be_frozen
+      expect(entry.args.first.first).to be_frozen
+      expect { entry.args.first << "changed" }.to raise_error(FrozenError)
+      expect { entry.kwargs[:shared].first.replace("changed") }.to raise_error(FrozenError)
+    end
+
+    it "raises the dedicated error with id and cause, without caching a failed decode" do
+      records << records.first.merge(id: "broken", context: "not Marshal")
+      entry = dead_tasks.find_by_id("broken")
+
+      2.times do
+        expect { entry.args }.to raise_error(Rage::Deferred::DeadTaskContextDeserializationError) { |error|
+          expect(error.message).to include("broken")
+          expect(error.cause).to be_a(TypeError)
+        }
+      end
+      expect(entry.id).to eq("broken")
+      expect(entry.exception_class).to eq("ArgumentError")
+      expect(dead_tasks.find_by_id("broken")).not_to be_nil
+    end
+
+    it "translates context accessor failures without caching a partial extraction" do
+      records << records.first.merge(id: "accessor-failure", context: Marshal.dump(["Task", [1], { key: 2 }]))
+      entry = dead_tasks.find_by_id("accessor-failure")
+      args_calls = 0
+      kwargs_calls = 0
+
+      allow(Rage::Deferred::Context).to receive(:get_args).and_wrap_original do |method, context|
+        args_calls += 1
+        method.call(context)
+      end
+      allow(Rage::Deferred::Context).to receive(:get_kwargs).and_wrap_original do |method, context|
+        kwargs_calls += 1
+        raise NoMethodError, "context accessor failed" if kwargs_calls == 1
+
+        method.call(context)
+      end
+
+      expect { entry.args }.to raise_error(Rage::Deferred::DeadTaskContextDeserializationError) { |error|
+        expect(error.message).to include("accessor-failure")
+        expect(error.cause).to be_a(NoMethodError)
+      }
+      expect(entry.kwargs).to eq({ key: 2 })
+      expect(entry.args).to eq([1])
+      expect(args_calls).to eq(2)
+      expect(kwargs_calls).to eq(2)
+    end
+
+    it "keeps missing referenced constants recoverable through metadata" do
+      stub_const("TemporaryDeadTaskArgument", Class.new)
+      dumped = Marshal.dump(["Task", [TemporaryDeadTaskArgument.new], {}])
+      hide_const("TemporaryDeadTaskArgument")
+      records << records.first.merge(id: "missing-constant", context: dumped)
+      entry = dead_tasks.find_by_id("missing-constant")
+
+      expect(entry.task_class).to eq("OlderTask")
+      expect { entry.args }.to raise_error(Rage::Deferred::DeadTaskContextDeserializationError) { |error|
+        expect(error.cause).to be_a(ArgumentError).or be_a(NameError)
+      }
+    end
+
+    it "retries decoding after a missing referenced constant is restored" do
+      stub_const("RestorableDeadTaskArgument", Class.new)
+      dumped = Marshal.dump(["Task", [RestorableDeadTaskArgument.new], {}])
+      hide_const("RestorableDeadTaskArgument")
+      records << records.first.merge(id: "restorable-constant", context: dumped)
+      entry = dead_tasks.find_by_id("restorable-constant")
+
+      expect { entry.args }.to raise_error(Rage::Deferred::DeadTaskContextDeserializationError)
+
+      stub_const("RestorableDeadTaskArgument", Class.new)
+      expect(entry.args.first).to be_a(RestorableDeadTaskArgument)
+    end
+  end
+end
+
+RSpec.describe Rage::Deferred, ".dead_tasks" do
+  before do
+    @memoized_values = {}
+    %i[@__backend @__dead_tasks @__queue].each do |name|
+      if described_class.instance_variable_defined?(name)
+        @memoized_values[name] = described_class.instance_variable_get(name)
+        described_class.remove_instance_variable(name)
+      end
+    end
+  end
+
+  after do
+    %i[@__backend @__dead_tasks @__queue].each do |name|
+      described_class.remove_instance_variable(name) if described_class.instance_variable_defined?(name)
+      described_class.instance_variable_set(name, @memoized_values[name]) if @memoized_values.key?(name)
+    end
+  end
+
+  let(:backend) { instance_double("Rage::Deferred::Backend") }
+
+  before do
+    allow(Rage.config.deferred).to receive(:backend).and_return(backend)
+  end
+
+  it "resolves the configured backend once and passes it directly to one memoized collection" do
+    expect(Rage::Deferred::DeadTasks).to receive(:new).with(backend).once.and_call_original
+    expect(backend).not_to receive(:each_dead_task)
+
+    first = described_class.dead_tasks
+
+    expect(described_class.dead_tasks).to equal(first)
+    expect(first.instance_variable_get(:@backend)).to equal(backend)
+    expect(Rage.config.deferred).to have_received(:backend).once
+  end
+
+  it "does not start traversal when creating unadvanced Enumerators" do
+    expect(backend).not_to receive(:each_dead_task)
+
+    first = described_class.dead_tasks.each
+    second = described_class.dead_tasks.each
+
+    expect(first).to be_a(Enumerator)
+    expect(second).not_to equal(first)
+  end
+
+  it "shares the backend when the queue initializes it first" do
+    queue = described_class.__queue
+    collection = described_class.dead_tasks
+
+    expect(queue.instance_variable_get(:@backend)).to equal(backend)
+    expect(collection.instance_variable_get(:@backend)).to equal(backend)
+    expect(Rage.config.deferred).to have_received(:backend).once
+  end
+
+  it "shares the backend when the collection initializes it first" do
+    collection = described_class.dead_tasks
+    queue = described_class.__queue
+
+    expect(collection.instance_variable_get(:@backend)).to equal(backend)
+    expect(queue.instance_variable_get(:@backend)).to equal(backend)
+    expect(Rage.config.deferred).to have_received(:backend).once
+  end
+end
+
+RSpec.describe Rage::Deferred::Backends::Nil, "dead-task traversal" do
+  it "is empty without creating persistence or background activity" do
+    backend = described_class.new
+    collection = Rage::Deferred::DeadTasks.new(backend)
+
+    expect(backend.method(:each_dead_task).parameters).to eq([])
+    expect(collection.to_a).to eq([])
+  end
+end
diff --git a/spec/spec_helper.rb b/spec/spec_helper.rb
index c6de6aca..913118f0 100644
--- a/spec/spec_helper.rb
+++ b/spec/spec_helper.rb
@@ -9,6 +9,7 @@
 require_relative "support/contexts/mocked_classes"
 require_relative "support/contexts/mocked_rage_routes"
 require_relative "support/custom_matchers"
+require_relative "support/shared_examples/dead_task_exact_lookup"
 
 RSpec.configure do |config|
   # Uncomment the line below to enable focused mode
diff --git a/spec/support/shared_examples/dead_task_exact_lookup.rb b/spec/support/shared_examples/dead_task_exact_lookup.rb
new file mode 100644
index 00000000..cd0f8a09
--- /dev/null
+++ b/spec/support/shared_examples/dead_task_exact_lookup.rb
@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+RSpec.shared_examples "a dead-task exact-lookup backend" do |empty:|
+  it "returns nil for a missing exact id" do
+    expect(backend.find_dead_task("missing-id")).to be_nil
+  end
+
+  if empty
+    it "remains empty and side-effect free" do
+      state = backend.instance_variables.to_h { |name| [name, backend.instance_variable_get(name)] }
+
+      expect(backend.find_dead_task("1700000001-1-1")).to be_nil
+      expect(backend.instance_variables.to_h { |name| [name, backend.instance_variable_get(name)] }).to eq(state)
+    end
+  else
+    it "returns an existing record with the shared public shape" do
+      expected = store_lookup_record(:existing)
+
+      expect(backend.find_dead_task(lookup_id)).to eq(expected)
+      expect(expected).to include(
+        :id, :task_class, :attempts, :enqueued_at, :failed_at,
+        :exception_class, :exception_message, :backtrace, :context
+      )
+    end
+
+    it "returns the newest frame-valid duplicate" do
+      expected = store_lookup_record(:duplicate)
+
+      expect(backend.find_dead_task(lookup_id)).to eq(expected)
+    end
+
+    it "retains the older frame-valid duplicate when the newer match is frame-invalid" do
+      expected = store_lookup_record(:frame_invalid_duplicate)
+
+      expect(backend.find_dead_task(lookup_id)).to eq(expected)
+    end
+
+    it "propagates a selected winner failure without falling back to an older duplicate" do
+      error_class, message = store_lookup_record(:selected_winner_failure)
+
+      expect { backend.find_dead_task(lookup_id) }.to raise_error(error_class, message)
+    end
+
+    it "returns a schema-incompatible newer frame without falling back" do
+      expected = store_lookup_record(:schema_incompatible_duplicate)
+
+      expect(backend.find_dead_task(lookup_id)).to eq(expected)
+    end
+  end
+end