diff --git a/app/services/conversations/unread_counts.rb b/app/services/conversations/unread_counts.rb
new file mode 100644
index 000000000..668395063
--- /dev/null
+++ b/app/services/conversations/unread_counts.rb
@@ -0,0 +1,4 @@
+module Conversations::UnreadCounts
+  READY_TTL = 24.hours.to_i
+  SET_TTL = 25.hours.to_i
+end
diff --git a/app/services/conversations/unread_counts/builder.rb b/app/services/conversations/unread_counts/builder.rb
new file mode 100644
index 000000000..54e466dbd
--- /dev/null
+++ b/app/services/conversations/unread_counts/builder.rb
@@ -0,0 +1,75 @@
+class Conversations::UnreadCounts::Builder
+  BATCH_SIZE = 1000
+
+  attr_reader :account
+
+  def initialize(account)
+    @account = account
+  end
+
+  def build_base!
+    store.clear_account!(account.id)
+    write_memberships(assignment: false)
+    store.mark_base_ready!(account.id)
+  end
+
+  def build_assignment!
+    store.clear_assignment!(account.id)
+    write_memberships(assignment: true)
+    store.mark_assignment_ready!(account.id)
+  end
+
+  def build_all!
+    build_base!
+    build_assignment!
+  end
+
+  private
+
+  def write_memberships(assignment:)
+    unread_conversations.in_batches(of: BATCH_SIZE) do |relation|
+      columns = %i[id inbox_id assignee_id cached_label_list team_id]
+      memberships = relation.pluck(*columns).map do |id, inbox_id, assignee_id, cached_label_list, team_id|
+        {
+          conversation_id: id,
+          inbox_id: inbox_id,
+          assignee_id: assignee_id,
+          team_id: team_id,
+          label_ids: label_ids_for(cached_label_list)
+        }
+      end
+
+      store.add_memberships(account_id: account.id, memberships: memberships, assignment: assignment)
+    end
+  end
+
+  def unread_conversations
+    account.conversations
+           .open
+           .joins(:messages)
+           .merge(Message.incoming.reorder(nil))
+           .where(messages: { account_id: account.id })
+           .where(unread_since_last_seen_condition)
+           .distinct
+  end
+
+  def unread_since_last_seen_condition
+    conversations = Conversation.arel_table
+    messages = Message.arel_table
+
+    conversations[:agent_last_seen_at].eq(nil).or(messages[:created_at].gt(conversations[:agent_last_seen_at]))
+  end
+
+  def label_ids_for(cached_label_list)
+    label_titles = cached_label_list.to_s.split(',').map(&:strip).compact_blank
+    labels_by_title.values_at(*label_titles).compact
+  end
+
+  def labels_by_title
+    @labels_by_title ||= account.labels.pluck(:title, :id).to_h
+  end
+
+  def store
+    ::Conversations::UnreadCounts::Store
+  end
+end
diff --git a/app/services/conversations/unread_counts/counter.rb b/app/services/conversations/unread_counts/counter.rb
new file mode 100644
index 000000000..f4126b61f
--- /dev/null
+++ b/app/services/conversations/unread_counts/counter.rb
@@ -0,0 +1,200 @@
+class Conversations::UnreadCounts::Counter
+  MANAGE_ALL_PERMISSION = 'conversation_manage'.freeze
+  UNASSIGNED_PERMISSION = 'conversation_unassigned_manage'.freeze
+  PARTICIPATING_PERMISSION = 'conversation_participating_manage'.freeze
+  BUILD_LOCK_TTL = 15.minutes.to_i
+  BUILD_WAIT_TIMEOUT = 30.seconds.to_i
+  BUILD_WAIT_INTERVAL = 0.1.seconds
+
+  attr_reader :account, :user
+
+  def initialize(account:, user:)
+    @account = account
+    @user = user
+  end
+
+  def perform
+    return empty_counts if permission_mode == :none
+
+    ensure_base_cache!
+    ensure_assignment_cache! if assignment_mode?
+
+    {
+      inboxes: unread_inbox_counts,
+      labels: unread_label_counts,
+      teams: unread_team_counts
+    }
+  end
+
+  private
+
+  def ensure_base_cache!
+    ensure_cache_ready!(
+      ready: -> { store.base_ready?(account.id) },
+      lock_key: base_build_lock_key
+    ) { ::Conversations::UnreadCounts::Builder.new(account).build_base! }
+  end
+
+  def ensure_assignment_cache!
+    ensure_cache_ready!(
+      ready: -> { store.assignment_ready?(account.id) },
+      lock_key: assignment_build_lock_key
+    ) { ::Conversations::UnreadCounts::Builder.new(account).build_assignment! }
+  end
+
+  def ensure_cache_ready!(ready:, lock_key:)
+    lock_manager = Redis::LockManager.new
+
+    loop do
+      return if ready.call
+
+      return if lock_manager.with_lock(lock_key, BUILD_LOCK_TTL) { yield unless ready.call }
+
+      wait_for_cache_ready(ready)
+    end
+  end
+
+  def wait_for_cache_ready(ready)
+    deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + BUILD_WAIT_TIMEOUT
+    sleep BUILD_WAIT_INTERVAL until ready.call || Process.clock_gettime(Process::CLOCK_MONOTONIC) >= deadline
+  end
+
+  def base_build_lock_key
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_BASE_BUILD_LOCK, account_id: account.id)
+  end
+
+  def assignment_build_lock_key
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_ASSIGNMENT_BUILD_LOCK, account_id: account.id)
+  end
+
+  def unread_inbox_counts
+    counts_for_grouped_keys(visible_inbox_ids.index_with { |inbox_id| inbox_keys_for_mode(inbox_id) })
+  end
+
+  def unread_label_counts
+    keys_by_id = Hash.new { |hash, key| hash[key] = [] }
+    sidebar_label_ids.each do |label_id|
+      visible_inbox_ids.each do |inbox_id|
+        keys_by_id[label_id].concat(label_inbox_keys_for_mode(label_id, inbox_id))
+      end
+    end
+
+    counts_for_grouped_keys(keys_by_id)
+  end
+
+  def unread_team_counts
+    keys_by_id = Hash.new { |hash, key| hash[key] = [] }
+    visible_team_ids.each do |team_id|
+      visible_inbox_ids.each do |inbox_id|
+        keys_by_id[team_id].concat(team_inbox_keys_for_mode(team_id, inbox_id))
+      end
+    end
+
+    counts_for_grouped_keys(keys_by_id)
+  end
+
+  def inbox_keys_for_mode(inbox_id)
+    case permission_mode
+    when :base
+      [store.inbox_key(account.id, inbox_id)]
+    when :unassigned_and_mine
+      [store.inbox_unassigned_key(account.id, inbox_id), store.inbox_assignee_key(account.id, inbox_id, user.id)]
+    when :mine
+      [store.inbox_assignee_key(account.id, inbox_id, user.id)]
+    end
+  end
+
+  def label_inbox_keys_for_mode(label_id, inbox_id)
+    case permission_mode
+    when :base
+      [store.label_inbox_key(account.id, label_id, inbox_id)]
+    when :unassigned_and_mine
+      [
+        store.label_inbox_unassigned_key(account.id, label_id, inbox_id),
+        store.label_inbox_assignee_key(account.id, label_id, inbox_id, user.id)
+      ]
+    when :mine
+      [store.label_inbox_assignee_key(account.id, label_id, inbox_id, user.id)]
+    end
+  end
+
+  def team_inbox_keys_for_mode(team_id, inbox_id)
+    case permission_mode
+    when :base
+      [store.team_inbox_key(account.id, team_id, inbox_id)]
+    when :unassigned_and_mine
+      [
+        store.team_inbox_unassigned_key(account.id, team_id, inbox_id),
+        store.team_inbox_assignee_key(account.id, team_id, inbox_id, user.id)
+      ]
+    when :mine
+      [store.team_inbox_assignee_key(account.id, team_id, inbox_id, user.id)]
+    end
+  end
+
+  def counts_for_grouped_keys(keys_by_id)
+    counts_by_key = store.counts_for_keys(keys_by_id.values.flatten)
+
+    keys_by_id.each_with_object({}) do |(id, keys), result|
+      count = keys.sum { |key| counts_by_key[key].to_i }
+      result[id.to_s] = count if count.positive?
+    end
+  end
+
+  def assignment_mode?
+    %i[unassigned_and_mine mine].include?(permission_mode)
+  end
+
+  def permission_mode
+    @permission_mode ||=
+      if !custom_role_agent? || permissions.include?(MANAGE_ALL_PERMISSION)
+        :base
+      elsif permissions.include?(UNASSIGNED_PERMISSION)
+        :unassigned_and_mine
+      elsif permissions.include?(PARTICIPATING_PERMISSION)
+        :mine
+      else
+        :none
+      end
+  end
+
+  def custom_role_agent?
+    account_user&.agent? && account_user.custom_role_id.present?
+  end
+
+  def permissions
+    account_user&.permissions || []
+  end
+
+  def account_user
+    @account_user ||= account.account_users.find_by(user_id: user.id)
+  end
+
+  def visible_inbox_ids
+    @visible_inbox_ids ||= if account_user&.administrator?
+                             account.inboxes.pluck(:id)
+                           else
+                             user.inboxes.where(account_id: account.id).pluck(:id)
+                           end
+  end
+
+  def sidebar_label_ids
+    @sidebar_label_ids ||= account.labels.where(show_on_sidebar: true).pluck(:id)
+  end
+
+  def visible_team_ids
+    @visible_team_ids ||= if account_user&.administrator?
+                            account.teams.pluck(:id)
+                          else
+                            user.teams.where(account_id: account.id).pluck(:id)
+                          end
+  end
+
+  def empty_counts
+    { inboxes: {}, labels: {}, teams: {} }
+  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
new file mode 100644
index 000000000..dbd01f235
--- /dev/null
+++ b/app/services/conversations/unread_counts/store.rb
@@ -0,0 +1,220 @@
+class Conversations::UnreadCounts::Store
+  class << self
+    def base_ready?(account_id)
+      Redis::Alfred.exists?(base_ready_key(account_id))
+    end
+
+    def assignment_ready?(account_id)
+      Redis::Alfred.exists?(assignment_ready_key(account_id))
+    end
+
+    def mark_base_ready!(account_id)
+      Redis::Alfred.set(base_ready_key(account_id), Time.current.to_i, ex: Conversations::UnreadCounts::READY_TTL)
+    end
+
+    def mark_assignment_ready!(account_id)
+      Redis::Alfred.set(assignment_ready_key(account_id), Time.current.to_i, ex: Conversations::UnreadCounts::READY_TTL)
+    end
+
+    def clear_account!(account_id)
+      delete_matching("#{account_prefix(account_id)}::*")
+    end
+
+    def clear_assignment!(account_id)
+      assignment_key_patterns(account_id).each { |pattern| delete_matching(pattern) }
+    end
+
+    def add_base_membership(account_id:, inbox_id:, label_ids:, conversation_id:, team_id: nil)
+      add_to_sets(base_keys(account_id, inbox_id, label_ids, team_id), conversation_id)
+    end
+
+    def remove_base_membership(account_id:, inbox_ids:, label_ids:, conversation_id:, team_ids: [])
+      keys = Array(inbox_ids).flat_map { |inbox_id| removable_base_keys(account_id, inbox_id, label_ids, team_ids) }
+      remove_from_sets(keys, conversation_id)
+    end
+
+    def add_assignment_membership(account_id:, conversation_id:, **membership)
+      add_to_sets(
+        assignment_keys(account_id, membership[:inbox_id], membership[:label_ids], membership[:assignee_id], membership[:team_id]),
+        conversation_id
+      )
+    end
+
+    def remove_assignment_membership(account_id:, conversation_id:, **membership)
+      keys = Array(membership[:inbox_ids]).flat_map do |inbox_id|
+        Array(membership[:assignee_ids]).flat_map do |assignee_id|
+          removable_assignment_keys(account_id, inbox_id, membership[:label_ids], assignee_id, membership[:team_ids])
+        end
+      end
+      remove_from_sets(keys, conversation_id)
+    end
+
+    def add_memberships(account_id:, memberships:, assignment: false)
+      return if memberships.blank?
+
+      Redis::Alfred.pipelined do |pipeline|
+        memberships.each do |membership|
+          keys = if assignment
+                   assignment_keys(account_id, membership[:inbox_id], membership[:label_ids], membership[:assignee_id], membership[:team_id])
+                 else
+                   base_keys(account_id, membership[:inbox_id], membership[:label_ids], membership[:team_id])
+                 end
+
+          keys.each do |key|
+            pipeline.sadd(key, membership[:conversation_id])
+            pipeline.expire(key, Conversations::UnreadCounts::SET_TTL)
+          end
+        end
+      end
+    end
+
+    def counts_for_keys(keys)
+      keys = keys.compact_blank
+      return {} if keys.blank?
+
+      counts = Redis::Alfred.pipelined do |pipeline|
+        keys.each { |key| pipeline.scard(key) }
+      end
+      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 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
+
+    private
+
+    def base_ready_key(account_id)
+      format(Redis::Alfred::UNREAD_CONVERSATIONS_BASE_READY, account_id: account_id)
+    end
+
+    def assignment_ready_key(account_id)
+      format(Redis::Alfred::UNREAD_CONVERSATIONS_ASSIGNMENT_READY, account_id: account_id)
+    end
+
+    def account_prefix(account_id)
+      format(Redis::Alfred::UNREAD_CONVERSATIONS_ACCOUNT_PREFIX, account_id: account_id)
+    end
+
+    def base_keys(account_id, inbox_id, label_ids, team_id = nil)
+      keys = [inbox_key(account_id, inbox_id)] + Array(label_ids).map { |label_id| label_inbox_key(account_id, label_id, inbox_id) }
+      keys << team_inbox_key(account_id, team_id, inbox_id) if team_id.present?
+      keys
+    end
+
+    def removable_base_keys(account_id, inbox_id, label_ids, team_ids)
+      keys = base_keys(account_id, inbox_id, label_ids)
+      keys.concat(Array(team_ids).compact_blank.map { |team_id| team_inbox_key(account_id, team_id, inbox_id) })
+    end
+
+    def assignment_keys(account_id, inbox_id, label_ids, assignee_id, team_id = nil)
+      keys = assignment_keys_without_team(account_id, inbox_id, label_ids, assignee_id)
+      keys << team_assignment_key(account_id, team_id, inbox_id, assignee_id) if team_id.present?
+      keys
+    end
+
+    def removable_assignment_keys(account_id, inbox_id, label_ids, assignee_id, team_ids)
+      keys = assignment_keys_without_team(account_id, inbox_id, label_ids, assignee_id)
+      keys.concat(Array(team_ids).compact_blank.map { |team_id| team_assignment_key(account_id, team_id, inbox_id, assignee_id) })
+    end
+
+    def assignment_keys_without_team(account_id, inbox_id, label_ids, assignee_id)
+      return assignee_keys(account_id, inbox_id, label_ids, assignee_id) if assignee_id.present?
+
+      unassigned_keys(account_id, inbox_id, label_ids)
+    end
+
+    def assignee_keys(account_id, inbox_id, label_ids, assignee_id)
+      [inbox_assignee_key(account_id, inbox_id, assignee_id)] +
+        Array(label_ids).map { |label_id| label_inbox_assignee_key(account_id, label_id, inbox_id, assignee_id) }
+    end
+
+    def unassigned_keys(account_id, inbox_id, label_ids)
+      [inbox_unassigned_key(account_id, inbox_id)] +
+        Array(label_ids).map { |label_id| label_inbox_unassigned_key(account_id, label_id, inbox_id) }
+    end
+
+    def team_assignment_key(account_id, team_id, inbox_id, assignee_id)
+      return team_inbox_assignee_key(account_id, team_id, inbox_id, assignee_id) if assignee_id.present?
+
+      team_inbox_unassigned_key(account_id, team_id, inbox_id)
+    end
+
+    def add_to_sets(keys, conversation_id)
+      write_to_sets(keys) { |pipeline, key| pipeline.sadd(key, conversation_id) }
+    end
+
+    def remove_from_sets(keys, conversation_id)
+      write_to_sets(keys) { |pipeline, key| pipeline.srem(key, conversation_id) }
+    end
+
+    def write_to_sets(keys)
+      keys = keys.compact_blank
+      return if keys.blank?
+
+      Redis::Alfred.pipelined do |pipeline|
+        keys.each do |key|
+          yield(pipeline, key)
+          pipeline.expire(key, Conversations::UnreadCounts::SET_TTL)
+        end
+      end
+    end
+
+    def delete_matching(pattern)
+      Redis::Alfred.scan_each(match: pattern, count: 1000) do |key|
+        Redis::Alfred.delete(key)
+      end
+    end
+
+    def assignment_key_patterns(account_id)
+      prefix = account_prefix(account_id)
+      [
+        assignment_ready_key(account_id),
+        "#{prefix}::INBOX::*::UNASSIGNED",
+        "#{prefix}::INBOX::*::ASSIGNEE::*",
+        "#{prefix}::LABEL::*::INBOX::*::UNASSIGNED",
+        "#{prefix}::LABEL::*::INBOX::*::ASSIGNEE::*",
+        "#{prefix}::TEAM::*::INBOX::*::UNASSIGNED",
+        "#{prefix}::TEAM::*::INBOX::*::ASSIGNEE::*"
+      ]
+    end
+  end
+end
diff --git a/lib/redis/alfred.rb b/lib/redis/alfred.rb
index e7e04ed60..1554b8806 100644
--- a/lib/redis/alfred.rb
+++ b/lib/redis/alfred.rb
@@ -40,6 +40,11 @@ module Redis::Alfred
       $alfred.with { |conn| conn.expire(key, seconds) }
     end
 
