From fe18dd84844f82b33abb5540c202abbef28bd9c4 Mon Sep 17 00:00:00 2001
From: Hampton Lintorn-Catlin <hampton@squareup.com>
Date: Thu, 27 Aug 2026 14:16:08 -0500
Subject: [PATCH 1/2] Notify a thread's participants (not just its opener) on
 every new comment

Every comment now fires the comment_created event, so a reply notifies
everyone already in the thread, not only the plan owner on the first
comment. SlackNotificationJob is debounced per-thread so a burst of
comments (e.g. an agent posting several in a row) collapses into one
Slack DM per recipient instead of one per comment, and a permanent
failure for one recipient no longer blocks the rest of the batch.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 app/jobs/slack_notification_job.rb           |  86 ++++++++++---
 config/initializers/coplan.rb                |   4 +-
 engine/app/models/coplan/comment.rb          |   4 +-
 spec/jobs/slack_notification_job_spec.rb     | 125 +++++++++++++++----
 spec/models/comment_spec.rb                  |  23 +---
 spec/models/coplan/comment_analytics_spec.rb |   4 +-
 6 files changed, 181 insertions(+), 65 deletions(-)

diff --git a/app/jobs/slack_notification_job.rb b/app/jobs/slack_notification_job.rb
index 63f1c62e..f54c6011 100644
--- a/app/jobs/slack_notification_job.rb
+++ b/app/jobs/slack_notification_job.rb
@@ -2,44 +2,94 @@ class SlackNotificationJob < ApplicationJob
   queue_as :default
 
   retry_on SlackClient::Error, wait: :polynomially_longer, attempts: 5
-  discard_on SlackClient::PermanentError
+
+  DEBOUNCE_WINDOW = 2.minutes
+  CACHE_EXPIRY = DEBOUNCE_WINDOW + 1.minute
+
+  # Coalesces a burst of comments on the same thread (e.g. an agent posting
+  # several in a row) into a single delayed job. The first call in a window
+  # records where the batch starts and schedules the send; later calls
+  # within the window are no-ops because a send is already scheduled — the
+  # eventual perform picks up everything created since the batch started.
+  def self.debounce(comment_thread_id:)
+    pending_key = pending_key(comment_thread_id)
+    return if Rails.cache.read(pending_key)
+
+    Rails.cache.write(pending_key, true, expires_in: CACHE_EXPIRY)
+    Rails.cache.write(batch_start_key(comment_thread_id), Time.current, expires_in: CACHE_EXPIRY)
+    set(wait: DEBOUNCE_WINDOW).perform_later(comment_thread_id: comment_thread_id)
+  end
+
+  def self.pending_key(comment_thread_id)
+    "slack_notification_job:pending:#{comment_thread_id}"
+  end
+
+  def self.batch_start_key(comment_thread_id)
+    "slack_notification_job:batch_start:#{comment_thread_id}"
+  end
 
   def perform(comment_thread_id:)
+    Rails.cache.delete(self.class.pending_key(comment_thread_id))
     return unless SlackClient.configured?
 
-    thread = CoPlan::CommentThread.find(comment_thread_id)
-    plan = thread.plan
-    coplan_author = plan.created_by_user
-    first_comment = thread.comments.order(:created_at, :id).first
+    thread = CoPlan::CommentThread.find_by(id: comment_thread_id)
+    return unless thread
 
-    return unless first_comment
-    return if first_comment.author_type == "human" && first_comment.author_id == coplan_author.id
+    batch_start = Rails.cache.read(self.class.batch_start_key(comment_thread_id)) || DEBOUNCE_WINDOW.ago
+    new_comments = thread.comments.kept.where("coplan_comments.created_at >= ?", batch_start).order(:created_at, :id).to_a
+    return if new_comments.empty?
 
-    return unless coplan_author.email.present?
+    plan = thread.plan
+    recipients = recipients_for(thread, plan, new_comments)
+    return if recipients.empty?
 
-    text = compose_message(thread, plan)
-    SlackClient.send_dm(email: coplan_author.email, text: text)
+    text = compose_message(thread, plan, new_comments)
+    recipients.each do |user|
+      next unless user.email.present?
+      send_dm(user, text)
+    end
   end
 
   private
 
