diff --git a/app/controllers/api/v1/accounts/conversations/unread_counts_controller.rb b/app/controllers/api/v1/accounts/conversations/unread_counts_controller.rb
new file mode 100644
index 000000000..d9f15613b
--- /dev/null
+++ b/app/controllers/api/v1/accounts/conversations/unread_counts_controller.rb
@@ -0,0 +1,16 @@
+class Api::V1::Accounts::Conversations::UnreadCountsController < Api::V1::Accounts::BaseController
+  before_action :ensure_unread_counts_enabled
+
+  def index
+    counts = ::Conversations::UnreadCounts::Counter.new(account: Current.account, user: Current.user).perform
+    render json: { payload: counts }
+  end
+
+  private
+
+  def ensure_unread_counts_enabled
+    return if Current.account.feature_enabled?('conversation_unread_counts')
+
+    render json: { error: I18n.t('errors.conversations.unread_counts.feature_not_enabled') }, status: :forbidden
+  end
+end
diff --git a/app/controllers/api/v1/accounts/conversations_controller.rb b/app/controllers/api/v1/accounts/conversations_controller.rb
index 6cc77cd54..2856c7817 100644
--- a/app/controllers/api/v1/accounts/conversations_controller.rb
+++ b/app/controllers/api/v1/accounts/conversations_controller.rb
@@ -162,6 +162,8 @@ class Api::V1::Accounts::ConversationsController < Api::V1::Accounts::BaseContro
     # rubocop:disable Rails/SkipsModelValidations
     @conversation.update_columns(updates)
     # rubocop:enable Rails/SkipsModelValidations
+
+    ::Conversations::UnreadCounts::Notifier.new(@conversation).perform
   end
 
   def should_update_last_seen?
diff --git a/app/dispatchers/async_dispatcher.rb b/app/dispatchers/async_dispatcher.rb
index 7416b7861..abf3ca354 100644
--- a/app/dispatchers/async_dispatcher.rb
+++ b/app/dispatchers/async_dispatcher.rb
@@ -17,6 +17,7 @@ class AsyncDispatcher < BaseDispatcher
       InstallationWebhookListener.instance,
       NotificationListener.instance,
       ParticipationListener.instance,
+      Conversations::UnreadCounts::Listener.instance,
       ReportingEventListener.instance,
       WebhookListener.instance
     ]
diff --git a/app/listeners/action_cable_listener.rb b/app/listeners/action_cable_listener.rb
index ff099618c..3bc221504 100644
--- a/app/listeners/action_cable_listener.rb
+++ b/app/listeners/action_cable_listener.rb
@@ -90,6 +90,15 @@ class ActionCableListener < BaseListener
     broadcast(account, tokens, CONVERSATION_UPDATED, conversation.push_event_data)
   end
 
+  def conversation_unread_count_changed(event)
+    account, inbox_members = ::Conversations::UnreadCounts::BroadcastScope.new(event).perform
+    return if account.blank? || !account.feature_enabled?('conversation_unread_counts')
+
+    tokens = user_tokens(account, inbox_members)
+
+    broadcast(account, tokens, CONVERSATION_UNREAD_COUNT_CHANGED, {})
+  end
+
   def conversation_typing_on(event)
     conversation = event.data[:conversation]
     account = conversation.account
diff --git a/app/models/account.rb b/app/models/account.rb
index b4cc03337..efaca8850 100644
--- a/app/models/account.rb
+++ b/app/models/account.rb
@@ -109,6 +109,7 @@ class Account < ApplicationRecord
 
   before_validation :validate_limit_keys
   after_create_commit :notify_creation
+  after_update_commit :clear_unread_conversation_counts_cache, if: :saved_change_to_feature_conversation_unread_counts?
   after_destroy :remove_account_sequences
 
   def agents
@@ -174,6 +175,10 @@ class Account < ApplicationRecord
     Rails.configuration.dispatcher.dispatch(ACCOUNT_CREATED, Time.zone.now, account: self)
   end
 
+  def clear_unread_conversation_counts_cache
+    ::Conversations::UnreadCounts::Store.clear_account!(id)
+  end
+
   trigger.after(:insert).for_each(:row) do
     "execute format('create sequence IF NOT EXISTS conv_dpid_seq_%s', NEW.id);"
   end
diff --git a/app/models/concerns/cache_keys.rb b/app/models/concerns/cache_keys.rb
index 3ad9bbadc..b37d7faa6 100644
--- a/app/models/concerns/cache_keys.rb
+++ b/app/models/concerns/cache_keys.rb
@@ -30,6 +30,7 @@ module CacheKeys
       update_cache_key_for_account(id, model.name.underscore)
     end
 
+    ::Conversations::UnreadCounts::Store.clear_account!(id)
     dispatch_cache_update_event
   end
 
diff --git a/app/models/conversation.rb b/app/models/conversation.rb
index 911cfdac6..0005ae4a4 100644
--- a/app/models/conversation.rb
+++ b/app/models/conversation.rb
@@ -121,6 +121,8 @@ class Conversation < ApplicationRecord
   after_update_commit :execute_after_update_commit_callbacks
   after_create_commit :notify_conversation_creation
   after_create_commit :load_attributes_created_by_db_triggers
+  before_destroy :set_unread_count_deletion_data
+  after_destroy_commit :notify_conversation_deletion
 
   delegate :auto_resolve_after, to: :account
 
@@ -270,6 +272,12 @@ class Conversation < ApplicationRecord
     dispatcher_dispatch(CONVERSATION_CREATED)
   end
 
+  def notify_conversation_deletion
+    return if @unread_count_deletion_data.blank?
+
+    Rails.configuration.dispatcher.dispatch(CONVERSATION_DELETED, Time.zone.now, conversation_data: @unread_count_deletion_data)
+  end
+
   def notify_conversation_updation
     return unless previous_changes.keys.present? && allowed_keys?
 
@@ -315,6 +323,17 @@ class Conversation < ApplicationRecord
                                                                        performed_by: Current.executed_by)
   end
 
+  def set_unread_count_deletion_data
+    @unread_count_deletion_data = {
+      id: id,
+      account_id: account_id,
+      inbox_id: inbox_id,
+      assignee_id: assignee_id,
+      team_id: team_id,
+      cached_label_list: cached_label_list
+    }
+  end
+
   def conversation_status_changed_to_open?
     return false unless open?
     # saved_change_to_status? method only works in case of update