+    # get expiry of a key in seconds
+    def ttl(key)
+      $alfred.with { |conn| conn.ttl(key) }
+    end
+
     # scan keys matching a pattern
     def scan_each(match: nil, count: 100, &)
       $alfred.with do |conn|
@@ -80,6 +85,10 @@ module Redis::Alfred
       $alfred.with { |conn| conn.lrem(key, count, value) }
     end
 
+    def pipelined(&)
+      $alfred.with { |conn| conn.pipelined(&) }
+    end
+
     # hash operations
 
     # add a key value to redis hash
diff --git a/lib/redis/lock_manager.rb b/lib/redis/lock_manager.rb
index 63063542c..4064d692c 100644
--- a/lib/redis/lock_manager.rb
+++ b/lib/redis/lock_manager.rb
@@ -49,6 +49,18 @@ class Redis::LockManager
     true
   end
 
+  def with_lock(key, timeout = LOCK_TIMEOUT)
+    return false unless lock(key, timeout)
+
+    begin
+      yield
+    ensure
+      unlock(key)
+    end
+
+    true
+  end
+
   # Checks if the given key is currently locked.
   #
   # === Parameters
diff --git a/lib/redis/redis_keys.rb b/lib/redis/redis_keys.rb
index 15ff553cc..812553844 100644
--- a/lib/redis/redis_keys.rb
+++ b/lib/redis/redis_keys.rb
@@ -9,6 +9,28 @@ module Redis::RedisKeys
   # Whether a conversation is muted ?
   CONVERSATION_MUTE_KEY = 'CONVERSATION::%<id>d::MUTED'.freeze
   CONVERSATION_DRAFT_MESSAGE = 'CONVERSATION::%<id>d::DRAFT_MESSAGE'.freeze
