diff --git a/sentry-rails/lib/sentry/rails/active_job.rb b/sentry-rails/lib/sentry/rails/active_job.rb
index df7f27ecd..d864e2311 100644
--- a/sentry-rails/lib/sentry/rails/active_job.rb
+++ b/sentry-rails/lib/sentry/rails/active_job.rb
@@ -5,13 +5,61 @@
 module Sentry
   module Rails
     module ActiveJobExtensions
+      SENTRY_PAYLOAD_KEY = "_sentry"
+
+      USER_FIELDS_WHITELIST = %w[id email username].freeze
+
       def perform_now
         if !Sentry.initialized? || already_supported_by_sentry_integration?
           super
         else
-          SentryReporter.record(self) do
-            super
+          SentryReporter.record(
+            self,
+            trace_headers: @_sentry_trace_headers,
+            user: @_sentry_user
+          ) { super }
+        end
+      end
+
+      def serialize
+        payload = super
+        return payload if !Sentry.initialized? || already_supported_by_sentry_integration?
+
+        begin
+          sentry_data = {}
+          if Sentry.configuration.rails.active_job_propagate_traces
+            headers = Sentry.get_trace_propagation_headers
+            sentry_data["trace_propagation_headers"] = headers if headers && !headers.empty?
           end
+
+          if Sentry.configuration.send_default_pii
+            user = Sentry.get_current_scope.user || {}
+            whitelisted = user.each_with_object({}) do |(k, v), acc|
+              acc[k.to_s] = v if USER_FIELDS_WHITELIST.include?(k.to_s)
+            end
+            sentry_data["user"] = whitelisted unless whitelisted.empty?
+          end
+
+          payload[SENTRY_PAYLOAD_KEY] = sentry_data unless sentry_data.empty?
+        rescue StandardError => e
+          Sentry.sdk_logger&.error("sentry-rails: failed to inject _sentry payload: #{e}")
+        end
+
+        payload
+      end
+
+      def deserialize(job_data)
+        super
+        return if !Sentry.initialized? || already_supported_by_sentry_integration?
+
+        begin
+          sentry_data = job_data[SENTRY_PAYLOAD_KEY]
+          return unless sentry_data
+
+          @_sentry_trace_headers = sentry_data["trace_propagation_headers"]
+          @_sentry_user = sentry_data["user"]
+        rescue StandardError => e
+          Sentry.sdk_logger&.error("sentry-rails: failed to extract _sentry payload: #{e}")
         end
       end
 
@@ -28,19 +76,67 @@ class SentryReporter
         }
 
         class << self
-          def record(job, &block)
+          def producer_callback_registered?
+            @producer_callback_registered ||= false
+          end
+
+          def producer_callback_registered!
+            @producer_callback_registered = true
+          end
+
+          def record_producer_span(job)
+            return yield if !Sentry.initialized? || job.already_supported_by_sentry_integration?
+
+            Sentry.with_child_span(op: "queue.publish", description: job.class.name) do |span|
+              if span
+                span.set_origin(SPAN_ORIGIN)
+                span.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_ID, job.job_id)
+                span.set_data(Sentry::Span::DataConventions::MESSAGING_DESTINATION_NAME, job.queue_name)
+              end
+              yield
+            end
+          end
+
+          def record(job, trace_headers: nil, user: nil, &block)
+            # Always give this thread a fresh hub cloned from the main hub so
+            # the job's events are fully isolated.  Save and restore whatever
+            # hub was on the thread before (e.g. the Rack request hub set by
+            # CaptureExceptions, or a stale hub left by a recycled thread-pool
+            # thread) so the outer context continues working correctly after
+            # the job finishes.
+            original_hub = Thread.current.thread_variable_get(Sentry::THREAD_LOCAL)
+            Sentry.clone_hub_to_current_thread
+
             Sentry.with_scope do |scope|
               begin
+                scope.set_user(user) if user && !user.empty?
                 scope.set_transaction_name(job.class.name, source: :task)
+                scope.set_tags(queue: job.queue_name)
+                scope.set_contexts(active_job: {
+                  job_class: job.class.name,
+                  job_id: job.job_id,
+                  queue: job.queue_name,
+                  provider_job_id: job.provider_job_id
+                })
 