diff --git a/app/services/conversations/unread_counts/broadcast_scope.rb b/app/services/conversations/unread_counts/broadcast_scope.rb
new file mode 100644
index 000000000..8e47050e3
--- /dev/null
+++ b/app/services/conversations/unread_counts/broadcast_scope.rb
@@ -0,0 +1,36 @@
+class Conversations::UnreadCounts::BroadcastScope
+  attr_reader :event
+
+  def initialize(event)
+    @event = event
+  end
+
+  def perform
+    return [conversation.account, conversation.inbox.members] if conversation.present?
+
+    deleted_conversation_scope
+  end
+
+  private
+
+  def conversation
+    event.data[:conversation]
+  end
+
+  def deleted_conversation_scope
+    conversation_data = event.data[:conversation_data]&.with_indifferent_access
+    return if conversation_data.blank?
+
+    account = Account.find_by(id: conversation_data[:account_id])
+    return if account.blank?
+
+    [account, inbox_members_for(account, conversation_data[:inbox_id])]
+  end
+
+  def inbox_members_for(account, inbox_id)
+    inbox = account.inboxes.find_by(id: inbox_id)
+    return User.none if inbox.blank?
+
+    inbox.members
+  end
+end
diff --git a/app/services/conversations/unread_counts/listener.rb b/app/services/conversations/unread_counts/listener.rb
new file mode 100644
index 000000000..28792d884
--- /dev/null
+++ b/app/services/conversations/unread_counts/listener.rb
@@ -0,0 +1,96 @@
+class Conversations::UnreadCounts::Listener < BaseListener
+  include Events::Types
+
+  def message_created(event)
+    message, = extract_message_and_account(event)
+    return unless message.incoming?
+    return unless message.account.feature_enabled?('conversation_unread_counts')
+
+    refresh(message.conversation)
+  end
+
+  def conversation_status_changed(event)
+    conversation, = extract_conversation_and_account(event)
+    refresh(conversation, event.data[:changed_attributes])
+  end
+
+  def conversation_updated(event)
+    return unless label_changed?(event.data[:changed_attributes])
+
+    conversation, = extract_conversation_and_account(event)
+    refresh(conversation, event.data[:changed_attributes])
+  end
+
+  def assignee_changed(event)
+    conversation, = extract_conversation_and_account(event)
+    refresh(conversation, event.data[:changed_attributes])
+  end
+
+  def team_changed(event)
+    conversation, = extract_conversation_and_account(event)
+    refresh(conversation, event.data[:changed_attributes])
+  end
+
+  def conversation_deleted(event)
+    conversation_data = event.data[:conversation_data]&.with_indifferent_access
+    return if conversation_data.blank?
+
+    account = Account.find_by(id: conversation_data[:account_id])
+    return unless account&.feature_enabled?('conversation_unread_counts')
+    return unless remove_deleted_conversation(account, conversation_data)
+
+    Rails.configuration.dispatcher.dispatch(CONVERSATION_UNREAD_COUNT_CHANGED, Time.zone.now, conversation_data: conversation_data.to_h)
+  end
+
+  private
+
+  def refresh(conversation, changed_attributes = nil)
+    ::Conversations::UnreadCounts::Notifier.new(conversation, changed_attributes: changed_attributes).perform
+  end
+
+  def remove_deleted_conversation(account, conversation_data)
+    return false unless store.base_ready?(account.id) || store.assignment_ready?(account.id)
+
+    removed = false
+    removed = remove_deleted_base_membership(account, conversation_data) || removed if store.base_ready?(account.id)
+    removed = remove_deleted_assignment_membership(account, conversation_data) || removed if store.assignment_ready?(account.id)
+    removed
+  end
+
+  def remove_deleted_base_membership(account, conversation_data)
+    store.remove_base_membership(
+      account_id: account.id,
+      inbox_ids: [conversation_data[:inbox_id]],
+      label_ids: label_ids_for(account, conversation_data[:cached_label_list]),
+      team_ids: [conversation_data[:team_id]],
+      conversation_id: conversation_data[:id]
+    )
+  end
+
+  def remove_deleted_assignment_membership(account, conversation_data)
+    store.remove_assignment_membership(
+      account_id: account.id,
+      inbox_ids: [conversation_data[:inbox_id]],
+      label_ids: label_ids_for(account, conversation_data[:cached_label_list]),
+      assignee_ids: [conversation_data[:assignee_id]],
+      team_ids: [conversation_data[:team_id]],
+      conversation_id: conversation_data[:id]
+    )
+  end
+
+  def label_ids_for(account, label_list)
+    label_titles = label_list.to_s.split(',').map(&:strip).compact_blank
+    account.labels.pluck(:title, :id).to_h.values_at(*label_titles).compact
+  end
+
+  def label_changed?(changed_attributes)
+    return false if changed_attributes.blank?
+
+    changed_attributes.key?('label_list') || changed_attributes.key?(:label_list) ||
+      changed_attributes.key?('cached_label_list') || changed_attributes.key?(:cached_label_list)
+  end
+
+  def store
+    ::Conversations::UnreadCounts::Store
+  end
+end
diff --git a/app/services/conversations/unread_counts/notifier.rb b/app/services/conversations/unread_counts/notifier.rb
new file mode 100644
index 000000000..652fbde3a
--- /dev/null
+++ b/app/services/conversations/unread_counts/notifier.rb
@@ -0,0 +1,19 @@
+class Conversations::UnreadCounts::Notifier
+  include Events::Types
+
+  attr_reader :conversation, :changed_attributes
+
+  def initialize(conversation, changed_attributes: nil)
+    @conversation = conversation
+    @changed_attributes = changed_attributes
+  end
+
+  def perform
+    return false unless conversation.account.feature_enabled?('conversation_unread_counts')
+
+    return false unless ::Conversations::UnreadCounts::Refresher.new(conversation, changed_attributes: changed_attributes).perform
+
+    Rails.configuration.dispatcher.dispatch(CONVERSATION_UNREAD_COUNT_CHANGED, Time.zone.now, conversation: conversation)
+    true
+  end
+end
diff --git a/app/services/conversations/unread_counts/refresher.rb b/app/services/conversations/unread_counts/refresher.rb
new file mode 100644
index 000000000..9456cab55
--- /dev/null
+++ b/app/services/conversations/unread_counts/refresher.rb
@@ -0,0 +1,178 @@
+class Conversations::UnreadCounts::Refresher
+  attr_reader :conversation, :changed_attributes
+
+  def initialize(conversation, changed_attributes: nil)
+    @conversation = conversation
+    @changed_attributes = changed_attributes.is_a?(Hash) ? changed_attributes : {}
+  end
+
+  def perform
+    return false unless base_ready? || assignment_ready?
+
+    before_memberships = store.memberships_for_keys(affected_cache_keys, conversation.id)
+    refresh_base_membership if base_ready?
+    refresh_assignment_membership if assignment_ready?
+    after_memberships = store.memberships_for_keys(affected_cache_keys, conversation.id)
+
+    before_memberships != after_memberships
+  end
+
+  private
+
+  def affected_cache_keys
+    keys = []
+    keys.concat(affected_base_keys) if base_ready?
+    keys.concat(affected_assignment_keys) if assignment_ready?
+    keys.uniq
+  end
+
+  def affected_base_keys
+    affected_inbox_ids.flat_map do |inbox_id|
+      [store.inbox_key(account.id, inbox_id)] +
+        affected_label_ids.map { |label_id| store.label_inbox_key(account.id, label_id, inbox_id) } +
+        affected_team_ids.map { |team_id| store.team_inbox_key(account.id, team_id, inbox_id) }
+    end
+  end
+
+  def affected_assignment_keys
+    affected_inbox_ids.flat_map do |inbox_id|
+      affected_assignee_ids.flat_map do |assignee_id|
+        assignment_keys_for(inbox_id, assignee_id)
+      end
+    end
+  end
+
+  def assignment_keys_for(inbox_id, assignee_id)
+    keys = assignee_id.present? ? assignee_keys_for(inbox_id, assignee_id) : unassigned_keys_for(inbox_id)
+
+    keys + affected_team_ids.map { |team_id| team_assignment_key_for(team_id, inbox_id, assignee_id) }
+  end
+
+  def assignee_keys_for(inbox_id, assignee_id)
+    [store.inbox_assignee_key(account.id, inbox_id, assignee_id)] +
+      affected_label_ids.map { |label_id| store.label_inbox_assignee_key(account.id, label_id, inbox_id, assignee_id) }
+  end
+
+  def unassigned_keys_for(inbox_id)
+    [store.inbox_unassigned_key(account.id, inbox_id)] +
+      affected_label_ids.map { |label_id| store.label_inbox_unassigned_key(account.id, label_id, inbox_id) }
+  end
+
+  def refresh_base_membership
+    store.remove_base_membership(
+      account_id: account.id,
+      inbox_ids: affected_inbox_ids,
+      label_ids: affected_label_ids,
+      team_ids: affected_team_ids,
+      conversation_id: conversation.id
+    )
+    return unless unread?
+
+    store.add_base_membership(
+      account_id: account.id,
+      inbox_id: conversation.inbox_id,
+      label_ids: current_label_ids,
+      team_id: conversation.team_id,
+      conversation_id: conversation.id
+    )
+  end
+
+  def refresh_assignment_membership
+    store.remove_assignment_membership(
+      account_id: account.id,
+      inbox_ids: affected_inbox_ids,
+      label_ids: affected_label_ids,
+      assignee_ids: affected_assignee_ids,
+      team_ids: affected_team_ids,
+      conversation_id: conversation.id
+    )
+    return unless unread?
+
+    store.add_assignment_membership(
+      account_id: account.id,
+      inbox_id: conversation.inbox_id,
+      label_ids: current_label_ids,
+      assignee_id: conversation.assignee_id,
+      team_id: conversation.team_id,
+      conversation_id: conversation.id
+    )
+  end
+
+  def unread?
+    # Sidebar unread counts intentionally track only open conversations.
+    return false unless conversation.open?
+
+    incoming_messages = conversation.messages.incoming.where(account_id: account.id)
+    if conversation.agent_last_seen_at
+      incoming_messages = incoming_messages.where(Message.arel_table[:created_at].gt(conversation.agent_last_seen_at))
+    end
+    incoming_messages.exists?
+  end
+
+  def affected_inbox_ids
+    [previous_value_for(:inbox_id), conversation.inbox_id].compact.uniq
+  end
+
+  def affected_assignee_ids
+    return [conversation.assignee_id].compact unless changed_attribute?(:assignee_id)
+
+    [previous_value_for(:assignee_id), conversation.assignee_id].uniq
+  end
+
+  def affected_label_ids
+    (previous_label_ids + current_label_ids).uniq
+  end
+
+  def affected_team_ids
+    [previous_value_for(:team_id), conversation.team_id].compact.uniq
+  end
+
+  def previous_label_ids
+    label_ids_for(previous_value_for(:label_list) || previous_value_for(:cached_label_list) || conversation.cached_label_list)
+  end
+
+  def current_label_ids
+    @current_label_ids ||= label_ids_for(conversation.cached_label_list)
+  end
+
+  def label_ids_for(label_list)
+    labels = label_list.is_a?(Array) ? label_list : label_list.to_s.split(',')
+    label_titles = labels.map(&:to_s).map(&:strip).compact_blank
+    labels_by_title.values_at(*label_titles).compact
+  end
+
+  def previous_value_for(attribute)
+    change = changed_attributes[attribute.to_s] || changed_attributes[attribute.to_sym]
+    change&.first
+  end
+
+  def changed_attribute?(attribute)
+    changed_attributes.key?(attribute.to_s) || changed_attributes.key?(attribute.to_sym)
+  end
+
+  def team_assignment_key_for(team_id, inbox_id, assignee_id)
+    return store.team_inbox_assignee_key(account.id, team_id, inbox_id, assignee_id) if assignee_id.present?
+
+    store.team_inbox_unassigned_key(account.id, team_id, inbox_id)
+  end
+
+  def labels_by_title
+    @labels_by_title ||= account.labels.pluck(:title, :id).to_h
+  end
+
+  def base_ready?
+    @base_ready ||= store.base_ready?(account.id)
+  end
+
+  def assignment_ready?
+    @assignment_ready ||= store.assignment_ready?(account.id)
+  end
+
+  def account
+    conversation.account
+  end
+
+  def store
+    ::Conversations::UnreadCounts::Store
+  end
+end
diff --git a/app/services/conversations/unread_counts/store.rb b/app/services/conversations/unread_counts/store.rb
index dbd01f235..e51fc5c09 100644
--- a/app/services/conversations/unread_counts/store.rb
+++ b/app/services/conversations/unread_counts/store.rb
@@ -1,4 +1,6 @@
 class Conversations::UnreadCounts::Store
