diff --git a/sentry-ruby/lib/sentry-ruby.rb b/sentry-ruby/lib/sentry-ruby.rb
index 52afec20e..7c846fb11 100644
--- a/sentry-ruby/lib/sentry-ruby.rb
+++ b/sentry-ruby/lib/sentry-ruby.rb
@@ -31,6 +31,7 @@
 require "sentry/cron/slug"
 require "sentry/vernier/profiler"
 require "sentry/metrics"
+require "sentry/job_extensions"
 
 [
   "sentry/rake",
diff --git a/sentry-ruby/lib/sentry/job_extensions.rb b/sentry-ruby/lib/sentry/job_extensions.rb
new file mode 100644
index 000000000..d9f16ab92
--- /dev/null
+++ b/sentry-ruby/lib/sentry/job_extensions.rb
@@ -0,0 +1,6 @@
+# frozen_string_literal: true
+
+require "sentry/job_extensions/extension"
+require "sentry/job_extensions/instrumentation"
+require "sentry/job_extensions/consumer"
+require "sentry/job_extensions/producer"
diff --git a/sentry-ruby/lib/sentry/job_extensions/consumer.rb b/sentry-ruby/lib/sentry/job_extensions/consumer.rb
new file mode 100644
index 000000000..20e2a6acc
--- /dev/null
+++ b/sentry-ruby/lib/sentry/job_extensions/consumer.rb
@@ -0,0 +1,92 @@
+# frozen_string_literal: true
+
+module Sentry
+  module JobExtensions
+    # @api private
+    class Consumer < Instrumentation
+      OP_NAME = "queue.process"
+
+      def apply_scope_data(scope, job)
+        data = extension.scope_data(job: job)
+        scope.set_contexts(data[:contexts]) if data[:contexts]
+        scope.set_tags(data[:tags]) if data[:tags]
+        scope.set_extras(data[:extra]) if data[:extra]
+        scope.set_user(data[:user]) if data[:user]
+      end
+
+      def start_transaction(scope, job, trace_headers: {})
+        options = {
+          name: scope.transaction_name,
+          source: scope.transaction_source,
+          op: OP_NAME,
+          origin: extension.span_origin(job: job)
+        }
+
+        transaction = if trace_headers.empty?
+          Sentry.start_transaction(**options)
+        else
+          Sentry.start_transaction(transaction: Sentry.continue_trace(trace_headers, **options), **options)
+        end
+        return unless transaction
+
+        scope.set_span(transaction)
+        safely("record the messaging data") do
+          set_messaging_data(transaction, extension.messaging_data(job: job))
+          set_retry_count(transaction, job)
+        end
+
+        transaction
+      end
+
+      def finish_transaction(transaction, status)
+        return if !transaction || transaction.timestamp
+
+        transaction.set_http_status(status)
+        transaction.finish
+      end
+
+      def report_exception?(job, exception)
+        extension.report_exception?(job: job, exception: exception)
+      end
+
+      def report_retried_exception?(job, exception)
+        extension.report_retried_exception?(job: job, exception: exception)
+      end
+
+      def capture_exception(job, exception, contexts: {}, tags: {}, extra: {})
+        error_data = additive_error_data(extension.error_data(job: job, exception: exception))
+
+        options = {
+          extra: error_data[:extra].merge(matching_keys(extra, error_data[:extra])),
+          tags: error_data[:tags].merge(matching_keys(tags, error_data[:tags])),
+          hint: { background: false }
+        }
+        contexts = error_data[:contexts].merge(matching_keys(contexts, error_data[:contexts]))
+        options[:contexts] = contexts unless contexts.empty?
+
+        extension.integration(job: job).capture_exception(exception, **options)
+      end
+
+      private
+
+      def additive_error_data(data)
+        scope = Sentry.get_current_scope
+
+        {
+          contexts: matching_keys(data.fetch(:contexts, {}), scope.contexts).to_h do |name, context|
+            scope_context = scope.contexts.fetch(name, {})
+            context = matching_keys(context, scope_context)
+            [name, context.merge(scope_context)]
+          end,
+          tags: matching_keys(data.fetch(:tags, {}), scope.tags).reject { |key, _value| scope.tags.key?(key) },
+          extra: matching_keys(data.fetch(:extra, {}), scope.extra).reject { |key, _value| scope.extra.key?(key) }
+        }
+      end
+
+      def matching_keys(data, reference)
+        keys = reference.keys.to_h { |key| [key.to_s, key] }
+        data.transform_keys { |key| keys.fetch(key.to_s, key) }
+      end
+    end
+  end
+end
diff --git a/sentry-ruby/lib/sentry/job_extensions/extension.rb b/sentry-ruby/lib/sentry/job_extensions/extension.rb
new file mode 100644
index 000000000..0b8e5ebf4
--- /dev/null
+++ b/sentry-ruby/lib/sentry/job_extensions/extension.rb
@@ -0,0 +1,154 @@
+# frozen_string_literal: true
+
+module Sentry
+  module JobExtensions
+    # Base class for objects that let a background-job backend enrich Sentry's
+    # job integrations.
+    #
+    # A backend gem subclasses this class and receives everything the extension
+    # depends on (configuration, collaborators) through its constructor. Sentry
+    # treats jobs as opaque objects: only the extension interprets the job it is
+    # given.
+    class Extension
+      # The extension's name, used to identify it in SDK log messages.
+      #
+      # @return [String]
+      attr_reader :name
+
+      # @param name [String, Symbol] a non-empty name identifying the backend,
+      #   stored as a frozen String
+      # @raise [ArgumentError] if the name is not a non-empty String or Symbol
+      def initialize(name:)
+        unless (name.is_a?(String) || name.is_a?(Symbol)) && !name.empty?
+          raise ArgumentError, "job extension name must be a non-empty String or Symbol, got #{name.inspect}"
+        end
+
+        @name = name.to_s.dup.freeze
+      end
+
+      # @!group Hooks
+      #
+      # Hooks are called directly by job integrations on the extension they
+      # are given. Each hook maps its input to data and must not mutate
+      # the objects it is given; the job integration applies the returned
+      # data. Override only the hooks the backend can supply.
+
+      # Data to apply to the scope the job is performed in. Called before the
+      # job is performed. A supplied user replaces the user the job
+      # integration sets on the scope.
+      #
+      # @param job [Object] the job being performed
+      # @return [Hash] a Hash with any of the keys +:contexts+ (a Hash of
+      #   context name to Hash), +:tags+ (a Hash), +:extra+ (a Hash) and
+      #   +:user+ (a Hash). Empty by default.
+      def scope_data(job:)
+        {}
+      end
+
+      # Data to add to the event captured for an exception raised by the job.
+      # Called when the job fails and its exception is about to be captured.
+      #
+      # Supplied tags and extra never replace a key already on the job's
+      # scope. Each supplied context is merged into the scope's context of the
+      # same name, keeping values already on the scope. Data the job integration captures
+      # explicitly with the exception takes precedence over this data.
+      #
+      # @param job [Object] the job that failed
+      # @param exception [Exception] the exception raised by the job
+      # @return [Hash] a Hash with any of the keys +:contexts+ (a Hash of
+      #   context name to Hash), +:tags+ (a Hash) and +:extra+ (a Hash).
+      #   Empty by default.
+      def error_data(job:, exception:)
+        {}
+      end
+
+      # The backend's own retry state for the job. Called when the job is
+      # enqueued, before it is performed, and when it fails. The hook may be
+      # called several times for the same attempt, so it must not change any
+      # state.
+      #
+      # @param job [Object] the job being enqueued, performed, or that failed
+      # @return [Hash] a Hash with the keys +:count+ (a non-negative Integer,
+      #   the number of retries already attempted) and +:will_retry+ (true or
+      #   false, whether the backend will retry the failed attempt). By
+      #   default +{ count: 0, will_retry: false }+.
+      def retry_info(job:)
+        { count: 0, will_retry: false }
+      end
+
+      # Whether the job integration captures the exception raised by the job.
+      # Called when the job fails. Declining to capture does not change the
+      # job's outcome: the job still fails and its transaction is still
+      # finished as failed.
+      #
+      # @param job [Object] the job that failed
+      # @param exception [Exception] the exception raised by the job
+      # @return [Boolean] true to capture the exception. By default true
+      #   only when +:will_retry+ from #retry_info is false.
+      def report_exception?(job:, exception:)
+        !retry_info(job: job)[:will_retry]
+      end
+
+      # Whether the job integration captures an exception that the job
+      # integration's own retry mechanism rescued and retries, so the backend
+      # never sees it. Called when the job integration schedules the retry.
+      #
+      # @param job [Object] the job being retried
+      # @param exception [Exception] the exception the job raised
+      # @return [Boolean] true to capture the exception. False by default.
+      def report_retried_exception?(job:, exception:)
+        false
+      end
+
+      # Adds trace propagation headers to the backend's job payload. Called
+      # when a job is enqueued.
+      #
+      # @param payload [Object] the backend's job payload
+      # @param headers [Hash{String => String}] the trace propagation headers
+      #   to carry, such as +sentry-trace+ and +baggage+
+      # @return [Object] a payload of the same class as the one given,
+      #   carrying the headers. By default the given payload, unchanged.
+      def inject_trace_headers(job:, payload:, headers:)
+        payload
+      end
+
+      # Reads the trace propagation headers carried by the backend's job
+      # payload. Called before the job is performed, to continue the trace
+      # started when it was enqueued.
+      #
+      # @param payload [Object] the backend's job payload
+      # @return [Hash{String => String}] the headers the payload carries. The
+      #   job continues the trace they describe; an empty Hash continues no
+      #   trace. Empty by default.
+      def extract_trace_headers(job:, payload:)
+        {}
+      end
+
+      # Data for the messaging attributes of the consumer span created when
+      # the job is performed. Also used for the producer span created when the
+      # job is enqueued unless #publish_messaging_data is overridden.
+      #
+      # @param job [Object] the job being enqueued or performed
+      # @return [Hash] a Hash with any of the keys +:system+ (a String naming
+      #   the backend), +:message_id+ (a String), +:destination+ (a String,
+      #   such as the queue name), +:receive_latency+ (a Numeric, the
+      #   milliseconds between enqueueing and the start of the job). Empty by
+      #   default.
+      def messaging_data(job:)
+        {}
+      end
+
+      # Data for the messaging attributes of the producer span created when
+      # the job is enqueued.
+      #
+      # @param job [Object] the job being enqueued
+      # @return [Hash] a Hash with the keys documented on #messaging_data. By
+      #   default the result of #messaging_data.
+      def publish_messaging_data(job:)
+        messaging_data(job: job)
+      end
+
+      # @!endgroup
+    end
+  end
+end
diff --git a/sentry-ruby/lib/sentry/job_extensions/instrumentation.rb b/sentry-ruby/lib/sentry/job_extensions/instrumentation.rb
new file mode 100644
index 000000000..71a23eee8
--- /dev/null
+++ b/sentry-ruby/lib/sentry/job_extensions/instrumentation.rb
@@ -0,0 +1,46 @@
+# frozen_string_literal: true
+
+module Sentry
+  module JobExtensions
+    # @api private
+    class Instrumentation
+      include LoggingHelper
+
+      MESSAGING_DATA_ATTRIBUTES = {
+        system: Span::DataConventions::MESSAGING_SYSTEM,
+        message_id: Span::DataConventions::MESSAGING_MESSAGE_ID,
+        destination: Span::DataConventions::MESSAGING_DESTINATION_NAME,
+        receive_latency: Span::DataConventions::MESSAGING_MESSAGE_RECEIVE_LATENCY
+      }.freeze
+
+      attr_reader :extension
+
+      def initialize(extension:)
+        @extension = extension
+      end
+
+      def safely(context, fallback: nil)
+        yield
+      rescue StandardError => e
+        log_error("Job extension #{extension.name.inspect} failed to #{context}", e)
+        fallback
+      end
+
+      private
+
+      def set_messaging_data(span, data)
+        data.slice(*MESSAGING_DATA_ATTRIBUTES.keys).each do |key, value|
+          span.set_data(MESSAGING_DATA_ATTRIBUTES[key], value)
+        end
+      end
+
+      def set_retry_count(span, job)
+        span.set_data(Span::DataConventions::MESSAGING_MESSAGE_RETRY_COUNT, extension.retry_info(job: job)[:count])
+      end
+
+      def sdk_logger
+        Sentry.sdk_logger if Sentry.initialized?
+      end
+    end
+  end
+end
diff --git a/sentry-ruby/lib/sentry/job_extensions/producer.rb b/sentry-ruby/lib/sentry/job_extensions/producer.rb
new file mode 100644
index 000000000..957b28b05
--- /dev/null
+++ b/sentry-ruby/lib/sentry/job_extensions/producer.rb
@@ -0,0 +1,41 @@
+# frozen_string_literal: true
+
+module Sentry
+  module JobExtensions
+    # @api private
+    class Producer < Instrumentation
+      OP_NAME = "queue.publish"
+
+      def record_span(job)
+        Sentry.with_child_span(op: OP_NAME) do |span|
+          safely("record the publish span") { describe_span(span, job) } if span
+
+          yield
+        end
+      end
+
+      def prepare_payload(job, payload)
+        payload
+      end
+
+      def inject_trace_headers(job, payload, headers)
+        extension.inject_trace_headers(job: job, payload: payload, headers: headers)
+      end
+
+      private
+
+      def describe_span(span, job)
+        span.set_description(extension.publish_description(job: job))
+        span.set_origin(extension.span_origin(job: job))
+        set_messaging_data(span, extension.publish_messaging_data(job: job))
+        set_retry_count(span, job)
+      end
+
+      class Delegated < Producer
+        def record_span(job)
+          yield
+        end
+      end
+    end
+  end
+end
diff --git a/sentry-ruby/spec/sentry/job_extensions/consumer_spec.rb b/sentry-ruby/spec/sentry/job_extensions/consumer_spec.rb
new file mode 100644
index 000000000..e4ac37e6b
--- /dev/null
+++ b/sentry-ruby/spec/sentry/job_extensions/consumer_spec.rb
@@ -0,0 +1,309 @@
+# frozen_string_literal: true
+
+require "sentry/integrable"
+
+module Sentry
+  module TestJobsIntegration
+    extend Sentry::Integrable
+
+    register_integration name: "test_jobs", version: "1.0"
+  end
+end
+
+RSpec.describe Sentry::JobExtensions::Consumer do
+  let(:job) { Struct.new(:id, :queue).new("job-1", "critical") }
+  let(:string_io) { StringIO.new }
+
+  def consumer(**hooks)
+    extension = Class.new(Sentry::JobExtensions::Extension) do
+      define_method(:span_origin) { |job:| "auto.queue.test" }
+      define_method(:integration) { |job:| Sentry::TestJobsIntegration }
+      hooks.each { |hook, body| define_method(hook, &body) }
+    end.new(name: "test_backend")
+    described_class.new(extension: extension)
+  end
+
+  def scope
+    Sentry.get_current_scope
+  end
+
+  def start_transaction(consumer, **options)
+    scope.set_transaction_name("TestJob", source: :task)
+    consumer.start_transaction(scope, job, **options)
+  end
+
+  def finished_transaction(consumer, **options)
+    transaction = start_transaction(consumer, **options)
+    consumer.finish_transaction(transaction, 200)
+    sentry_events.last
+  end
+
+  def capture(consumer, exception = RuntimeError.new("boom"), **options)
+    consumer.capture_exception(job, exception, **options)
+    sentry_events.last
+  end
+
+  def enriching_consumer
+    consumer(
+      messaging_data: ->(job:) { { system: "backend", message_id: job.id, destination: job.queue, receive_latency: 42.5 } },
+      retry_info: ->(job:) { { count: 3, will_retry: true } },
+      scope_data: ->(job:) { { contexts: { backend: { "jid" => "backend-jid" } }, tags: { "backend.queue" => "low" }, extra: { size: 12 } } }
+    )
+  end
+
+  let(:foreign_trace_id) { "a" * 32 }
+  let(:foreign_trace_headers) { { "sentry-trace" => "#{foreign_trace_id}-#{"b" * 16}-1" } }
+
+  before do
+    perform_basic_setup do |config|
+      config.traces_sample_rate = 1.0
+      config.sdk_logger = Logger.new(string_io)
+    end
+    setup_sentry_test
+  end
+
+  after do
+    teardown_sentry_test
+  end
+
+  describe "#start_transaction" do
+    it "starts a queue.process transaction on the scope named after the scope's transaction" do
+      transaction = start_transaction(consumer)
+
+      expect(scope.get_span).to eq(transaction)
+      expect(transaction.op).to eq("queue.process")
+      expect(transaction.origin).to eq("auto.queue.test")
+      expect(transaction.name).to eq("TestJob")
+      expect(transaction.source).to eq(:task)
+    end
+
+    it "records the extension's messaging data and retry count as span data" do
+      event = finished_transaction(enriching_consumer)
+
+      expect(event.contexts.dig(:trace, :data)).to include(
+        "messaging.system" => "backend",
+        "messaging.message.id" => "job-1",
+        "messaging.destination.name" => "critical",
+        "messaging.message.receive.latency" => 42.5,
+        "messaging.message.retry.count" => 3
+      )
+    end
+
+    it "keeps the transaction and logs the failure when the extension's messaging data raises" do
+      failing_consumer = consumer(messaging_data: ->(job:) { raise "broken messaging data" })
+
+      event = finished_transaction(failing_consumer)
+
+      expect(event.contexts.dig(:trace, :op)).to eq("queue.process")
+      expect(scope.get_span.timestamp).not_to be_nil
+      expect(string_io.string).to include("broken messaging data")
+    end
+
+    it "records the known messaging data and skips keys outside the messaging attributes" do
+      event = finished_transaction(
+        consumer(messaging_data: ->(job:) { { system: "backend", unknown_key: "value", destination: job.queue } })
+      )
+
+      expect(event.contexts.dig(:trace, :data)).to include(
+        "messaging.system" => "backend",
+        "messaging.destination.name" => "critical"
+      )
+      expect(event.contexts.dig(:trace, :data).values).not_to include("value")
+    end
+
+    context "when an extension collects latency only for sampled deliveries" do
+      let(:job) { Struct.new(:id, :queue, :enqueued_at).new("job-1", "critical", Time.now - 5) }
+      let(:sampling_consumer) do
+        consumer(
+          messaging_data: lambda do |job:|
+            data = { system: "backend", message_id: job.id, destination: job.queue }
+            data[:receive_latency] = (Time.now - job.enqueued_at) * 1_000 if Sentry.get_current_scope.get_span&.sampled
+            data
+          end
+        )
+      end
+
+      it "records latency for a sampled delivery when the caller is unsampled" do
+        scope.set_span(Sentry.start_transaction(name: "worker", op: "worker", sampled: false))
+
+        event = finished_transaction(sampling_consumer, trace_headers: foreign_trace_headers)
+
+        expect(event.contexts.dig(:trace, :data)).to include(
+          "messaging.message.receive.latency" => a_kind_of(Numeric)
+        )
+      end
+
+      it "omits latency from an unsampled delivery's error when the caller is sampled" do
+        scope.set_span(Sentry.start_transaction(name: "worker", op: "worker", sampled: true))
+        headers = { "sentry-trace" => "#{foreign_trace_id}-#{"b" * 16}-0" }
+        transaction = start_transaction(sampling_consumer, trace_headers: headers)
+
+        event = capture(sampling_consumer)
+        sampling_consumer.finish_transaction(transaction, 500)
+
+        expect(event.contexts.dig(:trace, :data)).to include(
+          "messaging.system" => "backend",
+          "messaging.message.id" => job.id,
+          "messaging.destination.name" => job.queue
+        )
+        expect(event.contexts.dig(:trace, :data)).not_to have_key("messaging.message.receive.latency")
+      end
+    end
+
+    it "continues the trace described by the trace headers" do
+      transaction = start_transaction(consumer, trace_headers: foreign_trace_headers)
+
+      expect(transaction.trace_id).to eq(foreign_trace_id)
+    end
+
+    it "keeps the scope's trace when the trace headers are empty" do
+      scope_trace_id = scope.propagation_context.trace_id
+
+      transaction = start_transaction(consumer, trace_headers: {})
+
+      expect(scope.propagation_context.trace_id).to eq(scope_trace_id)
+      expect(transaction.parent_span_id).to be_nil
+    end
+
+    it "returns nil when tracing is disabled" do
+      Sentry.configuration.traces_sample_rate = nil
+
+      expect(start_transaction(enriching_consumer)).to be_nil
+    end
+  end
+
+  describe "#finish_transaction" do
+    it "finishes the transaction with the status for the given HTTP status code" do
+      started = consumer
+      transaction = start_transaction(started)
+
+      started.finish_transaction(transaction, 500)
+
+      expect(sentry_events.last.contexts.dig(:trace, :status)).to eq("internal_error")
+    end
+
+    it "keeps the first completion status and sends the transaction only once" do
+      started = consumer
+      transaction = start_transaction(started)
+      started.finish_transaction(transaction, 200)
+      timestamp = transaction.timestamp
+
+      started.finish_transaction(transaction, 500)
+
+      expect(transaction.timestamp).to eq(timestamp)
+      expect(transaction.status).to eq("ok")
+      expect(sentry_events.size).to eq(1)
+      expect(sentry_events.last.contexts.dig(:trace, :status)).to eq("ok")
+    end
+
+    it "ignores a missing transaction" do
+      expect { consumer.finish_transaction(nil, 200) }.not_to raise_error
+    end
+  end
+
+  describe "#apply_scope_data" do
+    it "adds the extension's contexts, tags and extra to the scope" do
+      enriching_consumer.apply_scope_data(scope, job)
+
+      expect(scope.contexts[:backend]).to eq("jid" => "backend-jid")
+      expect(scope.tags).to include("backend.queue" => "low")
+      expect(scope.extra).to include(size: 12)
+    end
+
+    it "sets the user the extension supplies on the scope" do
+      consumer(scope_data: ->(job:) { { user: { "id" => "42", "segment" => "beta" } } }).apply_scope_data(scope, job)
+
+      expect(scope.user).to eq("id" => "42", "segment" => "beta")
+    end
+  end
+
+  describe "#report_exception?" do
+    let(:exception) { RuntimeError.new("boom") }
+
+    it "follows the extension's decision whether the backend will retry the job or not" do
+      reporting = consumer(report_exception?: ->(job:, exception:) { true }, retry_info: ->(job:) { { count: 0, will_retry: true } })
+      silent = consumer(report_exception?: ->(job:, exception:) { false }, retry_info: ->(job:) { { count: 0, will_retry: false } })
+
+      expect(reporting.report_exception?(job, exception)).to be(true)
+      expect(silent.report_exception?(job, exception)).to be(false)
+    end
+  end
+
+  describe "#report_retried_exception?" do
+    let(:exception) { RuntimeError.new("boom") }
+
+    it "follows the extension's decision" do
+      reporting = consumer(report_retried_exception?: ->(job:, exception:) { true })
+
+      expect(reporting.report_retried_exception?(job, exception)).to be(true)
+      expect(consumer.report_retried_exception?(job, exception)).to be(false)
+    end
+  end
+
+  describe "#capture_exception" do
+    def error_consumer
+      consumer(
+        error_data: lambda do |job:, exception:|
+          {
+            contexts: { backend_error: { "retried" => true }, job: { "id" => "backend", "attempt" => 2 } },
+            tags: { "backend.error" => "yes", queue: "backend-queue", job_id: "backend" },
+            extra: { attempts: 2, size: 99 }
+          }
+        end
+      )
+    end
+
+    it "captures the exception through the extension's integration" do
+      event = capture(consumer)
+
+      expect(event.exception.values.last.type).to eq("RuntimeError")
+      expect(event.exception.values.last.mechanism.type).to eq("test_jobs")
+      expect(event.sdk[:name]).to eq("sentry.ruby.test_jobs")
+    end
+
+    it "adds the extension's error data without replacing values already on the scope" do
+      scope.set_contexts(job: { "id" => "scope" })
+      scope.set_tags(queue: "default")
+      scope.set_extras(size: 12)
+
+      event = capture(error_consumer)
+
+      expect(event.contexts[:backend_error]).to eq("retried" => true)
+      expect(event.contexts[:job]).to eq("id" => "scope", "attempt" => 2)
+      expect(event.tags).to include("backend.error" => "yes", queue: "default")
+      expect(event.extra).to include(attempts: 2, size: 12)
+    end
+
+    it "keeps existing extension and user context values while adding error-specific fields" do
+      refining_consumer = consumer(
+        scope_data: ->(job:) { { contexts: { job: { "id" => "backend", "args" => ["raw"] } } } },
+        error_data: ->(job:, exception:) { { contexts: { job: { "args" => ["filtered"], "id" => "error", "attempt" => 2 } } } }
+      )
+      refining_consumer.apply_scope_data(scope, job)
+      scope.set_contexts(job: { "id" => "user" })
+
+      event = capture(refining_consumer)
+
+      expect(event.contexts[:job]).to eq("id" => "user", "args" => ["raw"], "attempt" => 2)
+    end
+
+    it "lets the caller's contexts, tags and extra win over the extension's error data" do
+      event = capture(error_consumer, contexts: { job: { "id" => "caller" } }, tags: { job_id: "caller" }, extra: { attempts: 5 })
+
+      expect(event.contexts[:job]).to eq("id" => "caller")
+      expect(event.tags).to include(job_id: "caller", "backend.error" => "yes")
+      expect(event.extra).to include(attempts: 5)
+    end
+
+    it "lets caller tags and extra override extension values with equivalent string or symbol keys" do
+      event = capture(error_consumer, tags: { "job_id" => "caller", :"backend.error" => "caller-error" }, extra: { "attempts" => 5 })
+
+      expect(event.tags).to eq(job_id: "caller", queue: "backend-queue", "backend.error" => "caller-error")
+      expect(event.extra).to eq(attempts: 5, size: 99)
+      expect(event.to_json_compatible).to include(
+        "tags" => { "job_id" => "caller", "queue" => "backend-queue", "backend.error" => "caller-error" },
+        "extra" => { "attempts" => 5, "size" => 99 }
+      )
+    end
+  end
+end
diff --git a/sentry-ruby/spec/sentry/job_extensions/producer_spec.rb b/sentry-ruby/spec/sentry/job_extensions/producer_spec.rb
new file mode 100644
index 000000000..cf9f238f2
--- /dev/null
+++ b/sentry-ruby/spec/sentry/job_extensions/producer_spec.rb
@@ -0,0 +1,101 @@
+# frozen_string_literal: true
+
+RSpec.describe Sentry::JobExtensions::Producer do
+  let(:job) { Struct.new(:id, :queue).new("job-1", "critical") }
+  let(:string_io) { StringIO.new }
+
+  def producer(**hooks)
+    extension = Class.new(Sentry::JobExtensions::Extension) do
+      define_method(:span_origin) { |job:| "auto.queue.test" }
+      define_method(:publish_description) { |job:| "TestJob" }
+      hooks.each { |hook, body| define_method(hook, &body) }
+    end.new(name: "test_backend")
+    described_class.new(extension: extension)
+  end
+
+  def enriching_producer
+    producer(
+      messaging_data: ->(job:) { { system: "backend", message_id: job.id, destination: job.queue, receive_latency: 42.5 } },
+      retry_info: ->(job:) { { count: 3, will_retry: true } }
+    )
+  end
+
+  def within_transaction
+    transaction = Sentry.start_transaction(name: "parent", op: "test")
+    Sentry.get_current_scope.set_span(transaction)
+    yield
+  ensure
+    transaction.finish
+  end
+
+  def publish_span
+    sentry_events.last.spans.find { |span| span[:op] == "queue.publish" }
+  end
+
+  def record_span(producer, &enqueue)
+    producer.record_span(job, &(enqueue || -> { :enqueued }))
+  end
+
+  before do
+    perform_basic_setup do |config|
+      config.traces_sample_rate = 1.0
+      config.sdk_logger = Logger.new(string_io)
+    end
+    setup_sentry_test
+  end
+
+  after do
+    teardown_sentry_test
+  end
+
+  describe "#record_span" do
+    it "records a queue.publish span with the extension's identity, messaging data and retry count around the enqueue" do
+      result = within_transaction { record_span(enriching_producer) }
+
+      expect(result).to eq(:enqueued)
+      expect(publish_span).to include(description: "TestJob", origin: "auto.queue.test")
+      expect(publish_span[:data]).to include(
+        "messaging.system" => "backend",
+        "messaging.message.id" => "job-1",
+        "messaging.destination.name" => "critical",
+        "messaging.message.receive.latency" => 42.5,
+        "messaging.message.retry.count" => 3
+      )
+    end
+
+    it "enqueues without a span when no transaction is active" do
+      expect(record_span(enriching_producer)).to eq(:enqueued)
+      expect(sentry_events).to be_empty
+    end
+
+    it "logs an instrumentation failure and enqueues the job exactly once" do
+      failing = producer(publish_messaging_data: ->(job:) { raise "boom from instrumentation" })
+      enqueued = 0
+
+      result = within_transaction { record_span(failing) { enqueued += 1 } }
+
+      expect(result).to eq(1)
+      expect(enqueued).to eq(1)
+      expect(string_io.string).to include("test_backend", "boom from instrumentation")
+    end
+
+    it "propagates a failing enqueue" do
+      expect do
+        within_transaction { record_span(enriching_producer) { raise "boom from enqueue" } }
+      end.to raise_error(RuntimeError, /boom from enqueue/)
+    end
+  end
+
+  describe "#inject_trace_headers" do
+    let(:payload) { { "job_class" => "TestJob" } }
+    let(:headers) { { "sentry-trace" => "trace" } }
+
+    it "returns the payload the extension built with the headers" do
+      extension_producer = producer(inject_trace_headers: ->(job:, payload:, headers:) { payload.merge("trace" => headers) })
+
+      expect(extension_producer.inject_trace_headers(job, payload, headers)).to eq(
+        "job_class" => "TestJob", "trace" => headers
+      )
+    end
+  end
+end
diff --git a/sentry-ruby/spec/sentry/job_extensions_spec.rb b/sentry-ruby/spec/sentry/job_extensions_spec.rb
new file mode 100644
index 000000000..eb2d67ef5
--- /dev/null
+++ b/sentry-ruby/spec/sentry/job_extensions_spec.rb
@@ -0,0 +1,58 @@
+# frozen_string_literal: true
+
+RSpec.describe "Sentry job extensions" do
+  describe "Sentry::JobExtensions::Extension" do
+    it "rejects a blank name" do
+      expect { Sentry::JobExtensions::Extension.new(name: "") }.to raise_error(ArgumentError)
+      expect { Sentry::JobExtensions::Extension.new(name: nil) }.to raise_error(ArgumentError)
+    end
+  end
+
+  describe "Sentry::JobExtensions::Extension hooks" do
+    let(:extension) { Sentry::JobExtensions::Extension.new(name: "noop") }
+    let(:job) { double("job") }
+    let(:exception) { RuntimeError.new("job failed") }
+    let(:payload) { { "class" => "HardJob", "args" => [1] } }
+
+    def retrying_extension(will_retry:)
+      Class.new(Sentry::JobExtensions::Extension) do
+        define_method(:retry_info) { |job:| { count: 1, will_retry: will_retry } }
+      end.new(name: "retrying")
+    end
+
+    it "supplies no scope, error or messaging data by default" do
+      expect(extension.scope_data(job: job)).to eq({})
+      expect(extension.error_data(job: job, exception: exception)).to eq({})
+      expect(extension.messaging_data(job: job)).to eq({})
+      expect(extension.publish_messaging_data(job: job)).to eq({})
+    end
+
+    it "reports no retries and no pending retry by default" do
+      expect(extension.retry_info(job: job)).to eq(count: 0, will_retry: false)
+    end
+
+    it "carries no trace headers and leaves the payload unchanged by default" do
+      expect(extension.extract_trace_headers(job: job, payload: payload)).to eq({})
+      expect(extension.inject_trace_headers(job: job, payload: payload, headers: { "sentry-trace" => "trace" })).to equal(payload)
+    end
+
+    it "reports an exception only when the backend will not retry the job" do
+      expect(retrying_extension(will_retry: false).report_exception?(job: job, exception: exception)).to be(true)
+      expect(retrying_extension(will_retry: true).report_exception?(job: job, exception: exception)).to be(false)
+    end
+
+    it "does not report exceptions the job integration retries by default" do
+      expect(extension.report_retried_exception?(job: job, exception: exception)).to be(false)
+    end
+
+    it "uses the messaging data for the publish span unless overridden" do
+      messaging = Class.new(Sentry::JobExtensions::Extension) do
+        def messaging_data(job:)
+          { system: "backend" }
+        end
+      end.new(name: "messaging")
+
+      expect(messaging.publish_messaging_data(job: job)).to eq(system: "backend")
+    end
+  end
+end