diff --git a/app/api/discussion_comment_api.rb b/app/api/discussion_comment_api.rb
index ffd70117f..72658f15b 100644
--- a/app/api/discussion_comment_api.rb
+++ b/app/api/discussion_comment_api.rb
@@ -149,7 +149,7 @@ class DiscussionCommentApi < Grape::API
     discussion_comment = task.all_comments.find(params[:task_comment_id])
     # discussion_comment.mark_discussion_completed
     # mark comment read for student
-    discussion_comment.mark_as_read(current_user, project.unit)
+    discussion_comment.mark_as_read(current_user)
 
     error!({ error: 'No discussion comment found for the given task' }, 403) if discussion_comment.nil?
 
diff --git a/app/models/comments/assessment_comment.rb b/app/models/comments/assessment_comment.rb
index 1b614b41f..60dd4d26b 100644
--- a/app/models/comments/assessment_comment.rb
+++ b/app/models/comments/assessment_comment.rb
@@ -1,6 +1,7 @@
 class AssessmentComment < TaskComment
   before_create do
     self.content_type = :assessment
+    self.attention_audience = :student
   end
 
   def serialize(user)
diff --git a/app/models/comments/comment_read_cursor.rb b/app/models/comments/comment_read_cursor.rb
new file mode 100644
index 000000000..70c725af6
--- /dev/null
+++ b/app/models/comments/comment_read_cursor.rb
@@ -0,0 +1,34 @@
+# frozen_string_literal: true
+
+class CommentReadCursor < ApplicationRecord
+  belongs_to :task
+  belongs_to :user
+  belongs_to :last_read_comment, class_name: 'TaskComment'
+
+  validates :task, :user, :last_read_comment, :read_at, presence: true
+  validate :last_read_comment_belongs_to_task
+
+  def self.advance(task:, user:, comment:, read_at: Time.current)
+    cursor = create_or_find_by!(task: task, user: user) do |new_cursor|
+      new_cursor.last_read_comment = comment
+      new_cursor.read_at = read_at
+    end
+
+    # A cursor is a high-water mark, so an older comment cannot move it backwards.
+    return cursor if cursor.last_read_comment_id >= comment.id
+
+    cursor.with_lock do
+      cursor.update!(last_read_comment: comment, read_at: read_at) if cursor.last_read_comment_id < comment.id
+    end
+
+    cursor
+  end
+
+  private
+
+  def last_read_comment_belongs_to_task
+    return if last_read_comment.nil? || last_read_comment.task_id == task_id
+
+    errors.add(:last_read_comment, 'must belong to the same task')
+  end
+end
diff --git a/app/models/comments/discuss_timeout_comment.rb b/app/models/comments/discuss_timeout_comment.rb
index a45b8aadd..e6a2e5881 100644
--- a/app/models/comments/discuss_timeout_comment.rb
+++ b/app/models/comments/discuss_timeout_comment.rb
@@ -2,6 +2,10 @@ class DiscussTimeoutComment < TaskComment
   WARNING_CONTENT_TYPE = 'discuss_timeout_warning'.freeze
   EXPIRED_CONTENT_TYPE = 'discuss_timeout_expired'.freeze
 
+  before_create do
+    self.attention_audience = :student
+  end
+
   def self.warning
     WARNING_CONTENT_TYPE
   end
diff --git a/app/models/comments/extension_comment.rb b/app/models/comments/extension_comment.rb
index abd9d1030..1a3be16f3 100644
--- a/app/models/comments/extension_comment.rb
+++ b/app/models/comments/extension_comment.rb
@@ -19,11 +19,8 @@ def assessed?
   # Make sure we can access super's version of mark_as_read for assess extension
   alias :super_mark_as_read :mark_as_read
 
-  # Allow individual staff and the student to read this... but stop
-  # the main tutor reading without assessing. As only the main tutor
-  # propagates reads, this will work as required - other staff cant
-  # make it read for the main tutor.
-  def mark_as_read(user, unit = self.unit)
+  # Do not let the recipient tutor mark the request as read before assessing it.
+  def mark_as_read(user)
     super if assessed? || user == project.student || user != recipient
   end
 
diff --git a/app/models/comments/scorm_comment.rb b/app/models/comments/scorm_comment.rb
index df7bcc9f5..0603b62a5 100644
--- a/app/models/comments/scorm_comment.rb
+++ b/app/models/comments/scorm_comment.rb
@@ -1,6 +1,7 @@
 class ScormComment < TaskComment
   before_create do
     self.content_type = :scorm
+    self.attention_audience = :student
   end
 
   def serialize(user)
diff --git a/app/models/comments/scorm_extension_comment.rb b/app/models/comments/scorm_extension_comment.rb
index 74bc9d0c8..ad0280e59 100644
--- a/app/models/comments/scorm_extension_comment.rb
+++ b/app/models/comments/scorm_extension_comment.rb
@@ -16,11 +16,8 @@ def assessed?
   # Make sure we can access super's version of mark_as_read for assess extension
   alias super_mark_as_read mark_as_read
 
-  # Allow individual staff and the student to read this... but stop
-  # the main tutor reading without assessing. As only the main tutor
-  # propagates reads, this will work as required - other staff cant
-  # make it read for the main tutor.
-  def mark_as_read(user, unit = self.unit)
+  # Do not let the recipient tutor mark the request as read before assessing it.
+  def mark_as_read(user)
     super if assessed? || user == project.student || user != recipient
   end
 
diff --git a/app/models/comments/task_checked_in_comment.rb b/app/models/comments/task_checked_in_comment.rb
index 36dfd4b13..cc17093c6 100644
--- a/app/models/comments/task_checked_in_comment.rb
+++ b/app/models/comments/task_checked_in_comment.rb
@@ -1,6 +1,7 @@
 class TaskCheckedInComment < TaskComment
   before_create do
     self.content_type = :checked_in