+  extend ::Conversations::UnreadCounts::StoreKeys
+
   class << self
     def base_ready?(account_id)
       Redis::Alfred.exists?(base_ready_key(account_id))
@@ -78,46 +80,14 @@ class Conversations::UnreadCounts::Store
       keys.zip(counts).to_h
     end
 
-    def inbox_key(account_id, inbox_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_INBOX, account_id: account_id, inbox_id: inbox_id)
-    end
+    def memberships_for_keys(keys, conversation_id)
+      keys = keys.compact_blank
+      return {} if keys.blank?
 
-    def label_inbox_key(account_id, label_id, inbox_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_LABEL_INBOX, account_id: account_id, label_id: label_id, inbox_id: inbox_id)
-    end
-
-    def team_inbox_key(account_id, team_id, inbox_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_TEAM_INBOX, account_id: account_id, team_id: team_id, inbox_id: inbox_id)
-    end
-
-    def inbox_unassigned_key(account_id, inbox_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_INBOX_UNASSIGNED, account_id: account_id, inbox_id: inbox_id)
-    end
-
-    def inbox_assignee_key(account_id, inbox_id, user_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_INBOX_ASSIGNEE, account_id: account_id, inbox_id: inbox_id, user_id: user_id)
-    end
-
-    def label_inbox_unassigned_key(account_id, label_id, inbox_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_LABEL_INBOX_UNASSIGNED, account_id: account_id, label_id: label_id, inbox_id: inbox_id)
-    end
-
-    def label_inbox_assignee_key(account_id, label_id, inbox_id, user_id)
-      format(
-        Redis::Alfred::UNREAD_CONVERSATIONS_LABEL_INBOX_ASSIGNEE,
-        account_id: account_id,
-        label_id: label_id,
-        inbox_id: inbox_id,
-        user_id: user_id
-      )
-    end
-
-    def team_inbox_unassigned_key(account_id, team_id, inbox_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_TEAM_INBOX_UNASSIGNED, account_id: account_id, team_id: team_id, inbox_id: inbox_id)
-    end
-
-    def team_inbox_assignee_key(account_id, team_id, inbox_id, user_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_TEAM_INBOX_ASSIGNEE, account_id: account_id, team_id: team_id, inbox_id: inbox_id, user_id: user_id)
+      memberships = Redis::Alfred.pipelined do |pipeline|
+        keys.each { |key| pipeline.sismember(key, conversation_id) }
+      end
+      keys.zip(memberships.map { |membership| membership == true || membership == 1 }).to_h
     end
 
     private
@@ -183,7 +153,16 @@ class Conversations::UnreadCounts::Store
     end
 
     def remove_from_sets(keys, conversation_id)
-      write_to_sets(keys) { |pipeline, key| pipeline.srem(key, conversation_id) }
+      keys = keys.compact_blank
+      return false if keys.blank?
+
+      results = Redis::Alfred.pipelined do |pipeline|
+        keys.each do |key|
+          pipeline.srem(key, conversation_id)
+          pipeline.expire(key, Conversations::UnreadCounts::SET_TTL)
+        end
+      end
+      results.each_slice(2).any? { |removed, _| removed == true || (removed.respond_to?(:to_i) && removed.to_i.positive?) }
     end
 
     def write_to_sets(keys)