-                transaction = Sentry.start_transaction(
+                transaction_options = {
                   name: scope.transaction_name,
                   source: scope.transaction_source,
                   op: OP_NAME,
                   origin: SPAN_ORIGIN
-                )
+                }
 
-                scope.set_span(transaction) if transaction
+                transaction = if trace_headers && !trace_headers.empty?
+                  continued = Sentry.continue_trace(trace_headers, **transaction_options)
+                  Sentry.start_transaction(transaction: continued, **transaction_options)
+                else
+                  Sentry.start_transaction(**transaction_options)
+                end
+
+                if transaction
+                  set_messaging_data(transaction, job)
+                  scope.set_span(transaction)
+                end
 
                 yield.tap do
                   finish_sentry_transaction(transaction, 200)
@@ -53,6 +149,28 @@ def record(job, &block)
                 raise
               end
             end
+          ensure
+            Thread.current.thread_variable_set(Sentry::THREAD_LOCAL, original_hub)
+          end
+
+          def set_messaging_data(transaction, job)
+            transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_ID, job.job_id)
+            transaction.set_data(Sentry::Span::DataConventions::MESSAGING_DESTINATION_NAME, job.queue_name)
+
+            if job.class.rescue_handlers.any?
+              transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RETRY_COUNT, [job.executions.to_i - 1, 0].max)
+            end
+
+            if (latency = compute_latency(job))
+              transaction.set_data(Sentry::Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY, latency)
+            end
+          end
+
+          def compute_latency(job)
+            return unless job.respond_to?(:enqueued_at) && job.enqueued_at
+
+            enqueued_time = job.enqueued_at.is_a?(String) ? Time.parse(job.enqueued_at) : job.enqueued_at
+            ((Time.now.to_f - enqueued_time.to_f) * 1000).round
           end
 
           def capture_exception(job, e)
diff --git a/sentry-rails/lib/sentry/rails/configuration.rb b/sentry-rails/lib/sentry/rails/configuration.rb
index a37c4446c..9e4c1e7ef 100644
--- a/sentry-rails/lib/sentry/rails/configuration.rb
+++ b/sentry-rails/lib/sentry/rails/configuration.rb
@@ -172,6 +172,11 @@ class Configuration
       # Set this option to true if you want Sentry to capture each retry failure
       attr_accessor :active_job_report_on_retry_error
 
+      # Whether we should inject trace propagation headers into the serialized job
+      # payload in order to have a connected trace between producer and consumer.
+      # Defaults to true. Set to false to opt out.
+      attr_accessor :active_job_propagate_traces
+
       # Configuration for structured logging feature
       # @return [StructuredLoggingConfiguration]
       attr_reader :structured_logging
@@ -193,6 +198,7 @@ def initialize
         @db_query_source_threshold_ms = 100
         @active_support_logger_subscription_items = Sentry::Rails::ACTIVE_SUPPORT_LOGGER_SUBSCRIPTION_ITEMS_DEFAULT.dup
         @active_job_report_on_retry_error = false
+        @active_job_propagate_traces = true
         @structured_logging = StructuredLoggingConfiguration.new
       end
     end
diff --git a/sentry-rails/lib/sentry/rails/railtie.rb b/sentry-rails/lib/sentry/rails/railtie.rb
index a86093768..a234e95a9 100644
--- a/sentry-rails/lib/sentry/rails/railtie.rb
+++ b/sentry-rails/lib/sentry/rails/railtie.rb
@@ -21,6 +21,13 @@ class Railtie < ::Rails::Railtie
       ActiveSupport.on_load(:active_job) do
         require "sentry/rails/active_job"
         prepend Sentry::Rails::ActiveJobExtensions
+
+        unless Sentry::Rails::ActiveJobExtensions::SentryReporter.producer_callback_registered?
+          around_enqueue do |job, block|
+            Sentry::Rails::ActiveJobExtensions::SentryReporter.record_producer_span(job, &block)
+          end
+          Sentry::Rails::ActiveJobExtensions::SentryReporter.producer_callback_registered!
+        end
       end
     end
 
diff --git a/sentry-rails/spec/active_job/shared_examples/tracing/consumer_transaction.rb b/sentry-rails/spec/active_job/shared_examples/tracing/consumer_transaction.rb
index 5cc2d1fb3..715a2d5b6 100644
--- a/sentry-rails/spec/active_job/shared_examples/tracing/consumer_transaction.rb
+++ b/sentry-rails/spec/active_job/shared_examples/tracing/consumer_transaction.rb
@@ -32,6 +32,43 @@ def perform
       expect(transaction.contexts.dig(:trace, :status)).to eq("ok")
     end
 