+    self.attention_audience = :none
   end
 
   after_create do
diff --git a/app/models/comments/task_comment.rb b/app/models/comments/task_comment.rb
index c74883d01..3dd8968dc 100644
--- a/app/models/comments/task_comment.rb
+++ b/app/models/comments/task_comment.rb
@@ -8,6 +8,8 @@ class TaskComment < ApplicationRecord
   include FileHelper
   include AuthorisationHelpers
 
+  enum :attention_audience, { none: 0, student: 1, staff: 2 }, prefix: :attention
+
   belongs_to :task, optional: false # Foreign key
   belongs_to :user, optional: false
   has_one :unit, through: :task
@@ -16,6 +18,10 @@ class TaskComment < ApplicationRecord
   belongs_to :recipient, class_name: 'User', optional: false
 
   has_many :comments_read_receipts, class_name: 'CommentsReadReceipts', dependent: :destroy, inverse_of: :task_comment
+  has_many :comment_read_cursors,
+           foreign_key: :last_read_comment_id,
+           inverse_of: :last_read_comment,
+           dependent: :restrict_with_exception
 
   # Can optionally be a reply to a comment
   belongs_to :task_comment, optional: true
@@ -29,12 +35,15 @@ class TaskComment < ApplicationRecord
   validates :comment, length: { minimum: 0, maximum: 4095, allow_blank: true }
   validate :valid_reply_to?, on: :create
 
+  before_validation :set_default_attention_audience, on: :create
+
   # After create, mark as read by user creating
   after_create do
     mark_as_read(self.user)
   end
 
   # Delete action - before dependent association
+  before_destroy :rewind_comment_read_cursors, prepend: true
   before_destroy :delete_associated_files
 
   def valid_reply_to?
@@ -76,8 +85,8 @@ def serialize(user)
     }
   end
 
-  def create_comment_read_receipt_entry(user)
-    comment_read_receipt = CommentsReadReceipts.find_or_create_by(user: user, task_comment: self)
+  def advance_read_cursor(user)
+    CommentReadCursor.advance(task: task, user: user, comment: self)
   end
 
   def comment
@@ -135,18 +144,30 @@ def attachment_mime_type
   end
 
   def remove_comment_read_entry(user)
-    CommentsReadReceipts.delete_all(user: user, task_comment: self)
-  end
+    cursor = CommentReadCursor.find_by(task_id: task_id, user_id: user.id)
+    return if cursor.nil? || cursor.last_read_comment_id < id
 
-  def mark_as_read(user, unit = self.unit)
-    return if read_by?(user) # avoid propagating if not needed
+    previous_comment_id = TaskComment
+                          .where(task_id: task_id)
+                          .where('id < ?', id)
+                          .maximum(:id)
 
-    if user == project.tutor_for(task.task_definition)
-      unit.staff.each do |staff_member|
-        create_comment_read_receipt_entry(staff_member.user)
-      end
+    if previous_comment_id.nil?
+      cursor.destroy!
     else
-      create_comment_read_receipt_entry(user)
+      cursor.update!(
+        last_read_comment_id: previous_comment_id,
+        read_at: Time.current
+      )
+    end
+  end
+
+  def mark_as_read(user)
+    assigned_tutor = project.tutor_for(task.task_definition)
+
+    CommentReadCursor.transaction do
+      advance_read_cursor(user) unless read_by?(user)
+      remove_unneeded_staff_cursors(assigned_tutor) if user == assigned_tutor
     end
   end
 
@@ -155,15 +176,61 @@ def mark_as_unread(user)
   end
 
   def new_for?(user)
-    !read_by? user
+    requires_attention_for?(user) && !read_by?(user)
   end
 
   def read_by?(user)
-    CommentsReadReceipts.find_by(user: user, task_comment: self).present?
+    return true if self.user == user || !requires_attention_for?(user)
+
+    cursor = CommentReadCursor.find_by(task_id: task_id, user_id: user.id)
+    cursor.present? && cursor.last_read_comment_id >= id
   end
 
   def time_read_by(user)
-    read_reciept = CommentsReadReceipts.find_by(user: user, task_comment: self)
-    read_reciept&.created_at
+    return nil unless requires_attention_for?(user)
+
+    cursor = CommentReadCursor.find_by(task_id: task_id, user_id: user.id)
+    cursor&.read_at if cursor&.last_read_comment_id.to_i >= id
+  end
+
+  def requires_attention_for?(user)
+    return true if attention_audience.nil?
+    return attention_student? if task.student_participant?(user)
+
+    attention_staff?
+  end
+
+  def rewind_comment_read_cursors
+    previous_comment_id = TaskComment
+                          .where(task_id: task_id)
+                          .where('id < ?', id)
+                          .maximum(:id)
+
+    cursors = CommentReadCursor.where(last_read_comment_id: id)
+    if previous_comment_id.nil?
+      cursors.delete_all
+    else
+      # A single comment can be the cursor for every teaching staff member.
+      # Keep destruction bounded to one SQL update.
+      # rubocop:disable Rails/SkipsModelValidations
+      cursors.update_all(
+        last_read_comment_id: previous_comment_id,
+        updated_at: Time.current
+      )
+      # rubocop:enable Rails/SkipsModelValidations
+    end
+  end
+
+  private
+
+  def remove_unneeded_staff_cursors(assigned_tutor)
+    retained_user_ids = task.student_participant_ids << assigned_tutor.id
+    CommentReadCursor.where(task_id: task_id).where.not(user_id: retained_user_ids).delete_all
+  end
+
+  def set_default_attention_audience
+    return if attention_audience.present? || user.nil? || task.nil?
+
+    self.attention_audience = user == task.project.student ? :staff : :student
   end
 end
