diff --git a/sentry-ruby/lib/sentry-ruby.rb b/sentry-ruby/lib/sentry-ruby.rb
index 12ed09d93..288a8c6c1 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/extension"
 
 [
   "sentry/rake",
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..1e62eb50f
--- /dev/null
+++ b/sentry-ruby/lib/sentry/job_extensions/extension.rb
@@ -0,0 +1,314 @@
+# frozen_string_literal: true
+
+module Sentry
+  module JobExtensions
+    # Base class for extensions that let a background-job backend enrich Sentry's
+    # job integrations. A backend gem subclasses it and overrides the hooks it can
+    # supply; jobs are opaque to Sentry and only interpreted by the extension.
+    class Extension
+      include LoggingHelper
+
+      PUBLISH_OP = "queue.publish"
+      PROCESS_OP = "queue.process"
+
+      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,
+        retry_count: Span::DataConventions::MESSAGING_MESSAGE_RETRY_COUNT
+      }.freeze
+
+      # The extension's name, used in SDK log messages.
+      # @return [Symbol]
+      attr_reader :name
+
+      # Creates an extension. Raises ArgumentError if the name is not a Symbol.
+      #
+      # @param name [Symbol]
+      # @param integration [Module] the integration that captures the job's exceptions
+      # @param origin [String] the origin of the job's spans
+      def initialize(name:, integration:, origin:)
+        raise ArgumentError, "job extension name must be a Symbol, got #{name.inspect}" unless name.is_a?(Symbol)
+
+        @name = name
+        @integration = integration
+        @origin = origin
+      end
+
+      # Wraps enqueueing a job in a `queue.publish` span.
+      #
+      # @param job [Object]
+      # @yieldreturn [Object] the result of enqueueing the job
+      # @return [Object] the block's result
+      def instrument_enqueue(job:, **context, &enqueue)
+        return enqueue.call unless safely("decide whether to record the producer span", fallback: true) { record_publish_span?(job: job, **context) }
+
+        Sentry.with_child_span(op: PUBLISH_OP) do |span|
+          if span
+            span.set_description(safely("describe the producer span") { publish_description(job: job, **context) })
+            span.set_origin(safely("set the producer span origin", fallback: @origin) { publish_span_origin(job: job, **context) })
+            set_messaging_data(span, safely("collect the producer messaging data", fallback: {}) { Hash(publish_messaging_data(job: job, **context)) })
+          end
+
+          enqueue.call
+        end
+      end
+
+      # Prepares the scope for performing the job and starts its `queue.process`
+      # transaction, continuing the trace the job was enqueued in.
+      #
+      # @param scope [Scope]
+      # @param job [Object]
+      # @return [Transaction, nil]
+      def start_job(scope:, job:, **context)
+        apply_scope_data(scope, safely("collect the scope data", fallback: {}) { Hash(scope_data(job: job, **context)) })
+
+        job_name = safely("name the transaction") { transaction_name(job: job, **context) }
+        scope.set_transaction_name(job_name, source: :task) if job_name
+
+        trace_headers = safely("extract trace headers", fallback: {}) { Hash(extract_trace_headers(job: job, **context)) }
+        origin = safely("set the transaction origin", fallback: @origin) { span_origin(job: job, **context) }
+        transaction = start_transaction(scope, trace_headers, origin)
+        return unless transaction
+
+        scope.set_span(transaction)
+        set_messaging_data(transaction, safely("collect the messaging data", fallback: {}) { Hash(messaging_data(job: job, **context)) })
+        transaction
+      end
+
+      # Finishes the job's transaction with the given status.
+      #
+      # @param transaction [Transaction, nil]
+      # @param status [Integer]
+      # @return [void]
+      def finish_job(transaction:, status:)
+        return if !transaction || transaction.timestamp
+
+        transaction.set_http_status(status)
+        transaction.finish
+      end
+
+      # Captures an exception raised by the job, enriched with the extension's error data.
+      #
+      # @param job [Object]
+      # @param exception [Exception]
+      # @return [Event, nil]
+      def capture_exception(job:, exception:, contexts: {}, tags: {}, extra: {}, **context)
+        data = safely("collect the error data", fallback: {}) { normalized_error_data(error_data(job: job, exception: exception, **context)) }
+        capturing_integration = safely("pick the capturing integration", fallback: @integration) { integration(job: job, **context) }
+
+        capturing_integration.capture_exception(exception, **merged_error_data(data, contexts: contexts, tags: tags, extra: extra))
+      end
+
+      # Runs a hook, logging its failure and returning `fallback` instead of raising.
+      #
+      # @param context [String] what the hook does, used in the log message
+      # @return [Object]
+      def safely(context, fallback: nil)
+        yield
+      rescue StandardError => e
+        log_failure(context, e)
+        fallback
+      end
+
+      ##### Hooks #####
+      #
+      # Called by job integrations with the job and any extra keyword context.
+      # Hooks must not mutate the objects they are given.
+
+      # Returns whether to record a `queue.publish` span when the job is enqueued.
+      #
+      # @param job [Object]
+      # @return [Boolean]
+      def record_publish_span?(job:, **)
+        true
+      end
+
+      # Returns data to apply to the scope the job is performed in.
+      #
+      # @param job [Object]
+      # @return [Hash] with optional `:contexts`, `:tags`, `:extra` and `:user`
+      def scope_data(job:, **)
+        {}
+      end
+
+      # Returns data to add to the event captured for a failed job. Data already on
+      # the scope or passed explicitly by the job integration takes precedence.
+      #
+      # @param job [Object]
+      # @param exception [Exception]
+      # @return [Hash] with optional `:contexts`, `:tags` and `:extra`
+      def error_data(job:, exception:, **)
+        {}
+      end
+
+      # Returns whether the backend will retry the failed job.
+      #
+      # @param job [Object]
+      # @return [Boolean]
+      def will_retry?(job:, **)
+        false
+      end
+
+      # Returns whether to capture the job's exception. By default only when the
+      # backend won't retry the job.
+      #
+      # @param job [Object]
+      # @param exception [Exception]
+      # @return [Boolean]
+      def report_exception?(job:, exception:, **context)
+        !will_retry?(job: job, exception: exception, **context)
+      end
+
+      # Returns whether to capture an exception that the job integration's own
+      # retry mechanism retries, so the backend never sees it.
+      #
+      # @param job [Object]
+      # @param exception [Exception]
+      # @return [Boolean]
+      def report_retried_exception?(job:, exception:, **)
+        false
+      end
+
+      # Returns the backend's job payload carrying the trace propagation headers.
+      #
+      # @param job [Object]
+      # @param payload [Object]
+      # @param headers [Hash{String => String}]
+      # @return [Object] a payload of the same class as the one given
+      def inject_trace_headers(job:, payload:, headers:, **)
+        payload
+      end
+
+      # Returns the trace propagation headers the job carries, so that performing
+      # it continues the trace it was enqueued in.
+      #
+      # @param job [Object]
+      # @return [Hash{String => String}]
+      def extract_trace_headers(job:, **)
+        {}
+      end
+
+      # Returns messaging attributes for the job's spans.
+      #
+      # @param job [Object]
+      # @return [Hash] with optional `:system`, `:message_id`, `:destination`,
+      #   `:receive_latency` and `:retry_count`
+      def messaging_data(job:, **)
+        {}
+      end
+
+      # Returns messaging attributes for the `queue.publish` span. Defaults to
+      # {#messaging_data}.
+      #
+      # @param job [Object]
+      # @return [Hash]
+      def publish_messaging_data(job:, **context)
+        messaging_data(job: job, **context)
+      end
+
+      # Returns the name of the job's transaction. Defaults to the job's class name.
+      #
+      # @param job [Object]
+      # @return [String, nil]
+      def transaction_name(job:, **)
+        job.class.name
+      end
+
+      # Returns the description of the `queue.publish` span. Defaults to the job's class name.
+      #
+      # @param job [Object]
+      # @return [String, nil]
+      def publish_description(job:, **)
+        job.class.name
+      end
+
+      # Returns the origin of the job's transaction. Defaults to the extension's origin.
+      #
+      # @param job [Object]
+      # @return [String]
+      def span_origin(job:, **)
+        @origin
+      end
+
+      # Returns the origin of the `queue.publish` span. Defaults to {#span_origin}.
+      #
+      # @param job [Object]
+      # @return [String]
+      def publish_span_origin(job:, **context)
+        span_origin(job: job, **context)
+      end
+
+      # Returns the integration that captures the job's exceptions. Defaults to
+      # the extension's integration.
+      #
+      # @param job [Object]
+      # @return [Module]
+      def integration(job:, **)
+        @integration
+      end
+
+      private
+
+      def start_transaction(scope, trace_headers, origin)
+        options = {
+          name: scope.transaction_name,
+          source: scope.transaction_source,
+          op: PROCESS_OP,
+          origin: origin
+        }
+
+        if trace_headers.empty?
+          Sentry.start_transaction(**options)
+        else
+          Sentry.start_transaction(transaction: Sentry.continue_trace(trace_headers, **options), **options)
+        end
+      end
+
+      def apply_scope_data(scope, data)
+        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 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 normalized_error_data(data)
+        data = Hash(data)
+
+        {
+          contexts: Hash(data[:contexts]).transform_values { |context| Hash(context) },
+          tags: Hash(data[:tags]),
+          extra: Hash(data[:extra])
+        }
+      end
+
+      def merged_error_data(data, contexts:, tags:, extra:)
+        scope = Sentry.get_current_scope
+
+        merged_contexts = data.fetch(:contexts, {}).to_h { |context_name, context| [context_name, context.merge(scope.contexts.fetch(context_name, {}))] }.merge(contexts)
+        options = {
+          extra: data.fetch(:extra, {}).reject { |key, _value| scope.extra.key?(key) }.merge(extra),
+          tags: data.fetch(:tags, {}).reject { |key, _value| scope.tags.key?(key) }.merge(tags),
+          hint: { background: false }
+        }
+        options[:contexts] = merged_contexts unless merged_contexts.empty?
+        options
+      end
+
+      def log_failure(context, error)
+        log_error("Job extension #{name.inspect} failed to #{context}", error)
+      end
+
+      def sdk_logger
+        Sentry.sdk_logger if Sentry.initialized?
+      end
+    end
+  end
+end
diff --git a/sentry-ruby/spec/sentry/job_extensions/extension_spec.rb b/sentry-ruby/spec/sentry/job_extensions/extension_spec.rb
new file mode 100644
index 000000000..dbce9d9f4
--- /dev/null
+++ b/sentry-ruby/spec/sentry/job_extensions/extension_spec.rb
@@ -0,0 +1,427 @@
+# 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::Extension do
+  let(:job) { Struct.new(:id, :queue).new("job-1", "critical") }
+  let(:exception) { RuntimeError.new("boom") }
+  let(:string_io) { StringIO.new }
+  let(:foreign_trace_id) { "a" * 32 }
+  let(:foreign_trace_headers) { { "sentry-trace" => "#{foreign_trace_id}-#{"b" * 16}-1" } }
+
+  def build_extension(**hooks)
+    Class.new(described_class) do
+      hooks.each { |hook, body| define_method(hook, &body) }
+    end.new(name: :test_backend, integration: Sentry::TestJobsIntegration, origin: "auto.queue.test")
+  end
+
+  def enriching_extension
+    build_extension(
+      messaging_data: ->(job:, **) { { system: "backend", message_id: job.id, destination: job.queue, receive_latency: 42.5, retry_count: 3 } },
+      scope_data: ->(job:, **) { { contexts: { backend: { "jid" => "backend-jid" } }, tags: { "backend.queue" => "low" }, extra: { size: 12 } } }
+    )
+  end
+
+  def tracing_extension(**hooks)
+    build_extension(extract_trace_headers: ->(job:, trace_headers: {}, **) { trace_headers }, **hooks)
+  end
+
+  def scope
+    Sentry.get_current_scope
+  end
+
+  def within_transaction
+    transaction = Sentry.start_transaction(name: "parent", op: "test")
+    scope.set_span(transaction)
+    yield
+  ensure
+    transaction.finish
+  end
+
+  def publish_span
+    sentry_events.last.spans.find { |span| span[:op] == "queue.publish" }
+  end
+
+  def finished_job(extension, **context)
+    transaction = extension.start_job(scope: scope, job: job, **context)
+    extension.finish_job(transaction: transaction, status: 200)
+    sentry_events.last
+  end
+
+  def capture(extension, **options)
+    extension.capture_exception(job: job, exception: exception, **options)
+    sentry_events.last
+  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 "#initialize" do
+    it "keeps the Symbol name as given" do
+      expect(build_extension.name).to be(:test_backend)
+    end
+
+    it "rejects a name that is not a Symbol" do
+      expect { described_class.new(name: "test_backend", integration: Sentry::TestJobsIntegration, origin: "auto.queue.test") }
+        .to raise_error(ArgumentError, /must be a Symbol/)
+      expect { described_class.new(name: nil, integration: Sentry::TestJobsIntegration, origin: "auto.queue.test") }
+        .to raise_error(ArgumentError, /must be a Symbol/)
+    end
+  end
+
+  describe "hooks" do
+    let(:extension) { build_extension }
+    let(:payload) { { "class" => "HardJob", "args" => [1] } }
+
+    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 "carries no trace headers and leaves the payload unchanged by default" do
+      expect(extension.extract_trace_headers(job: job)).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(build_extension(will_retry?: ->(**) { false }).report_exception?(job: job, exception: exception)).to be(true)
+      expect(build_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 = build_extension(messaging_data: ->(job:, **) { { system: "backend" } })
+
+      expect(messaging.publish_messaging_data(job: job)).to eq(system: "backend")
+    end
+
+    it "ignores keywords a hook does not read" do
+      expect(extension.scope_data(job: job, queue: "low")).to eq({})
+      expect(extension.report_exception?(job: job, exception: exception, attempt: 2)).to be(true)
+    end
+
+    it "passes the keywords an integration supplies to the hooks that read them" do
+      queue_aware = build_extension(scope_data: ->(job:, queue:, **) { { tags: { queue: queue } } })
+
+      queue_aware.start_job(scope: scope, job: job, queue: "low")
+
+      expect(scope.tags).to include(queue: "low")
+    end
+  end
+
+  describe "#instrument_enqueue" do
+    it "records a queue.publish span with the extension's identity and messaging data around the enqueue" do
+      extension = build_extension(
+        publish_description: ->(**) { "TestJob" },
+        publish_messaging_data: ->(job:, **) { { system: "backend", message_id: job.id, destination: job.queue, retry_count: 3 } }
+      )
+
+      result = within_transaction { extension.instrument_enqueue(job: job) { :enqueued } }
+
+      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.retry.count" => 3
+      )
+    end
+
+    it "gives the publish span its own origin when the extension supplies one" do
+      extension = build_extension(publish_span_origin: ->(**) { Sentry::Span::DEFAULT_SPAN_ORIGIN })
+
+      within_transaction { extension.instrument_enqueue(job: job) { :enqueued } }
+
+      expect(publish_span).to include(origin: "manual")
+    end
+
+    it "enqueues without a span when no transaction is active" do
+      expect(build_extension.instrument_enqueue(job: job) { :enqueued }).to eq(:enqueued)
+      expect(sentry_events).to be_empty
+    end
+
+    it "enqueues without a span when the extension does not record one" do
+      extension = build_extension(record_publish_span?: ->(**) { false })
+
+      result = within_transaction { extension.instrument_enqueue(job: job) { :enqueued } }
+
+      expect(result).to eq(:enqueued)
+      expect(publish_span).to be_nil
+    end
+
+    it "logs an instrumentation failure and enqueues the job exactly once" do
+      failing = build_extension(publish_messaging_data: ->(**) { raise "boom from instrumentation" })
+      enqueued = 0
+
+      result = within_transaction { failing.instrument_enqueue(job: job) { 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 "enqueues the job exactly once when a data hook returns something other than a Hash" do
+      extension = build_extension(publish_messaging_data: ->(**) { nil })
+      enqueued = 0
+
+      within_transaction { extension.instrument_enqueue(job: job) { enqueued += 1 } }
+
+      expect(enqueued).to eq(1)
+      expect(publish_span).not_to be_nil
+    end
+
+    it "propagates a failing enqueue" do
+      expect do
+        within_transaction { build_extension.instrument_enqueue(job: job) { raise "boom from enqueue" } }
+      end.to raise_error(RuntimeError, /boom from enqueue/)
+    end
+  end
+
+  describe "#start_job" do
+    it "starts a queue.process transaction on the scope named by the extension" do
+      extension = build_extension(transaction_name: ->(**) { "TestJob" })
+
+      transaction = extension.start_job(scope: scope, job: job)
+
+      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 "adds the extension's contexts, tags and extra to the scope" do
+      enriching_extension.start_job(scope: scope, job: 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
+      build_extension(scope_data: ->(**) { { user: { "id" => "42", "segment" => "beta" } } }).start_job(scope: scope, job: job)
+
+      expect(scope.user).to eq("id" => "42", "segment" => "beta")
+    end
+
+    it "records the extension's messaging data and retry count as span data" do
+      event = finished_job(enriching_extension)
+
+      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 "records the known messaging data and skips keys outside the messaging attributes" do
+      event = finished_job(build_extension(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
+
+    it "keeps the transaction and logs the failure when a hook raises" do
+      failing = build_extension(
+        scope_data: ->(**) { raise "broken scope data" },
+        messaging_data: ->(**) { raise "broken messaging data" }
+      )
+
+      event = finished_job(failing)
+
+      expect(event.contexts.dig(:trace, :op)).to eq("queue.process")
+      expect(string_io.string).to include("broken scope data", "broken messaging data")
+    end
+
+    it "starts the transaction when data hooks return something other than a Hash" do
+      extension = build_extension(
+        scope_data: ->(**) { nil },
+        extract_trace_headers: ->(**) { nil },
+        messaging_data: ->(**) { "not a hash" }
+      )
+
+      event = finished_job(extension)
+
+      expect(event.contexts.dig(:trace, :op)).to eq("queue.process")
+      expect(string_io.string).to include("Job extension :test_backend failed to collect the messaging data")
+    end
+
+    it "continues the trace described by the extracted trace headers" do
+      transaction = tracing_extension.start_job(scope: scope, job: job, trace_headers: foreign_trace_headers)
+
+      expect(transaction.trace_id).to eq(foreign_trace_id)
+    end
+
+    it "keeps the scope's trace when there are no trace headers" do
+      scope_trace_id = scope.propagation_context.trace_id
+
+      transaction = tracing_extension.start_job(scope: scope, job: job)
+
+      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(enriching_extension.start_job(scope: scope, job: job)).to be_nil
+    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_extension) do
+        tracing_extension(
+          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_job(sampling_extension, 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 = sampling_extension.start_job(scope: scope, job: job, trace_headers: headers)
+
+        event = capture(sampling_extension)
+        sampling_extension.finish_job(transaction: transaction, status: 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
+  end
+
+  describe "#finish_job" do
+    let(:extension) { build_extension }
+
+    it "finishes the transaction with the status for the given HTTP status code" do
+      transaction = extension.start_job(scope: scope, job: job)
+
+      extension.finish_job(transaction: transaction, status: 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
+      transaction = extension.start_job(scope: scope, job: job)
+      extension.finish_job(transaction: transaction, status: 200)
+
+      extension.finish_job(transaction: transaction, status: 500)
+
+      expect(transaction.status).to eq("ok")
+      expect(sentry_events.size).to eq(1)
+    end
+
+    it "ignores a missing transaction" do
+      expect { extension.finish_job(transaction: nil, status: 200) }.not_to raise_error
+    end
+  end
+
+  describe "#capture_exception" do
+    def error_extension
+      build_extension(
+        error_data: lambda do |**|
+          {
+            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(build_extension)
+
+      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_extension)
+
+      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 "lets the caller's contexts, tags and extra win over the extension's error data" do
+      event = capture(error_extension, 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 "captures the exception with the caller's data through its own integration when a hook raises" do
+      failing = build_extension(error_data: ->(**) { raise "broken error data" })
+
+      event = capture(failing, tags: { job_id: "caller" })
+
+      expect(event.exception.values.last.mechanism.type).to eq("test_jobs")
+      expect(event.tags).to include(job_id: "caller")
+      expect(string_io.string).to include("broken error data")
+    end
+
+    it "captures the exception with the caller's data when the error data hook returns nil values" do
+      extension = build_extension(error_data: ->(**) { { contexts: { backend: nil }, tags: nil, extra: nil } })
+
+      event = capture(extension, tags: { job_id: "caller" })
+
+      expect(event.exception.values.last.type).to eq("RuntimeError")
+      expect(event.tags).to include(job_id: "caller")
+    end
+
+    it "captures the exception with the caller's data when the error data hook returns something other than a Hash" do
+      event = capture(build_extension(error_data: ->(**) { nil }), tags: { job_id: "caller" })
+
+      expect(event.exception.values.last.type).to eq("RuntimeError")
+      expect(event.tags).to include(job_id: "caller")
+    end
+  end
+end