+  UNREAD_CONVERSATIONS_ACCOUNT_PREFIX = 'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d'.freeze
+  UNREAD_CONVERSATIONS_BASE_READY = 'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::READY::BASE'.freeze
+  UNREAD_CONVERSATIONS_ASSIGNMENT_READY = 'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::READY::ASSIGNMENT'.freeze
+  UNREAD_CONVERSATIONS_BASE_BUILD_LOCK = 'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::BUILD_LOCK::BASE'.freeze
+  UNREAD_CONVERSATIONS_ASSIGNMENT_BUILD_LOCK = 'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::BUILD_LOCK::ASSIGNMENT'.freeze
+  UNREAD_CONVERSATIONS_INBOX = 'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::INBOX::%<inbox_id>d'.freeze
+  UNREAD_CONVERSATIONS_LABEL_INBOX =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::LABEL::%<label_id>d::INBOX::%<inbox_id>d'.freeze
+  UNREAD_CONVERSATIONS_TEAM_INBOX =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::TEAM::%<team_id>d::INBOX::%<inbox_id>d'.freeze
+  UNREAD_CONVERSATIONS_INBOX_UNASSIGNED =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::INBOX::%<inbox_id>d::UNASSIGNED'.freeze
+  UNREAD_CONVERSATIONS_INBOX_ASSIGNEE =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::INBOX::%<inbox_id>d::ASSIGNEE::%<user_id>d'.freeze
+  UNREAD_CONVERSATIONS_LABEL_INBOX_UNASSIGNED =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::LABEL::%<label_id>d::INBOX::%<inbox_id>d::UNASSIGNED'.freeze
+  UNREAD_CONVERSATIONS_LABEL_INBOX_ASSIGNEE =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::LABEL::%<label_id>d::INBOX::%<inbox_id>d::ASSIGNEE::%<user_id>d'.freeze
+  UNREAD_CONVERSATIONS_TEAM_INBOX_UNASSIGNED =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::TEAM::%<team_id>d::INBOX::%<inbox_id>d::UNASSIGNED'.freeze
+  UNREAD_CONVERSATIONS_TEAM_INBOX_ASSIGNEE =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::TEAM::%<team_id>d::INBOX::%<inbox_id>d::ASSIGNEE::%<user_id>d'.freeze
 
   ## User Keys
   # SSO Auth Tokens