+    it "sets queue scope tag on the consumer transaction" do
+      successful_job.set(queue: "important").perform_later
+      drain
+
+      transaction = sentry_events.find { |e| e.is_a?(Sentry::TransactionEvent) }
+      expect(transaction).not_to be_nil
+      expect(transaction.tags[:queue]).to eq("important")
+    end
+
+    it "sets active_job context on the consumer transaction" do
+      successful_job.perform_later
+      drain
+
+      transaction = sentry_events.find { |e| e.is_a?(Sentry::TransactionEvent) }
+      expect(transaction).not_to be_nil
+
+      ctx = transaction.contexts[:active_job]
+      expect(ctx).not_to be_nil
+      expect(ctx[:job_class]).to eq(successful_job.name)
+      expect(ctx[:job_id]).to be_a(String).and(satisfy { |v| !v.empty? })
+      expect(ctx[:queue]).to eq("default")
+    end
+
+    it "sets active_job context on the error event" do
+      expect do
+        failing_job.perform_later
+        drain
+      end.to raise_error(RuntimeError, /boom from tracing spec/)
+
+      error_event = sentry_events.find { |e| e.is_a?(Sentry::ErrorEvent) }
+      expect(error_event).not_to be_nil
+
+      ctx = error_event.contexts[:active_job]
+      expect(ctx).not_to be_nil
+      expect(ctx[:job_class]).to eq(failing_job.name)
+    end
+
     it "records a db.sql.active_record child span when the job performs a query" do
       query_job = job_fixture do
         def perform
