diff --git a/engine/app/jobs/concerns/coplan/debounced_job.rb b/engine/app/jobs/concerns/coplan/debounced_job.rb
new file mode 100644
index 00000000..a4c95aa0
--- /dev/null
+++ b/engine/app/jobs/concerns/coplan/debounced_job.rb
@@ -0,0 +1,292 @@
+module CoPlan
+  # Coalesces a burst of same-key events into a single delayed job.
+  #
+  # The shape of the problem: something fires an event per occurrence — a
+  # comment created, a plan edited — and a host wants one notification for
+  # the whole burst instead of one per event. Include this in the host's own
+  # ActiveJob class, call `debounce` from the event callback, and implement
+  # `perform_batch`; the first event in a window schedules one delayed job
+  # and later events inside the window are no-ops. When the job finally runs
+  # it is handed the timestamp the batch started at so it can query for
+  # everything since.
+  #
+  #   class DigestJob < ApplicationJob
+  #     include CoPlan::DebouncedJob
+  #
+  #     retry_on Net::OpenTimeout, wait: :polynomially_longer, attempts: 5
+  #     debounces_with window: 2.minutes, retry_horizon: 30.minutes
+  #
+  #     def perform_batch(key:, batch_start:)
+  #       comments = CommentThread.find(key).comments.where(created_at: batch_start..)
+  #       deliver(comments)
+  #     end
+  #   end
+  #
+  #   # in a notification_handler, or a model callback:
+  #   DigestJob.debounce(key: comment.comment_thread_id, event_at: comment.created_at)
+  #
+  # This deliberately knows nothing about comments, threads or delivery
+  # channels. It owns exactly three things — the atomic claim, the batch
+  # boundary, and when the claim is released — because those are the three
+  # things hand-rolled debouncers keep getting wrong.
+  #
+  # Requires a `Rails.cache` store that honours `unless_exist:` on write
+  # (MemoryStore, RedisCacheStore, MemCacheStore, SolidCache all do). A
+  # store that ignores it — notably `:null_store`, the Rails default in the
+  # test environment — degrades to "every event enqueues its own job".
+  module DebouncedJob
+    extend ActiveSupport::Concern
+
+    # How long the first event waits for the rest of its burst.
+    DEFAULT_WINDOW = 1.minute
+
+    # How far past the scheduled run the claim and the batch boundary must
+    # still be readable. This is not the debounce window: it exists to
+    # outlive the *job's own retry backoff*, so a job on its last retry
+    # still finds the boundary it was enqueued with instead of quietly
+    # falling back to a narrower one and dropping the front of its batch.
+    # 30 minutes comfortably clears five attempts of `:polynomially_longer`
+    # (~6 minutes); jobs with a longer retry policy should raise it.
+    DEFAULT_RETRY_HORIZON = 30.minutes
+
+    included do
+      class_attribute :debounce_window, instance_writer: false, default: DEFAULT_WINDOW
+      class_attribute :debounce_retry_horizon, instance_writer: false, default: DEFAULT_RETRY_HORIZON
+    end
+
+    class_methods do
+      # Configures the two durations. They are separate on purpose: the
+      # window is a product decision ("how long do we wait for the rest of
+      # the burst"), the retry horizon is an operational one ("how long can
+      # this job still be in flight"). Tying cache TTLs to the window alone
+      # is what expires batch state out from under a retrying job.
+      def debounces_with(window: nil, retry_horizon: nil)
+        self.debounce_window = validate_duration!(window, :window) if window
+        self.debounce_retry_horizon = validate_duration!(retry_horizon, :retry_horizon) if retry_horizon
+      end
+
+      # TTL for both the claim and the batch boundary. Always longer than
+      # the window by the full retry horizon.
+      def debounce_state_ttl
+        debounce_window + debounce_retry_horizon
+      end
+
+      # Claims a pending slot for `key` and, if this caller won it,
+      # schedules the one delayed job for the burst. Returns true if this
+      # call started a batch, false if a batch was already pending.
+      #
+      # `event_at` is the triggering event's own timestamp — the record's
+      # `created_at`, not `Time.current`. It is required and unforgiving on
+      # purpose. `debounce` runs after the event is already persisted (often
+      # from a callback or another job), so call time is always *later* than
+      # the event; using it as the boundary excludes the very event that
+      # started the batch from the batch's own query.
+      def debounce(key:, event_at:, **arguments)
+        boundary = coerce_event_at(event_at)
+
+        # One atomic write-if-absent is the whole claim. A read-then-write
+        # would let two callers both see "no batch pending" and both
+        # enqueue, delivering the burst twice.
+        claimed = Rails.cache.write(
+          debounce_pending_cache_key(key), true,
+          expires_in: debounce_state_ttl, unless_exist: true
+        )
+
+        unless claimed
+          # Still waiting out its window: the eventual perform_batch query
+          # is "since batch_start" and naturally picks this event up, no
+          # action needed. Already running perform_batch, though, means the
+          # query may have already executed before this event landed —
+          # record it so perform schedules a catch-up run instead of
+          # silently dropping it (see the rearm branch in #perform).
+          mark_dirty(key, boundary) if running?(key)
+          return false
+        end
+
+        # Fold in anything a previous run marked dirty but didn't live long
+        # enough to rearm for itself (see #perform) — release_debounce
+        # deliberately leaves the dirty marker alone so a later claim can
+        # still pick it up instead of the event being lost between the two.
+        leftover = consume_dirty(key)
+        effective_boundary = leftover && leftover < boundary ? leftover : boundary
+
+        Rails.cache.write(debounce_batch_start_cache_key(key), effective_boundary.iso8601(9), expires_in: debounce_state_ttl)
+        schedule_run(key, arguments)
+      end
+
+      # The timestamp the batch for `key` started at, or nil if no batch
+      # state exists.
+      def batch_start_for(key)
+        raw = Rails.cache.read(debounce_batch_start_cache_key(key))
+        raw && Time.zone.parse(raw)
+      end
+
+      def debounce_pending?(key)
+        Rails.cache.exist?(debounce_pending_cache_key(key))
+      end
+
+      # Ends the batch for `key`, so the next event starts a new one.
+      # Called for you after `perform_batch` returns with nothing left
+      # dirty; the ordering matters, see the comment inside. Deliberately
+      # does not touch the dirty marker — see `debounce`'s leftover fold-in.
+      def release_debounce(key)
+        # Boundary first, claim second. The other order leaves a moment
+        # where a new event can win the claim and write its own boundary,
+        # only for this call to delete it a line later.
+        Rails.cache.delete(debounce_batch_start_cache_key(key))
+        Rails.cache.delete(debounce_pending_cache_key(key))
+      end
+
+      # True while a job for `key` is actively inside `perform_batch` — a
+      # narrower window than the claim itself, which also spans the wait.
+      # `debounce` uses this to tell "still waiting, will naturally be
+      # picked up" apart from "query may have already run, could be missed".
+      def running?(key)
+        Rails.cache.exist?(debounce_running_cache_key(key))
+      end
+
+      def mark_running(key)
+        Rails.cache.write(debounce_running_cache_key(key), true, expires_in: debounce_state_ttl)
+      end
+
+      def clear_running(key)
+        Rails.cache.delete(debounce_running_cache_key(key))
+      end
+
+      # Records that an event arrived while perform_batch was running and
+      # may not have made it into that run's query. Keeps the earliest such
+      # event so a second and third arrival during the same run don't
+      # shadow the first.
+      def mark_dirty(key, event_at)
+        cache_key = debounce_dirty_cache_key(key)
+        current = Rails.cache.read(cache_key)
+        earliest = current ? [ Time.zone.parse(current), event_at ].min : event_at
+        Rails.cache.write(cache_key, earliest.iso8601(9), expires_in: debounce_state_ttl)
+      end
+
+      # Returns and clears the dirty boundary, or nil if nothing arrived
+      # mid-run.
+      def consume_dirty(key)
+        cache_key = debounce_dirty_cache_key(key)
+        raw = Rails.cache.read(cache_key)
+        return nil unless raw
+
+        Rails.cache.delete(cache_key)
+        Time.zone.parse(raw)
+      end
+
+      # Re-arms the claim this job already holds for a follow-up run,
+      # instead of releasing and re-claiming. Release-then-reclaim would
+      # open a gap where an unrelated concurrent `debounce` call could win
+      # the fresh claim with its own (later) boundary and skip over the
+      # event that's still unhandled here.
+      def rearm(key, event_at, arguments)
+        Rails.cache.write(debounce_batch_start_cache_key(key), event_at.iso8601(9), expires_in: debounce_state_ttl)
+        schedule_run(key, arguments)
+      end
+
+      def debounce_pending_cache_key(key)
+        "coplan:debounced_job:#{debounce_namespace}:pending:#{key}"
+      end
+
+      def debounce_batch_start_cache_key(key)
+        "coplan:debounced_job:#{debounce_namespace}:batch_start:#{key}"
+      end
+
+      def debounce_running_cache_key(key)
+        "coplan:debounced_job:#{debounce_namespace}:running:#{key}"
+      end
+
+      def debounce_dirty_cache_key(key)
+        "coplan:debounced_job:#{debounce_namespace}:dirty:#{key}"
+      end
+
+      # Namespaces cache keys per job class, so two debounced jobs keyed on
+      # the same record don't fight over one claim.
+      def debounce_namespace
+        name.presence || to_s
+      end
+
+      private
+
+      def coerce_event_at(event_at)
+        unless event_at.respond_to?(:to_time)
+          raise ArgumentError,
+            "#{self}.debounce requires event_at: the triggering event's own timestamp " \
+            "(e.g. record.created_at), got #{event_at.inspect}"
+        end
+
+        event_at.to_time.utc
+      end
+
+      def validate_duration!(value, label)
+        unless value.respond_to?(:to_i) && value.to_i.positive?
+          raise ArgumentError, "#{self}.debounces_with #{label}: must be a positive duration, got #{value.inspect}"
+        end
+
+        value
+      end
+
+      # Releases the claim if the job never actually made it onto the
+      # queue — a raised adapter error, or a halted enqueue callback
+      # returning an unsuccessfully-enqueued job — so a stuck claim doesn't
+      # block every `debounce` call for this key until debounce_state_ttl
+      # expires with no job to show for it.
+      def schedule_run(key, arguments)
+        job = set(wait: debounce_window).perform_later(key: key, **arguments)
+        return true if job&.successfully_enqueued?
+
+        release_debounce(key)
+        false
+      rescue StandardError
+        release_debounce(key)
+        raise
+      end
+    end
+
+    # Including jobs implement `perform_batch`, not `perform` — this owns
+    # `perform` so the claim is always released (or rearmed) in the right
+    # place.
+    def perform(key:, **arguments)
+      self.class.mark_running(key)
+      perform_batch(key: key, batch_start: debounce_batch_start(key), **arguments)
+
+      # An event that arrived while perform_batch was running may already
+      # be too late for the query it just ran — rearm the same claim for a
+      # follow-up instead of releasing, so it isn't silently dropped.
+      if (dirty_at = self.class.consume_dirty(key))
+        self.class.rearm(key, dirty_at, arguments)
+      else
+        self.class.release_debounce(key)
+      end
+    ensure
+      # Cleared unconditionally, including on a raise from perform_batch
+      # (retry_on leaves the claim itself held — see release_debounce not
+      # being called above on that path). A retry re-queries batch_start
+      # fresh, so it naturally covers anything that arrived in the gap;
+      # `running?` only needs to be accurate for the window it's actually
+      # protecting.
+      self.class.clear_running(key)
+    end
+
+    private
+
+    def debounce_batch_start(key)
+      self.class.batch_start_for(key) || missing_batch_start(key)
+    end
+
+    # Should not happen: the boundary outlives the window by the whole
+    # retry horizon. If it does, the state was lost rather than the batch
+    # being empty, so fall back to the widest window we could have retained
+    # and say so loudly. Erring wide risks a duplicate; erring narrow drops
+    # events silently, which is the failure nobody notices.
+    def missing_batch_start(key)
+      Rails.logger.warn(
+        "#{self.class}: batch start for #{key.inspect} expired before the job ran; " \
+        "falling back to #{self.class.debounce_state_ttl.inspect} ago. " \
+        "Raise retry_horizon above this job's retry backoff."
+      )
+      self.class.debounce_state_ttl.ago
+    end
+  end
+end
diff --git a/spec/jobs/coplan/debounced_job_spec.rb b/spec/jobs/coplan/debounced_job_spec.rb
new file mode 100644
index 00000000..7b8f5374
--- /dev/null
+++ b/spec/jobs/coplan/debounced_job_spec.rb
@@ -0,0 +1,391 @@
+require "rails_helper"
+
+# A transient failure, so the test job can exercise its own retry policy.
+class DebouncedJobSpecError < StandardError; end
+
+# Stand-in for a host's job. Deliberately touches nothing but the concern:
+# the primitive is supposed to know nothing about comments or delivery.
+class DebouncedJobSpecJob < CoPlan::ApplicationJob
+  include CoPlan::DebouncedJob
+
+  retry_on DebouncedJobSpecError, wait: 5.minutes, attempts: 3
+
+  debounces_with window: 2.minutes, retry_horizon: 30.minutes
+
+  cattr_accessor :batches, default: []
+  cattr_accessor :failures_remaining, default: 0
+
+  def perform_batch(key:, batch_start:)
+    if self.class.failures_remaining.positive?
+      self.class.failures_remaining -= 1
+      raise DebouncedJobSpecError, "transient"
+    end
+
+    self.class.batches << { key: key, batch_start: batch_start, performed_at: Time.current }
+  end
+end
+
+RSpec.describe CoPlan::DebouncedJob, type: :job do
+  include ActiveJob::TestHelper
+
+  let(:job) { DebouncedJobSpecJob }
+  let(:key) { "thread-abc" }
+
+  # The test environment runs :null_store, which silently ignores
+  # `unless_exist:` — swap in a real store so claims actually claim.
+  around do |example|
+    original_cache = Rails.cache
+    Rails.cache = ActiveSupport::Cache::MemoryStore.new
+    example.run
+    Rails.cache = original_cache
+  end
+
+  before do
+    DebouncedJobSpecJob.batches = []
+    DebouncedJobSpecJob.failures_remaining = 0
+  end
+
+  describe "the batch boundary" do
+    # Failure mode 1: using claim time as the boundary. `debounce` always
+    # runs after the triggering event is persisted, so Time.current is
+    # later than the event's created_at and the batch's own query excludes
+    # the event that started it.
+    it "records the triggering event's timestamp, not the time debounce was called" do
+      freeze_time do
+        event_at = 90.seconds.ago
+
+        expect(job.debounce(key: key, event_at: event_at)).to be(true)
+
+        expect(job.batch_start_for(key)).to eq(event_at)
+        expect(job.batch_start_for(key)).to be < Time.current
+      end
+    end
+
+    it "hands that same timestamp to perform_batch, so the triggering event is inside its own batch" do
+      freeze_time do
+        event_at = 90.seconds.ago
+        job.debounce(key: key, event_at: event_at)
+
+        travel 2.minutes
+        job.perform_now(key: key)
+
+        batch = DebouncedJobSpecJob.batches.sole
+        expect(batch[:batch_start]).to eq(event_at)
+        expect(batch[:batch_start]).to be < batch[:performed_at]
+      end
+    end
+
+    it "refuses to guess a boundary when event_at is missing" do
+      expect { job.debounce(key: key, event_at: nil) }
+        .to raise_error(ArgumentError, /triggering event's own timestamp/)
+      expect(enqueued_jobs).to be_empty
+    end
+
+    it "keeps the first event's boundary when later events land inside the window" do
+      freeze_time do
+        first_at = 10.seconds.ago
+        job.debounce(key: key, event_at: first_at)
+
+        travel 30.seconds
+        expect(job.debounce(key: key, event_at: Time.current)).to be(false)
+
+        expect(job.batch_start_for(key)).to eq(first_at)
+      end
+    end
+  end
+
+  # Failure mode 2: a read-then-write claim. Two workers both read "no batch
+  # pending", both write, both enqueue — the burst is delivered twice. The
+  # threaded example below is the end-to-end assertion; the two after it pin
+  # the structure, since MRI rarely interleaves a read-then-write on demand.
+  describe "claiming the batch" do
+    it "lets exactly one of many concurrent callers win the claim" do
+      event_at = 1.minute.ago
+      results = []
+      mutex = Mutex.new
+      start_gate = Queue.new
+
+      threads = 25.times.map do
+        Thread.new do
+          start_gate.pop # all 25 launch together, into the same claim
+          won = job.debounce(key: key, event_at: event_at)
+          mutex.synchronize { results << won }
+        end
+      end
+      25.times { start_gate << :go }
+      threads.each(&:join)
+
+      expect(results.count(true)).to eq(1)
+      expect(results.size).to eq(25)
+      expect(enqueued_jobs.count { |j| j["job_class"] == "DebouncedJobSpecJob" }).to eq(1)
+    end
+
+    it "claims with a write-if-absent and never reads the pending key to decide" do
+      allow(Rails.cache).to receive(:read).and_call_original
+      allow(Rails.cache).to receive(:write).and_call_original
+
+      job.debounce(key: key, event_at: 1.minute.ago)
+
+      # The claim itself never reads: the atomic write's return value is the
+      # only input. (A read of the *dirty* key does happen here, to fold in
+      # anything left over from a previous run — see "leftover dirty state"
+      # below — but that's unrelated to who wins the claim.)
+      expect(Rails.cache).not_to have_received(:read).with(job.debounce_pending_cache_key(key))
+      expect(Rails.cache).to have_received(:write).with(
+        job.debounce_pending_cache_key(key), true, hash_including(unless_exist: true)
+      ).once
+    end
+
+    # The atomic write's return value is the only thing that decides who
+    # owns the batch. A read-then-write claim would look at the empty cache
+    # here, conclude it had won, and enqueue a duplicate.
+    it "loses the claim purely on the write's return value, not on cache contents" do
+      pending_key = job.debounce_pending_cache_key(key)
+      allow(Rails.cache).to receive(:write).and_call_original
+      allow(Rails.cache).to receive(:write)
+        .with(pending_key, true, hash_including(unless_exist: true)).and_return(false)
+
+      expect(Rails.cache.read(pending_key)).to be_nil
+      expect(job.debounce(key: key, event_at: 1.minute.ago)).to be(false)
+
+      expect(enqueued_jobs).to be_empty
+      expect(job.batch_start_for(key)).to be_nil
+    end
+
+    it "enqueues one delayed job for the burst and no-ops for the rest" do
+      freeze_time do
+        expect {
+          job.debounce(key: key, event_at: Time.current)
+        }.to have_enqueued_job(DebouncedJobSpecJob).with(key: key).at(2.minutes.from_now)
+
+        expect {
+          3.times { job.debounce(key: key, event_at: Time.current) }
+        }.not_to have_enqueued_job(DebouncedJobSpecJob)
+      end
+    end
+
+    it "keys claims per job class and per key" do
+      job.debounce(key: "one", event_at: 1.minute.ago)
+
+      expect(job.debounce_pending?("one")).to be(true)
+      expect(job.debounce_pending?("two")).to be(false)
+      expect(job.debounce_pending_cache_key("one")).to include("DebouncedJobSpecJob")
+    end
+  end
+
+  describe "state lifetime" do
+    # Failure mode 3: TTLs tied to the debounce window. A job that retries
+    # for minutes finds its batch state already gone and silently narrows
+    # its query, dropping the front of the batch.
+    it "sizes state TTL as the window plus the full retry horizon" do
+      expect(job.debounce_state_ttl).to eq(32.minutes)
+      expect(job.debounce_state_ttl).to be > job.debounce_window
+      expect(job.debounce_state_ttl).to be > job.debounce_retry_horizon
+    end
+
+    it "keeps the boundary and the claim readable long past the debounce window" do
+      freeze_time do
+        event_at = Time.current
+        job.debounce(key: key, event_at: event_at)
+
+        travel 20.minutes # far past the 2-minute window, inside the retry horizon
+
+        expect(job.batch_start_for(key)).to eq(event_at)
+        expect(job.debounce_pending?(key)).to be(true)
+      end
+    end
+
+    it "holds the claim across a retry and gives the retry the original boundary" do
+      freeze_time do
+        event_at = 30.seconds.ago
+        job.debounce(key: key, event_at: event_at)
+        DebouncedJobSpecJob.failures_remaining = 1
+
+        travel 2.minutes
+        job.perform_now(key: key) # raises inside perform_batch, retry_on re-enqueues
+
+        expect(DebouncedJobSpecJob.batches).to be_empty
+        expect(job.debounce_pending?(key)).to be(true)
+
+        travel 20.minutes # the retry finally runs, way past the window
+        job.perform_now(key: key)
+
+        expect(DebouncedJobSpecJob.batches.sole[:batch_start]).to eq(event_at)
+      end
+    end
+
+    it "warns and falls back to the widest retained window if state is somehow lost" do
+      freeze_time do
+        job.debounce(key: key, event_at: Time.current)
+
+        travel 33.minutes # past state TTL entirely
+        expect(job.batch_start_for(key)).to be_nil
+
+        allow(Rails.logger).to receive(:warn)
+        job.perform_now(key: key)
+
+        expect(Rails.logger).to have_received(:warn).with(/expired before the job ran/)
+        expect(DebouncedJobSpecJob.batches.sole[:batch_start]).to eq(32.minutes.ago)
+      end
+    end
+  end
+
+  describe "releasing the claim" do
+    it "releases only after perform_batch completes, so a mid-flight event cannot open a second batch" do
+      freeze_time do
+        job.debounce(key: key, event_at: Time.current)
+
+        observed = nil
+        allow_any_instance_of(DebouncedJobSpecJob).to receive(:perform_batch) do
+          observed = job.debounce_pending?(key)
+        end
+
+        job.perform_now(key: key)
+
+        expect(observed).to be(true)
+        expect(job.debounce_pending?(key)).to be(false)
+      end
+    end
+
+    it "leaves the claim held when the batch raises" do
+      allow_any_instance_of(DebouncedJobSpecJob).to receive(:perform_batch).and_raise("delivery is down")
+
+      job.debounce(key: key, event_at: 1.minute.ago)
+      expect { job.perform_now(key: key) }.to raise_error("delivery is down")
+
+      expect(job.debounce_pending?(key)).to be(true)
+      expect(job.batch_start_for(key)).to be_present
+    end
+
+    it "lets the next event start a fresh batch once the previous one finished" do
+      freeze_time do
+        first_at = Time.current
+        job.debounce(key: key, event_at: first_at)
+        travel 2.minutes
+        job.perform_now(key: key)
+
+        travel 1.minute
+        later_at = Time.current
+        expect(job.debounce(key: key, event_at: later_at)).to be(true)
+        expect(job.batch_start_for(key)).to eq(later_at)
+      end
+    end
+  end
+
+  # Failure mode 4: an event lands after perform_batch has already queried
+  # but before the claim is released. The claim being held makes a plain
+  # `debounce` call at that moment a no-op, and the query that already ran
+  # can't retroactively include it — without a rearm, it's gone for good.
+  describe "events arriving mid-run" do
+    it "rearms a follow-up run for an event that arrives while perform_batch is executing" do
+      freeze_time do
+        job.debounce(key: key, event_at: 30.seconds.ago)
+        travel 2.minutes
+
+        late_arrival = Time.current
+        allow_any_instance_of(DebouncedJobSpecJob).to receive(:perform_batch) do
+          # Simulates a comment landing after this job's own query already ran.
+          expect(job.debounce(key: key, event_at: late_arrival)).to be(false)
+        end
+
+        expect {
+          job.perform_now(key: key)
+        }.to have_enqueued_job(DebouncedJobSpecJob).with(key: key)
+
+        expect(job.debounce_pending?(key)).to be(true)
+        expect(job.batch_start_for(key)).to eq(late_arrival)
+      end
+    end
+
+    it "does not rearm when nothing arrives while perform_batch is executing" do
+      freeze_time do
+        job.debounce(key: key, event_at: Time.current)
+        travel 2.minutes
+
+        job.perform_now(key: key)
+
+        expect(job.debounce_pending?(key)).to be(false)
+        # Only the original debounce's own enqueue — no follow-up rearm.
+        expect(enqueued_jobs.count { |j| j["job_class"] == "DebouncedJobSpecJob" }).to eq(1)
+      end
+    end
+
+    it "keeps the earliest of several mid-run arrivals as the follow-up boundary" do
+      freeze_time do
+        job.debounce(key: key, event_at: Time.current)
+        travel 2.minutes
+
+        earlier = Time.current
+        later = Time.current + 1.second
+        allow_any_instance_of(DebouncedJobSpecJob).to receive(:perform_batch) do
+          job.debounce(key: key, event_at: later)
+          job.debounce(key: key, event_at: earlier)
+        end
+
+        job.perform_now(key: key)
+
+        expect(job.batch_start_for(key)).to eq(earlier)
+      end
+    end
+
+    it "does not treat an event still waiting out the window as mid-run" do
+      freeze_time do
+        job.debounce(key: key, event_at: Time.current)
+
+        travel 10.seconds
+        job.debounce(key: key, event_at: Time.current) # still waiting, perform hasn't started
+
+        travel 2.minutes
+        job.perform_now(key: key)
+
+        expect(job.debounce_pending?(key)).to be(false) # released normally, no rearm needed
+      end
+    end
+  end
+
+  # Failure mode 5: `debounce` writes the claim, then perform_later fails
+  # (adapter error, or a halted enqueue callback) before actually scheduling
+  # anything. Without cleanup, every event for this key silently no-ops
+  # until debounce_state_ttl expires, even though no job exists to run them.
+  describe "when enqueueing fails" do
+    it "releases the claim and re-raises when the queue adapter raises" do
+      configured_job = double("configured_job")
+      allow(job).to receive(:set).and_return(configured_job)
+      allow(configured_job).to receive(:perform_later).and_raise(StandardError, "queue unavailable")
+
+      expect {
+        job.debounce(key: key, event_at: 1.minute.ago)
+      }.to raise_error(StandardError, "queue unavailable")
+
+      expect(job.debounce_pending?(key)).to be(false)
+      expect(job.batch_start_for(key)).to be_nil
+    end
+
+    it "releases the claim when perform_later returns a job that wasn't actually enqueued" do
+      configured_job = double("configured_job")
+      allow(job).to receive(:set).and_return(configured_job)
+      allow(configured_job).to receive(:perform_later).and_return(instance_double(DebouncedJobSpecJob, successfully_enqueued?: false))
+
+      expect(job.debounce(key: key, event_at: 1.minute.ago)).to be(false)
+
+      expect(job.debounce_pending?(key)).to be(false)
+      expect(job.batch_start_for(key)).to be_nil
+    end
+  end
+
+  describe "configuration" do
+    it "defaults to the concern's window and retry horizon" do
+      default_job = Class.new(CoPlan::ApplicationJob) { include CoPlan::DebouncedJob }
+
+      expect(default_job.debounce_window).to eq(CoPlan::DebouncedJob::DEFAULT_WINDOW)
+      expect(default_job.debounce_retry_horizon).to eq(CoPlan::DebouncedJob::DEFAULT_RETRY_HORIZON)
+    end
+
+    it "rejects a non-positive window or retry horizon" do
+      klass = Class.new(CoPlan::ApplicationJob) { include CoPlan::DebouncedJob }
+
+      expect { klass.debounces_with(window: -1.minute) }.to raise_error(ArgumentError, /positive duration/)
+      expect { klass.debounces_with(retry_horizon: "soon") }.to raise_error(ArgumentError, /positive duration/)
+    end
+  end
+end