diff --git a/spec/enterprise/services/conversations/unread_counts/counter_spec.rb b/spec/enterprise/services/conversations/unread_counts/counter_spec.rb
new file mode 100644
index 000000000..cbb2d6c98
--- /dev/null
+++ b/spec/enterprise/services/conversations/unread_counts/counter_spec.rb
@@ -0,0 +1,72 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::Counter do
+  let(:account) { create(:account) }
+  let(:agent) { create(:user, account: account, role: :agent) }
+  let(:other_agent) { create(:user, account: account, role: :agent) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:label) { create(:label, account: account, title: 'support', show_on_sidebar: true) }
+  let(:team) { create(:team, account: account, allow_auto_assign: false) }
+  let(:account_user) { account.account_users.find_by(user: agent) }
+  let(:store) { Conversations::UnreadCounts::Store }
+
+  before do
+    create(:inbox_member, user: agent, inbox: inbox)
+    create(:team_member, user: agent, team: team)
+  end
+
+  after do
+    store.clear_account!(account.id)
+  end
+
+  it 'uses base counts for custom roles with conversation_manage permission' do
+    account_user.update!(custom_role: create(:custom_role, account: account, permissions: ['conversation_manage']))
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], assignee: other_agent, team: team)
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], team: team)
+
+    result = described_class.new(account: account, user: agent).perform
+
+    expect(result[:inboxes]).to eq(inbox.id.to_s => 2)
+    expect(result[:labels]).to eq(label.id.to_s => 2)
+    expect(result[:teams]).to eq(team.id.to_s => 2)
+    expect(store.assignment_ready?(account.id)).to be(false)
+  end
+
+  it 'counts assigned and unassigned conversations for conversation_unassigned_manage permission' do
+    account_user.update!(custom_role: create(:custom_role, account: account, permissions: ['conversation_unassigned_manage']))
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], assignee: agent, team: team)
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], team: team)
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], assignee: other_agent, team: team)
+
+    result = described_class.new(account: account, user: agent).perform
+
+    expect(result[:inboxes]).to eq(inbox.id.to_s => 2)
+    expect(result[:labels]).to eq(label.id.to_s => 2)
+    expect(result[:teams]).to eq(team.id.to_s => 2)
+    expect(store.assignment_ready?(account.id)).to be(true)
+  end
+
+  it 'counts only assigned conversations for conversation_participating_manage permission' do
+    account_user.update!(custom_role: create(:custom_role, account: account, permissions: ['conversation_participating_manage']))
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], assignee: agent, team: team)
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], team: team)
+
+    result = described_class.new(account: account, user: agent).perform
+
+    expect(result[:inboxes]).to eq(inbox.id.to_s => 1)
+    expect(result[:labels]).to eq(label.id.to_s => 1)
+    expect(result[:teams]).to eq(team.id.to_s => 1)
+    expect(store.assignment_ready?(account.id)).to be(true)
+  end
+
+  it 'returns zero for custom roles without conversation permissions' do
+    account_user.update!(custom_role: create(:custom_role, account: account, permissions: []))
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], assignee: agent, team: team)
+
+    result = described_class.new(account: account, user: agent).perform
+
+    expect(result).to eq(inboxes: {}, labels: {}, teams: {})
+    expect(store.base_ready?(account.id)).to be(false)
+    expect(store.assignment_ready?(account.id)).to be(false)
+  end
+end
diff --git a/spec/lib/redis/lock_manager_spec.rb b/spec/lib/redis/lock_manager_spec.rb
index 208594322..8e5feafb0 100644
--- a/spec/lib/redis/lock_manager_spec.rb
+++ b/spec/lib/redis/lock_manager_spec.rb
@@ -35,6 +35,28 @@ RSpec.describe Redis::LockManager do
     end
   end
 