diff --git a/sentry-rails/spec/active_job/shared_examples/tracing/distributed_tracing.rb b/sentry-rails/spec/active_job/shared_examples/tracing/distributed_tracing.rb
new file mode 100644
index 000000000..8dac36d55
--- /dev/null
+++ b/sentry-rails/spec/active_job/shared_examples/tracing/distributed_tracing.rb
@@ -0,0 +1,9 @@
+# frozen_string_literal: true
+
+RSpec.shared_examples "an ActiveJob backend that supports distributed tracing" do
+  it_behaves_like "an ActiveJob backend that emits a producer span on enqueue"
+  it_behaves_like "an ActiveJob backend that propagates trace context through the job payload"
+  it_behaves_like "an ActiveJob backend that records messaging span data on the consumer transaction"
+  it_behaves_like "an ActiveJob backend that propagates Sentry user context through job payloads"
+  it_behaves_like "an ActiveJob backend that isolates Sentry context per worker thread"
+end
diff --git a/sentry-rails/spec/active_job/shared_examples/tracing/messaging_span_data.rb b/sentry-rails/spec/active_job/shared_examples/tracing/messaging_span_data.rb
new file mode 100644
index 000000000..e3b9069c1
--- /dev/null
+++ b/sentry-rails/spec/active_job/shared_examples/tracing/messaging_span_data.rb
@@ -0,0 +1,90 @@
+# frozen_string_literal: true
+
+RSpec.shared_examples "an ActiveJob backend that records messaging span data on the consumer transaction" do
+  include ActiveSupport::Testing::TimeHelpers
+
+  let(:successful_job) do
+    job_fixture do
+      def perform; end
+    end
+  end
+
+  let(:configure_sentry) { proc { |config| config.traces_sample_rate = 1.0 } }
+
+  it "records messaging.message.id and messaging.destination.name on the consumer transaction" do
+    successful_job.set(queue: "critical").perform_later
+    drain
+
+    data = consumer_transaction.contexts.dig(:trace, :data)
+    expect(data["messaging.message.id"]).to be_a(String).and(satisfy { |v| !v.empty? })
+    expect(data["messaging.destination.name"]).to eq("critical")
+  end
+
+  it "omits messaging.message.retry.count for non-retryable jobs" do
+    successful_job.perform_later
+    drain
+
+    data = consumer_transaction.contexts.dig(:trace, :data)
+    expect(data).not_to have_key("messaging.message.retry.count")
+  end
+
+  context "when the job is retryable" do
+    let(:retryable_job) do
+      job_fixture do
+        retry_on StandardError, attempts: 3, wait: 0
+
+        def perform; end
+      end
+    end
+
+    it "records messaging.message.retry.count = 0 on the first execution" do
+      retryable_job.perform_later
+      drain
+
+      data = consumer_transaction.contexts.dig(:trace, :data)
+      expect(data["messaging.message.retry.count"]).to eq(0)
+    end
+
+    it "records messaging.message.retry.count across real retried executions", skip: RAILS_VERSION < 6.0 do
+      # Mirrors sentry-sidekiq's convention (see sentry-sidekiq's
+      # error_handler.rb): retry.count is the producer-side retry counter as
+      # observed when the consumer starts, NOT a "this is retry N" index.
+      # On attempt 1 ActiveJob has not yet incremented executions, so we
+      # report 0; on attempt 2 executions is 1 (set by the prior run), still
+      # max(1 - 1, 0) = 0; on attempt 3 executions is 2 → 1.
+      retried_job = job_fixture do
+        retry_on StandardError, attempts: 3, wait: 0
+
+        def perform
+          raise StandardError, "trigger retry" if executions < 3
+        end
+      end
+
+      retried_job.perform_later
+      drain
+
+      consumer_txns = transactions.select { |t| t.contexts.dig(:trace, :op) == "queue.active_job" }
+      retry_counts = consumer_txns.map { |t| t.contexts.dig(:trace, :data, "messaging.message.retry.count") }
+      expect(retry_counts).to eq([0, 0, 1])
+    end
+  end
+
+  it "records messaging.message.receive.latency in milliseconds", skip: RAILS_VERSION < 6.1 do
+    successful_job.perform_later
+
+    # Older Rails versions truncate Time.now to whole seconds inside `travel`
+    # (no `with_usec:` option until 7.0+), so the measured latency can be up
+    # to ~999ms below the travel delta. Widen the tolerance accordingly.
+    if RAILS_VERSION > 7.0
+      travel(5.seconds, with_usec: true) { drain }
+      tolerance = 50
+    else
+      travel(5.seconds) { drain }
+      tolerance = 1100
+    end
+
+    latency = consumer_transaction.contexts.dig(:trace, :data, "messaging.message.receive.latency")
+    expect(latency).to be_a(Integer)
+    expect(latency).to be_within(tolerance).of(5_000)
+  end
+end
diff --git a/sentry-rails/spec/active_job/shared_examples/tracing/producer_span.rb b/sentry-rails/spec/active_job/shared_examples/tracing/producer_span.rb
new file mode 100644
index 000000000..9a52460f5
--- /dev/null
+++ b/sentry-rails/spec/active_job/shared_examples/tracing/producer_span.rb
@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+RSpec.shared_examples "an ActiveJob backend that emits a producer span on enqueue" do
+  let(:successful_job) do
+    job_fixture do
+      def perform; end
+    end
+  end
+
+  context "with traces_sample_rate = 1.0" do
+    let(:configure_sentry) { proc { |config| config.traces_sample_rate = 1.0 } }
+
+    it "adds a queue.publish child span to the active parent transaction" do
+      within_parent_transaction do
+        successful_job.set(queue: "events").perform_later
+      end
+
+      parent = transactions.find { |t| t.contexts.dig(:trace, :op) == "test" }
+      expect(parent).not_to be_nil
+
+      publish_span = parent.spans.find { |s| s[:op] == "queue.publish" }
+      expect(publish_span).not_to be_nil
+      expect(publish_span[:description]).to eq(successful_job.name)
+      expect(publish_span[:origin]).to eq("auto.queue.active_job")
+      expect(publish_span[:data]["messaging.message.id"]).to be_a(String).and(satisfy { |v| !v.empty? })
+      expect(publish_span[:data]["messaging.destination.name"]).to eq("events")
+      expect(publish_span[:timestamp]).not_to be_nil
+    end
+
+    it "does not raise or capture an orphan span when no parent transaction is active" do
+      expect { successful_job.perform_later }.not_to raise_error
+
+      orphan_publish = transactions.flat_map(&:spans).find { |s| s[:op] == "queue.publish" }
+      expect(orphan_publish).to be_nil
+    end
+  end
+
+  context "with traces_sample_rate = 0" do
+    let(:configure_sentry) { proc { |config| config.traces_sample_rate = 0 } }
+
+    it "does not capture a queue.publish span" do
+      within_parent_transaction do
+        successful_job.perform_later
+      end
+
+      publish_spans = transactions.flat_map(&:spans).select { |s| s[:op] == "queue.publish" }
+      expect(publish_spans).to be_empty
+    end
+  end
+end
diff --git a/sentry-rails/spec/active_job/shared_examples/tracing/trace_propagation.rb b/sentry-rails/spec/active_job/shared_examples/tracing/trace_propagation.rb
new file mode 100644
index 000000000..61c872019
--- /dev/null
+++ b/sentry-rails/spec/active_job/shared_examples/tracing/trace_propagation.rb
@@ -0,0 +1,83 @@
+# frozen_string_literal: true
+
+RSpec.shared_examples "an ActiveJob backend that propagates trace context through the job payload" do
+  let(:successful_job) do
+    job_fixture do
+      def perform; end
+    end
+  end
+
+  let(:configure_sentry) { proc { |config| config.traces_sample_rate = 1.0 } }
+
+  it "produces a consumer transaction whose trace_id matches the parent transaction" do
+    parent_trace_id = nil
+    publish_span_id = nil
+
+    within_parent_transaction do |parent|
+      parent_trace_id = parent.trace_id
+      successful_job.perform_later
+      publish_span_id = parent.span_recorder.spans.find { |s| s.op == "queue.publish" }&.span_id
+    end
+
+    drain
+
+    expect(consumer_transaction).not_to be_nil
+    expect(consumer_transaction.contexts.dig(:trace, :trace_id)).to eq(parent_trace_id)
+    expect(consumer_transaction.contexts.dig(:trace, :parent_span_id)).to eq(publish_span_id)
+  end
+
+  it "captures a consumer transaction without raising when no parent transaction was active at enqueue" do
+    expect { successful_job.perform_later }.not_to raise_error
+    expect { drain }.not_to raise_error
+
+    expect(consumer_transaction).not_to be_nil
+    expect(consumer_transaction.contexts.dig(:trace, :trace_id)).to be_a(String)
+  end
+
+  it "survives a JSON round-trip of the serialized payload" do
+    parent_trace_id = nil
+
+    within_parent_transaction do |parent|
+      parent_trace_id = parent.trace_id
+      payload = successful_job.new.serialize
+      round_tripped = JSON.parse(JSON.generate(payload))
+      ::ActiveJob::Base.execute(round_tripped)
+    end
+
+    expect(consumer_transaction).not_to be_nil
+    expect(consumer_transaction.contexts.dig(:trace, :trace_id)).to eq(parent_trace_id)
+  end
+
+  context "when active_job_propagate_traces is false" do
+    let(:configure_sentry) do
+      proc do |config|
+        config.traces_sample_rate = 1.0
+        config.rails.active_job_propagate_traces = false
+      end
+    end
+
+    it "does not inject trace headers into the job payload" do
+      within_parent_transaction do
+        successful_job.perform_later
+      end
+
+      payload = queue_adapter.enqueued_jobs.last
+      sentry_payload = (payload[:_sentry_full_payload] || payload)["_sentry"]
+      expect(sentry_payload&.dig("trace_propagation_headers")).to be_nil
+    end
+
+    it "starts a new unconnected consumer transaction" do
+      parent_trace_id = nil
+
+      within_parent_transaction do |parent|
+        parent_trace_id = parent.trace_id
+        successful_job.perform_later
+      end
+
+      drain
+
+      expect(consumer_transaction).not_to be_nil
+      expect(consumer_transaction.contexts.dig(:trace, :trace_id)).not_to eq(parent_trace_id)
+    end
+  end
+end
diff --git a/sentry-rails/spec/active_job/shared_examples/tracing/user_propagation.rb b/sentry-rails/spec/active_job/shared_examples/tracing/user_propagation.rb
new file mode 100644
index 000000000..8619f743e
--- /dev/null
+++ b/sentry-rails/spec/active_job/shared_examples/tracing/user_propagation.rb
@@ -0,0 +1,95 @@
+# frozen_string_literal: true
+
+RSpec.shared_examples "an ActiveJob backend that propagates Sentry user context through job payloads" do
+  let(:successful_job) do
+    job_fixture do
+      def perform; end
+    end
+  end
+
+  let(:failing_job) do
+    job_fixture do
+      def perform
+        raise "boom from user_propagation spec"
+      end
+    end
+  end
+
+  let(:full_user) do
+    {
+      id: "u1",
+      email: "alice@example.com",
+      username: "alice",
+      ip_address: "1.2.3.4",
+      segment: "vip"
+    }
+  end
+
+  context "when send_default_pii is true" do
+    let(:configure_sentry) do
+      proc do |config|
+        config.traces_sample_rate = 1.0
+        config.send_default_pii = true
+      end
+    end
+
+    it "propagates only id, email, and username to the consumer transaction" do
+      Sentry.set_user(full_user)
+
+      successful_job.perform_later
+
+      # Simulate the cross-process boundary by clearing the producer scope
+      # before the consumer runs. Without this the consumer's with_scope
+      # inherits the user from the test thread and the test cannot tell
+      # whether propagation actually happened.
+      Sentry.set_user({})
+
+      drain
+
+      expect(consumer_transaction).not_to be_nil
+      expect(consumer_transaction.user).to eq(
+        "id" => "u1",
+        "email" => "alice@example.com",
+        "username" => "alice"
+      )
+    end
+
+    it "propagates the whitelisted user to a captured error event" do
+      Sentry.set_user(full_user)
+
+      failing_job.perform_later
+      Sentry.set_user({})
+
+      expect { drain }.to raise_error(RuntimeError, /boom from user_propagation spec/)
+
+      error_event = sentry_events.find { |e| e.is_a?(Sentry::ErrorEvent) }
+      expect(error_event).not_to be_nil
+      expect(error_event.user).to eq(
+        "id" => "u1",
+        "email" => "alice@example.com",
+        "username" => "alice"
+      )
+    end
+  end
+
+  context "when send_default_pii is false" do
+    let(:configure_sentry) do
+      proc do |config|
+        config.traces_sample_rate = 1.0
+        config.send_default_pii = false
+      end
+    end
+
+    it "does not propagate user context to the consumer transaction" do
+      Sentry.set_user(full_user)
+
+      successful_job.perform_later
+      Sentry.set_user({})
+
+      drain
+
+      expect(consumer_transaction).not_to be_nil
+      expect(consumer_transaction.user).to eq({})
+    end
+  end
+end
diff --git a/sentry-rails/spec/active_job/shared_examples/tracing/worker_hub_isolation.rb b/sentry-rails/spec/active_job/shared_examples/tracing/worker_hub_isolation.rb
new file mode 100644
index 000000000..85315369e
--- /dev/null
+++ b/sentry-rails/spec/active_job/shared_examples/tracing/worker_hub_isolation.rb
@@ -0,0 +1,36 @@
+# frozen_string_literal: true
+
+RSpec.shared_examples "an ActiveJob backend that isolates Sentry context per worker thread" do
+  let(:configure_sentry) { proc { |config| config.traces_sample_rate = 1.0 } }
+
+  it "creates an isolated hub per worker thread when run concurrently" do
+    job_a = job_fixture do
+      def perform
+        Sentry.get_current_scope.set_tags(job: "A")
+      end
+    end
+
+    job_b = job_fixture do
+      def perform
+        Sentry.get_current_scope.set_tags(job: "B")
+      end
+    end
+
+    Sentry.get_current_scope.set_tags(test_thread: true)
+
+    worker_thread { job_a.perform_now }.join
+    worker_thread { job_b.perform_now }.join
+
+    txn_a = transactions.find { |t| t.tags[:job] == "A" }
+    txn_b = transactions.find { |t| t.tags[:job] == "B" }
+
+    expect(txn_a).not_to be_nil
+    expect(txn_b).not_to be_nil
+    expect(txn_a.tags[:job]).to eq("A")
+    expect(txn_b.tags[:job]).to eq("B")
+
+    # The test thread's own scope is unchanged.
+    expect(Sentry.get_current_scope.tags[:test_thread]).to be_truthy
+    expect(Sentry.get_current_scope.tags).not_to have_key(:job)
+  end
+end
diff --git a/sentry-rails/spec/active_job/support/harness.rb b/sentry-rails/spec/active_job/support/harness.rb
index 4e489fb20..c66c03eb3 100644
--- a/sentry-rails/spec/active_job/support/harness.rb
+++ b/sentry-rails/spec/active_job/support/harness.rb
@@ -1,5 +1,30 @@
 # frozen_string_literal: true
 