-  def compose_message(thread, plan)
-    comment_body = first_comment_body(thread).truncate(300)
+  # Everyone who's part of the conversation: the plan owner, plus anyone
+  # who's commented in the thread. A participant is skipped only if every
+  # comment in this batch is their own — if someone else said something
+  # new, they still hear about it even if they also commented in the batch.
+  def recipients_for(thread, plan, new_comments)
+    participant_ids = thread.comments.where(author_type: %w[human local_agent]).distinct.pluck(:author_id)
+    participant_ids = (participant_ids + [ plan.created_by_user_id ]).uniq
+
+    users_by_id = CoPlan::User.where(id: participant_ids).index_by(&:id)
+    participant_ids.filter_map do |user_id|
+      next unless new_comments.any? { |c| c.author_id != user_id }
+      users_by_id[user_id]
+    end
+  end
+
+  # A permanent failure for one recipient (e.g. no Slack account for their
+  # email) shouldn't stop the rest of the batch from being notified.
+  def send_dm(user, text)
+    SlackClient.send_dm(email: user.email, text: text)
+  rescue SlackClient::PermanentError => e
+    Rails.logger.warn("[SlackNotificationJob] skipping #{user.id}: #{e.message}")
+  end
+
+  def compose_message(thread, plan, new_comments)
     plan_url = CoPlan::Engine.routes.url_helpers.plan_url(plan, **default_url_options)
+    latest_body = (new_comments.last.body_markdown || "").truncate(300)
 
-    lines = [ "New comment on *#{plan.title}*:" ]
+    lines = [
+      new_comments.size == 1 ? "New comment on *#{plan.title}*:" : "#{new_comments.size} new comments on *#{plan.title}*:"
+    ]
     if thread.anchor_text.present?
       lines << "> _#{thread.anchor_text.truncate(120)}_"
     end
-    lines << "> #{comment_body}"
+    lines << "> #{latest_body}"
     lines << plan_url
     lines.join("\n")
   end
 
-  def first_comment_body(thread)
-    thread.comments.order(:created_at, :id).first&.body_markdown || ""
-  end
-
   def default_url_options
     Rails.application.config.action_mailer.default_url_options || { host: "localhost", port: 3000 }
   end
diff --git a/config/initializers/coplan.rb b/config/initializers/coplan.rb
index 7c1b2f82..8e6d9e29 100644
--- a/config/initializers/coplan.rb
+++ b/config/initializers/coplan.rb
@@ -43,7 +43,9 @@
   config.notification_handler = ->(event, payload) {
     case event
     when :comment_created
-      SlackNotificationJob.perform_later(comment_thread_id: payload[:comment_thread_id])
+      # Debounced: an agent posting several comments in a row should produce
+      # one Slack DM per recipient, not one per comment.
+      SlackNotificationJob.debounce(comment_thread_id: payload[:comment_thread_id])
     end
   }
 
diff --git a/engine/app/models/coplan/comment.rb b/engine/app/models/coplan/comment.rb
index b15d3101..4d151d7d 100644
--- a/engine/app/models/coplan/comment.rb
+++ b/engine/app/models/coplan/comment.rb
@@ -11,7 +11,7 @@ class Comment < ApplicationRecord
     validates :agent_name, length: { maximum: 20 }, allow_nil: true
 
     before_save :rewrite_plain_mentions, if: :body_markdown_changed?
-    after_create_commit :notify_plan_author, if: :first_comment_in_thread?
+    after_create_commit :notify_thread_participants
     after_create_commit :track_comment_created
     # Runs on save (not just create) so adding a mention via edit also
     # notifies. ProcessMentions uses find_or_create_by to dedupe.
@@ -54,7 +54,7 @@ def first_comment_in_thread?
       @first_comment_in_thread = comment_thread.comments.count == 1
     end
 
-    def notify_plan_author
+    def notify_thread_participants
       CoPlan::NotificationJob.perform_later("comment_created", { comment_thread_id: comment_thread_id })
     end
 