+  describe '#with_lock' do
+    it 'yields when the lock is acquired and releases the lock' do
+      yielded = false
+
+      expect(lock_manager.with_lock(lock_key) { yielded = true }).to be true
+      expect(yielded).to be true
+      expect(lock_manager.locked?(lock_key)).to be false
+    end
+
+    it 'returns false without yielding when the lock is already acquired' do
+      lock_manager.lock(lock_key)
+
+      expect { |block| lock_manager.with_lock(lock_key, &block) }.not_to yield_control
+      expect(lock_manager.with_lock(lock_key) { raise 'should not run' }).to be false
+    end
+
+    it 'releases the lock when the block raises' do
+      expect { lock_manager.with_lock(lock_key) { raise 'boom' } }.to raise_error('boom')
+      expect(lock_manager.locked?(lock_key)).to be false
+    end
+  end
+
   describe '#locked?' do
     it 'returns true if a key is locked' do
       lock_manager.lock(lock_key)
diff --git a/spec/rails_helper.rb b/spec/rails_helper.rb
index d863dd58c..e83db4046 100644
--- a/spec/rails_helper.rb
+++ b/spec/rails_helper.rb
@@ -71,6 +71,7 @@ RSpec.configure do |config|
   config.include FileUploadHelpers
   config.include CsvSpecHelpers
   config.include InstagramSpecHelpers
+  config.include ConversationsUnreadCountsHelpers
   config.include Devise::Test::IntegrationHelpers, type: :request
   config.include ActiveSupport::Testing::TimeHelpers
   config.include ActionCable::TestHelper