+# Rails 5.2's TestAdapter stores a minimal hash per enqueued job (only job
+# class, args, queue) and its instantiate_job recreates jobs via `.new(*args)`
+# — never calling our `deserialize` override.  That means the `_sentry`
+# payload injected by `serialize` is silently discarded before the consumer
+# ever sees it, breaking distributed-tracing propagation.
+#
+# This adapter subclass calls `job.serialize` a second time after `super` has
+# stored the minimal hash and saves the full output alongside it.  The drain
+# then drives each job through `ActiveJob::Base.execute(full_payload)`, which
+# goes through the normal deserialize → perform_now path and picks up the
+# Sentry trace headers and user context that were captured at enqueue time.
+class Rails52FullPayloadTestAdapter < ::ActiveJob::QueueAdapters::TestAdapter
+  def enqueue(job)
+    prev = enqueued_jobs.length
+    super
+    enqueued_jobs.last[:_sentry_full_payload] = job.serialize if enqueued_jobs.length > prev
+  end
+
+  def enqueue_at(job, timestamp)
+    prev = enqueued_jobs.length
+    super
+    enqueued_jobs.last[:_sentry_full_payload] = job.serialize if enqueued_jobs.length > prev
+  end
+end
+
 RSpec.shared_context "active_job backend harness" do |adapter:|
   let(:adapter) { adapter }
   let(:configure_sentry) { proc { } }