diff --git a/spec/jobs/slack_notification_job_spec.rb b/spec/jobs/slack_notification_job_spec.rb
index 9d965650..edf7b5f9 100644
--- a/spec/jobs/slack_notification_job_spec.rb
+++ b/spec/jobs/slack_notification_job_spec.rb
@@ -2,18 +2,19 @@
 
 RSpec.describe SlackNotificationJob, type: :job do
   let(:plan_author) { create(:coplan_user, email: "author@example.com") }
-  let(:commenter) { create(:coplan_user) }
+  let(:commenter) { create(:coplan_user, email: "commenter@example.com") }
   let(:plan) { create(:plan, created_by_user: plan_author) }
 
   let(:thread_record) do
     create(:comment_thread, plan: plan,
       plan_version: plan.current_plan_version, created_by_user: commenter)
   end
-  let!(:first_comment) do
-    thread_record.comments.create!(
-      author_type: "human",
-      author_id: commenter.id, body_markdown: "A comment body."
-    )
+
+  around do |example|
+    original_cache = Rails.cache
+    Rails.cache = ActiveSupport::Cache::MemoryStore.new
+    example.run
+    Rails.cache = original_cache
   end
 
   before do
@@ -22,7 +23,14 @@
   end
 
   describe "#perform" do
-    it "sends a DM to the plan author" do
+    let!(:first_comment) do
+      thread_record.comments.create!(
+        author_type: "human",
+        author_id: commenter.id, body_markdown: "A comment body."
+      )
+    end
+
+    it "sends a DM to the plan author for the thread's first comment" do
       described_class.perform_now(comment_thread_id: thread_record.id)
 
       expect(SlackClient).to have_received(:send_dm).with(
@@ -42,15 +50,6 @@
       )
     end
 
-    it "includes first comment body in the message" do
-      described_class.perform_now(comment_thread_id: thread_record.id)
-
-      expect(SlackClient).to have_received(:send_dm).with(
-        email: plan_author.email,
-        text: a_string_including("A comment body.")
-      )
-    end
-
     it "skips notification when first comment author is the plan author" do
       first_comment.update_columns(author_id: plan_author.id)
 
@@ -67,7 +66,7 @@
       expect(SlackClient).to have_received(:send_dm)
     end
 
-    it "skips notification when thread has no comments" do
+    it "skips notification when the thread has no comments in the batch window" do
       first_comment.destroy!
 
       described_class.perform_now(comment_thread_id: thread_record.id)
@@ -82,26 +81,102 @@
 
       expect(SlackClient).not_to have_received(:send_dm)
     end
+  end
+
+  describe "replies" do
+    let(:other_participant) { create(:coplan_user, email: "other@example.com") }
+
+    before do
+      # An hour-old comment, clearly outside this batch — simulates it
+      # having already been notified about in an earlier run.
+      travel_to 1.hour.ago do
+        thread_record.comments.create!(
+          author_type: "human", author_id: commenter.id, body_markdown: "First comment."
+        )
+      end
+      Rails.cache.write(described_class.batch_start_key(thread_record.id), 30.minutes.ago)
+    end
+
+    it "notifies the plan author and prior participants, excluding the replier" do
+      thread_record.comments.create!(
+        author_type: "human", author_id: other_participant.id, body_markdown: "Someone else chimes in."
+      )
+
+      described_class.perform_now(comment_thread_id: thread_record.id)
+
+      expect(SlackClient).to have_received(:send_dm).with(email: plan_author.email, text: anything)
+      expect(SlackClient).to have_received(:send_dm).with(email: commenter.email, text: anything)
+      expect(SlackClient).not_to have_received(:send_dm).with(email: other_participant.email, text: anything)
+    end
+
+    it "does not notify the replier about their own reply" do
+      thread_record.comments.create!(
+        author_type: "human", author_id: commenter.id, body_markdown: "Following up on my own thread."
+      )
+
+      described_class.perform_now(comment_thread_id: thread_record.id)
+
+      expect(SlackClient).not_to have_received(:send_dm).with(email: commenter.email, text: anything)
+    end
+  end
+
+  describe ".debounce" do
+    it "schedules exactly one delayed job for a burst of comments on the same thread" do
+      expect {
+        3.times { described_class.debounce(comment_thread_id: thread_record.id) }
+      }.to have_enqueued_job(described_class).exactly(1).times.on_queue("default")
+    end
+
+    it "coalesces every comment created during the debounce window into one message per recipient" do
+      Rails.cache.write(described_class.batch_start_key(thread_record.id), 30.minutes.ago)
+      other_participant = create(:coplan_user)
+      thread_record.comments.create!(author_type: "human", author_id: other_participant.id, body_markdown: "First.")
+      thread_record.comments.create!(author_type: "human", author_id: other_participant.id, body_markdown: "Second.")
+      thread_record.comments.create!(author_type: "human", author_id: other_participant.id, body_markdown: "Third.")
+
+      described_class.perform_now(comment_thread_id: thread_record.id)
+
+      expect(SlackClient).to have_received(:send_dm).with(
+        email: plan_author.email,
+        text: a_string_including("3 new comments").and(a_string_including("Third."))
+      ).once
+    end
+
+    it "allows a new burst after the previous one has been sent" do
+      thread_record.comments.create!(author_type: "human", author_id: commenter.id, body_markdown: "First.")
+
+      described_class.debounce(comment_thread_id: thread_record.id)
+      described_class.perform_now(comment_thread_id: thread_record.id)
 