diff --git a/app/services/conversations/unread_counts/store_keys.rb b/app/services/conversations/unread_counts/store_keys.rb
new file mode 100644
index 000000000..bbb6d101a
--- /dev/null
+++ b/app/services/conversations/unread_counts/store_keys.rb
@@ -0,0 +1,43 @@
+module Conversations::UnreadCounts::StoreKeys
+  def inbox_key(account_id, inbox_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_INBOX, account_id: account_id, inbox_id: inbox_id)
+  end
+
+  def label_inbox_key(account_id, label_id, inbox_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_LABEL_INBOX, account_id: account_id, label_id: label_id, inbox_id: inbox_id)
+  end
+
+  def team_inbox_key(account_id, team_id, inbox_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_TEAM_INBOX, account_id: account_id, team_id: team_id, inbox_id: inbox_id)
+  end
+
+  def inbox_unassigned_key(account_id, inbox_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_INBOX_UNASSIGNED, account_id: account_id, inbox_id: inbox_id)
+  end
+
+  def inbox_assignee_key(account_id, inbox_id, user_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_INBOX_ASSIGNEE, account_id: account_id, inbox_id: inbox_id, user_id: user_id)
+  end
+
+  def label_inbox_unassigned_key(account_id, label_id, inbox_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_LABEL_INBOX_UNASSIGNED, account_id: account_id, label_id: label_id, inbox_id: inbox_id)
+  end
+
+  def label_inbox_assignee_key(account_id, label_id, inbox_id, user_id)
+    format(
+      Redis::Alfred::UNREAD_CONVERSATIONS_LABEL_INBOX_ASSIGNEE,
+      account_id: account_id,
+      label_id: label_id,
+      inbox_id: inbox_id,
+      user_id: user_id
+    )
+  end
+
+  def team_inbox_unassigned_key(account_id, team_id, inbox_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_TEAM_INBOX_UNASSIGNED, account_id: account_id, team_id: team_id, inbox_id: inbox_id)
+  end
+
+  def team_inbox_assignee_key(account_id, team_id, inbox_id, user_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_TEAM_INBOX_ASSIGNEE, account_id: account_id, team_id: team_id, inbox_id: inbox_id, user_id: user_id)
+  end
+end
diff --git a/config/features.yml b/config/features.yml
index c469fed90..03105588b 100644
--- a/config/features.yml
+++ b/config/features.yml
@@ -17,10 +17,10 @@
   display_name: Facebook Channel
   enabled: true
   help_url: https://chwt.app/hc/fb
-- name: channel_twitter
-  display_name: Twitter Channel
-  enabled: true
-  deprecated: true
+- name: conversation_unread_counts
+  display_name: Conversation Unread Counts
+  enabled: false
+  chatwoot_internal: true
 - name: ip_lookup
   display_name: IP Lookup
   enabled: false
diff --git a/config/locales/en.yml b/config/locales/en.yml
index 8c66632d8..18b721caf 100644
--- a/config/locales/en.yml
+++ b/config/locales/en.yml
@@ -81,6 +81,9 @@ en:
     saml:
       feature_not_enabled: SAML feature not enabled for this account
       sso_not_enabled: SAML SSO is not enabled for this installation
+    conversations:
+      unread_counts:
+        feature_not_enabled: Conversation unread counts feature not enabled for this account
     data_import:
       data_type:
         invalid: Invalid data type
diff --git a/config/routes.rb b/config/routes.rb
index 355491d5b..e7f8a4745 100644
--- a/config/routes.rb
+++ b/config/routes.rb
@@ -133,6 +133,7 @@ Rails.application.routes.draw do
             collection do
               get :meta
               get :search
+              get :unread_counts, to: 'conversations/unread_counts#index'
               post :filter
             end
             scope module: :conversations do
diff --git a/db/migrate/20260508000000_repurpose_channel_twitter_flag_for_conversation_unread_counts.rb b/db/migrate/20260508000000_repurpose_channel_twitter_flag_for_conversation_unread_counts.rb
new file mode 100644
index 000000000..4c0bff5bd
--- /dev/null
+++ b/db/migrate/20260508000000_repurpose_channel_twitter_flag_for_conversation_unread_counts.rb
@@ -0,0 +1,20 @@
+class RepurposeChannelTwitterFlagForConversationUnreadCounts < ActiveRecord::Migration[7.1]
+  def up
+    # The channel_twitter flag (deprecated) has been renamed to conversation_unread_counts.
+    # Disable it on any accounts that had channel_twitter enabled so the repurposed
+    # flag starts in its intended default-off state.
+    Account.feature_conversation_unread_counts.find_each(batch_size: 100) do |account|
+      account.disable_features(:conversation_unread_counts)
+      account.save!(validate: false)
+    end
+
+    # Remove the stale channel_twitter entry from ACCOUNT_LEVEL_FEATURE_DEFAULTS.
+    # ConfigLoader only adds new flags; it never removes renamed ones.
+    config = InstallationConfig.find_by(name: 'ACCOUNT_LEVEL_FEATURE_DEFAULTS')
+    return if config&.value.blank?
+
+    config.value = config.value.reject { |feature| feature['name'] == 'channel_twitter' }
+    config.save!
+    GlobalConfig.clear_cache
+  end
+end
diff --git a/lib/events/types.rb b/lib/events/types.rb
index d742232c4..171649a7d 100644
--- a/lib/events/types.rb
+++ b/lib/events/types.rb
@@ -16,6 +16,7 @@ module Events::Types
   # conversation events
   CONVERSATION_CREATED = 'conversation.created'
   CONVERSATION_UPDATED = 'conversation.updated'
+  CONVERSATION_DELETED = 'conversation.deleted'
   CONVERSATION_READ = 'conversation.read'
   CONVERSATION_BOT_HANDOFF = 'conversation.bot_handoff'
   # FIXME: deprecate the opened and resolved events in future in favor of status changed event.
@@ -26,6 +27,7 @@ module Events::Types
 
   CONVERSATION_STATUS_CHANGED = 'conversation.status_changed'
   CONVERSATION_CONTACT_CHANGED = 'conversation.contact_changed'
+  CONVERSATION_UNREAD_COUNT_CHANGED = 'conversation.unread_count_changed'
   ASSIGNEE_CHANGED = 'assignee.changed'
   TEAM_CHANGED = 'team.changed'
   CONVERSATION_TYPING_ON = 'conversation.typing_on'
diff --git a/spec/controllers/api/v1/accounts/conversations_controller_spec.rb b/spec/controllers/api/v1/accounts/conversations_controller_spec.rb
index 19d080b47..f8fd446d2 100644
--- a/spec/controllers/api/v1/accounts/conversations_controller_spec.rb
+++ b/spec/controllers/api/v1/accounts/conversations_controller_spec.rb
@@ -101,6 +101,76 @@ RSpec.describe 'Conversations API', type: :request do
     end
   end
 