@@ -8,12 +33,33 @@
     make_basic_app(&configure_sentry)
     setup_sentry_test
 
-    ::ActiveJob::Base.queue_adapter = adapter
+    # Rails 5.2's TestAdapter discards the full serialize output (including the
+    # _sentry payload) when deferring jobs.  Use our augmented subclass instead
+    # so the drain can replay jobs through the proper deserialize path.
+    #
+    # NOTE: In Rails 5.2 test specs, ActiveJob::TestHelper installs a
+    # _test_adapter on ActiveJob::Base via an outer around hook (before_setup).
+    # The queue_adapter class method returns _test_adapter when present, so we
+    # must use enable_test_adapter (not queue_adapter=) to override it.
+    if RAILS_VERSION < 6.0 && adapter == :test
+      @_original_test_adapter = ::ActiveJob::Base._test_adapter
+      ::ActiveJob::Base.enable_test_adapter(Rails52FullPayloadTestAdapter.new)
+    else
+      ::ActiveJob::Base.queue_adapter = adapter
+    end
 
     boot_adapter(adapter)
 
     example.run
   ensure
+    if RAILS_VERSION < 6.0 && adapter == :test
+      if @_original_test_adapter
+        ::ActiveJob::Base.enable_test_adapter(@_original_test_adapter)
+      else
+        ::ActiveJob::Base.disable_test_adapter
+      end
+    end
+
     reset_adapter(adapter)
 
     teardown_sentry_test