diff --git a/app/models/comments/task_discussed_comment.rb b/app/models/comments/task_discussed_comment.rb
index 5c4ee3a71..f756e54f1 100644
--- a/app/models/comments/task_discussed_comment.rb
+++ b/app/models/comments/task_discussed_comment.rb
@@ -1,6 +1,7 @@
 class TaskDiscussedComment < TaskComment
   before_create do
     self.content_type = :discussed_in_class
+    self.attention_audience = :none
   end
 
   after_create do
diff --git a/app/models/comments/task_status_comment.rb b/app/models/comments/task_status_comment.rb
index a224279db..16bf2d90b 100644
--- a/app/models/comments/task_status_comment.rb
+++ b/app/models/comments/task_status_comment.rb
@@ -3,6 +3,7 @@ class TaskStatusComment < TaskComment
 
   before_create do
     self.content_type = :status
+    self.attention_audience = :none
   end
 
   after_create do
diff --git a/app/models/overseer_assessment.rb b/app/models/overseer_assessment.rb
index d01a3210e..6dc0b808a 100644
--- a/app/models/overseer_assessment.rb
+++ b/app/models/overseer_assessment.rb
@@ -38,14 +38,17 @@ def self.student_notification_grace_period
          AND assessment_comments.type = 'AssessmentComment'
       SQL
       .joins(<<~SQL.squish)
-        LEFT JOIN comments_read_receipts student_read_receipts
-          ON student_read_receipts.task_comment_id = assessment_comments.id
-         AND student_read_receipts.user_id = projects.user_id
+        LEFT JOIN comment_read_cursors student_read_cursor
+          ON student_read_cursor.task_id = assessment_comments.task_id
+         AND student_read_cursor.user_id = projects.user_id
       SQL
       .where(status: statuses[:failed], student_notified_at: nil)
       .where(users: { receive_task_notifications: true })
       .where('overseer_assessments.updated_at <= ?', notification_cutoff)
