diff --git a/README.md b/README.md
index 728f9971f..5117e23c6 100644
--- a/README.md
+++ b/README.md
@@ -34,6 +34,7 @@ Solid Queue can be used with SQL databases such as MySQL, PostgreSQL, or SQLite,
 - [Batch jobs](#batch-jobs)
   - [Batch progress and counters](#batch-progress-and-counters)
   - [Batch maintenance](#batch-maintenance)
+    - [Stalled batches](#stalled-batches)
   - [Clearing batches](#clearing-batches)
   - [Upgrading existing installations](#upgrading-existing-installations)
 - [Puma plugin](#puma-plugin)
@@ -763,7 +764,7 @@ Batches track `total_jobs`, `completed_jobs`, `failed_jobs` and `pending_jobs`,
 
 ### Batch maintenance
 
-Batch completion is normally detected as jobs finish, without ever locking the batch row outside a single once-per-batch moment. A few edge cases can't trigger that detection: jobs removed via bulk discards (which delete jobs without callbacks), a process that crashed after enqueueing jobs but before starting its batch, or a completion whose callback enqueueing failed and rolled back.
+Batch completion is normally detected as jobs finish, without ever locking the batch row outside a single once-per-batch moment. A couple of edge cases can't trigger that detection: jobs removed via bulk discards (which delete jobs without callbacks), or a completion whose callback enqueueing failed and rolled back.
 
 The dispatcher sweeps these up automatically via `SolidQueue::Batch.sweep_stalled`, as part of its regular maintenance (every `concurrency_maintenance_interval` seconds, sharing a single maintenance timer and database connection). If you disable `batch_maintenance` (or don't run a dispatcher), you can run the sweep yourself, for example as a [recurring task](#recurring-tasks):
 
@@ -773,6 +774,24 @@ batch_maintenance:
   schedule: every 5 minutes
 ```
 
+#### Stalled batches
+
+Maintenance only completes batches their creator *sealed*—batches `SolidQueue::Batch.enqueue` finished filling. A batch created outside any transaction is sealed in the same write as its row, so a process dying mid-creation can't leave an unsealed batch behind. One created inside a transaction can: sealing waits for the commit, and a transaction that's still open, rolled back, or died uncommitted leaves the batch unsealed—indistinguishably so.
+
+Completing an unsealed batch would lose work: it finishes with whatever happened to have landed, fires its callbacks, and enqueues still pending on the transaction then raise `SolidQueue::Batch::AlreadyFinished`. Rather than guess whether more jobs are coming, the sweep emits a `stalled_batch.solid_queue` event for each one and waits.
+
+An unsealed batch doesn't wait forever. One still unsealed a day after creation came from a transaction that will never commit, so it ends the way its data did: the sweep removes it, as if it was never created. No callbacks fire, and nothing accumulates or needs monitoring. Jobs that reached the queue before the rollback have already run and stay untouched—the same behaviour as jobs enqueued outside a batch in a rolled-back transaction.
+
+In between, batches are yours to inspect or resolve early:
+
+```ruby
+SolidQueue::Batch.stalled                          # unsealed for more than 5 minutes
+SolidQueue::Batch.stalled.find_each(&:start)       # complete them yourself
+SolidQueue::Batch.stalled.each(&:destroy)          # or discard them early
+```
+
+Jobs already in a stalled batch run normally—only the batch's own completion waits. Keep transactions that create batches shorter than the expiry window (`expire_after` on `sweep_stalled`, one day by default): a transaction that outlives it and then commits finds its batch gone and raises `SolidQueue::Batch::AlreadyFinished` from the deferred enqueues.
+
 ### Clearing batches
 
 Finished, non-failed batches are cleared with `SolidQueue::Batch.clear_finished_in_batches` after `config.solid_queue.clear_finished_jobs_after`, but only when you invoke it. Failed batches are kept, like failed jobs, so you can inspect them. Installing Solid Queue configures [a recurring task](#recurring-tasks) that clears finished jobs every hour; you can add a matching entry for batches to your `recurring.yml`:
diff --git a/app/models/solid_queue/batch.rb b/app/models/solid_queue/batch.rb
index ad94387ca..2c1741e32 100644
--- a/app/models/solid_queue/batch.rb
+++ b/app/models/solid_queue/batch.rb
@@ -63,11 +63,22 @@ def enqueue(&block)
         raise AlreadyFinished, "Can't enqueue an already finished batch"
       end
 
+      # Decided before opening our own transaction: with none open anywhere,
+      # nothing outside this method can roll the batch back or defer its jobs.
+      seal_on_create = no_enclosing_transactions?
+
       transaction do
         save! if new_record?
 
         self.class.wrap_in_batch_context(id) { block&.call(self) }
 
+        # Sealing in the same transaction as the batch row makes the crashed
+        # creator unrepresentable: either nothing commits, or a sealed batch
+        # and its jobs commit together. Only batches that got jobs qualify;
+        # one with none may still be waiting on deferred enqueues, and sealed
+        # empty means complete.
+        seal_if_filled if seal_on_create
+
         if ActiveRecord.respond_to?(:after_all_transactions_commit)
           ActiveRecord.after_all_transactions_commit { start }
         end
@@ -106,6 +117,14 @@ def mark_as_enqueued
         Batch.where(id: id, enqueued_at: nil).update_all(enqueued_at: Time.current)
       end
 
+      def seal_if_filled
+        Batch.where(id: id, enqueued_at: nil).where.not(total_jobs: 0).update_all(enqueued_at: Time.current)
+      end
+
+      def no_enclosing_transactions?
+        ActiveRecord.respond_to?(:all_open_transactions) && ActiveRecord.all_open_transactions.none?
+      end
+
       def finalize
         reload
 
diff --git a/app/models/solid_queue/batch/sweepable.rb b/app/models/solid_queue/batch/sweepable.rb
index 1d4d0f20a..67b2b996d 100644
--- a/app/models/solid_queue/batch/sweepable.rb
+++ b/app/models/solid_queue/batch/sweepable.rb
@@ -2,22 +2,38 @@
 
 module SolidQueue
   class Batch
-    # Repairs batches that the regular completion detection can't finish on
-    # its own: jobs removed via bulk discards, processes that crashed after
-    # enqueueing jobs but before starting their batch, or completions whose
-    # callback enqueueing failed and rolled back.
+    # Repairs batches that the regular completion detection can't finish on its
+    # own: jobs removed via bulk discards, or completions whose callback
+    # enqueueing failed and rolled back.
+    #
+    # Repair here never invents state. A batch is only completed once its creator
+    # sealed it by calling #start, because "sealed" is the only thing that makes
+    # an empty batch meaningfully complete rather than merely unfilled. Batches
+    # that were never sealed are reported, not finished—see #report_stalled_batches.
     module Sweepable
       extend ActiveSupport::Concern
 
+      included do
+        scope :unsealed, -> { unfinished.where(enqueued_at: nil) }
+      end
+
       class_methods do
-        def sweep_stalled(stalled_for: 5.minutes, batch_size: 500)
-          SolidQueue.instrument(:sweep_stalled_batches, stalled_for: stalled_for, stale_executions: 0, finished_batches: 0, started_batches: 0) do |payload|
+        def sweep_stalled(stalled_for: 5.minutes, expire_after: 1.day, batch_size: 500)
+          SolidQueue.instrument(:sweep_stalled_batches, stalled_for: stalled_for, expire_after: expire_after, stale_executions: 0, finished_batches: 0, expired_batches: 0, stalled_batches: 0) do |payload|
             payload[:stale_executions] = sweep_stale_executions(batch_size:)
             payload[:finished_batches] = finish_stalled_batches(batch_size:)
-            payload[:started_batches] = start_stalled_batches(stalled_for:, batch_size:)
+            payload[:expired_batches] = expire_abandoned_batches(expire_after:, batch_size:)
+            payload[:stalled_batches] = report_stalled_batches(stalled_for:, batch_size:)
           end
         end
 
+        # Batches their creator never sealed, and hasn't sealed for a while. Use
+        # this to find them, and SolidQueue::Batch#start to adopt one deliberately
+        # once you've established its creator is gone for good.
+        def stalled(stalled_for: 5.minutes)
+          unsealed.where(created_at: ...stalled_for.ago)
+        end
+
         private
           # BatchExecution rows represent outstanding work. A row for a resolved
           # job violates that invariant, so remove it immediately; destroy's
@@ -35,7 +51,9 @@ def sweep_stale_executions(batch_size:)
             swept
           end
 
-          # A started batch with no tracking rows left can finish
+          # A sealed batch with no tracking rows left can finish. Sealed is the
+          # load-bearing word: its creator got far enough to declare the batch
+          # complete, so an empty one really is done.
           def finish_stalled_batches(batch_size:)
             finished = 0
 
@@ -47,16 +65,48 @@ def finish_stalled_batches(batch_size:)
             finished
           end
 
-          # A batch that crashed between creation and start never got enqueued
-          def start_stalled_batches(stalled_for:, batch_size:)
-            started = 0
+          # A batch unsealed long past any reasonable transaction was created in
+          # a transaction that will never commit: its data rolled back, so the
+          # batch ends the same way—removed, as if never created. Not completing
+          # it means no callbacks ever fire over rolled-back work; not keeping
+          # it means nothing accumulates or needs monitoring. Jobs that reached
+          # the queue before the rollback already ran and stay untouched, just
+          # like jobs enqueued outside a batch in a rolled-back transaction.
+          #
+          # A transaction that outlives expire_after and then commits raises
+          # AlreadyFinished from its deferred enqueues: loud, and without
+          # firing success callbacks over lost work.
+          def expire_abandoned_batches(expire_after:, batch_size:)
+            expired = 0
+
+            unsealed.where(created_at: ...expire_after.ago).find_each(batch_size: batch_size) do |batch|
+              expired += 1
+              batch.destroy
+            end
+
+            expired
+          end
+
+          # Batches created outside any transaction seal in the same write as
+          # their row, so a crashed creator can't leave one behind. An unsealed
+          # batch therefore came from inside a transaction—one still filling it,
+          # rolled back, or died uncommitted—or is still waiting on deferred
+          # enqueues. Completing any of those loses work: the batch finishes
+          # with whatever happened to have landed, fires its callbacks, and the
+          # real enqueues then raise AlreadyFinished. Since "still coming" and
+          # "never coming" are indistinguishable here, report them and let an
+          # operator decide, rather than guessing and reporting success for
+          # work that never ran.
+          def report_stalled_batches(stalled_for:, batch_size:)
+            stalled_batches = stalled(stalled_for: stalled_for)
+            count = stalled_batches.count
+            return 0 if count.zero?
 
-            unfinished.where(enqueued_at: nil).where(created_at: ...stalled_for.ago).find_each(batch_size: batch_size) do |batch|
-              started += 1
-              batch.start
+            stalled_batches.find_each(batch_size: batch_size) do |batch|
+              SolidQueue.instrument(:stalled_batch, batch_id: batch.id, created_at: batch.created_at, total_jobs: batch.total_jobs)
             end
 
-            started
+            count
           end
       end
     end
diff --git a/lib/solid_queue/log_subscriber.rb b/lib/solid_queue/log_subscriber.rb
index 6806b853e..d55a3aa91 100644
--- a/lib/solid_queue/log_subscriber.rb
+++ b/lib/solid_queue/log_subscriber.rb
@@ -47,7 +47,7 @@ def finish_batch(event)
   end
 
   def sweep_stalled_batches(event)
-    debug formatted_event(event, action: "Sweep stalled batches", **event.payload.slice(:stale_executions, :finished_batches, :started_batches))
+    debug formatted_event(event, action: "Sweep stalled batches", **event.payload.slice(:stale_executions, :finished_batches, :expired_batches, :stalled_batches))
   end
 
   def batch_progress_error(event)
diff --git a/test/integration/batch_lifecycle_test.rb b/test/integration/batch_lifecycle_test.rb
index d5bda2870..0d38ca138 100644
--- a/test/integration/batch_lifecycle_test.rb
+++ b/test/integration/batch_lifecycle_test.rb
@@ -127,6 +127,92 @@ def perform
     assert_finished_in_order(job!(job1), batch1.reload)
   end
 
+  test "a batch filled from a transaction that outlives the stalled window still completes correctly" do
+    skip if Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 1
+
+    ApplicationJob.enqueue_after_transaction_commit = true
+
+    batch = nil
+    JobResult.transaction do
+      JobResult.create!(queue_name: "default", status: "")
+
+      batch = SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("late")) do
+        AddToBufferJob.perform_later("late")
+      end
+
+      # The batch row commits before this transaction does, so maintenance
+      # running elsewhere can reach it while the enqueues are still pending
+      SolidQueue::Batch.where(id: batch.id).update_all(created_at: 10.minutes.ago)
+      SolidQueue::Batch.sweep_stalled
+
+      assert_not batch.reload.finished?, "maintenance must not complete a batch still being filled"
+    end
+
+    assert_equal 1, batch.reload.total_jobs
+
+    @dispatcher.start
+    @worker.start
+
+    wait_for_batches_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(5.seconds)
+
+    assert batch.reload.finished?
+    assert_equal [ "late", "late: 1 jobs succeeded!" ].sort, JobBuffer.values.sort
+  end
+
+  test "a batch from a rolled-back transaction never reports success" do
+    skip if Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 1
+
+    batch = nil
+    JobResult.transaction do
+      JobResult.create!(queue_name: "default", status: "")
+
+      batch = SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("phantom")) do
+        AddToBufferJob.perform_later("rolled back")
+      end
+
+      raise ActiveRecord::Rollback
+    end
+
+    # Maintenance finds the leftover batch old and unstarted, and leaves it alone
+    SolidQueue::Batch.where(id: batch.id).update_all(created_at: 10.minutes.ago)
+    SolidQueue::Batch.sweep_stalled
+
+    @dispatcher.start
+    @worker.start
+    wait_for_jobs_to_finish_for(5.seconds)
+
+    assert_not batch.reload.finished?
+    assert_nil batch.enqueued_at
+    assert_not_includes JobBuffer.values, "phantom: 1 jobs succeeded!"
+  end
+
+  test "a batch whose creator dies right after enqueue is already complete" do
+    skip unless ActiveRecord.respond_to?(:all_open_transactions)
+
+    # The deferred start never runs, as if the process died the moment enqueue returned
+    SolidQueue::Batch.any_instance.stubs(:start)
+
+    jobs_in_transaction = nil
+    batch = SolidQueue::Batch.enqueue(on_success: BatchOnSuccessJob.new("sealed")) do
+      AddToBufferJob.perform_later("sealed")
+      jobs_in_transaction = SolidQueue::Job.count
+    end
+
+    skip "enqueues are deferred here, so sealing waits for start" if jobs_in_transaction.zero?
+
+    assert batch.reload.enqueued?, "the batch sealed in the same commit as its row"
+
+    @dispatcher.start
+    @worker.start
+
+    wait_for_batches_to_finish_for(5.seconds)
+    wait_for_jobs_to_finish_for(5.seconds)
+
+    assert batch.reload.finished?
+    assert_equal [ "sealed", "sealed: 1 jobs succeeded!" ].sort, JobBuffer.values.sort
+  end
+
   test "prebuilt jobs capture their batch before enqueue is deferred" do
     skip if Rails::VERSION::MAJOR == 7 && Rails::VERSION::MINOR == 1
 
diff --git a/test/models/solid_queue/batch_test.rb b/test/models/solid_queue/batch_test.rb
index 8dab723d7..ea2c764b2 100644
--- a/test/models/solid_queue/batch_test.rb
+++ b/test/models/solid_queue/batch_test.rb
@@ -473,18 +473,140 @@ def perform; end
     assert_equal 3, batch.completed_jobs
   end
 
-  test "sweep_stalled starts batches whose creating process died before starting them" do
-    batch = SolidQueue::Batch.enqueue { NiceJob.perform_later("world") }
+  test "a batch created outside any transaction is sealed with its own row" do
+    skip "sealing on create needs ActiveRecord.all_open_transactions" unless ActiveRecord.respond_to?(:all_open_transactions)
+
+    # Simulate the creator dying before the deferred start runs
+    SolidQueue::Batch.any_instance.stubs(:start)
+
+    jobs_in_transaction = nil
+    batch = SolidQueue::Batch.enqueue do
+      NiceJob.perform_later("world")
+      jobs_in_transaction = SolidQueue::Job.count
+    end
+
+    # Configurations that defer enqueues put no jobs in the batch's own
+    # transaction, so sealing correctly waits for start there
+    skip "enqueues are deferred here, so there was nothing to seal over" if jobs_in_transaction.zero?
+
+    assert batch.reload.enqueued?, "batch should be sealed even though start never ran"
+    assert_empty SolidQueue::Batch.stalled(stalled_for: 0.seconds)
+  end
+
+  test "a batch created inside a transaction is not sealed until it commits" do
+    SolidQueue::Batch.any_instance.stubs(:start)
+
+    batch = nil
+    JobResult.transaction do
+      JobResult.create!(queue_name: "default", status: "")
+      batch = SolidQueue::Batch.enqueue { NiceJob.perform_later("world") }
+    end
+
+    assert_nil batch.reload.enqueued_at, "sealing is left to start when a transaction encloses the batch"
+  end
+
+  test "a batch with no jobs is not sealed on create" do
+    # total_jobs 0 can mean enqueues deferred to a commit that hasn't happened;
+    # sealed empty would mean complete, so sealing must wait for start
+    SolidQueue::Batch.any_instance.stubs(:start)
 
-    # Simulate a process that crashed after committing jobs but before start
+    batch = SolidQueue::Batch.enqueue { }
+
+    assert_nil batch.reload.enqueued_at
+  end
+
+  test "sweep_stalled reports batches that were never sealed instead of completing them" do
+    batch = SolidQueue::Batch.enqueue(on_success: BatchCompletionJob) { NiceJob.perform_later("world") }
+
+    # A batch whose creator never called start: either it died, or it's still
+    # filling the batch from a transaction that hasn't committed
     batch.update_columns(enqueued_at: nil, created_at: 10.minutes.ago)
     batch.jobs.sole.finished!
+    SolidQueue::Job.where(class_name: "BatchCompletionJob").delete_all
+
+    events = []
+    callback = ->(*args) { events << ActiveSupport::Notifications::Event.new(*args) }
+    ActiveSupport::Notifications.subscribed(callback, "stalled_batch.solid_queue") do
+      SolidQueue::Batch.sweep_stalled
+    end
 
     assert_not batch.reload.finished?
+    assert_nil batch.enqueued_at
+    assert_empty SolidQueue::Job.where(class_name: "BatchCompletionJob")
+
+    assert_equal 1, events.size
+    assert_equal batch.id, events.sole.payload[:batch_id]
+  end
+
+  test "sweep_stalled removes unsealed batches abandoned past the expiry window" do
+    skip "Rails 7.1 seals batches on create via after_commit" unless ActiveRecord.respond_to?(:all_open_transactions)
+
+    batch = nil
+    JobResult.transaction do
+      JobResult.create!(queue_name: "default", status: "")
+      batch = SolidQueue::Batch.enqueue(on_success: BatchCompletionJob) { NiceJob.perform_later("world") }
+      raise ActiveRecord::Rollback
+    end
+
+    skip "the rollback removed everything here" if SolidQueue::Batch.find_by(id: batch.id).nil?
+
+    batch.update_columns(created_at: 2.days.ago)
+
+    SolidQueue::Batch.sweep_stalled
+
+    assert_nil SolidQueue::Batch.find_by(id: batch.id)
+    assert_empty SolidQueue::Job.where(class_name: "BatchCompletionJob"), "no callbacks for a rolled-back batch"
+  end
+
+  test "sweep_stalled leaves unsealed batches younger than the expiry window" do
+    batch = SolidQueue::Batch.enqueue { NiceJob.perform_later("world") }
+    batch.update_columns(enqueued_at: nil, created_at: 10.minutes.ago)
+
+    SolidQueue::Batch.sweep_stalled
+
+    assert SolidQueue::Batch.exists?(batch.id)
+    assert_nil batch.reload.enqueued_at
+  end
+
+  test "sweep_stalled counts stalled batches in its payload" do
+    batch = SolidQueue::Batch.enqueue { NiceJob.perform_later("world") }
+    batch.update_columns(enqueued_at: nil, created_at: 10.minutes.ago)
+
+    payload = nil
+    callback = ->(*args) { payload = ActiveSupport::Notifications::Event.new(*args).payload }
+    ActiveSupport::Notifications.subscribed(callback, "sweep_stalled_batches.solid_queue") do
+      SolidQueue::Batch.sweep_stalled
+    end
+
+    assert_equal 1, payload[:stalled_batches]
+  end
+
+  test "stalled finds unsealed batches and start adopts one deliberately" do
+    batch = SolidQueue::Batch.enqueue(on_success: BatchCompletionJob) { NiceJob.perform_later("world") }
+    batch.update_columns(enqueued_at: nil, created_at: 10.minutes.ago)
+    batch.jobs.sole.finished!
+    SolidQueue::Job.where(class_name: "BatchCompletionJob").delete_all
+
+    assert_equal [ batch ], SolidQueue::Batch.stalled.to_a
+
+    batch.start
+
+    assert batch.reload.finished?
+    assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count
+  end
+
+  test "sweep_stalled still finishes sealed batches with no tracking rows left" do
+    batch = SolidQueue::Batch.enqueue(on_success: BatchCompletionJob) { NiceJob.perform_later("world") }
+    SolidQueue::Job.where(class_name: "BatchCompletionJob").delete_all
+
+    # Sealed by its creator, but its tracking row was removed without callbacks
+    assert batch.reload.enqueued?
+    batch.batch_executions.delete_all
 
     SolidQueue::Batch.sweep_stalled
 
     assert batch.reload.finished?
+    assert_equal 1, SolidQueue::Job.where(class_name: "BatchCompletionJob").count
   end
 
   test "conflict-discarded jobs count the same for single and bulk enqueues" do