@@ -32,15 +78,35 @@ def reset_adapter(_adapter)
   def drain(at: nil)
     case adapter
     when :test
-      if RAILS_VERSION < 6.0
-        # Rails 5.2: perform_enqueued_jobs always requires a block and only runs
-        # jobs enqueued *inside* the block. Manually flush already-enqueued jobs.
-        jobs = queue_adapter.enqueued_jobs.dup
-        queue_adapter.enqueued_jobs.clear
-        jobs.each { |payload| send(:instantiate_job, payload).perform_now }
-      else
-        kwargs = at ? { at: at } : {}
-        perform_enqueued_jobs(**kwargs)
+      # Loop until the queue is empty so retries (which re-enqueue during a
+      # drain pass) are cascaded through to completion. Both Rails 5.2's
+      # manual flush and Rails 6+'s perform_enqueued_jobs(no block) operate
+      # on a snapshot, so a single pass would only run jobs that existed
+      # before draining started.
+      loop do
+        break if queue_adapter.enqueued_jobs.empty?
+
+        if RAILS_VERSION < 6.0
+          # Rails 5.2: perform_enqueued_jobs always requires a block and only
+          # runs jobs enqueued *inside* the block. Manually flush already-
+          # enqueued jobs. When using Rails52FullPayloadTestAdapter, each
+          # payload also carries a :_sentry_full_payload key with the complete
+          # serialize output. Drive those jobs through Base.execute so our
+          # deserialize override runs and populates @_sentry_trace_headers /
+          # @_sentry_user before perform_now.
+          jobs = queue_adapter.enqueued_jobs.dup
+          queue_adapter.enqueued_jobs.clear
+          jobs.each do |payload|
+            if (full = payload[:_sentry_full_payload])
+              ::ActiveJob::Base.execute(full)
+            else
+              send(:instantiate_job, payload).perform_now
+            end
+          end
+        else
+          kwargs = at ? { at: at } : {}
+          perform_enqueued_jobs(**kwargs)
+        end
       end
     else
       raise NotImplementedError, "active_job backend harness has no drain strategy for adapter: #{adapter.inspect}"