diff --git a/spec/services/conversations/unread_counts/builder_spec.rb b/spec/services/conversations/unread_counts/builder_spec.rb
new file mode 100644
index 000000000..4b3aec230
--- /dev/null
+++ b/spec/services/conversations/unread_counts/builder_spec.rb
@@ -0,0 +1,93 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::Builder do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:label) { create(:label, account: account, title: 'urgent', show_on_sidebar: true) }
+  let(:assignee) { create(:user, account: account, role: :agent) }
+  let(:team) { create(:team, account: account, allow_auto_assign: false) }
+  let(:store) { Conversations::UnreadCounts::Store }
+
+  after do
+    store.clear_account!(account.id)
+  end
+
+  describe '#build_base!' do
+    it 'stores unread open conversations by inbox and label inbox' do
+      unread_conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title], team: team)
+      create_read_conversation
+      create_resolved_unread_conversation
+
+      described_class.new(account).build_base!
+
+      expect(store.base_ready?(account.id)).to be(true)
+      expect(redis_set_members(store.inbox_key(account.id, inbox.id))).to contain_exactly(unread_conversation.id.to_s)
+      expect(redis_set_members(store.label_inbox_key(account.id, label.id, inbox.id))).to contain_exactly(unread_conversation.id.to_s)
+      expect(redis_set_members(store.team_inbox_key(account.id, team.id, inbox.id))).to contain_exactly(unread_conversation.id.to_s)
+    end
+
+    it 'clears assignment-aware cache data before rebuilding base data' do
+      assigned_conversation = create_unread_conversation(
+        account: account,
+        inbox: inbox,
+        labels: [label.title],
+        assignee: assignee,
+        team: team
+      )
+
+      described_class.new(account).build_assignment!
+      described_class.new(account).build_base!
+
+      expect(store.assignment_ready?(account.id)).to be(false)
+      expect(redis_set_members(store.inbox_assignee_key(account.id, inbox.id, assignee.id))).to be_empty
+      expect(redis_set_members(store.inbox_key(account.id, inbox.id))).to contain_exactly(assigned_conversation.id.to_s)
+    end
+  end
+
+  describe '#build_assignment!' do
+    it 'stores unread open conversations by unassigned and assignee dimensions' do
+      assigned_conversation = create_unread_conversation(
+        account: account,
+        inbox: inbox,
+        labels: [label.title],
+        assignee: assignee,
+        team: team
+      )
+      unassigned_conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title], team: team)
+
+      described_class.new(account).build_assignment!
+
+      expect(store.assignment_ready?(account.id)).to be(true)
+      expect(redis_set_members(store.inbox_assignee_key(account.id, inbox.id, assignee.id))).to contain_exactly(assigned_conversation.id.to_s)
+      expect(redis_set_members(store.label_inbox_assignee_key(account.id, label.id, inbox.id, assignee.id))).to contain_exactly(
+        assigned_conversation.id.to_s
+      )
+      expect(redis_set_members(store.team_inbox_assignee_key(account.id, team.id, inbox.id, assignee.id))).to contain_exactly(
+        assigned_conversation.id.to_s
+      )
+      expect(redis_set_members(store.inbox_unassigned_key(account.id, inbox.id))).to contain_exactly(unassigned_conversation.id.to_s)
+      expect(redis_set_members(store.label_inbox_unassigned_key(account.id, label.id, inbox.id))).to contain_exactly(
+        unassigned_conversation.id.to_s
+      )
+      expect(redis_set_members(store.team_inbox_unassigned_key(account.id, team.id, inbox.id))).to contain_exactly(
+        unassigned_conversation.id.to_s
+      )
+    end
+  end
+
+  def create_read_conversation
+    conversation = create(:conversation, account: account, inbox: inbox, agent_last_seen_at: 1.minute.from_now)
+    create(:message, account: account, inbox: inbox, conversation: conversation, message_type: :incoming)
+    conversation
+  end
+
+  def create_resolved_unread_conversation
+    conversation = create_unread_conversation(account: account, inbox: inbox)
+    conversation.update!(status: :resolved)
+    conversation
+  end
+
+  def redis_set_members(key)
+    Redis::Alfred.pipelined { |pipeline| pipeline.smembers(key) }.first
+  end
+end
diff --git a/spec/services/conversations/unread_counts/counter_spec.rb b/spec/services/conversations/unread_counts/counter_spec.rb
new file mode 100644
index 000000000..bfd10436e
--- /dev/null
+++ b/spec/services/conversations/unread_counts/counter_spec.rb
@@ -0,0 +1,95 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::Counter do
+  let(:account) { create(:account) }
+  let(:agent) { create(:user, account: account, role: :agent) }
+  let(:admin) { create(:user, account: account, role: :administrator) }
+  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(:hidden_label) { create(:label, account: account, title: 'internal', show_on_sidebar: false) }
+  let(:visible_team) { create(:team, account: account, allow_auto_assign: false) }
+  let(:store) { Conversations::UnreadCounts::Store }
+
+  before do
+    create(:inbox_member, user: agent, inbox: visible_inbox)
+    create(:team_member, user: agent, team: visible_team)
+  end
+
+  after do
+    store.clear_account!(account.id)
+  end
+
+  it 'builds the base cache on demand' do
+    create_unread_conversation(account: account, inbox: visible_inbox, labels: [label.title], team: visible_team)
+
+    described_class.new(account: account, user: agent).perform
+
+    expect(store.base_ready?(account.id)).to be(true)
+  end
+
+  it 'uses a Redis lock while building the base cache on demand' do
+    lock_key = "UNREAD_CONVERSATIONS::V1::ACCOUNT::#{account.id}::BUILD_LOCK::BASE"
+    lock_manager = instance_double(Redis::LockManager)
+    allow(Redis::LockManager).to receive(:new).and_return(lock_manager)
+    allow(lock_manager).to receive(:with_lock).with(lock_key, described_class::BUILD_LOCK_TTL).and_yield.and_return(true)
+
+    create_unread_conversation(account: account, inbox: visible_inbox, labels: [label.title], team: visible_team)
+
+    described_class.new(account: account, user: agent).perform
+
+    expect(lock_manager).to have_received(:with_lock).with(lock_key, described_class::BUILD_LOCK_TTL)
+  end
+
+  it 'waits instead of rebuilding when another process owns the base build lock' do
+    lock_manager = instance_double(Redis::LockManager, with_lock: false)
+    counter = described_class.new(account: account, user: agent)
+
+    allow(Redis::LockManager).to receive(:new).and_return(lock_manager)
+    allow(counter).to receive(:wait_for_cache_ready) { store.mark_base_ready!(account.id) }
+    expect(Conversations::UnreadCounts::Builder).not_to receive(:new)
+
+    counter.perform
+
+    expect(counter).to have_received(:wait_for_cache_ready)
+    expect(store.base_ready?(account.id)).to be(true)
+  end
+
+  it 'counts unread conversations only across inboxes visible to a normal agent' do
+    create_unread_conversation(account: account, inbox: visible_inbox, labels: [label.title], team: visible_team)
+    create_unread_conversation(account: account, inbox: hidden_inbox, labels: [label.title], team: visible_team)
+
+    result = described_class.new(account: account, user: agent).perform
+
+    expect(result).to eq(
+      inboxes: { visible_inbox.id.to_s => 1 },
+      labels: { label.id.to_s => 1 },
+      teams: { visible_team.id.to_s => 1 }
+    )
+  end
+
+  it 'counts unread conversations across all account inboxes for admins' do
+    create_unread_conversation(account: account, inbox: visible_inbox, labels: [label.title], team: visible_team)
+    create_unread_conversation(account: account, inbox: hidden_inbox, labels: [label.title], team: visible_team)
+
+    result = described_class.new(account: account, user: admin).perform
+
+    expect(result).to eq(
+      inboxes: { visible_inbox.id.to_s => 1, hidden_inbox.id.to_s => 1 },
+      labels: { label.id.to_s => 2 },
+      teams: { visible_team.id.to_s => 2 }
+    )
+  end
+
+  it 'does not return zero counts or labels hidden from the sidebar' do
+    create_unread_conversation(account: account, inbox: visible_inbox, labels: [hidden_label.title], team: visible_team)
+
+    result = described_class.new(account: account, user: agent).perform
+
+    expect(result).to eq(
+      inboxes: { visible_inbox.id.to_s => 1 },
+      labels: {},
+      teams: { visible_team.id.to_s => 1 }
+    )
+  end
+end
diff --git a/spec/services/conversations/unread_counts/store_spec.rb b/spec/services/conversations/unread_counts/store_spec.rb
new file mode 100644
index 000000000..652c74e46
--- /dev/null
+++ b/spec/services/conversations/unread_counts/store_spec.rb
@@ -0,0 +1,183 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::Store do
+  let(:account_id) { 1 }
+  let(:inbox_id) { 2 }
+  let(:label_id) { 3 }
+  let(:user_id) { 4 }
+  let(:conversation_id) { 5 }
+  let(:team_id) { 6 }
+
+  after do
+    described_class.clear_account!(account_id)
+  end
+
+  describe 'key builders' do
+    it 'builds base keys using the Redis key naming convention' do
+      expect(described_class.inbox_key(account_id, inbox_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::INBOX::2'
+      )
+      expect(described_class.label_inbox_key(account_id, label_id, inbox_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::LABEL::3::INBOX::2'
+      )
+      expect(described_class.team_inbox_key(account_id, team_id, inbox_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::TEAM::6::INBOX::2'
+      )
+    end
+
+    it 'builds assignment-aware keys using the Redis key naming convention' do
+      expect(described_class.inbox_unassigned_key(account_id, inbox_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::INBOX::2::UNASSIGNED'
+      )
+      expect(described_class.inbox_assignee_key(account_id, inbox_id, user_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::INBOX::2::ASSIGNEE::4'
+      )
+      expect(described_class.label_inbox_unassigned_key(account_id, label_id, inbox_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::LABEL::3::INBOX::2::UNASSIGNED'
+      )
+      expect(described_class.label_inbox_assignee_key(account_id, label_id, inbox_id, user_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::LABEL::3::INBOX::2::ASSIGNEE::4'
+      )
+      expect(described_class.team_inbox_unassigned_key(account_id, team_id, inbox_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::TEAM::6::INBOX::2::UNASSIGNED'
+      )
+      expect(described_class.team_inbox_assignee_key(account_id, team_id, inbox_id, user_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::TEAM::6::INBOX::2::ASSIGNEE::4'
+      )
+    end
+  end
+
+  describe 'ready markers' do
+    it 'tracks base and assignment readiness independently' do
+      expect(described_class.base_ready?(account_id)).to be(false)
+      expect(described_class.assignment_ready?(account_id)).to be(false)
+
+      described_class.mark_base_ready!(account_id)
+      described_class.mark_assignment_ready!(account_id)
+
+      expect(described_class.base_ready?(account_id)).to be(true)
+      expect(described_class.assignment_ready?(account_id)).to be(true)
+      expect(ttl_for('UNREAD_CONVERSATIONS::V1::ACCOUNT::1::READY::BASE')).to be_within(5).of(Conversations::UnreadCounts::READY_TTL)
+      expect(ttl_for('UNREAD_CONVERSATIONS::V1::ACCOUNT::1::READY::ASSIGNMENT')).to be_within(5).of(Conversations::UnreadCounts::READY_TTL)
+    end
+  end
+
+  describe 'set operations' do
+    it 'adds, counts, and removes base memberships' 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.counts_for_keys(base_keys)).to eq(
+        described_class.inbox_key(account_id, inbox_id) => 1,
+        described_class.label_inbox_key(account_id, label_id, inbox_id) => 1,
+        described_class.team_inbox_key(account_id, team_id, inbox_id) => 1
+      )
+      expect(base_keys.map { |key| ttl_for(key) }).to all(be_within(5).of(Conversations::UnreadCounts::SET_TTL))
+
+      described_class.remove_base_membership(
+        account_id: account_id,
+        inbox_ids: [inbox_id],
+        label_ids: [label_id],
+        team_ids: [team_id],
+        conversation_id: conversation_id
+      )
+
+      expect(described_class.counts_for_keys(base_keys).values).to all(eq(0))
+    end
+
+    it 'adds, counts, and removes assignment-aware memberships' do
+      described_class.add_assignment_membership(
+        account_id: account_id,
+        inbox_id: inbox_id,
+        label_ids: [label_id],
+        assignee_id: user_id,
+        team_id: team_id,
+        conversation_id: conversation_id
+      )
+
+      expect(described_class.counts_for_keys(assignment_keys)).to eq(
+        described_class.inbox_assignee_key(account_id, inbox_id, user_id) => 1,
+        described_class.label_inbox_assignee_key(account_id, label_id, inbox_id, user_id) => 1,
+        described_class.team_inbox_assignee_key(account_id, team_id, inbox_id, user_id) => 1
+      )
+      expect(assignment_keys.map { |key| ttl_for(key) }).to all(be_within(5).of(Conversations::UnreadCounts::SET_TTL))
+
+      described_class.remove_assignment_membership(
+        account_id: account_id,
+        inbox_ids: [inbox_id],
+        label_ids: [label_id],
+        assignee_ids: [user_id],
+        team_ids: [team_id],
+        conversation_id: conversation_id
+      )
+
+      expect(described_class.counts_for_keys(assignment_keys).values).to all(eq(0))
+    end
+
+    it 'sets expiry on bulk membership writes' do
+      described_class.add_memberships(
+        account_id: account_id,
+        memberships: [{
+          inbox_id: inbox_id,
+          label_ids: [label_id],
+          team_id: team_id,
+          conversation_id: conversation_id
+        }]
+      )
+
+      expect(base_keys.map { |key| ttl_for(key) }).to all(be_within(5).of(Conversations::UnreadCounts::SET_TTL))
+    end
+
+    it 'clears all account memberships' do
+      described_class.mark_base_ready!(account_id)
+      described_class.mark_assignment_ready!(account_id)
+      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
+      )
+      described_class.add_assignment_membership(
+        account_id: account_id,
+        inbox_id: inbox_id,
+        label_ids: [label_id],
+        assignee_id: user_id,
+        team_id: team_id,
+        conversation_id: conversation_id
+      )
+
+      described_class.clear_account!(account_id)
+
+      expect(described_class.base_ready?(account_id)).to be(false)
+      expect(described_class.assignment_ready?(account_id)).to be(false)
+      expect(described_class.counts_for_keys(base_keys).values).to all(eq(0))
+      expect(described_class.counts_for_keys(assignment_keys).values).to all(eq(0))
+    end
+  end
+
+  def base_keys
+    [
+      described_class.inbox_key(account_id, inbox_id),
+      described_class.label_inbox_key(account_id, label_id, inbox_id),
+      described_class.team_inbox_key(account_id, team_id, inbox_id)
+    ]
+  end
+
+  def assignment_keys
+    [
+      described_class.inbox_assignee_key(account_id, inbox_id, user_id),
+      described_class.label_inbox_assignee_key(account_id, label_id, inbox_id, user_id),
+      described_class.team_inbox_assignee_key(account_id, team_id, inbox_id, user_id)
+    ]
+  end
+
+  def ttl_for(key)
+    Redis::Alfred.ttl(key)
+  end
+end
diff --git a/spec/support/conversations_unread_counts_helpers.rb b/spec/support/conversations_unread_counts_helpers.rb
new file mode 100644
index 000000000..e1a98d034
--- /dev/null
+++ b/spec/support/conversations_unread_counts_helpers.rb
@@ -0,0 +1,11 @@
+module ConversationsUnreadCountsHelpers
+  def create_unread_conversation(account:, inbox:, labels: [], assignee: nil, team: nil)
+    create(:team_member, user: assignee, team: team) if assignee.present? && team.present? && !team.members.exists?(assignee.id)
+
+    conversation = create(:conversation, account: account, inbox: inbox, assignee: assignee, team: team, agent_last_seen_at: 1.hour.ago)
+    conversation.update_labels(labels) if labels.present?
+
+    create(:message, account: account, inbox: inbox, conversation: conversation, message_type: :incoming, created_at: 5.minutes.ago)
+    conversation
+  end
+end