-      .where('student_read_receipts.id IS NULL')
+      .where(
+        'student_read_cursor.last_read_comment_id IS NULL ' \
+        'OR student_read_cursor.last_read_comment_id < assessment_comments.id'
+      )
       .where(<<~SQL.squish)
         assessment_comments.id = (
           SELECT latest_comment.id
diff --git a/app/models/project.rb b/app/models/project.rb
index 785f9b39b..8f18df990 100644
--- a/app/models/project.rb
+++ b/app/models/project.rb
@@ -281,14 +281,18 @@ def reference_date
 
   def task_details_for_shallow_serializer(user)
     teaching_breaks = unit.teaching_period&.breaks.to_a
+    attention_audience = TaskComment.attention_audiences.fetch(user == student ? 'student' : 'staff')
 
     tasks
       .joins(:task_status)
-      .joins("LEFT JOIN task_comments ON task_comments.task_id = tasks.id AND (task_comments.type IS NULL OR task_comments.type <> 'TaskStatusComment')")
-      .joins("LEFT JOIN comments_read_receipts crr ON crr.task_comment_id = task_comments.id AND crr.user_id = #{user.id}")
+      .joins('LEFT JOIN tasks comment_tasks ON comment_tasks.id = tasks.id ' \
+             'OR (tasks.group_submission_id IS NOT NULL ' \
+             'AND comment_tasks.group_submission_id = tasks.group_submission_id)')
+      .joins("LEFT JOIN task_comments ON task_comments.task_id = comment_tasks.id AND (task_comments.attention_audience IS NULL OR task_comments.attention_audience = #{attention_audience}) AND (task_comments.type IS NULL OR task_comments.type <> 'TaskStatusComment')")
+      .joins("LEFT JOIN comment_read_cursors crc ON crc.task_id = task_comments.task_id AND crc.user_id = #{user.id.to_i}")
       .joins('LEFT OUTER JOIN task_similarities ON tasks.id = task_similarities.task_id')
       .select(
-        'SUM(case when crr.user_id is null AND NOT task_comments.id is null then 1 else 0 end) as number_unread', 'project_id', 'tasks.id as id',
+        'SUM(case when (crc.last_read_comment_id IS NULL OR task_comments.id > crc.last_read_comment_id) AND NOT task_comments.id is null then 1 else 0 end) as number_unread', 'project_id', 'tasks.id as id',
         'task_definition_id', 'task_statuses.id as status_id',
         'completion_date', 'times_assessed', 'submission_date', 'tasks.grade as grade', 'quality_pts', 'include_in_portfolio', 'grade',
         'SUM(case when task_similarities.flagged then 1 else 0 end) as similar_to_count'
diff --git a/app/models/task.rb b/app/models/task.rb
index 9a535b5a1..96813ddb3 100644
--- a/app/models/task.rb
+++ b/app/models/task.rb
@@ -131,6 +131,7 @@ def specific_permission_hash(role, perm_hash, _other)
   has_one :overflow_task_claim, dependent: :destroy
 
   has_many :comments, class_name: 'TaskComment', dependent: :destroy, inverse_of: :task
+  has_many :comment_read_cursors, dependent: :destroy, inverse_of: :task
   has_many :task_similarities, class_name: 'TaskSimilarity', dependent: :destroy, inverse_of: :task
   has_many :reverse_jplag_similarities, class_name: 'JplagTaskSimilarity', dependent: :destroy, inverse_of: :other_task, foreign_key: 'other_task_id'
   has_many :reverse_moss_similarities, class_name: 'MossTaskSimilarity', dependent: :destroy, inverse_of: :other_task, foreign_key: 'other_task_id'
@@ -226,8 +227,17 @@ def all_comments
   end
 
   def mark_comments_as_read(user, comments)
+    latest_comment_by_task = {}
+
     comments.each do |comment|
-      comment.mark_as_read(user, unit)
+      next unless comment.requires_attention_for?(user)
+
+      current = latest_comment_by_task[comment.task_id]
+      latest_comment_by_task[comment.task_id] = comment if current.nil? || current.id < comment.id
+    end
+
+    latest_comment_by_task.each_value do |comment|
+      comment.mark_as_read(user)
     end
   end
 
@@ -241,16 +251,25 @@ def comments_for_user(user)
     TaskComment
       .joins('JOIN users AS authors ON authors.id = task_comments.user_id')
       .joins('JOIN users AS recipients ON recipients.id = task_comments.recipient_id')
-      .joins("LEFT JOIN comments_read_receipts u_crr ON u_crr.task_comment_id = task_comments.id AND u_crr.user_id = #{user.id}")
-      .joins("LEFT JOIN comments_read_receipts r_crr ON r_crr.task_comment_id = task_comments.id AND r_crr.user_id = recipients.id")
+      .joins(
+        "LEFT JOIN comment_read_cursors user_cursor " \
+        "ON user_cursor.task_id = task_comments.task_id AND user_cursor.user_id = #{user.id.to_i}"
+      )
+      .joins(
+        'LEFT JOIN comment_read_cursors recipient_cursor ' \
+        'ON recipient_cursor.task_id = task_comments.task_id ' \
+        'AND recipient_cursor.user_id = recipients.id'
+      )
       .where('task_comments.task_id = :task_id', task_id: self.id)
       .order('created_at ASC')
       .select(
         'task_comments.id AS id',
         'task_comments.comment AS comment',
         'task_comments.content_type AS content_type',
-        "case when u_crr.created_at IS NULL then 1 else 0 end AS is_new",
-        'r_crr.created_at AS recipient_read_time',
+        'CASE WHEN user_cursor.last_read_comment_id IS NULL ' \
+        'OR task_comments.id > user_cursor.last_read_comment_id THEN 1 ELSE 0 END AS is_new',
+        'CASE WHEN task_comments.id <= recipient_cursor.last_read_comment_id ' \
+        'THEN recipient_cursor.read_at ELSE NULL END AS recipient_read_time',
         'task_comments.created_at AS created_at',
         'authors.id AS author_id',
         'authors.first_name AS author_first_name',
@@ -569,6 +588,16 @@ def group_task?
     !group_submission.nil? || !task_definition.group_set.nil?
   end
 
+  def student_participant_ids
+    return [project.user_id] if group_submission.nil?
+
+    group_submission.projects.distinct.pluck(:user_id)
+  end
+
+  def student_participant?(user)
+    user.present? && student_participant_ids.include?(user.id)
+  end
+
   def active_overflow_task_claim
     claim = overflow_task_claim
     return nil unless claim
@@ -874,7 +903,11 @@ def assess(task_status, assessor, assess_date = Time.zone.now, recursive_fix = f
           # Since we are calling this assess method again, we recursively check for more dependent tasks that need to be updated
           task.assess(TaskStatus.fix_and_resubmit, assessor, assess_date, recursive_fix)
           task.add_status_comment(assessor, TaskStatus.fix_and_resubmit)
-          task.add_text_comment(assessor, "**Automated comment**: A prerequisite task was updated to Fix and Resubmit, so this task was updated as well. You may need to review and update the prerequisite before resubmitting.")
+          task.add_text_comment(
+            assessor,
+            "**Automated comment**: A prerequisite task was updated to Fix and Resubmit, so this task was updated as well. You may need to review and update the prerequisite before resubmitting.",
+            attention_audience: :student
+          )
         end
       end
 
@@ -974,7 +1007,7 @@ def weight
     task_definition.weighting.to_f
   end
 
-  def add_text_comment(user, text, reply_to_id = nil)
+  def add_text_comment(user, text, reply_to_id = nil, attention_audience: nil)
     text = text.strip
     return nil if user.nil? || text.nil? || text.empty?
 
@@ -992,6 +1025,7 @@ def add_text_comment(user, text, reply_to_id = nil)
     comment.content_type = :text
     comment.recipient = user == project.student ? project.tutor_for(task_definition) : project.student
     comment.reply_to_id = reply_to_id
+    comment.attention_audience = attention_audience if attention_audience.present?
     comment.save!
 
     comment
@@ -1075,7 +1109,7 @@ def add_discussion_comment(user, prompts)
       raise "Error attaching uploaded file." unless discussion.add_prompt(prompt, index)
     end
 
-    discussion.mark_as_read(user, unit)
+    discussion.mark_as_read(user)
 
     logger.info(discussion)
     return discussion
@@ -1614,7 +1648,11 @@ def convert_submission_to_pdf(source_folder: FileHelper.student_work_dir(:new),
     rescue => e
       SubmissionHistory.clear_document_previews(self)
       trigger_transition trigger: 'fix', by_user: project.tutor_for(task_definition)
-      add_text_comment project.tutor_for(task_definition), "**Automated Comment**: Something went wrong with your submission. Check the files and resubmit this task. #{e.message}"
+      add_text_comment(
+        project.tutor_for(task_definition),
+        "**Automated Comment**: Something went wrong with your submission. Check the files and resubmit this task. #{e.message}",
+        attention_audience: :student
+      )
       raise e
     ensure
       # Ensure latex aux file is removed - if broken will cause issues for next submission in sidekiq
diff --git a/app/models/unit.rb b/app/models/unit.rb
index 2ebe9c58d..75c2cb269 100644
--- a/app/models/unit.rb
+++ b/app/models/unit.rb
@@ -2549,18 +2549,35 @@ def tutorial_enrolment_subquery
   # Return all tasks from the database for this unit and given user
   #
   def get_all_tasks_for(user, my_tutorials_only = false)
+    staff_attention = TaskComment.attention_audiences.fetch('staff')
+    unread_comment = '(COALESCE(crc.last_read_comment_id, 0) < task_comments.id ' \
+                     'AND COALESCE(tutor_crc.last_read_comment_id, 0) < task_comments.id)'
     result =  student_tasks.
               joins(:task_status).
               joins("LEFT OUTER JOIN (#{tutorial_enrolment_subquery}) as sq ON sq.project_id = projects.id AND (sq.tutorial_stream_id = task_definitions.tutorial_stream_id OR sq.tutorial_stream_id IS NULL)").
-              joins("LEFT JOIN task_comments ON task_comments.task_id = tasks.id AND (task_comments.type IS NULL OR task_comments.type <> 'TaskStatusComment') AND (task_comments.content_type IS NULL OR (task_comments.content_type <> 'plan' AND task_comments.content_type <> 'discussed_in_class'))").
-              joins("LEFT JOIN comments_read_receipts crr ON crr.task_comment_id = task_comments.id AND crr.user_id = #{user.id}").
+              joins("LEFT JOIN unit_roles task_tutor_roles ON task_tutor_roles.id = COALESCE(sq.unit_role_id, #{main_convenor_id.to_i})").
+              joins(
+                'LEFT JOIN tasks comment_tasks ON comment_tasks.id = tasks.id ' \
+                'OR (tasks.group_submission_id IS NOT NULL ' \
+                'AND comment_tasks.group_submission_id = tasks.group_submission_id)'
+              ).
+              joins(
+                "LEFT JOIN task_comments ON task_comments.task_id = comment_tasks.id " \
+                "AND (task_comments.attention_audience IS NULL OR task_comments.attention_audience = #{staff_attention}) " \
+                "AND (task_comments.type IS NULL OR task_comments.type <> 'TaskStatusComment') " \
+                "AND (task_comments.content_type IS NULL OR (task_comments.content_type <> 'plan' " \
+                "AND task_comments.content_type <> 'discussed_in_class'))"
+              ).
+              joins("LEFT JOIN comment_read_cursors crc ON crc.task_id = task_comments.task_id AND crc.user_id = #{user.id.to_i}").
+              joins('LEFT JOIN comment_read_cursors tutor_crc ON tutor_crc.task_id = task_comments.task_id ' \
+                    'AND tutor_crc.user_id = task_tutor_roles.user_id').
               joins("LEFT JOIN task_pins ON task_pins.task_id = tasks.id AND task_pins.user_id = #{user.id}").
               joins('LEFT OUTER JOIN task_similarities ON tasks.id = task_similarities.task_id').
               select(
                 'sq.tutorial_id AS tutorial_id',
                 'sq.tutorial_stream_id AS tutorial_stream_id',
                 'tasks.id',
-                "SUM(case when crr.user_id is null AND NOT task_comments.id is null then 1 else 0 end) as number_unread",
+                "SUM(case when #{unread_comment} AND task_comments.id IS NOT NULL then 1 else 0 end) as number_unread",
                 'COUNT(distinct task_pins.task_id) != 0 as pinned',
                 "SUM(case when task_comments.date_extension_assessed IS NULL AND task_comments.type = 'ExtensionComment' AND NOT task_comments.id IS NULL THEN 1 ELSE 0 END) > 0 as has_extensions",
                 'project_id',
@@ -2621,8 +2638,10 @@ def tasks_awaiting_feedback(user)
   # student comment -- whichever is newer.
   #
   def tasks_for_task_inbox(user, my_students_only = false)
+    unread_comment = '(COALESCE(crc.last_read_comment_id, 0) < task_comments.id ' \
+                     'AND COALESCE(tutor_crc.last_read_comment_id, 0) < task_comments.id)'
     get_all_tasks_for(user, my_students_only)
-      .having('task_statuses.id IN (:ids) OR COUNT(task_pins.task_id) > 0 OR SUM(case when crr.user_id is null AND NOT task_comments.id is null then 1 else 0 end) > 0', ids: [TaskStatus.ready_for_feedback, TaskStatus.need_help])
+      .having("task_statuses.id IN (:ids) OR COUNT(task_pins.task_id) > 0 OR SUM(case when #{unread_comment} AND task_comments.id IS NOT NULL then 1 else 0 end) > 0", ids: [TaskStatus.ready_for_feedback, TaskStatus.need_help])
       .order('pinned DESC, submission_date ASC, MAX(task_comments.created_at) ASC, task_definition_id ASC')
   end
 
diff --git a/app/models/user.rb b/app/models/user.rb
index 67a50d187..9349c3324 100644
--- a/app/models/user.rb
+++ b/app/models/user.rb
@@ -176,6 +176,7 @@ def token_for_text?(a_token, token_type)
   has_many    :engagements, dependent: :restrict_with_exception, inverse_of: :user
   has_many    :engagement_comments, dependent: :restrict_with_exception, inverse_of: :user
   has_many    :auth_tokens, dependent: :destroy, inverse_of: :user
+  has_many    :comment_read_cursors, dependent: :destroy, inverse_of: :user
   has_many    :user_oauth_tokens, dependent: :destroy, inverse_of: :user
   has_many    :user_oauth_states, dependent: :destroy, inverse_of: :user
   has_one     :webcal, dependent: :destroy, inverse_of: :user
diff --git a/app/sidekiq/accept_overseer_job.rb b/app/sidekiq/accept_overseer_job.rb
index 59ddd753f..5227e287d 100644
--- a/app/sidekiq/accept_overseer_job.rb
+++ b/app/sidekiq/accept_overseer_job.rb
@@ -108,7 +108,11 @@ def perform(task_id, _output_path, docker_image_name_tag, submission, assessment
         task.add_status_comment(task.project.tutor_for(task.task_definition), failure_status)
         oa.update!(result_task_status: failure_status.status_key.to_s)
       end
-      task.add_text_comment(task.project.tutor_for(task.task_definition), "**Automated comment**: Some tests did not pass for this submission. Please review the Overseer report, verify your output, and resubmit.")
+      task.add_text_comment(
+        task.project.tutor_for(task.task_definition),
+        "**Automated comment**: Some tests did not pass for this submission. Please review the Overseer report, verify your output, and resubmit.",
+        attention_audience: :student
+      )
     end
 
     FileUtils.rm_rf(work_dir)
diff --git a/app/sidekiq/execute_communication_set_job.rb b/app/sidekiq/execute_communication_set_job.rb
index 5f99bd3fe..0edf567b8 100644
--- a/app/sidekiq/execute_communication_set_job.rb
+++ b/app/sidekiq/execute_communication_set_job.rb
@@ -256,7 +256,7 @@ def execute_task_comment_action(action, projects, unit, rule)
         }
       end
 
-      comment = task.add_text_comment(comment_author, rendered_comment)
+      comment = task.add_text_comment(comment_author, rendered_comment, attention_audience: :student)
 
       if comment.nil?
         next {
diff --git a/db/migrate/20260728051502_create_comment_read_cursors.rb b/db/migrate/20260728051502_create_comment_read_cursors.rb
new file mode 100644
index 000000000..82cdf101a
--- /dev/null
+++ b/db/migrate/20260728051502_create_comment_read_cursors.rb
@@ -0,0 +1,44 @@
+class CreateCommentReadCursors < ActiveRecord::Migration[8.0]
+  def up
+    create_cursor_table
+    backfill_cursors
+  end
+
+  def down
+    drop_table :comment_read_cursors
+  end
+
+  private
+
+  def create_cursor_table
+    create_table :comment_read_cursors do |t|
+      t.references :task, null: false
+      t.references :user, null: false
+      t.references :last_read_comment, null: false
+      t.datetime :read_at, null: false
+
+      t.timestamps
+      t.index [:task_id, :user_id], unique: true
+    end
+  end
+
+  def backfill_cursors
+    say_with_time 'Backfilling comment read cursors' do
+      execute <<~SQL
+        INSERT INTO comment_read_cursors
+          (task_id, user_id, last_read_comment_id, read_at, created_at, updated_at)
+        SELECT
+          task_comments.task_id,
+          comments_read_receipts.user_id,
+          MAX(comments_read_receipts.task_comment_id),
+          MAX(comments_read_receipts.updated_at),
+          MIN(comments_read_receipts.created_at),
+          MAX(comments_read_receipts.updated_at)
+        FROM comments_read_receipts
+        INNER JOIN task_comments
+          ON task_comments.id = comments_read_receipts.task_comment_id
+        GROUP BY task_comments.task_id, comments_read_receipts.user_id
+      SQL
+    end
+  end
+end
diff --git a/db/migrate/20260729020310_add_attention_audience_to_task_comments.rb b/db/migrate/20260729020310_add_attention_audience_to_task_comments.rb
new file mode 100644
index 000000000..469a93c0e
--- /dev/null
+++ b/db/migrate/20260729020310_add_attention_audience_to_task_comments.rb
@@ -0,0 +1,5 @@
+class AddAttentionAudienceToTaskComments < ActiveRecord::Migration[8.0]
+  def change
+    add_column :task_comments, :attention_audience, :integer
+  end
+end
diff --git a/db/schema.rb b/db/schema.rb
index 2a0252e4c..5d2aa8722 100644
--- a/db/schema.rb
+++ b/db/schema.rb
@@ -10,7 +10,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema[8.0].define(version: 2026_07_28_033837) do
+ActiveRecord::Schema[8.0].define(version: 2026_07_29_020310) do
   create_table "activity_types", charset: "utf8mb4", collation: "utf8mb4_general_ci", force: :cascade do |t|
     t.string "name", null: false
     t.string "abbreviation", null: false
@@ -63,6 +63,19 @@
     t.index ["tutor_id"], name: "index_chip_usages_on_tutor_id"
   end
 
+  create_table "comment_read_cursors", charset: "utf8mb4", collation: "utf8mb4_general_ci", force: :cascade do |t|
+    t.bigint "task_id", null: false
+    t.bigint "user_id", null: false
+    t.bigint "last_read_comment_id", null: false
+    t.datetime "read_at", null: false
+    t.datetime "created_at", null: false
+    t.datetime "updated_at", null: false
+    t.index ["last_read_comment_id"], name: "index_comment_read_cursors_on_last_read_comment_id"
+    t.index ["task_id", "user_id"], name: "index_comment_read_cursors_on_task_id_and_user_id", unique: true
+    t.index ["task_id"], name: "index_comment_read_cursors_on_task_id"
+    t.index ["user_id"], name: "index_comment_read_cursors_on_user_id"
+  end
+
   create_table "comments_read_receipts", charset: "utf8mb4", collation: "utf8mb4_general_ci", force: :cascade do |t|
     t.bigint "task_comment_id", null: false
     t.bigint "user_id", null: false
@@ -555,6 +568,7 @@
     t.bigint "reply_to_id"
     t.bigint "commentable_id"
     t.string "commentable_type"
+    t.integer "attention_audience"
     t.index ["assessor_id"], name: "index_task_comments_on_assessor_id"
     t.index ["commentable_type", "commentable_id"], name: "index_task_comments_on_commentable_type_and_commentable_id"
     t.index ["discussion_comment_id"], name: "index_task_comments_on_discussion_comment_id"
diff --git a/lib/tasks/maintenance.rake b/lib/tasks/maintenance.rake
index d7053b6e2..8986d838c 100644
--- a/lib/tasks/maintenance.rake
+++ b/lib/tasks/maintenance.rake
@@ -79,7 +79,11 @@ namespace :maintenance do
     tutor = task.project.tutor_for(task.task_definition)
 
     task.trigger_transition(trigger: 'fix', by_user: tutor)
-    task.add_text_comment(tutor, "**Automated Comment**: Something went wrong with compiling your submission. Please resubmit the task.")
+    task.add_text_comment(
+      tutor,
+      "**Automated Comment**: Something went wrong with compiling your submission. Please resubmit the task.",
+      attention_audience: :student
+    )
   rescue StandardError => e
     Rails.logger.error "Failed to move task #{task.id} to fix and add automated comment!\n#{e.message}"
   end
@@ -88,7 +92,11 @@ namespace :maintenance do
     tutor = task.project.tutor_for(task.task_definition)
 
     task.trigger_transition(trigger: 'fix', by_user: tutor)
-    task.add_text_comment(tutor, "**Automated Comment**: Something went wrong while running the automated tests for this submission. Please resubmit the task.")
+    task.add_text_comment(
+      tutor,
+      "**Automated Comment**: Something went wrong while running the automated tests for this submission. Please resubmit the task.",
+      attention_audience: :student
+    )
   rescue StandardError => e
     Rails.logger.error "Failed to move task #{task.id} to fix and add Overseer automated comment!\n#{e.message}"
   end
diff --git a/test/models/comment_read_cursor_test.rb b/test/models/comment_read_cursor_test.rb
new file mode 100644
index 000000000..adee6236f
--- /dev/null
+++ b/test/models/comment_read_cursor_test.rb
@@ -0,0 +1,187 @@
+require 'test_helper'
+
+class CommentReadCursorTest < ActiveSupport::TestCase
+  def test_comment_reads_use_one_cursor_per_task_and_user
+    project = FactoryBot.create(:project)
+    task = project.task_for_task_definition(project.unit.task_definitions.first)
+    reader = project.student
+    author = project.unit.main_convenor_user
+
+    comments = [
+      task.add_text_comment(author, 'First'),
+      task.add_text_comment(author, 'Second'),
+      task.add_text_comment(author, 'Third')
+    ]
+
+    task.mark_comments_as_read(reader, comments)
+
+    cursor = CommentReadCursor.find_by!(task: task, user: reader)
+    assert_equal comments.last.id, cursor.last_read_comment_id
+    assert_equal 1, CommentReadCursor.where(task: task, user: reader).count
+    assert(comments.none? { |comment| comment.new_for?(reader) })
+
+    CommentReadCursor.advance(
+      task: task,
+      user: reader,
+      comment: comments.first
+    )
+    assert_equal comments.last.id, cursor.reload.last_read_comment_id
+
+    comments.second.mark_as_unread(reader)
+
+    assert_not comments.first.new_for?(reader)
+    assert comments.second.new_for?(reader)
+    assert comments.last.new_for?(reader)
+  end
+
+  def test_assigned_tutor_reading_a_comment_removes_other_staff_cursors
+    project = FactoryBot.create(:project)
+    task = project.task_for_task_definition(project.unit.task_definitions.first)
+    assigned_tutor = project.tutor_for(task.task_definition)
+    other_staff = FactoryBot.create(:user, :tutor)
+    project.unit.employ_staff(other_staff, Role.tutor)
+    tutor_comment = task.add_text_comment(assigned_tutor, 'Some feedback')
+    tutor_comment.mark_as_read(project.student)
+    comment = task.add_text_comment(project.student, 'A question')
+
+    comment.mark_as_read(other_staff)
+    assert_equal comment.id, CommentReadCursor.find_by!(task: task, user: other_staff).last_read_comment_id
+
+    comment.mark_as_read(assigned_tutor)
+
+    assert comment.read_by?(assigned_tutor)
+    assert_nil CommentReadCursor.find_by(task: task, user: other_staff)
+    assert_equal tutor_comment.id, CommentReadCursor.find_by!(task: task, user: project.student).last_read_comment_id
+  end
+
+  def test_assigned_tutor_cursor_hides_a_task_from_other_staff_inboxes
+    project = FactoryBot.create(:project)
+    unit = project.unit
+    task_definition = unit.task_definitions.first
+    assigned_tutor = FactoryBot.create(:user, :tutor)
+    assigned_tutor_role = unit.employ_staff(assigned_tutor, Role.tutor)
+    tutorial = FactoryBot.create(
+      :tutorial,
+      unit: unit,
+      campus: project.campus,
+      tutorial_stream: task_definition.tutorial_stream,
+      unit_role: assigned_tutor_role
+    )
+    project.enrol_in(tutorial)
+    other_staff = FactoryBot.create(:user, :tutor)
+    unit.employ_staff(other_staff, Role.tutor)
+    task = project.task_for_task_definition(task_definition)
+    comment = task.add_text_comment(project.student, 'Please review this')
+
+    assert_includes unit.tasks_for_task_inbox(other_staff).map(&:task_id), task.id
+
+    comment.mark_as_read(other_staff)
+    assert_equal comment.id, CommentReadCursor.find_by!(task: task, user: other_staff).last_read_comment_id
+
+    comment.mark_as_read(assigned_tutor)
+
+    assert_nil CommentReadCursor.find_by(task: task, user: other_staff)
+    assert_not_includes unit.tasks_for_task_inbox(other_staff).map(&:task_id), task.id
+  end
+
+  def test_group_students_have_independent_unread_comment_cursors
+    unit = FactoryBot.create(
+      :unit,
+      group_sets: 1,
+      groups: [{ gs: 0, students: 2 }],
+      group_tasks: [{ idx: 0, gs: 0 }]
+    )
+    task_definition = unit.task_definitions.first
+    projects = unit.groups.first.projects.to_a
+    first_task = projects.first.task_for_task_definition(task_definition)
+    second_task = projects.second.task_for_task_definition(task_definition)
+    tutor = projects.first.tutor_for(task_definition)
+
+    comment = first_task.add_text_comment(tutor, 'Feedback for the group')
+    second_task.reload
+
+    assert comment.new_for?(projects.first.student)
+    assert comment.new_for?(projects.second.student)
+    assert_equal 1, unread_count(projects.first, task_definition)
+    assert_equal 1, unread_count(projects.second, task_definition)
+
+    first_task.mark_comments_as_read(projects.first.student, first_task.all_comments)
+
+    assert_not comment.new_for?(projects.first.student)
+    assert comment.new_for?(projects.second.student)
+    assert_equal 0, unread_count(projects.first, task_definition)
+    assert_equal 1, unread_count(projects.second, task_definition)
+  end
+
+  def test_destroying_the_cursor_comment_rewinds_each_users_cursor
+    project = FactoryBot.create(:project)
+    task = project.task_for_task_definition(project.unit.task_definitions.first)
+    other_staff = FactoryBot.create(:user, :tutor)
+    project.unit.employ_staff(other_staff, Role.tutor)
+    author = project.student
+    readers = [project.tutor_for(task.task_definition), other_staff]
+    previous_comment = task.add_text_comment(author, 'First')
+    cursor_comment = task.add_text_comment(author, 'Second')
+
+    readers.each { |reader| cursor_comment.mark_as_read(reader) }
+
+    cursor_comment.destroy!
+
+    readers.each do |reader|
+      cursor = CommentReadCursor.find_by!(task: task, user: reader)
+      assert_equal previous_comment.id, cursor.last_read_comment_id
+    end
+  end
+
+  def test_destroying_the_only_comment_removes_its_cursors
+    project = FactoryBot.create(:project)
+    task = project.task_for_task_definition(project.unit.task_definitions.first)
+    reader = project.student
+    comment = task.add_text_comment(project.unit.main_convenor_user, 'Only comment')
+
+    comment.mark_as_read(reader)
+    comment.destroy!
+
+    assert_nil CommentReadCursor.find_by(task: task, user: reader)
+  end
+
+  def test_only_staff_attention_comments_count_in_the_tutor_inbox
+    project = FactoryBot.create(:project)
+    task = project.task_for_task_definition(project.unit.task_definitions.first)
+    tutor = project.tutor_for(task.task_definition)
+
+    student_comment = task.add_text_comment(project.student, 'Please review this')
+    status_comment = task.add_status_comment(project.student, TaskStatus.ready_for_feedback)
+    assessment_comment = AssessmentComment.create!(
+      task: task,
+      user: tutor,
+      recipient: project.student,
+      comment: 'Automated assessment complete'
+    )
+
+    assert student_comment.attention_staff?
+    assert status_comment.attention_none?
+    assert assessment_comment.attention_student?
+    assert student_comment.new_for?(tutor)
+    assert_not status_comment.new_for?(tutor)
+    assert_not assessment_comment.new_for?(tutor)
+    assert_nil CommentReadCursor.find_by(task: task, user: tutor)
+
+    inbox_task = project.unit.tasks_for_task_inbox(tutor).find { |item| item.task_id == task.id }
+    assert_not_nil inbox_task
+    assert_equal 1, inbox_task.number_unread.to_i
+
+    task.mark_comments_as_read(tutor, task.comments)
+
+    cursor = CommentReadCursor.find_by!(task: task, user: tutor)
+    assert_equal student_comment.id, cursor.last_read_comment_id
+    assert_not(project.unit.tasks_for_task_inbox(tutor).any? { |item| item.task_id == task.id })
+  end
+
+  private
+
+  def unread_count(project, task_definition)
+    project.task_details_for_shallow_serializer(project.student)
+           .find { |task| task[:task_definition_id] == task_definition.id }[:num_new_comments].to_i
+  end
+end
diff --git a/test/models/task_definition_test.rb b/test/models/task_definition_test.rb
index 16b17bf9d..1651c994b 100644
--- a/test/models/task_definition_test.rb
+++ b/test/models/task_definition_test.rb
@@ -661,8 +661,8 @@ def test_overdue_tasks_update_to_assess_in_portfolio
     task1 = student.task_for_task_definition(td1)
     task2 = student.task_for_task_definition(td2)
 
-    task1.comments.delete_all
-    task2.comments.delete_all
+    task1.comments.destroy_all
+    task2.comments.destroy_all
 
     task1.update(task_status_id: TaskStatus.time_exceeded.id)
 
diff --git a/test/models/task_test.rb b/test/models/task_test.rb
index 496d3feb1..bbf68256b 100644
--- a/test/models/task_test.rb
+++ b/test/models/task_test.rb
@@ -1866,7 +1866,7 @@ def test_prerequisite_tasks_change_to_fix_and_resubmit
     task3.trigger_transition(trigger: 'ready_for_feedback', by_user: unit.main_convenor_user)
     task4.trigger_transition(trigger: 'ready_for_feedback', by_user: unit.main_convenor_user)
 
-    task2.comments.delete_all
+    task2.comments.destroy_all
 
     task1.reload
     task2.reload
diff --git a/test/models/unit_model_test.rb b/test/models/unit_model_test.rb
index b53659ea0..3f8fe0abe 100644
--- a/test/models/unit_model_test.rb
+++ b/test/models/unit_model_test.rb
@@ -1262,8 +1262,8 @@ def test_overdue_tasks_update_to_assess_in_portfolio
     task1 = student.task_for_task_definition(td1)
     task2 = student.task_for_task_definition(td2)
 
-    task1.comments.delete_all
-    task2.comments.delete_all
+    task1.comments.destroy_all
+    task2.comments.destroy_all
 
     task1.update(task_status_id: TaskStatus.time_exceeded.id)