-    it "enqueues on the default queue" do
-      thread_record # eagerly create to avoid callback enqueue inside expect block
       expect {
-        described_class.perform_later(comment_thread_id: thread_record.id)
+        described_class.debounce(comment_thread_id: thread_record.id)
       }.to have_enqueued_job(described_class).on_queue("default")
     end
+  end
 
-    it "discards permanent Slack errors without retrying" do
-      allow(SlackClient).to receive(:send_dm).and_raise(SlackClient::PermanentError, "users_not_found")
+  describe "error handling" do
+    let!(:first_comment) do
+      thread_record.comments.create!(
+        author_type: "human", author_id: commenter.id, body_markdown: "First comment."
+      )
+    end
+
+    it "skips a recipient with a permanent Slack error but keeps notifying others" do
+      other_participant = create(:coplan_user, email: "other@example.com")
+      thread_record.comments.create!(
+        author_type: "human", author_id: other_participant.id, body_markdown: "Reply."
+      )
+      allow(SlackClient).to receive(:send_dm).with(email: plan_author.email, text: anything)
+        .and_raise(SlackClient::PermanentError, "users_not_found")
 
       expect {
         described_class.perform_now(comment_thread_id: thread_record.id)
       }.not_to raise_error
 
-      expect(SlackClient).to have_received(:send_dm).once
+      expect(SlackClient).to have_received(:send_dm).with(email: commenter.email, text: anything)
     end
 
-    it "retries on transient Slack errors" do
-      thread_record # eagerly create
+    it "retries the whole batch on a transient Slack error" do
       allow(SlackClient).to receive(:send_dm).and_raise(SlackClient::Error, "ratelimited")
 
       expect {
diff --git a/spec/models/comment_spec.rb b/spec/models/comment_spec.rb
index 8454c546..ca9646dd 100644
--- a/spec/models/comment_spec.rb
+++ b/spec/models/comment_spec.rb
@@ -52,27 +52,16 @@
       }.to have_enqueued_job(CoPlan::NotificationJob)
     end
 