+  describe 'GET /api/v1/accounts/{account.id}/conversations/unread_counts' do
+    context 'when it is an unauthenticated user' do
+      it 'returns unauthorized' do
+        get "/api/v1/accounts/#{account.id}/conversations/unread_counts"
+
+        expect(response).to have_http_status(:unauthorized)
+      end
+    end
+
+    context 'when it is an authenticated user' do
+      let(:agent) { create(:user, account: account, role: :agent) }
+      let(:visible_inbox) { create(:inbox, account: account) }
+      let(:hidden_inbox) { create(:inbox, account: account) }
+      let(:label) { create(:label, account: account, title: 'billing', show_on_sidebar: true) }
+      let(:team) { create(:team, account: account, allow_auto_assign: false) }
+
+      before do
+        create(:inbox_member, user: agent, inbox: visible_inbox)
+        create(:team_member, user: agent, team: team)
+      end
+
+      after do
+        Conversations::UnreadCounts::Store.clear_account!(account.id)
+      end
+
+      context 'when conversation unread counts feature is enabled' do
+        before do
+          account.enable_features!(:conversation_unread_counts)
+        end
+
+        it 'returns unread conversation counts scoped to the signed-in user' do
+          create_unread_conversation(account: account, inbox: visible_inbox, labels: [label.title])
+          create_unread_conversation(account: account, inbox: hidden_inbox, labels: [label.title])
+
+          get "/api/v1/accounts/#{account.id}/conversations/unread_counts",
+              headers: agent.create_new_auth_token,
+              as: :json
+
+          expect(response).to have_http_status(:success)
+          expect(response.parsed_body['payload']).to eq(
+            'inboxes' => { visible_inbox.id.to_s => 1 },
+            'labels' => { label.id.to_s => 1 },
+            'teams' => {}
+          )
+        end
+
+        it 'returns unread team conversation counts scoped to the signed-in user' do
+          create_unread_conversation(account: account, inbox: visible_inbox, team: team)
+          create_unread_conversation(account: account, inbox: hidden_inbox, team: team)
+
+          get "/api/v1/accounts/#{account.id}/conversations/unread_counts",
+              headers: agent.create_new_auth_token,
+              as: :json
+
+          expect(response).to have_http_status(:success)
+          expect(response.parsed_body['payload']['teams']).to eq(team.id.to_s => 1)
+        end
+      end
+
+      it 'returns forbidden when conversation unread counts feature is disabled' do
+        get "/api/v1/accounts/#{account.id}/conversations/unread_counts",
+            headers: agent.create_new_auth_token,
+            as: :json
+
+        expect(response).to have_http_status(:forbidden)
+        expect(response.parsed_body['error']).to eq('Conversation unread counts feature not enabled for this account')
+      end
+    end
+  end
+
   describe 'GET /api/v1/accounts/{account.id}/conversations/search' do
     context 'when it is an unauthenticated user' do
       it 'returns unauthorized' do
@@ -777,6 +847,23 @@ RSpec.describe 'Conversations API', type: :request do
         expect(conversation.reload.agent_last_seen_at).to be > initial_last_seen
       end
 
+      it 'refreshes unread count cache when conversation is marked read' do
+        account.enable_features!(:conversation_unread_counts)
+        conversation.update!(agent_last_seen_at: 1.hour.ago)
+        create(:message, account: account, inbox: conversation.inbox, conversation: conversation, message_type: :incoming, created_at: 5.minutes.ago)
+        Conversations::UnreadCounts::Builder.new(account).build_base!
+
+        post "/api/v1/accounts/#{account.id}/conversations/#{conversation.display_id}/update_last_seen",
+             headers: agent.create_new_auth_token,
+             as: :json
+
+        inbox_key = Conversations::UnreadCounts::Store.inbox_key(account.id, conversation.inbox_id)
+        expect(response).to have_http_status(:success)
+        expect(Conversations::UnreadCounts::Store.counts_for_keys([inbox_key])).to eq(inbox_key => 0)
+      ensure
+        Conversations::UnreadCounts::Store.clear_account!(account.id)
+      end
+
       it 'updates both if one timestamp is old even when the other is recent' do
         conversation.update!(assignee_id: agent.id, agent_last_seen_at: 2.hours.ago, assignee_last_seen_at: 30.minutes.ago)
         # Ensure all messages are older than assignee_last_seen_at (no unread messages)
@@ -847,6 +934,22 @@ RSpec.describe 'Conversations API', type: :request do
         expect(conversation.reload.agent_last_seen_at).to eq(last_seen_at)
         expect(conversation.reload.assignee_last_seen_at).to eq(last_seen_at)
       end
+
+      it 'refreshes unread count cache when conversation is marked unread' do
+        account.enable_features!(:conversation_unread_counts)
+        conversation.update!(agent_last_seen_at: 1.minute.from_now, assignee_last_seen_at: 1.minute.from_now)
+        Conversations::UnreadCounts::Builder.new(account).build_base!
+
+        post "/api/v1/accounts/#{account.id}/conversations/#{conversation.display_id}/unread",
+             headers: agent.create_new_auth_token,
+             as: :json
+
+        inbox_key = Conversations::UnreadCounts::Store.inbox_key(account.id, conversation.inbox_id)
+        expect(response).to have_http_status(:success)
+        expect(Conversations::UnreadCounts::Store.counts_for_keys([inbox_key])).to eq(inbox_key => 1)
+      ensure
+        Conversations::UnreadCounts::Store.clear_account!(account.id)
+      end
     end
   end
 
diff --git a/spec/controllers/super_admin/accounts_controller_spec.rb b/spec/controllers/super_admin/accounts_controller_spec.rb
index 6b7e6eeed..e4ff81a08 100644
--- a/spec/controllers/super_admin/accounts_controller_spec.rb
+++ b/spec/controllers/super_admin/accounts_controller_spec.rb
@@ -32,6 +32,10 @@ RSpec.describe 'Super Admin accounts API', type: :request do
       create(:team, account: account)
     end
 
+    after do
+      Conversations::UnreadCounts::Store.clear_account!(account.id)
+    end
+
     context 'when it is an unauthenticated user' do
       it 'returns unauthorized' do
         post "/super_admin/accounts/#{account.id}/reset_cache"
@@ -52,6 +56,21 @@ RSpec.describe 'Super Admin accounts API', type: :request do
         range = now_timestamp..(now_timestamp + 10)
         expect(account.reload.cache_keys.values.all? { |v| range.cover?(v.to_i) }).to be(true)
       end
+
+      it 'clears conversation unread count cache' do
+        inbox = account.inboxes.first
+        store = Conversations::UnreadCounts::Store
+        inbox_key = store.inbox_key(account.id, inbox.id)
+        store.mark_base_ready!(account.id)
+        store.add_base_membership(account_id: account.id, inbox_id: inbox.id, label_ids: [], conversation_id: 1)
+
+        sign_in(super_admin, scope: :super_admin)
+        post "/super_admin/accounts/#{account.id}/reset_cache"
+
+        expect(response).to have_http_status(:redirect)
+        expect(store.base_ready?(account.id)).to be(false)
+        expect(store.counts_for_keys([inbox_key])).to eq(inbox_key => 0)
+      end
     end
   end
 
diff --git a/spec/listeners/action_cable_listener_spec.rb b/spec/listeners/action_cable_listener_spec.rb
index 8b18f1582..cdb9a93cb 100644
--- a/spec/listeners/action_cable_listener_spec.rb
+++ b/spec/listeners/action_cable_listener_spec.rb
@@ -231,4 +231,68 @@ describe ActionCableListener do
       listener.conversation_updated(event)
     end
   end
