diff --git a/ruby/CHANGELOG.md b/ruby/CHANGELOG.md
index ff57759aa..51691aa46 100644
--- a/ruby/CHANGELOG.md
+++ b/ruby/CHANGELOG.md
@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 
 - Allow `River::Workers#add` to register a kind and a work block, using the client's default retry and timeout policies. [PR #1486](https://github.com/riverqueue/river/pull/1486).
+- Added `Client#job_complete_tx(job)` to complete a running job in the caller's Active Record or Sequel transaction, including pending metadata and output. An active transaction on the driver's connection is required. Completion rolls back with the transaction; once committed, it takes precedence over subsequent worker errors and emits a completion event when the worker finishes. [PR #1490](https://github.com/riverqueue/river/pull/1490).
 
 ## [0.13.0] - 2026-10-08
 
diff --git a/ruby/docs/README.md b/ruby/docs/README.md
index 1c9c7dece..d1a21bba1 100644
--- a/ruby/docs/README.md
+++ b/ruby/docs/README.md
@@ -206,6 +206,45 @@ end
 
 The equivalent works inside `ActiveRecord::Base.transaction` with the Active Record driver.
 
+### Transactional completion
+
+Call `job.client.job_complete_tx(job)` inside a transaction to commit application
+writes and job completion together (the equivalent of Go's `JobCompleteTx`):
+
+```ruby
+class FulfillOrderWorker
+  def work(job)
+    DB.transaction do
+      DB[:orders].where(id: job.args.fetch("order_id")).update(fulfilled: true)
+      job.output = {fulfilled: true}
+      job.client.job_complete_tx(job)
+    end
+  end
+end
+```
+
+With Active Record, use it inside a `ApplicationRecord.transaction` block. As
+with insertion, the transaction must use the driver's connection on the same
+thread. No explicit transaction argument is needed. Calling `job_complete_tx`
+without an active transaction on that connection raises `River::Error` without
+changing the job.
+
+`job_complete_tx` accepts the running `River::Job` passed to the worker and returns
+an updated `River::JobRow`, leaving the original job unchanged. Set output and
+metadata before calling it so they're included in the transaction. A missing
+job raises `River::NotFoundError`; a pending cancellation raises
+`River::JobCancelError`, allowing the transaction to roll back.
+Metadata added afterward, including persisted logs, is saved separately when
+the worker finishes.
+
+Make completion the last operation in the transaction and let the transaction
+commit before returning from `work`. Propagate errors that roll back the
+transaction so River retries the job. If you swallow a rollback and return
+successfully, River completes the job normally. Once completion commits, later
+worker errors cannot retry or cancel the job. River emits `:job_completed` when
+the worker finishes, and skips its usual finalization hooks for the already
+completed job.
+
 ### [Bulk insertion](https://riverqueue.com/docs/inserting-many-jobs)
 
 `#insert_many` inserts a batch atomically and returns one `River::JobInsertResult` per input. Use `River::InsertManyParams` when jobs need different options:
diff --git a/ruby/driver/riverqueue-activerecord/lib/driver.rb b/ruby/driver/riverqueue-activerecord/lib/driver.rb
index 2f45e0d9e..736352f2f 100644
--- a/ruby/driver/riverqueue-activerecord/lib/driver.rb
+++ b/ruby/driver/riverqueue-activerecord/lib/driver.rb
@@ -48,6 +48,11 @@ def errors = {}
       end
     end
 
+    # Reports whether this thread has a transaction on the configured connection.
+    def in_transaction?
+      @connection_class.connection_pool.with_connection(&:transaction_open?)
+    end
+
     def job_get_by_id(id)
       if @is_sqlite
         row = sqlite_job_rows("WHERE id = ? LIMIT 1", [id]).first
diff --git a/ruby/driver/riverqueue-activerecord/spec/connection_class_spec.rb b/ruby/driver/riverqueue-activerecord/spec/connection_class_spec.rb
index db59a5655..a075cfe3f 100644
--- a/ruby/driver/riverqueue-activerecord/spec/connection_class_spec.rb
+++ b/ruby/driver/riverqueue-activerecord/spec/connection_class_spec.rb
@@ -69,6 +69,34 @@ class SelectedRiverConnection < ActiveRecord::Base
         expect(@selected.job_list).to be_empty
       end
 
+      it "completes jobs in the selected connection's transaction" do
+        row = @client.insert(River::JobArgsHash.new(:selected, {})).job
+        job = River::Job.new(@client, @selected.job_claim(id: row.id, attempted_by: "test"))
+
+        SelectedRiverConnection.transaction do
+          @client.job_complete_tx(job)
+          raise ActiveRecord::Rollback
+        end
+
+        expect(@client.job_get(row.id)).to have_attributes(finalized_at: nil, state: "running")
+
+        SelectedRiverConnection.transaction { @client.job_complete_tx(job) }
+
+        expect(@client.job_get(row.id)).to have_attributes(finalized_at: be_a(Time), state: "completed")
+        expect(@primary.job_list).to be_empty
+      end
+
+      it "rejects job completion in an unrelated Base transaction" do
+        row = @client.insert(River::JobArgsHash.new(:selected, {})).job
+        job = River::Job.new(@client, @selected.job_claim(id: row.id, attempted_by: "test"))
+
+        ActiveRecord::Base.transaction do
+          expect { @client.job_complete_tx(job) }.to raise_error(River::Error, /requires an active transaction/)
+        end
+
+        expect(@client.job_get(row.id)).to have_attributes(finalized_at: nil, state: "running")
+      end
+
       if backend == :postgres
         it "detects capabilities again when the selected class replaces its pool" do
           capabilities = @selected.postgres_capabilities
diff --git a/ruby/driver/riverqueue-sequel/lib/driver.rb b/ruby/driver/riverqueue-sequel/lib/driver.rb
index a2c576e50..14b8eac49 100644
--- a/ruby/driver/riverqueue-sequel/lib/driver.rb
+++ b/ruby/driver/riverqueue-sequel/lib/driver.rb
@@ -31,6 +31,11 @@ def initialize(db)
       end
     end
 
+    # Reports whether this thread has a transaction on the configured database.
+    def in_transaction?
+      @db.in_transaction?
+    end
+
     def job_get_by_id(id)
       if @is_sqlite
         row = sqlite_job_rows("WHERE id = ? LIMIT 1", id).first
diff --git a/ruby/driver/riverqueue-sequel/spec/client_spec.rb b/ruby/driver/riverqueue-sequel/spec/client_spec.rb
index 17cf40f4a..4195d33b2 100644
--- a/ruby/driver/riverqueue-sequel/spec/client_spec.rb
+++ b/ruby/driver/riverqueue-sequel/spec/client_spec.rb
@@ -35,6 +35,21 @@
       it_behaves_like "Postgres finalized job list plans" if adapter == :postgres
       it_behaves_like "Postgres rescue concurrency" if adapter == :postgres
       it_behaves_like "dedicated worker process"
+
+      it "rejects job completion in an unrelated database's transaction" do
+        other_database = Sequel.sqlite
+        client = River::Client.new(@driver)
+        row = client.insert(River::JobArgsHash.new(:transactional, {})).job
+        job = River::Job.new(client, @driver.job_claim(id: row.id, attempted_by: "test"))
+
+        other_database.transaction do
+          expect { client.job_complete_tx(job) }.to raise_error(River::Error, /requires an active transaction/)
+        end
+
+        expect(client.job_get(row.id)).to have_attributes(finalized_at: nil, state: "running")
+      ensure
+        other_database&.disconnect
+      end
     end
   end
 end
diff --git a/ruby/lib/client.rb b/ruby/lib/client.rb
index 02c3dae8b..2f43d8d02 100644
--- a/ruby/lib/client.rb
+++ b/ruby/lib/client.rb
@@ -182,6 +182,37 @@ def job_cancel(id)
       @driver.job_cancel(id) || raise(NotFoundError, "job not found: #{id}")
     end
 
+    # Completes a running Job and returns its updated JobRow, including pending
+    # metadata and output. The original JobRow is not modified.
+    #
+    # Requires and joins the caller's transaction on the same Active Record
+    # connection or Sequel database. Commit before returning from work;
+    # propagate rollback errors so the runtime can retry the job. A committed
+    # completion takes precedence over any subsequent worker error.
+    #
+    # Raises NotFoundError for a missing job, JobCancelError for a pending
+    # cancellation, and Error if there is no active transaction or the job is
+    # no longer running or completed.
+    def job_complete_tx(job)
+      raise ArgumentError, "job must be a River::Job" unless job.is_a?(Job)
+      raise ArgumentError, "job must belong to this client" unless job.client.equal?(self)
+      raise Error, "job must be running" unless job.row.state == JOB_STATE_RUNNING
+      raise Error, "job_complete_tx requires an active transaction on the driver's connection" unless @driver.in_transaction?
+
+      # Mark the attempt before the write: an interrupt can arrive after the
+      # database commits but before the driver returns. The runtime checks the
+      # persisted state because the caller may still roll back this transaction.
+      job.__completion_attempted = true
+      now = @time_now_utc.call
+      completed = @driver.job_complete(id: job.row.id, finalized_at: now, metadata: job.metadata_updates, now: now)
+      raise JobCancelError if completed == :cancelled
+
+      completed ||= job_get(job.row.id)
+      raise Error, "job must be running" unless completed.state == JOB_STATE_COMPLETED
+
+      completed
+    end
+
     # Deletes a job by ID and returns its former JobRow.
     #
     # Raises NotFoundError if the job does not exist and JobRunningError if it
diff --git a/ruby/lib/client_runtime.rb b/ruby/lib/client_runtime.rb
index dbf82328b..92ac7fa43 100644
--- a/ruby/lib/client_runtime.rb
+++ b/ruby/lib/client_runtime.rb
@@ -303,6 +303,8 @@ def wake
           finish_work(row.id)
         end
 
+        return [:completed, nil] if finish_transactional_completion(job, timing)
+
         finalize_hooks = @config.plugins.any? { |plugin| plugin.respond_to?(:job_finalize) }
         if finalize_hooks
           raise JobCancelError if @driver.job_get_cancelled_ids([row.id]).include?(row.id)
@@ -322,19 +324,22 @@ def wake
         end
 
         [:completed, nil]
-      rescue JobSnoozeError => error
-        job.__capture_resumable_metadata!
-        [finish_snoozed(row, job, error, timing), error]
-      rescue JobCancelError => error
-        job.__capture_resumable_metadata!
-        finish_failed(row, job, error, timing, cancelled: true)
-        [:cancelled, error]
-      rescue Interrupted => error
-        job.__capture_resumable_metadata!
-        [finish_interrupted(row, job, timing), error]
       rescue => error
+        return [:completed, nil] if finish_transactional_completion(job, timing)
+
         job.__capture_resumable_metadata!
-        [finish_failed(row, job, error, timing, worker: worker), error]
+        outcome = case error
+        when JobSnoozeError
+          finish_snoozed(row, job, error, timing)
+        when JobCancelError
+          finish_failed(row, job, error, timing, cancelled: true)
+          :cancelled
+        when Interrupted
+          finish_interrupted(row, job, timing)
+        else
+          finish_failed(row, job, error, timing, worker: worker)
+        end
+        [outcome, error]
       ensure
         @mutex.synchronize do
           @running.delete(row.id)
@@ -410,6 +415,21 @@ def wake
       (updated&.state == JOB_STATE_CANCELLED) ? :cancelled : :snoozed
     end
 
+    private def finish_transactional_completion(job, timing)
+      return false unless job.__completion_attempted
+
+      completed = @driver.job_get_by_id(job.row.id)
+      return false unless completed && completed.state == JOB_STATE_COMPLETED
+
+      # Work middleware (including persisted logging) can add metadata after the
+      # worker commits. Save it without changing the committed state or timestamp.
+      metadata = job.metadata_updates
+      completed = @driver.job_metadata_merge(job.row.id, metadata) || completed unless metadata.empty?
+
+      publish(EVENT_JOB_COMPLETED, completed, timing)
+      true
+    end
+
     private def finish_work(id)
       @mutex.synchronize do
         entry = @running[id]
diff --git a/ruby/lib/worker.rb b/ruby/lib/worker.rb
index 082595537..ab91d09ac 100644
--- a/ruby/lib/worker.rb
+++ b/ruby/lib/worker.rb
@@ -81,6 +81,9 @@ def work(job)
   # A job being worked, with access to its persisted row and attempt-local
   # metadata changes.
   class Job
+    # Internal runtime hint; persisted state determines whether completion committed.
+    attr_accessor :__completion_attempted
+
     # Client working this job.
     attr_reader :client
 
@@ -89,6 +92,7 @@ class Job
 
     def initialize(client, row)
       @client = client
+      @__completion_attempted = false
       @logger = nil
       @metadata_updates = {}
       @row = row
diff --git a/ruby/sig/client.rbs b/ruby/sig/client.rbs
index b0652d648..f729e810f 100644
--- a/ruby/sig/client.rbs
+++ b/ruby/sig/client.rbs
@@ -24,6 +24,7 @@ module River
       ?scheduled_at: Time?, ?state: jobStateInitialInput?, ?tags: Array[String]?, ?unique_opts: UniqueOpts?) -> JobInsertResult
     def insert_many: (Array[jobArgs | InsertManyParams]) -> Array[JobInsertResult]
     def job_cancel: (Integer) -> JobRow
+    def job_complete_tx: (Job) -> JobRow
     def job_delete: (Integer) -> JobRow
     def job_delete_many: (?JobListParams?, ?after: untyped, ?after_id: untyped,
       ?ids: untyped, ?kinds: untyped, ?limit: untyped, ?metadata: untyped, ?priorities: untyped,
diff --git a/ruby/sig/driver.rbs b/ruby/sig/driver.rbs
index 00853c97a..c84b2f71f 100644
--- a/ruby/sig/driver.rbs
+++ b/ruby/sig/driver.rbs
@@ -1,6 +1,7 @@
 module River
   interface _Driver
     def respond_to?: (Symbol, ?bool) -> bool
+    def in_transaction?: () -> bool
     def init_driver: () -> void
     def migration_backend: () -> Symbol
     def migration_connection: [A] () { (untyped) -> A } -> A
diff --git a/ruby/sig/runtime.rbs b/ruby/sig/runtime.rbs
index ed40bf87d..a7315fbc9 100644
--- a/ruby/sig/runtime.rbs
+++ b/ruby/sig/runtime.rbs
@@ -94,11 +94,13 @@ module River
 
   class Job
     RESUMABLE_CURSOR_UNSET: Object
+    @__completion_attempted: bool
     @client: Client
     @logger: untyped
     @metadata_updates: Hash[String, untyped]
     @resumable_state: ResumableState
     @row: JobRow
+    attr_accessor __completion_attempted: bool
     attr_reader client: Client
     attr_reader row: JobRow
     def initialize: (Client, JobRow) -> void
@@ -380,6 +382,7 @@ module River
     def finish_failed: (JobRow, Job, Exception, JobTiming, ?cancelled: bool, ?worker: untyped) -> untyped
     def finish_interrupted: (JobRow, Job, JobTiming) -> untyped
     def finish_snoozed: (JobRow, Job, JobSnoozeError, JobTiming) -> untyped
+    def finish_transactional_completion: (Job, JobTiming) -> bool
     def finish_work: (Integer) -> untyped
     def invoke_plugins: (Symbol, *untyped) -> Array[untyped]
     def invoke_worker: (untyped, Job) -> untyped
diff --git a/ruby/spec/client_driver_shared_examples.rb b/ruby/spec/client_driver_shared_examples.rb
index b94077329..af7c01987 100644
--- a/ruby/spec/client_driver_shared_examples.rb
+++ b/ruby/spec/client_driver_shared_examples.rb
@@ -4,6 +4,7 @@
 require "riverqueue/testing"
 require_relative "runtime_draining_shared_examples"
 require_relative "runtime_finishing_shared_examples"
+require_relative "transactional_completion_shared_examples"
 
 RSpec.shared_examples "Postgres state update races" do
   it "skips a concurrent retry when selecting jobs for bulk deletion" do
@@ -239,6 +240,7 @@
 RSpec.shared_examples "client driver end to end" do
   it_behaves_like "cancellation while draining"
   it_behaves_like "externally claimed job finalization"
+  it_behaves_like "transactional job completion"
   [false, true].product([false, true]).each do |with_cursor, rollback|
     it "resumes after a #{rollback ? "rolled-back" : "committed"} #{with_cursor ? "cursor" : "step"} checkpoint" do
       received = []
diff --git a/ruby/spec/transactional_completion_shared_examples.rb b/ruby/spec/transactional_completion_shared_examples.rb
new file mode 100644
index 000000000..0b9bf415b
--- /dev/null
+++ b/ruby/spec/transactional_completion_shared_examples.rb
@@ -0,0 +1,268 @@
+# frozen_string_literal: true
+
+require "riverqueue/testing"
+
+RSpec.shared_examples "transactional job completion" do
+  let(:completion_plugins) { [] }
+  let(:completion_worker) { Object.new }
+  let(:completion_client) do
+    River::Client.new(@driver, config: River::Config.new(
+      plugins: completion_plugins, workers: River::Workers.new.add(:transactional, completion_worker)
+    ))
+  end
+
+  def claim_completion_job
+    row = insert_completion_job
+    River::Job.new(completion_client, @driver.job_claim(id: row.id, attempted_by: "test"))
+  end
+
+  # Open an application transaction directly, rather than using River's driver
+  # transaction wrapper, to exercise the public Active Record and Sequel APIs.
+  def completion_transaction(&block)
+    if @driver.respond_to?(:connection_class)
+      @driver.connection_class.transaction(requires_new: true, &block)
+    else
+      @driver.instance_variable_get(:@db).transaction(savepoint: true, &block)
+    end
+  end
+
+  def insert_completion_job
+    completion_client.insert(River::JobArgsHash.new(:transactional, {}), metadata: {"existing" => true}).job
+  end
+
+  it "commits completion, pending metadata, and other writes in the caller's transaction" do
+    events = completion_client.subscribe(:job_completed, :job_failed)
+    transaction = method(:completion_transaction)
+    check_events = -> { expect { events.pop(true) }.to raise_error(ThreadError) }
+    snapshots = []
+    completion_worker.define_singleton_method(:work) do |job|
+      job.output = {"result" => 42}
+      transaction.call do
+        job.client.insert(River::JobArgsHash.new(:followup, {}))
+        snapshots << job.client.job_complete_tx(job)
+        snapshots << job.row
+        check_events.call
+      end
+    end
+    row = insert_completion_job
+
+    result = River::Testing.perform_job(completion_client, row.id)
+
+    expect(result).to have_attributes(error: nil, outcome: :completed)
+    expect(result.job).to have_attributes(
+      finalized_at: be_a(Time), metadata: include("existing" => true, "output" => {"result" => 42}), state: "completed"
+    )
+    expect(snapshots[0]).to have_attributes(finalized_at: result.job.finalized_at, state: "completed")
+    expect(snapshots[1]).to have_attributes(finalized_at: nil, state: "running")
+    expect(completion_client.job_list(kinds: [:followup]).jobs.length).to eq(1)
+    expect(events.pop(true)).to have_attributes(kind: :job_completed, job: have_attributes(id: row.id, state: "completed"))
+    expect { events.pop(true) }.to raise_error(ThreadError)
+  ensure
+    events&.close
+  end
+
+  it "rolls back completion and other writes and retries a failed attempt" do
+    events = completion_client.subscribe(:job_completed, :job_failed)
+    transaction = method(:completion_transaction)
+    completion_worker.define_singleton_method(:work) do |job|
+      transaction.call do
+        job.client.insert(River::JobArgsHash.new(:followup, {}))
+        job.client.job_complete_tx(job)
+        raise "roll back the work"
+      end
+    end
+    row = insert_completion_job
+
+    result = River::Testing.perform_job(completion_client, row.id)
+
+    expect(result).to have_attributes(error: have_attributes(message: "roll back the work"), outcome: :retried)
+    expect(result.job).to have_attributes(finalized_at: nil, errors: contain_exactly(have_attributes(error: "roll back the work")))
+    expect(result.job.state).not_to eq("completed")
+    expect(completion_client.job_list(kinds: [:followup]).jobs).to be_empty
+    expect(events.pop(true).kind).to eq(:job_failed)
+    expect { events.pop(true) }.to raise_error(ThreadError)
+  ensure
+    events&.close
+  end
+
+  [RuntimeError.new("after commit"), River::JobCancelError.new, River::JobSnoozeError.new(60), River::ClientRuntime::Interrupted.new].each do |error|
+    it "keeps a committed completion when the worker raises #{error.class}" do
+      events = completion_client.subscribe(:job_completed, :job_failed, :job_cancelled, :job_snoozed, :job_interrupted)
+      transaction = method(:completion_transaction)
+      completion_worker.define_singleton_method(:work) do |job|
+        transaction.call { job.client.job_complete_tx(job) }
+        raise error
+      end
+      row = insert_completion_job
+
+      result = River::Testing.perform_job(completion_client, row.id)
+
+      expect(result).to have_attributes(error: nil, outcome: :completed)
+      expect(result.job).to have_attributes(attempt: 1, state: "completed")
+      expect(Array(result.job.errors)).to be_empty
+      expect(events.pop(true)).to have_attributes(kind: :job_completed, job: have_attributes(id: row.id, state: "completed"))
+      expect { events.pop(true) }.to raise_error(ThreadError)
+    ensure
+      events&.close
+    end
+  end
+
+  it "allows normal runtime completion after the caller deliberately rolls back" do
+    transaction = method(:completion_transaction)
+    completion_worker.define_singleton_method(:work) do |job|
+      transaction.call do
+        job.client.job_complete_tx(job)
+        raise job.client.driver.rollback_exception
+      end
+      job.output = "after rollback"
+    end
+
+    result = River::Testing.perform_job(completion_client, insert_completion_job.id)
+
+    expect(result).to have_attributes(error: nil, outcome: :completed)
+    expect(result.job).to have_attributes(metadata: include("output" => "after rollback"), state: "completed")
+  end
+
+  it "honors cancellation and rolls back the caller's other writes" do
+    transaction = method(:completion_transaction)
+    completion_worker.define_singleton_method(:work) do |job|
+      job.client.job_cancel(job.id)
+      transaction.call do
+        job.client.insert(River::JobArgsHash.new(:followup, {}))
+        job.client.job_complete_tx(job)
+      end
+    end
+
+    result = River::Testing.perform_job(completion_client, insert_completion_job.id)
+
+    expect(result).to have_attributes(error: be_a(River::JobCancelError), outcome: :cancelled)
+    expect(result.job.state).to eq("cancelled")
+    expect(completion_client.job_list(kinds: [:followup]).jobs).to be_empty
+  end
+
+  it "returns an already committed completion without changing its timestamp or metadata" do
+    job = claim_completion_job
+    first = completion_transaction { completion_client.job_complete_tx(job) }
+    job.output = "too late"
+
+    repeated = completion_transaction { completion_client.job_complete_tx(job) }
+
+    expect(repeated).to have_attributes(finalized_at: first.finalized_at, metadata: first.metadata, state: "completed")
+  end
+
+  it "rejects jobs that are not running before writing" do
+    job = River::Job.new(completion_client, insert_completion_job)
+
+    expect { completion_client.job_complete_tx(job) }.to raise_error(River::Error, "job must be running")
+    expect(completion_client.job_get(job.id).state).to eq("available")
+  end
+
+  it "rejects stale running jobs whose persisted state changed" do
+    job = claim_completion_job
+    completion_client.job_update(job.id, state: :scheduled)
+
+    expect { completion_transaction { completion_client.job_complete_tx(job) } }.to raise_error(River::Error, "job must be running")
+    expect(completion_client.job_get(job.id).state).to eq("scheduled")
+  end
+
+  it "rejects jobs belonging to another client" do
+    job = claim_completion_job
+
+    expect { River::Client.new(@driver).job_complete_tx(job) }.to raise_error(ArgumentError, "job must belong to this client")
+    expect(completion_client.job_get(job.id).state).to eq("running")
+  end
+
+  it "rejects job rows and IDs instead of a running Job" do
+    row = insert_completion_job
+
+    [row, row.id].each do |argument|
+      expect { completion_client.job_complete_tx(argument) }.to raise_error(ArgumentError, "job must be a River::Job")
+    end
+  end
+
+  it "rejects completion outside a transaction without changing the job" do
+    job = claim_completion_job
+    job.output = "not saved"
+
+    expect { completion_client.job_complete_tx(job) }.to raise_error(
+      River::Error, "job_complete_tx requires an active transaction on the driver's connection"
+    )
+
+    expect(completion_client.job_get(job.id)).to have_attributes(
+      finalized_at: nil, metadata: job.row.metadata, state: "running"
+    )
+  end
+
+  it "reports a missing job without publishing a completion event" do
+    transaction = method(:completion_transaction)
+    completion_worker.define_singleton_method(:work) do |job|
+      job.client.driver.job_delete_if_running(job.id)
+      transaction.call { job.client.job_complete_tx(job) }
+    end
+    events = completion_client.subscribe(:job_completed)
+    row = insert_completion_job
+
+    result = River::Testing.perform_job(completion_client, row.id)
+
+    expect(result.error).to be_a(River::NotFoundError)
+    expect { events.pop(true) }.to raise_error(ThreadError)
+  ensure
+    events&.close
+  end
+
+  it "reports the committed snapshot if the job is deleted before saving later metadata" do
+    transaction = method(:completion_transaction)
+    completion_worker.define_singleton_method(:work) do |job|
+      job.output = 42
+      transaction.call { job.client.job_complete_tx(job) }
+    end
+    @driver.define_singleton_method(:job_metadata_merge) do |id, metadata|
+      job_delete(id)
+      super(id, metadata)
+    end
+    events = completion_client.subscribe(:job_completed)
+    row = insert_completion_job
+
+    result = River::Testing.perform_job(completion_client, row.id)
+
+    expect(result).to have_attributes(error: nil, job: nil, outcome: :completed)
+    expect(events.pop(true).job).to have_attributes(id: row.id, metadata: include("output" => 42), state: "completed")
+  ensure
+    events&.close
+  end
+
+  context "with work middleware and finalization hooks" do
+    let(:finalize_calls) { [] }
+    let(:completion_plugins) do
+      calls = finalize_calls
+      finalizer = Object.new
+      finalizer.define_singleton_method(:job_finalize) do |*args|
+        calls << args
+        :delete
+      end
+      [River::JobPersistedLogging::Plugin.new, finalizer]
+    end
+
+    [false, true].each do |raise_after_commit|
+      it "preserves logs and skips normal finalization hooks with raise_after_commit=#{raise_after_commit}" do
+        transaction = method(:completion_transaction)
+        completion_worker.define_singleton_method(:work) do |job|
+          job.logger << "before commit\n"
+          transaction.call { job.client.job_complete_tx(job) }
+          job.logger << "after commit\n"
+          raise "after commit" if raise_after_commit
+        end
+        events = completion_client.subscribe(:job_completed)
+
+        result = River::Testing.perform_job(completion_client, insert_completion_job.id)
+
+        expect(result).to have_attributes(error: nil, outcome: :completed)
+        expect(result.job.metadata.fetch("river:log")).to eq([{"attempt" => 1, "log" => "before commit\nafter commit\n"}])
+        expect(events.pop(true).job.metadata).to eq(result.job.metadata)
+        expect(finalize_calls).to be_empty
+      ensure
+        events&.close
+      end
+    end
+  end
+end
diff --git a/ruby/spec/upstream_runtime_spec.rb b/ruby/spec/upstream_runtime_spec.rb
index 0e1909f47..222bc17fb 100644
--- a/ruby/spec/upstream_runtime_spec.rb
+++ b/ruby/spec/upstream_runtime_spec.rb
@@ -7,11 +7,13 @@
 require_relative "row_decoding_shared_examples"
 require_relative "runtime_draining_shared_examples"
 require_relative "runtime_finishing_shared_examples"
+require_relative "transactional_completion_shared_examples"
 
 RSpec.describe "upstream runtime parity", database: :sqlite do
   it_behaves_like "SQLite corrupt job runtime"
   it_behaves_like "cancellation while draining"
   it_behaves_like "externally claimed job finalization"
+  it_behaves_like "transactional job completion"
 
   before do
     @database = Sequel.sqlite