-    it "does not enqueue NotificationJob for subsequent comments" do
+    # Every comment fires the event now, not just the thread opener — a
+    # reply should notify anyone already in the thread. It's up to the
+    # notification_handler (Slack DM job, debounced) to decide who actually
+    # hears about it and to coalesce a burst into one message.
+    it "enqueues NotificationJob for subsequent comments too" do
       create(:comment, comment_thread: thread_record)
 
       expect {
         create(:comment, comment_thread: thread_record)
-      }.not_to have_enqueued_job(CoPlan::NotificationJob)
-    end
-
-    # Regression for COPLAN-30: `first_comment_in_thread?` used to compare
-    # UUIDs with `id < ?`, which is not insertion-ordered. A reply whose UUID
-    # sorts before the opener's was wrongly treated as the thread opener,
-    # firing a duplicate "new comment thread" notification for the plan author.
-    it "does not enqueue NotificationJob for a reply whose UUID sorts before the opener" do
-      high_id = "ffffffff-ffff-ffff-ffff-ffffffffffff"
-      low_id  = "00000000-0000-0000-0000-000000000001"
-
-      create(:comment, comment_thread: thread_record, id: high_id)
-
-      expect {
-        create(:comment, comment_thread: thread_record, id: low_id)
-      }.not_to have_enqueued_job(CoPlan::NotificationJob)
+      }.to have_enqueued_job(CoPlan::NotificationJob)
     end
   end
 
diff --git a/spec/models/coplan/comment_analytics_spec.rb b/spec/models/coplan/comment_analytics_spec.rb
index fda5efec..9b1ebac0 100644
--- a/spec/models/coplan/comment_analytics_spec.rb
+++ b/spec/models/coplan/comment_analytics_spec.rb
@@ -43,8 +43,8 @@
     expect(payload[:properties][:is_first_in_thread]).to be(false)
   end
 
-  # `first_comment_in_thread?` (used by notify_plan_author) compares UUID
-  # strings with `id < ?`, which is not insertion-ordered. The analytics
+  # `first_comment_in_thread?` (used by notify_thread_participants) compares
+  # UUID strings with `id < ?`, which is not insertion-ordered. The analytics
   # path uses a total-count check instead, so a reply whose UUID happens
   # to sort before the existing first comment still records is_first=false.
   it "is not fooled by a reply whose UUID sorts before earlier comments" do

From 16705adb51856f8a17816651c4fd7043fcdb6832 Mon Sep 17 00:00:00 2001
From: Hampton Lintorn-Catlin <hampton@squareup.com>
Date: Thu, 27 Aug 2026 14:50:41 -0500
Subject: [PATCH 2/2] Fix batch_start timing and duplicate-DM bugs surfaced by
 review

- batch_start was Time.current at debounce-call time, which is always
  after the triggering comment's own created_at (the comment must
  exist to have enqueued the job that calls debounce). That silently
  excluded the triggering comment from its own notification. Now
  passed through from the comment's created_at instead.
- Debounce's pending-flag check-and-set was read-then-write, not
  atomic; two concurrent calls could both schedule a job. Use
  Rails.cache's unless_exist: true instead.
- The pending flag was cleared at the top of perform, before any work
  happened, so a comment arriving mid-retry could start an overlapping
  batch that clobbered this one's cached batch_start. Now only cleared
  after a full successful send.
- A transient error partway through the recipient loop caused the
  whole job to retry, re-sending to recipients who'd already
  succeeded. Track already-notified recipients in cache and skip them
  on retry.
- Align soft-delete scoping: recipients_for now uses .kept like the
  new_comments query, so a participant whose only comment was deleted
  doesn't stay a permanent participant asymmetrically.
---
 app/jobs/slack_notification_job.rb       | 48 ++++++++++++++++++++----
 config/initializers/coplan.rb            |  5 ++-
 engine/app/models/coplan/comment.rb      |  7 +++-
 spec/jobs/slack_notification_job_spec.rb | 43 +++++++++++++++++++++
 4 files changed, 93 insertions(+), 10 deletions(-)

diff --git a/app/jobs/slack_notification_job.rb b/app/jobs/slack_notification_job.rb
index f54c6011..e8925513 100644
--- a/app/jobs/slack_notification_job.rb
+++ b/app/jobs/slack_notification_job.rb
@@ -5,18 +5,30 @@ class SlackNotificationJob < ApplicationJob
 
   DEBOUNCE_WINDOW = 2.minutes
   CACHE_EXPIRY = DEBOUNCE_WINDOW + 1.minute