+
+  describe '#conversation_unread_count_changed' do
+    let(:event_name) { :'conversation.unread_count_changed' }
+    let!(:agent_without_inbox_access) { create(:user, account: account, role: :agent) }
+    let!(:event) { Events::Base.new(event_name, Time.zone.now, conversation: conversation) }
+
+    before do
+      account.enable_features!(:conversation_unread_counts)
+    end
+
+    it 'sends a lightweight refresh event to inbox agents and admins' do
+      expect(conversation.inbox.reload.inbox_members.count).to eq(1)
+
+      expect(ActionCableBroadcastJob).to receive(:perform_later).with(
+        a_collection_containing_exactly(agent.pubsub_token, admin.pubsub_token),
+        'conversation.unread_count_changed',
+        {
+          account_id: account.id
+        }
+      )
+
+      listener.conversation_unread_count_changed(event)
+    end
+
+    it 'does not broadcast unread count refresh to agents outside the inbox' do
+      expect(ActionCableBroadcastJob).not_to receive(:perform_later).with(
+        array_including(agent_without_inbox_access.pubsub_token),
+        anything,
+        anything
+      )
+
+      listener.conversation_unread_count_changed(event)
+    end
+
+    it 'does not broadcast when conversation unread counts feature is disabled' do
+      account.disable_features!(:conversation_unread_counts)
+
+      expect(ActionCableBroadcastJob).not_to receive(:perform_later)
+
+      listener.conversation_unread_count_changed(event)
+    end
+
+    it 'supports deleted conversation data' do
+      event = Events::Base.new(
+        event_name,
+        Time.zone.now,
+        conversation_data: {
+          id: conversation.id,
+          account_id: account.id,
+          inbox_id: conversation.inbox_id
+        }
+      )
+
+      expect(ActionCableBroadcastJob).to receive(:perform_later).with(
+        a_collection_containing_exactly(agent.pubsub_token, admin.pubsub_token),
+        'conversation.unread_count_changed',
+        {
+          account_id: account.id
+        }
+      )
+
+      listener.conversation_unread_count_changed(event)
+    end
+  end
 end
diff --git a/spec/models/account_spec.rb b/spec/models/account_spec.rb
index 76dbbcba2..38ca9694a 100644
--- a/spec/models/account_spec.rb
+++ b/spec/models/account_spec.rb
@@ -50,6 +50,44 @@ RSpec.describe Account do
     end
   end
 
+  describe 'conversation unread counts feature flag' do
+    let(:account) { create(:account) }
+    let(:inbox) { create(:inbox, account: account) }
+    let(:store) { Conversations::UnreadCounts::Store }
+    let(:inbox_key) { store.inbox_key(account.id, inbox.id) }
+
+    after do
+      store.clear_account!(account.id)
+    end
+
+    it 'clears unread count cache when the feature is enabled' do
+      build_unread_count_cache
+
+      account.enable_features!(:conversation_unread_counts)
+
+      expect(store.base_ready?(account.id)).to be(false)
+      expect(store.assignment_ready?(account.id)).to be(false)
+      expect(store.counts_for_keys([inbox_key])).to eq(inbox_key => 0)
+    end
+
+    it 'clears unread count cache when the feature is disabled' do
+      account.enable_features!(:conversation_unread_counts)
+      build_unread_count_cache
+
+      account.disable_features!(:conversation_unread_counts)
+
+      expect(store.base_ready?(account.id)).to be(false)
+      expect(store.assignment_ready?(account.id)).to be(false)
+      expect(store.counts_for_keys([inbox_key])).to eq(inbox_key => 0)
+    end
+
+    def build_unread_count_cache
+      store.mark_base_ready!(account.id)
+      store.mark_assignment_ready!(account.id)
+      store.add_base_membership(account_id: account.id, inbox_id: inbox.id, label_ids: [], conversation_id: 1)
+    end
+  end
+
   describe 'inbound_email_domain' do
     let(:account) { create(:account) }
 
diff --git a/spec/models/conversation_spec.rb b/spec/models/conversation_spec.rb
index 0bf90859e..58d64ea94 100644
--- a/spec/models/conversation_spec.rb
+++ b/spec/models/conversation_spec.rb
@@ -717,6 +717,25 @@ RSpec.describe Conversation do
 
       expect { notification.reload }.to raise_error ActiveRecord::RecordNotFound
     end
+
+    it 'dispatches conversation deleted event with unread count cache data' do
+      allow(Rails.configuration.dispatcher).to receive(:dispatch)
+
+      conversation.destroy!
+
+      expect(Rails.configuration.dispatcher).to have_received(:dispatch).with(
+        'conversation.deleted',
+        kind_of(Time),
+        conversation_data: {
+          id: conversation.id,
+          account_id: conversation.account_id,
+          inbox_id: conversation.inbox_id,
+          assignee_id: conversation.assignee_id,
+          team_id: conversation.team_id,
+          cached_label_list: conversation.cached_label_list
+        }
+      )
+    end
   end
 
   describe 'validate invalid referer url' do