@@ -53,4 +119,28 @@ def job_fixture(name = nil, &block)
     stub_const(name, klass)
     klass
   end
+
+  def transactions
+    sentry_events.select { |e| e.is_a?(Sentry::TransactionEvent) }
+  end
+
+  def consumer_transaction
+    transactions.find { |t| t.contexts.dig(:trace, :op) == "queue.active_job" }
+  end
+
+  def within_parent_transaction(name: "parent.test", op: "test")
+    txn = Sentry.start_transaction(name: name, op: op)
+    Sentry.get_current_scope.set_span(txn) if txn
+    yield(txn)
+  ensure
+    txn&.finish
+  end
+
+  # Hook used by the worker_hub_isolation shared example. The default
+  # is a plain Thread.new — adapters that need extra setup (e.g. an
+  # isolated database per worker thread, like :solid_queue on SQLite)
+  # override this to wrap the block in their isolation scope.
+  def worker_thread(&block)
+    Thread.new(&block)
+  end
 end
diff --git a/sentry-rails/spec/active_job/test_adapter_spec.rb b/sentry-rails/spec/active_job/test_adapter_spec.rb
index 4d5e704de..bde362193 100644
--- a/sentry-rails/spec/active_job/test_adapter_spec.rb
+++ b/sentry-rails/spec/active_job/test_adapter_spec.rb
@@ -6,4 +6,5 @@
   include_context "active_job backend harness", adapter: :test
 
   it_behaves_like "a Sentry-instrumented ActiveJob backend"
+  it_behaves_like "an ActiveJob backend that supports distributed tracing"
 end
diff --git a/sentry-rails/spec/sentry/rails/tracing/active_storage_subscriber_spec.rb b/sentry-rails/spec/sentry/rails/tracing/active_storage_subscriber_spec.rb
index 54b4eda9c..b1a3618e9 100644
--- a/sentry-rails/spec/sentry/rails/tracing/active_storage_subscriber_spec.rb
+++ b/sentry-rails/spec/sentry/rails/tracing/active_storage_subscriber_spec.rb
@@ -42,14 +42,13 @@
 
       request_transaction = transport.events.last.to_h
       expect(request_transaction[:type]).to eq("transaction")
-      expect(request_transaction[:spans].count).to eq(2)
-
-      span = request_transaction[:spans][1]
-      expect(span[:op]).to eq("file.service_upload.active_storage")
-      expect(span[:origin]).to eq("auto.file.rails")
-      expect(span[:description]).to eq("Disk")
-      expect(span.dig(:data, :key)).to be_nil
-      expect(span[:trace_id]).to eq(request_transaction.dig(:contexts, :trace, :trace_id))
+
+      upload_span = request_transaction[:spans].find { |s| s[:op] == "file.service_upload.active_storage" }
+      expect(upload_span).not_to be_nil
+      expect(upload_span[:origin]).to eq("auto.file.rails")
+      expect(upload_span[:description]).to eq("Disk")
+      expect(upload_span.dig(:data, :key)).to be_nil
+      expect(upload_span[:trace_id]).to eq(request_transaction.dig(:contexts, :trace, :trace_id))
     end
   end
 
@@ -71,10 +70,10 @@
 
       request_transaction = transport.events.last.to_h
       expect(request_transaction[:type]).to eq("transaction")
-      expect(request_transaction[:spans].count).to eq(2)
 
-      span = request_transaction[:spans][1]
-      expect(span.dig(:data, :key)).to eq(p.cover.key)
+      upload_span = request_transaction[:spans].find { |s| s[:op] == "file.service_upload.active_storage" }
+      expect(upload_span).not_to be_nil
+      expect(upload_span.dig(:data, :key)).to eq(p.cover.key)
     end
   end