+  # retry_on below can span several minutes across its 5 attempts of
+  # polynomially-longer backoff — give batch/notified state room to outlive
+  # every retry, not just the debounce wait, or a late retry falls back to
+  # the wrong window or forgets who it already notified.
+  RETRY_STATE_EXPIRY = 30.minutes
 
   # Coalesces a burst of comments on the same thread (e.g. an agent posting
   # several in a row) into a single delayed job. The first call in a window
   # records where the batch starts and schedules the send; later calls
   # within the window are no-ops because a send is already scheduled — the
   # eventual perform picks up everything created since the batch started.
-  def self.debounce(comment_thread_id:)
+  #
+  # batch_start is the triggering comment's own created_at, not the dispatch
+  # time this method runs at — this method is always called from a job
+  # that's already been enqueued and dequeued for a comment that already
+  # exists, so Time.current here would be later than that comment's
+  # created_at and silently exclude it from its own notification.
+  def self.debounce(comment_thread_id:, comment_created_at: nil)
     pending_key = pending_key(comment_thread_id)
-    return if Rails.cache.read(pending_key)
+    started = Rails.cache.write(pending_key, true, expires_in: CACHE_EXPIRY, unless_exist: true)
+    return unless started
 
-    Rails.cache.write(pending_key, true, expires_in: CACHE_EXPIRY)
-    Rails.cache.write(batch_start_key(comment_thread_id), Time.current, expires_in: CACHE_EXPIRY)
+    Rails.cache.write(batch_start_key(comment_thread_id), comment_created_at, expires_in: RETRY_STATE_EXPIRY)
+    Rails.cache.delete(notified_key(comment_thread_id))
     set(wait: DEBOUNCE_WINDOW).perform_later(comment_thread_id: comment_thread_id)
   end
 
@@ -28,8 +40,22 @@ def self.batch_start_key(comment_thread_id)
     "slack_notification_job:batch_start:#{comment_thread_id}"
   end
 
+  def self.notified_key(comment_thread_id)
+    "slack_notification_job:notified:#{comment_thread_id}"
+  end
+
   def perform(comment_thread_id:)
+    send_batch(comment_thread_id)
+    # Only release the pending flag once the batch fully sends. A transient
+    # error (see retry_on above) skips this line, keeping the flag set so a
+    # comment arriving mid-retry doesn't start a second, overlapping batch
+    # that clobbers this one's batch_start/notified cache state.
     Rails.cache.delete(self.class.pending_key(comment_thread_id))
+  end
+
+  private
+
+  def send_batch(comment_thread_id)
     return unless SlackClient.configured?
 
     thread = CoPlan::CommentThread.find_by(id: comment_thread_id)
@@ -44,20 +70,28 @@ def perform(comment_thread_id:)
     return if recipients.empty?
 
     text = compose_message(thread, plan, new_comments)
+    notified_key = self.class.notified_key(comment_thread_id)
+    already_notified = Rails.cache.read(notified_key) || []
+
+    # A transient error retries the whole job (see retry_on above); track who
+    # already got a DM this batch so a retry doesn't double-notify recipients
+    # that succeeded before the recipient that raised.
     recipients.each do |user|
       next unless user.email.present?
+      next if already_notified.include?(user.id)
+
       send_dm(user, text)
+      already_notified << user.id
+      Rails.cache.write(notified_key, already_notified, expires_in: RETRY_STATE_EXPIRY)
     end
   end
 
-  private
-
   # Everyone who's part of the conversation: the plan owner, plus anyone
   # who's commented in the thread. A participant is skipped only if every
   # comment in this batch is their own — if someone else said something
   # new, they still hear about it even if they also commented in the batch.
   def recipients_for(thread, plan, new_comments)
-    participant_ids = thread.comments.where(author_type: %w[human local_agent]).distinct.pluck(:author_id)
+    participant_ids = thread.comments.kept.where(author_type: %w[human local_agent]).distinct.pluck(:author_id)
     participant_ids = (participant_ids + [ plan.created_by_user_id ]).uniq
 
     users_by_id = CoPlan::User.where(id: participant_ids).index_by(&:id)