diff --git a/spec/services/conversations/unread_counts/listener_spec.rb b/spec/services/conversations/unread_counts/listener_spec.rb
new file mode 100644
index 000000000..fbb0a0835
--- /dev/null
+++ b/spec/services/conversations/unread_counts/listener_spec.rb
@@ -0,0 +1,164 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::Listener do
+  let(:listener) { described_class.instance }
+  let(:account) { create(:account) }
+  let(:conversation) { create(:conversation, account: account) }
+  let(:notifier) { instance_double(Conversations::UnreadCounts::Notifier, perform: true) }
+
+  before do
+    allow(Conversations::UnreadCounts::Notifier).to receive(:new).and_return(notifier)
+  end
+
+  it 'refreshes unread counts when an incoming message is created' do
+    account.enable_features!(:conversation_unread_counts)
+    message = create(:message, account: account, inbox: conversation.inbox, conversation: conversation, message_type: :incoming)
+    event = Events::Base.new('message.created', Time.zone.now, message: message)
+
+    listener.message_created(event)
+
+    expect(Conversations::UnreadCounts::Notifier).to have_received(:new).with(conversation, changed_attributes: nil)
+    expect(notifier).to have_received(:perform)
+  end
+
+  it 'ignores outgoing message creation' do
+    message = create(:message, account: account, inbox: conversation.inbox, conversation: conversation, message_type: :outgoing)
+    event = Events::Base.new('message.created', Time.zone.now, message: message)
+
+    listener.message_created(event)
+
+    expect(Conversations::UnreadCounts::Notifier).not_to have_received(:new)
+  end
+
+  it 'ignores incoming message creation when conversation unread counts are disabled' do
+    message = create(:message, account: account, inbox: conversation.inbox, conversation: conversation, message_type: :incoming)
+    event = Events::Base.new('message.created', Time.zone.now, message: message)
+
+    expect(message).not_to receive(:conversation)
+
+    listener.message_created(event)
+
+    expect(Conversations::UnreadCounts::Notifier).not_to have_received(:new)
+  end
+
+  it 'refreshes unread counts when conversation status changes' do
+    changed_attributes = { 'status' => %w[open resolved] }
+    event = Events::Base.new('conversation.status_changed', Time.zone.now, conversation: conversation, changed_attributes: changed_attributes)
+
+    listener.conversation_status_changed(event)
+
+    expect(Conversations::UnreadCounts::Notifier).to have_received(:new).with(conversation, changed_attributes: changed_attributes)
+    expect(notifier).to have_received(:perform)
+  end
+
+  it 'refreshes unread counts when labels change' do
+    changed_attributes = { label_list: [%w[old], %w[new]] }
+    event = Events::Base.new('conversation.updated', Time.zone.now, conversation: conversation, changed_attributes: changed_attributes)
+
+    listener.conversation_updated(event)
+
+    expect(Conversations::UnreadCounts::Notifier).to have_received(:new).with(conversation, changed_attributes: changed_attributes)
+    expect(notifier).to have_received(:perform)
+  end
+
+  it 'ignores conversation updates unrelated to unread count dimensions' do
+    event = Events::Base.new('conversation.updated', Time.zone.now, conversation: conversation, changed_attributes: { priority: [nil, 'high'] })
+
+    listener.conversation_updated(event)
+
+    expect(Conversations::UnreadCounts::Notifier).not_to have_received(:new)
+  end
+
+  it 'refreshes unread counts when assignee changes' do
+    changed_attributes = { assignee_id: [nil, 1] }
+    event = Events::Base.new('assignee.changed', Time.zone.now, conversation: conversation, changed_attributes: changed_attributes)
+
+    listener.assignee_changed(event)
+
+    expect(Conversations::UnreadCounts::Notifier).to have_received(:new).with(conversation, changed_attributes: changed_attributes)
+    expect(notifier).to have_received(:perform)
+  end
+
+  it 'refreshes unread counts when team changes' do
+    changed_attributes = { team_id: [nil, 1] }
+    event = Events::Base.new('team.changed', Time.zone.now, conversation: conversation, changed_attributes: changed_attributes)
+
+    listener.team_changed(event)
+
+    expect(Conversations::UnreadCounts::Notifier).to have_received(:new).with(conversation, changed_attributes: changed_attributes)
+    expect(notifier).to have_received(:perform)
+  end
+
+  it 'removes unread count memberships when a conversation is deleted' do
+    account.enable_features!(:conversation_unread_counts)
+    label = create(:label, account: account)
+    team = create(:team, account: account)
+    assignee = create(:user, account: account)
+    create(:team_member, team: team, user: assignee)
+    conversation.update!(assignee_id: assignee.id, team: team)
+    conversation.update_labels([label.title])
+    conversation.reload
+    conversation_data = deleted_conversation_data(conversation)
+    store.mark_base_ready!(account.id)
+    store.mark_assignment_ready!(account.id)
+    store.add_base_membership(
+      account_id: account.id,
+      inbox_id: conversation.inbox_id,
+      label_ids: [label.id],
+      team_id: team.id,
+      conversation_id: conversation.id
+    )
+    store.add_assignment_membership(
+      account_id: account.id,
+      inbox_id: conversation.inbox_id,
+      label_ids: [label.id],
+      assignee_id: assignee.id,
+      team_id: team.id,
+      conversation_id: conversation.id
+    )
+    allow(Rails.configuration.dispatcher).to receive(:dispatch)
+
+    listener.conversation_deleted(Events::Base.new('conversation.deleted', Time.zone.now, conversation_data: conversation_data))
+
+    expect(store.counts_for_keys(deleted_base_keys(conversation, label, team)).values).to all(eq(0))
+    expect(store.counts_for_keys(deleted_assignment_keys(conversation, label, team, assignee)).values).to all(eq(0))
+    expect(Rails.configuration.dispatcher).to have_received(:dispatch).with(
+      'conversation.unread_count_changed',
+      kind_of(Time),
+      conversation_data: conversation_data.stringify_keys
+    )
+  ensure
+    store.clear_account!(account.id)
+  end
+
+  def deleted_conversation_data(conversation)
+    {
+      id: conversation.id,
+      account_id: conversation.account_id,
+      inbox_id: conversation.inbox_id,
+      assignee_id: conversation.assignee_id,
+      team_id: conversation.team_id,
+      cached_label_list: conversation.cached_label_list
+    }
+  end
+
+  def deleted_base_keys(conversation, label, team)
+    [
+      store.inbox_key(account.id, conversation.inbox_id),
+      store.label_inbox_key(account.id, label.id, conversation.inbox_id),
+      store.team_inbox_key(account.id, team.id, conversation.inbox_id)
+    ]
+  end
+
+  def deleted_assignment_keys(conversation, label, team, assignee)
+    [
+      store.inbox_assignee_key(account.id, conversation.inbox_id, assignee.id),
+      store.label_inbox_assignee_key(account.id, label.id, conversation.inbox_id, assignee.id),
+      store.team_inbox_assignee_key(account.id, team.id, conversation.inbox_id, assignee.id)
+    ]
+  end
+
+  def store
+    Conversations::UnreadCounts::Store
+  end
+end
diff --git a/spec/services/conversations/unread_counts/notifier_spec.rb b/spec/services/conversations/unread_counts/notifier_spec.rb
new file mode 100644
index 000000000..1b35d37f6
--- /dev/null
+++ b/spec/services/conversations/unread_counts/notifier_spec.rb
@@ -0,0 +1,48 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::Notifier do
+  let!(:conversation) { create(:conversation) }
+  let(:refresher) { instance_double(Conversations::UnreadCounts::Refresher, perform: refresh_result) }
+  let(:refresh_result) { true }
+
+  before do
+    conversation.account.enable_features!(:conversation_unread_counts)
+    allow(Conversations::UnreadCounts::Refresher).to receive(:new).and_return(refresher)
+    allow(Rails.configuration.dispatcher).to receive(:dispatch)
+  end
+
+  it 'dispatches unread count changed event after a successful refresh' do
+    described_class.new(conversation).perform
+
+    expect(Rails.configuration.dispatcher).to have_received(:dispatch).with(
+      'conversation.unread_count_changed',
+      kind_of(Time),
+      conversation: conversation
+    )
+  end
+
+  context 'when refresh does not change unread count memberships' do
+    let(:refresh_result) { false }
+
+    it 'does not dispatch unread count changed event' do
+      described_class.new(conversation).perform
+
+      expect(Rails.configuration.dispatcher).not_to have_received(:dispatch)
+    end
+  end
+
+  context 'when conversation unread counts feature is disabled' do
+    before do
+      conversation.account.disable_features!(:conversation_unread_counts)
+      allow(Conversations::UnreadCounts::Store).to receive(:clear_account!)
+    end
+
+    it 'does not refresh, clear cache, or dispatch unread count changed event' do
+      described_class.new(conversation).perform
+
+      expect(Conversations::UnreadCounts::Refresher).not_to have_received(:new)
+      expect(Conversations::UnreadCounts::Store).not_to have_received(:clear_account!)
+      expect(Rails.configuration.dispatcher).not_to have_received(:dispatch)
+    end
+  end
+end
diff --git a/spec/services/conversations/unread_counts/refresher_spec.rb b/spec/services/conversations/unread_counts/refresher_spec.rb
new file mode 100644
index 000000000..9b22c2186
--- /dev/null
+++ b/spec/services/conversations/unread_counts/refresher_spec.rb
@@ -0,0 +1,145 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::Refresher do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:label) { create(:label, account: account, title: 'urgent', show_on_sidebar: true) }
+  let(:new_label) { create(:label, account: account, title: 'billing', show_on_sidebar: true) }
+  let(:team) { create(:team, account: account, allow_auto_assign: false) }
+  let(:new_team) { create(:team, account: account, allow_auto_assign: false) }
+  let(:assignee) { create(:user, account: account, role: :agent) }
+  let(:other_assignee) { create(:user, account: account, role: :agent) }
+  let(:store) { Conversations::UnreadCounts::Store }
+
+  after do
+    store.clear_account!(account.id)
+  end
+
+  it 'does not update redis when unread caches are not ready' do
+    conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title])
+
+    expect(described_class.new(conversation).perform).to be(false)
+    expect(store.counts_for_keys([store.inbox_key(account.id, inbox.id)])).to eq(store.inbox_key(account.id, inbox.id) => 0)
+  end
+
+  it 'adds an unread conversation to base cache' do
+    conversation = create(:conversation, account: account, inbox: inbox, agent_last_seen_at: 1.hour.ago)
+    conversation.update_labels([label.title])
+    store.mark_base_ready!(account.id)
+
+    create(:message, account: account, inbox: inbox, conversation: conversation, message_type: :incoming, created_at: 5.minutes.ago)
+
+    expect(described_class.new(conversation.reload).perform).to be(true)
+
+    expect(store.counts_for_keys(base_keys)).to eq(
+      store.inbox_key(account.id, inbox.id) => 1,
+      store.label_inbox_key(account.id, label.id, inbox.id) => 1
+    )
+  end
+
+  it 'returns false when refresh does not change unread counts' do
+    conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title])
+    Conversations::UnreadCounts::Builder.new(account).build_base!
+
+    create(:message, account: account, inbox: inbox, conversation: conversation, message_type: :incoming)
+
+    expect(described_class.new(conversation.reload).perform).to be(false)
+    expect(store.counts_for_keys(base_keys)).to eq(
+      store.inbox_key(account.id, inbox.id) => 1,
+      store.label_inbox_key(account.id, label.id, inbox.id) => 1
+    )
+  end
+
+  it 'removes a conversation from base cache when it becomes read' do
+    conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title])
+    Conversations::UnreadCounts::Builder.new(account).build_base!
+
+    conversation.update!(agent_last_seen_at: 1.minute.from_now)
+    expect(described_class.new(conversation.reload).perform).to be(true)
+
+    expect(store.counts_for_keys(base_keys).values).to all(eq(0))
+  end
+
+  it 'moves base label membership when labels change' do
+    conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title])
+    Conversations::UnreadCounts::Builder.new(account).build_base!
+
+    conversation.update_labels([new_label.title])
+    expect(described_class.new(conversation.reload, changed_attributes: { label_list: [[label.title], [new_label.title]] }).perform).to be(true)
+
+    expect(store.counts_for_keys([
+                                   store.label_inbox_key(account.id, label.id, inbox.id),
+                                   store.label_inbox_key(account.id, new_label.id, inbox.id)
+                                 ])).to eq(
+                                   store.label_inbox_key(account.id, label.id, inbox.id) => 0,
+                                   store.label_inbox_key(account.id, new_label.id, inbox.id) => 1
+                                 )
+  end
+
+  it 'moves base team membership when team changes' do
+    conversation = create_unread_conversation(account: account, inbox: inbox, team: team)
+    Conversations::UnreadCounts::Builder.new(account).build_base!
+
+    conversation.update!(team: new_team)
+    expect(described_class.new(conversation.reload, changed_attributes: { team_id: [team.id, new_team.id] }).perform).to be(true)
+
+    expect(store.counts_for_keys([
+                                   store.team_inbox_key(account.id, team.id, inbox.id),
+                                   store.team_inbox_key(account.id, new_team.id, inbox.id)
+                                 ])).to eq(
+                                   store.team_inbox_key(account.id, team.id, inbox.id) => 0,
+                                   store.team_inbox_key(account.id, new_team.id, inbox.id) => 1
+                                 )
+  end
+
+  it 'moves assignment-aware membership when assignee changes' do
+    conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title], assignee: assignee)
+    Conversations::UnreadCounts::Builder.new(account).build_assignment!
+
+    conversation.update!(assignee: other_assignee)
+    described_class.new(conversation.reload, changed_attributes: { assignee_id: [assignee.id, other_assignee.id] }).perform
+
+    expect(store.counts_for_keys([
+                                   store.inbox_assignee_key(account.id, inbox.id, assignee.id),
+                                   store.inbox_assignee_key(account.id, inbox.id, other_assignee.id)
+                                 ])).to eq(
+                                   store.inbox_assignee_key(account.id, inbox.id, assignee.id) => 0,
+                                   store.inbox_assignee_key(account.id, inbox.id, other_assignee.id) => 1
+                                 )
+  end
+
+  it 'does not remove unassigned membership when assignee did not change' do
+    conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title], assignee: assignee)
+    Conversations::UnreadCounts::Builder.new(account).build_assignment!
+    allow(store).to receive(:remove_assignment_membership).and_call_original
+
+    conversation.update_labels([new_label.title])
+    described_class.new(conversation.reload, changed_attributes: { label_list: [[label.title], [new_label.title]] }).perform
+
+    expect(store).to have_received(:remove_assignment_membership).with(hash_including(assignee_ids: [assignee.id]))
+  end
+
+  it 'moves assignment-aware team membership when team changes' do
+    create(:team_member, user: assignee, team: new_team)
+    conversation = create_unread_conversation(account: account, inbox: inbox, assignee: assignee, team: team)
+    Conversations::UnreadCounts::Builder.new(account).build_assignment!
+
+    conversation.update!(team: new_team)
+    described_class.new(conversation.reload, changed_attributes: { team_id: [team.id, new_team.id] }).perform
+
+    expect(store.counts_for_keys([
+                                   store.team_inbox_assignee_key(account.id, team.id, inbox.id, assignee.id),
+                                   store.team_inbox_assignee_key(account.id, new_team.id, inbox.id, assignee.id)
+                                 ])).to eq(
+                                   store.team_inbox_assignee_key(account.id, team.id, inbox.id, assignee.id) => 0,
+                                   store.team_inbox_assignee_key(account.id, new_team.id, inbox.id, assignee.id) => 1
+                                 )
+  end
+
+  def base_keys
+    [
+      store.inbox_key(account.id, inbox.id),
+      store.label_inbox_key(account.id, label.id, inbox.id)
+    ]
+  end
+end
diff --git a/spec/services/conversations/unread_counts/store_spec.rb b/spec/services/conversations/unread_counts/store_spec.rb
index 652c74e46..fcf21c985 100644
--- a/spec/services/conversations/unread_counts/store_spec.rb
+++ b/spec/services/conversations/unread_counts/store_spec.rb
@@ -90,6 +90,23 @@ RSpec.describe Conversations::UnreadCounts::Store do
       expect(described_class.counts_for_keys(base_keys).values).to all(eq(0))
     end
 
+    it 'checks memberships for a conversation across keys' do
+      described_class.add_base_membership(
+        account_id: account_id,
+        inbox_id: inbox_id,
+        label_ids: [label_id],
+        team_id: team_id,
+        conversation_id: conversation_id
+      )
+
+      expect(described_class.memberships_for_keys(base_keys, conversation_id)).to eq(
+        described_class.inbox_key(account_id, inbox_id) => true,
+        described_class.label_inbox_key(account_id, label_id, inbox_id) => true,
+        described_class.team_inbox_key(account_id, team_id, inbox_id) => true
+      )
+      expect(described_class.memberships_for_keys(base_keys, 999).values).to all(be(false))
+    end
+
     it 'adds, counts, and removes assignment-aware memberships' do
       described_class.add_assignment_membership(
         account_id: account_id,