diff --git a/config/initializers/coplan.rb b/config/initializers/coplan.rb
index 8e6d9e29..de7f8df7 100644
--- a/config/initializers/coplan.rb
+++ b/config/initializers/coplan.rb
@@ -45,7 +45,10 @@
     when :comment_created
       # Debounced: an agent posting several comments in a row should produce
       # one Slack DM per recipient, not one per comment.
-      SlackNotificationJob.debounce(comment_thread_id: payload[:comment_thread_id])
+      SlackNotificationJob.debounce(
+        comment_thread_id: payload[:comment_thread_id],
+        comment_created_at: payload[:comment_created_at]
+      )
     end
   }
 
diff --git a/engine/app/models/coplan/comment.rb b/engine/app/models/coplan/comment.rb
index 4d151d7d..6b6b45de 100644
--- a/engine/app/models/coplan/comment.rb
+++ b/engine/app/models/coplan/comment.rb
@@ -49,13 +49,16 @@ def first_comment_in_thread?
       # IDs are random UUIDs, not insertion-ordered, so we can't compare them
       # with `id < ?`. after_create_commit guarantees the row is persisted, so
       # a total count of 1 reliably means this comment opened the thread.
-      # Memoized: both the notify and analytics callbacks ask.
+      # Memoized since the analytics callback re-asks in the same request.
       return @first_comment_in_thread if defined?(@first_comment_in_thread)
       @first_comment_in_thread = comment_thread.comments.count == 1
     end
 
     def notify_thread_participants
-      CoPlan::NotificationJob.perform_later("comment_created", { comment_thread_id: comment_thread_id })
+      CoPlan::NotificationJob.perform_later(
+        "comment_created",
+        { comment_thread_id: comment_thread_id, comment_created_at: created_at }
+      )
     end
 
     def track_comment_created
diff --git a/spec/jobs/slack_notification_job_spec.rb b/spec/jobs/slack_notification_job_spec.rb
index edf7b5f9..419fda8f 100644
--- a/spec/jobs/slack_notification_job_spec.rb
+++ b/spec/jobs/slack_notification_job_spec.rb
@@ -152,6 +152,25 @@
         described_class.debounce(comment_thread_id: thread_record.id)
       }.to have_enqueued_job(described_class).on_queue("default")
     end
+
+    it "includes the comment that triggered the batch, not just later ones" do
+      # Regression test: batch_start must be the triggering comment's own
+      # created_at, not Time.current at the moment debounce runs — by the
+      # time this method is called, the comment already exists (that's what
+      # enqueued the job that led here), so Time.current would be later than
+      # created_at and would wrongly exclude this very comment.
+      comment = thread_record.comments.create!(
+        author_type: "human", author_id: commenter.id, body_markdown: "Kicks off the batch."
+      )
+
+      described_class.debounce(comment_thread_id: thread_record.id, comment_created_at: comment.created_at)
+      described_class.perform_now(comment_thread_id: thread_record.id)
+
+      expect(SlackClient).to have_received(:send_dm).with(
+        email: plan_author.email,
+        text: a_string_including("Kicks off the batch.")
+      )
+    end
   end
 
   describe "error handling" do
@@ -183,5 +202,29 @@
         described_class.perform_now(comment_thread_id: thread_record.id)
       }.to have_enqueued_job(described_class)
     end
+
+    it "does not double-notify a recipient who already succeeded when a transient error forces a retry" do
+      other_participant = create(:coplan_user, email: "other@example.com")
+      thread_record.comments.create!(
+        author_type: "human", author_id: other_participant.id, body_markdown: "Reply."
+      )
+
+      call_log = []
+      raised = false
+      allow(SlackClient).to receive(:send_dm) do |email:, text:|
+        call_log << email
+        if email == other_participant.email && !raised
+          raised = true
+          raise SlackClient::Error, "ratelimited"
+        end
+      end
+
+      described_class.perform_now(comment_thread_id: thread_record.id) # fails partway through
+      described_class.perform_now(comment_thread_id: thread_record.id) # simulated retry
+
+      expect(call_log.tally[plan_author.email]).to eq(1)
+      expect(call_log.tally[commenter.email]).to eq(1)
+      expect(call_log.tally[other_participant.email]).to eq(2) # one failed attempt, one success
+    end
   end
 end