diff --git a/assets/l10n/app_en.arb b/assets/l10n/app_en.arb
index 8a005225d8..1d9a1ad6d5 100644
--- a/assets/l10n/app_en.arb
+++ b/assets/l10n/app_en.arb
@@ -342,12 +342,9 @@
   "@serverUrlValidationErrorUnsupportedScheme": {
     "description": "Error message when URL has an unsupported scheme."
   },
-  "markAsReadLabel": "Mark {num, plural, =1{1 message} other{{num} messages}} as read",
-  "@markAsReadLabel": {
-    "description": "Button text to mark messages as read.",
-    "placeholders": {
-      "num": {"type": "int", "example": "4"}
-    }
+  "markAllAsReadLabel": "Mark all messages as read",
+  "@markAllAsReadLabel": {
+    "description": "Button text to mark messages as read."
   },
   "markAsReadComplete": "Marked {num, plural, =1{1 message} other{{num} messages}} as read.",
   "@markAsReadComplete": {
diff --git a/lib/api/model/events.dart b/lib/api/model/events.dart
index 12e9a702c5..c47b66a78f 100644
--- a/lib/api/model/events.dart
+++ b/lib/api/model/events.dart
@@ -1,6 +1,5 @@
 import 'package:json_annotation/json_annotation.dart';
 
-import 'initial_snapshot.dart';
 import 'model.dart';
 
 part 'events.g.dart';
@@ -35,6 +34,13 @@ sealed class Event {
           case 'update': return RealmUserUpdateEvent.fromJson(json);
           default: return UnexpectedEvent.fromJson(json);
         }
+      case 'stream':
+        switch (json['op'] as String) {
+          case 'create': return StreamCreateEvent.fromJson(json);
+          case 'delete': return StreamDeleteEvent.fromJson(json);
+          // TODO(#182): case 'update': …
+          default: return UnexpectedEvent.fromJson(json);
+        }
       case 'subscription':
         switch (json['op'] as String) {
           case 'add': return SubscriptionAddEvent.fromJson(json);
@@ -44,13 +50,8 @@ sealed class Event {
           case 'peer_remove': return SubscriptionPeerRemoveEvent.fromJson(json);
           default: return UnexpectedEvent.fromJson(json);
         }
-      case 'stream':
-        switch (json['op'] as String) {
-          case 'create': return StreamCreateEvent.fromJson(json);
-          case 'delete': return StreamDeleteEvent.fromJson(json);
-          // TODO(#182): case 'update': …
-          default: return UnexpectedEvent.fromJson(json);
-        }
+      // case 'muted_topics': … // TODO(#422) we ignore this feature on older servers
+      case 'user_topic': return UserTopicEvent.fromJson(json);
       case 'message': return MessageEvent.fromJson(json);
       case 'update_message': return UpdateMessageEvent.fromJson(json);
       case 'delete_message': return DeleteMessageEvent.fromJson(json);
@@ -195,7 +196,7 @@ class CustomProfileFieldsEvent extends Event {
 ///
 /// The corresponding API docs are in several places for
 /// different values of `op`; see subclasses.
-abstract class RealmUserEvent extends Event {
+sealed class RealmUserEvent extends Event {
   @override
   @JsonKey(includeToJson: true)
   String get type => 'realm_user';
@@ -308,6 +309,57 @@ class RealmUserUpdateEvent extends RealmUserEvent {
   Map<String, dynamic> toJson() => _$RealmUserUpdateEventToJson(this);
 }
 
+/// A Zulip event of type `stream`.
+///
+/// The corresponding API docs are in several places for
+/// different values of `op`; see subclasses.
+sealed class StreamEvent extends Event {
+  @override
+  @JsonKey(includeToJson: true)
+  String get type => 'stream';
+
+  String get op;
+
+  StreamEvent({required super.id});
+}
+
+/// A [StreamEvent] with op `create`: https://zulip.com/api/get-events#stream-create
+@JsonSerializable(fieldRename: FieldRename.snake)
+class StreamCreateEvent extends StreamEvent {
+  @override
+  String get op => 'create';
+
+  final List<ZulipStream> streams;
+
+  StreamCreateEvent({required super.id, required this.streams});
+
+  factory StreamCreateEvent.fromJson(Map<String, dynamic> json) =>
+    _$StreamCreateEventFromJson(json);
+
+  @override
+  Map<String, dynamic> toJson() => _$StreamCreateEventToJson(this);
+}
+
+/// A [StreamEvent] with op `delete`: https://zulip.com/api/get-events#stream-delete
+@JsonSerializable(fieldRename: FieldRename.snake)
+class StreamDeleteEvent extends StreamEvent {
+  @override
+  String get op => 'delete';
+
+  final List<ZulipStream> streams;
+
+  StreamDeleteEvent({required super.id, required this.streams});
+
+  factory StreamDeleteEvent.fromJson(Map<String, dynamic> json) =>
+    _$StreamDeleteEventFromJson(json);
+
+  @override
+  Map<String, dynamic> toJson() => _$StreamDeleteEventToJson(this);
+}
+
+// TODO(#182) StreamUpdateEvent, for a [StreamEvent] with op `update`:
+//   https://zulip.com/api/get-events#stream-update
+
 /// A Zulip event of type `subscription`.
 ///
 /// The corresponding API docs are in several places for
@@ -492,57 +544,33 @@ class SubscriptionPeerRemoveEvent extends SubscriptionEvent {
   Map<String, dynamic> toJson() => _$SubscriptionPeerRemoveEventToJson(this);
 }
 
-/// A Zulip event of type `stream`.
-///
-/// The corresponding API docs are in several places for
-/// different values of `op`; see subclasses.
-abstract class StreamEvent extends Event {
-  @override
-  @JsonKey(includeToJson: true)
-  String get type => 'stream';
-
-  String get op;
-
-  StreamEvent({required super.id});
-}
-
-/// A [StreamEvent] with op `create`: https://zulip.com/api/get-events#stream-create
+/// A Zulip event of type `user_topic`: https://zulip.com/api/get-events#user_topic
 @JsonSerializable(fieldRename: FieldRename.snake)
-class StreamCreateEvent extends StreamEvent {
+class UserTopicEvent extends Event {
   @override
-  String get op => 'create';
-
-  final List<ZulipStream> streams;
-
-  StreamCreateEvent({required super.id, required this.streams});
-
-  factory StreamCreateEvent.fromJson(Map<String, dynamic> json) =>
-    _$StreamCreateEventFromJson(json);
-
-  @override
-  Map<String, dynamic> toJson() => _$StreamCreateEventToJson(this);
-}
-
-/// A [StreamEvent] with op `delete`: https://zulip.com/api/get-events#stream-delete
-@JsonSerializable(fieldRename: FieldRename.snake)
-class StreamDeleteEvent extends StreamEvent {
-  @override
-  String get op => 'delete';
+  @JsonKey(includeToJson: true)
+  String get type => 'user_topic';
 
-  final List<ZulipStream> streams;
+  final int streamId;
+  final String topicName;
+  final int lastUpdated;
+  final UserTopicVisibilityPolicy visibilityPolicy;
 
-  StreamDeleteEvent({required super.id, required this.streams});
+  UserTopicEvent({
+    required super.id,
+    required this.streamId,
+    required this.topicName,
+    required this.lastUpdated,
+    required this.visibilityPolicy,
+  });
 
-  factory StreamDeleteEvent.fromJson(Map<String, dynamic> json) =>
-    _$StreamDeleteEventFromJson(json);
+  factory UserTopicEvent.fromJson(Map<String, dynamic> json) =>
+    _$UserTopicEventFromJson(json);
 
   @override
-  Map<String, dynamic> toJson() => _$StreamDeleteEventToJson(this);
+  Map<String, dynamic> toJson() => _$UserTopicEventToJson(this);
 }
 
-// TODO(#182) StreamUpdateEvent, for a [StreamEvent] with op `update`:
-//   https://zulip.com/api/get-events#stream-update
-
 /// A Zulip event of type `message`: https://zulip.com/api/get-events#message
 // TODO use [JsonSerializable] here too, using its customization features,
 //   in order to skip the boilerplate in [fromJson] and [toJson].
diff --git a/lib/api/model/events.g.dart b/lib/api/model/events.g.dart
index e378d3cfef..b7257df594 100644
--- a/lib/api/model/events.g.dart
+++ b/lib/api/model/events.g.dart
@@ -175,6 +175,36 @@ const _$UserRoleEnumMap = {
   UserRole.unknown: null,
 };
 
+StreamCreateEvent _$StreamCreateEventFromJson(Map<String, dynamic> json) =>
+    StreamCreateEvent(
+      id: json['id'] as int,
+      streams: (json['streams'] as List<dynamic>)
+          .map((e) => ZulipStream.fromJson(e as Map<String, dynamic>))
+          .toList(),
+    );
+
+Map<String, dynamic> _$StreamCreateEventToJson(StreamCreateEvent instance) =>
+    <String, dynamic>{
+      'id': instance.id,
+      'type': instance.type,
+      'streams': instance.streams,
+    };
+
+StreamDeleteEvent _$StreamDeleteEventFromJson(Map<String, dynamic> json) =>
+    StreamDeleteEvent(
+      id: json['id'] as int,
+      streams: (json['streams'] as List<dynamic>)
+          .map((e) => ZulipStream.fromJson(e as Map<String, dynamic>))
+          .toList(),
+    );
+
+Map<String, dynamic> _$StreamDeleteEventToJson(StreamDeleteEvent instance) =>
+    <String, dynamic>{
+      'id': instance.id,
+      'type': instance.type,
+      'streams': instance.streams,
+    };
+
 SubscriptionAddEvent _$SubscriptionAddEventFromJson(
         Map<String, dynamic> json) =>
     SubscriptionAddEvent(
@@ -285,35 +315,33 @@ Map<String, dynamic> _$SubscriptionPeerRemoveEventToJson(
       'user_ids': instance.userIds,
     };
 
-StreamCreateEvent _$StreamCreateEventFromJson(Map<String, dynamic> json) =>
-    StreamCreateEvent(
+UserTopicEvent _$UserTopicEventFromJson(Map<String, dynamic> json) =>
+    UserTopicEvent(
       id: json['id'] as int,
-      streams: (json['streams'] as List<dynamic>)
-          .map((e) => ZulipStream.fromJson(e as Map<String, dynamic>))
-          .toList(),
+      streamId: json['stream_id'] as int,
+      topicName: json['topic_name'] as String,
+      lastUpdated: json['last_updated'] as int,
+      visibilityPolicy: $enumDecode(
+          _$UserTopicVisibilityPolicyEnumMap, json['visibility_policy']),
     );
 
-Map<String, dynamic> _$StreamCreateEventToJson(StreamCreateEvent instance) =>
+Map<String, dynamic> _$UserTopicEventToJson(UserTopicEvent instance) =>
     <String, dynamic>{
       'id': instance.id,
       'type': instance.type,
-      'streams': instance.streams,
+      'stream_id': instance.streamId,
+      'topic_name': instance.topicName,
+      'last_updated': instance.lastUpdated,
+      'visibility_policy': instance.visibilityPolicy,
     };
 
-StreamDeleteEvent _$StreamDeleteEventFromJson(Map<String, dynamic> json) =>
-    StreamDeleteEvent(
-      id: json['id'] as int,
-      streams: (json['streams'] as List<dynamic>)
-          .map((e) => ZulipStream.fromJson(e as Map<String, dynamic>))
-          .toList(),
-    );
-
-Map<String, dynamic> _$StreamDeleteEventToJson(StreamDeleteEvent instance) =>
-    <String, dynamic>{
-      'id': instance.id,
-      'type': instance.type,
-      'streams': instance.streams,
-    };
+const _$UserTopicVisibilityPolicyEnumMap = {
+  UserTopicVisibilityPolicy.none: 0,
+  UserTopicVisibilityPolicy.muted: 1,
+  UserTopicVisibilityPolicy.unmuted: 2,
+  UserTopicVisibilityPolicy.followed: 3,
+  UserTopicVisibilityPolicy.unknown: null,
+};
 
 UpdateMessageEvent _$UpdateMessageEventFromJson(Map<String, dynamic> json) =>
     UpdateMessageEvent(
diff --git a/lib/api/model/initial_snapshot.dart b/lib/api/model/initial_snapshot.dart
index 198852d971..794f04be4c 100644
--- a/lib/api/model/initial_snapshot.dart
+++ b/lib/api/model/initial_snapshot.dart
@@ -24,6 +24,8 @@ class InitialSnapshot {
 
   final List<CustomProfileField> customProfileFields;
 
+  // final List<…> mutedTopics; // TODO(#422) we ignore this feature on older servers
+
   final Map<String, RealmEmojiItem> realmEmoji;
 
   final List<RecentDmConversation> recentPrivateConversations;
@@ -42,6 +44,8 @@ class InitialSnapshot {
   // TODO(server-5) remove pre-5.0 comment
   final UserSettings? userSettings; // TODO(server-5)
 
+  final List<UserTopicItem>? userTopics; // TODO(server-6)
+
   final Map<String, RealmDefaultExternalAccount> realmDefaultExternalAccounts;
 
   final int maxFileUploadSizeMib;
@@ -88,6 +92,7 @@ class InitialSnapshot {
     required this.unreadMsgs,
     required this.streams,
     required this.userSettings,
+    required this.userTopics,
     required this.realmDefaultExternalAccounts,
     required this.maxFileUploadSizeMib,
     required this.realmUsers,
@@ -125,34 +130,6 @@ class RealmDefaultExternalAccount {
   Map<String, dynamic> toJson() => _$RealmDefaultExternalAccountToJson(this);
 }
 
-/// An item in `realm_emoji`.
-///
-/// For docs, search for "realm_emoji:"
-/// in <https://zulip.com/api/register-queue>.
-@JsonSerializable(fieldRename: FieldRename.snake)
-class RealmEmojiItem {
-  final String id;
-  final String name;
-  final String sourceUrl;
-  final String? stillUrl;
-  final bool deactivated;
-  final int? authorId;
-
-  RealmEmojiItem({
-    required this.id,
-    required this.name,
-    required this.sourceUrl,
-    required this.stillUrl,
-    required this.deactivated,
-    required this.authorId,
-  });
-
-  factory RealmEmojiItem.fromJson(Map<String, dynamic> json) =>
-    _$RealmEmojiItemFromJson(json);
-
-  Map<String, dynamic> toJson() => _$RealmEmojiItemToJson(this);
-}
-
 /// An item in `recent_private_conversations`.
 ///
 /// For docs, search for "recent_private_conversations:"
@@ -205,45 +182,29 @@ class UserSettings {
   static final Iterable<String> debugKnownNames = _$UserSettingsFieldMap.keys;
 }
 
-/// The name of a user setting that has a property in [UserSettings].
+/// An item in the `user_topics` snapshot.
 ///
-/// In Zulip event-handling code (for [UserSettingsUpdateEvent]),
-/// we switch exhaustively on a value of this type
-/// to ensure that every setting in [UserSettings] responds to the event.
-@JsonEnum(fieldRename: FieldRename.snake, alwaysCreate: true)
-enum UserSettingName {
-  twentyFourHourTime,
-  displayEmojiReactionUsers,
-  emojiset;
-
-  /// Get a [UserSettingName] from a raw, snake-case string we recognize, else null.
-  ///
-  /// Example:
-  ///   'display_emoji_reaction_users' -> UserSettingName.displayEmojiReactionUsers
-  static UserSettingName? fromRawString(String raw) => _byRawString[raw];
-
-  // _$…EnumMap is thanks to `alwaysCreate: true` and `fieldRename: FieldRename.snake`
-  static final _byRawString = _$UserSettingNameEnumMap
-    .map((key, value) => MapEntry(value, key));
-}
+/// For docs, search for "user_topics:"
+/// in <https://zulip.com/api/register-queue>.
+@JsonSerializable(fieldRename: FieldRename.snake)
+class UserTopicItem {
+  final int streamId;
+  final String topicName;
+  final int lastUpdated;
+  @JsonKey(unknownEnumValue: UserTopicVisibilityPolicy.unknown)
+  final UserTopicVisibilityPolicy visibilityPolicy;
+
+  UserTopicItem({
+    required this.streamId,
+    required this.topicName,
+    required this.lastUpdated,
+    required this.visibilityPolicy,
+  });
+
+  factory UserTopicItem.fromJson(Map<String, dynamic> json) =>
+    _$UserTopicItemFromJson(json);
 
-/// As in [UserSettings.emojiset].
-@JsonEnum(fieldRename: FieldRename.kebab, alwaysCreate: true)
-enum Emojiset {
-  google,
-  googleBlob,
-  twitter,
-  text;
-
-  /// Get an [Emojiset] from a raw string. Throws if the string is unrecognized.
-  ///
-  /// Example:
-  ///   'google-blob' -> Emojiset.googleBlob
-  static Emojiset fromRawString(String raw) => _byRawString[raw]!;
-
-  // _$…EnumMap is thanks to `alwaysCreate: true` and `fieldRename: FieldRename.kebab`
-  static final _byRawString = _$EmojisetEnumMap
-    .map((key, value) => MapEntry(value, key));
+  Map<String, dynamic> toJson() => _$UserTopicItemToJson(this);
 }
 
 /// The `unread_msgs` snapshot.
diff --git a/lib/api/model/initial_snapshot.g.dart b/lib/api/model/initial_snapshot.g.dart
index 3815547a65..23687879ff 100644
--- a/lib/api/model/initial_snapshot.g.dart
+++ b/lib/api/model/initial_snapshot.g.dart
@@ -41,6 +41,9 @@ InitialSnapshot _$InitialSnapshotFromJson(Map<String, dynamic> json) =>
           ? null
           : UserSettings.fromJson(
               json['user_settings'] as Map<String, dynamic>),
+      userTopics: (json['user_topics'] as List<dynamic>?)
+          ?.map((e) => UserTopicItem.fromJson(e as Map<String, dynamic>))
+          .toList(),
       realmDefaultExternalAccounts:
           (json['realm_default_external_accounts'] as Map<String, dynamic>).map(
         (k, e) => MapEntry(
@@ -77,6 +80,7 @@ Map<String, dynamic> _$InitialSnapshotToJson(InitialSnapshot instance) =>
       'unread_msgs': instance.unreadMsgs,
       'streams': instance.streams,
       'user_settings': instance.userSettings,
+      'user_topics': instance.userTopics,
       'realm_default_external_accounts': instance.realmDefaultExternalAccounts,
       'max_file_upload_size_mib': instance.maxFileUploadSizeMib,
       'realm_users': instance.realmUsers,
@@ -102,26 +106,6 @@ Map<String, dynamic> _$RealmDefaultExternalAccountToJson(
       'url_pattern': instance.urlPattern,
     };
 
-RealmEmojiItem _$RealmEmojiItemFromJson(Map<String, dynamic> json) =>
-    RealmEmojiItem(
-      id: json['id'] as String,
-      name: json['name'] as String,
-      sourceUrl: json['source_url'] as String,
-      stillUrl: json['still_url'] as String?,
-      deactivated: json['deactivated'] as bool,
-      authorId: json['author_id'] as int?,
-    );
-
-Map<String, dynamic> _$RealmEmojiItemToJson(RealmEmojiItem instance) =>
-    <String, dynamic>{
-      'id': instance.id,
-      'name': instance.name,
-      'source_url': instance.sourceUrl,
-      'still_url': instance.stillUrl,
-      'deactivated': instance.deactivated,
-      'author_id': instance.authorId,
-    };
-
 RecentDmConversation _$RecentDmConversationFromJson(
         Map<String, dynamic> json) =>
     RecentDmConversation(
@@ -163,6 +147,32 @@ const _$EmojisetEnumMap = {
   Emojiset.text: 'text',
 };
 
+UserTopicItem _$UserTopicItemFromJson(Map<String, dynamic> json) =>
+    UserTopicItem(
+      streamId: json['stream_id'] as int,
+      topicName: json['topic_name'] as String,
+      lastUpdated: json['last_updated'] as int,
+      visibilityPolicy: $enumDecode(
+          _$UserTopicVisibilityPolicyEnumMap, json['visibility_policy'],
+          unknownValue: UserTopicVisibilityPolicy.unknown),
+    );
+
+Map<String, dynamic> _$UserTopicItemToJson(UserTopicItem instance) =>
+    <String, dynamic>{
+      'stream_id': instance.streamId,
+      'topic_name': instance.topicName,
+      'last_updated': instance.lastUpdated,
+      'visibility_policy': instance.visibilityPolicy,
+    };
+
+const _$UserTopicVisibilityPolicyEnumMap = {
+  UserTopicVisibilityPolicy.none: 0,
+  UserTopicVisibilityPolicy.muted: 1,
+  UserTopicVisibilityPolicy.unmuted: 2,
+  UserTopicVisibilityPolicy.followed: 3,
+  UserTopicVisibilityPolicy.unknown: null,
+};
+
 UnreadMessagesSnapshot _$UnreadMessagesSnapshotFromJson(
         Map<String, dynamic> json) =>
     UnreadMessagesSnapshot(
@@ -240,9 +250,3 @@ Map<String, dynamic> _$UnreadHuddleSnapshotToJson(
       'user_ids_string': instance.userIdsString,
       'unread_message_ids': instance.unreadMessageIds,
     };
-
-const _$UserSettingNameEnumMap = {
-  UserSettingName.twentyFourHourTime: 'twenty_four_hour_time',
-  UserSettingName.displayEmojiReactionUsers: 'display_emoji_reaction_users',
-  UserSettingName.emojiset: 'emojiset',
-};
diff --git a/lib/api/model/model.dart b/lib/api/model/model.dart
index 730b28ff89..25b0e1debb 100644
--- a/lib/api/model/model.dart
+++ b/lib/api/model/model.dart
@@ -4,6 +4,8 @@ import 'package:flutter_color_models/flutter_color_models.dart';
 import 'package:json_annotation/json_annotation.dart';
 
 import '../../widgets/color.dart';
+import 'events.dart';
+import 'initial_snapshot.dart';
 import 'reaction.dart';
 
 export 'reaction.dart';
@@ -109,6 +111,76 @@ class CustomProfileFieldExternalAccountData {
   Map<String, dynamic> toJson() => _$CustomProfileFieldExternalAccountDataToJson(this);
 }
 
+/// An item in [InitialSnapshot.realmEmoji] or [RealmEmojiUpdateEvent].
+///
+/// For docs, search for "realm_emoji:"
+/// in <https://zulip.com/api/register-queue>.
+@JsonSerializable(fieldRename: FieldRename.snake)
+class RealmEmojiItem {
+  final String id;
+  final String name;
+  final String sourceUrl;
+  final String? stillUrl;
+  final bool deactivated;
+  final int? authorId;
+
+  RealmEmojiItem({
+    required this.id,
+    required this.name,
+    required this.sourceUrl,
+    required this.stillUrl,
+    required this.deactivated,
+    required this.authorId,
+  });
+
+  factory RealmEmojiItem.fromJson(Map<String, dynamic> json) =>
+    _$RealmEmojiItemFromJson(json);
+
+  Map<String, dynamic> toJson() => _$RealmEmojiItemToJson(this);
+}
+
+
+/// The name of a user setting that has a property in [UserSettings].
+///
+/// In Zulip event-handling code (for [UserSettingsUpdateEvent]),
+/// we switch exhaustively on a value of this type
+/// to ensure that every setting in [UserSettings] responds to the event.
+@JsonEnum(fieldRename: FieldRename.snake, alwaysCreate: true)
+enum UserSettingName {
+  twentyFourHourTime,
+  displayEmojiReactionUsers,
+  emojiset;
+
+  /// Get a [UserSettingName] from a raw, snake-case string we recognize, else null.
+  ///
+  /// Example:
+  ///   'display_emoji_reaction_users' -> UserSettingName.displayEmojiReactionUsers
+  static UserSettingName? fromRawString(String raw) => _byRawString[raw];
+
+  // _$…EnumMap is thanks to `alwaysCreate: true` and `fieldRename: FieldRename.snake`
+  static final _byRawString = _$UserSettingNameEnumMap
+    .map((key, value) => MapEntry(value, key));
+}
+
+/// As in [UserSettings.emojiset].
+@JsonEnum(fieldRename: FieldRename.kebab, alwaysCreate: true)
+enum Emojiset {
+  google,
+  googleBlob,
+  twitter,
+  text;
+
+  /// Get an [Emojiset] from a raw string. Throws if the string is unrecognized.
+  ///
+  /// Example:
+  ///   'google-blob' -> Emojiset.googleBlob
+  static Emojiset fromRawString(String raw) => _byRawString[raw]!;
+
+  // _$…EnumMap is thanks to `alwaysCreate: true` and `fieldRename: FieldRename.kebab`
+  static final _byRawString = _$EmojisetEnumMap
+    .map((key, value) => MapEntry(value, key));
+}
+
 /// As in [InitialSnapshot.realmUsers], [InitialSnapshot.realmNonActiveUsers], and [InitialSnapshot.crossRealmBots].
 ///
 /// In the Zulip API, the items in realm_users, realm_non_active_users, and
@@ -473,6 +545,21 @@ enum _StreamColorVariant {
   barBackground,
 }
 
+@JsonEnum(fieldRename: FieldRename.snake, valueField: "apiValue")
+enum UserTopicVisibilityPolicy {
+  none(apiValue: 0),
+  muted(apiValue: 1),
+  unmuted(apiValue: 2), // TODO(server-7) newly added
+  followed(apiValue: 3), // TODO(server-8) newly added
+  unknown(apiValue: null);
+
+  const UserTopicVisibilityPolicy({required this.apiValue});
+
+  final int? apiValue;
+
+  int? toJson() => apiValue;
+}
+
 /// As in the get-messages response.
 ///
 /// https://zulip.com/api/get-messages#response
diff --git a/lib/api/model/model.g.dart b/lib/api/model/model.g.dart
index 2269da8673..fd1e355d69 100644
--- a/lib/api/model/model.g.dart
+++ b/lib/api/model/model.g.dart
@@ -70,6 +70,26 @@ Map<String, dynamic> _$CustomProfileFieldExternalAccountDataToJson(
       'url_pattern': instance.urlPattern,
     };
 
+RealmEmojiItem _$RealmEmojiItemFromJson(Map<String, dynamic> json) =>
+    RealmEmojiItem(
+      id: json['id'] as String,
+      name: json['name'] as String,
+      sourceUrl: json['source_url'] as String,
+      stillUrl: json['still_url'] as String?,
+      deactivated: json['deactivated'] as bool,
+      authorId: json['author_id'] as int?,
+    );
+
+Map<String, dynamic> _$RealmEmojiItemToJson(RealmEmojiItem instance) =>
+    <String, dynamic>{
+      'id': instance.id,
+      'name': instance.name,
+      'source_url': instance.sourceUrl,
+      'still_url': instance.stillUrl,
+      'deactivated': instance.deactivated,
+      'author_id': instance.authorId,
+    };
+
 User _$UserFromJson(Map<String, dynamic> json) => User(
       userId: json['user_id'] as int,
       deliveryEmailStaleDoNotUse: json['delivery_email'] as String?,
@@ -343,6 +363,19 @@ Map<String, dynamic> _$DmMessageToJson(DmMessage instance) => <String, dynamic>{
           const DmRecipientListConverter().toJson(instance.displayRecipient),
     };
 
+const _$UserSettingNameEnumMap = {
+  UserSettingName.twentyFourHourTime: 'twenty_four_hour_time',
+  UserSettingName.displayEmojiReactionUsers: 'display_emoji_reaction_users',
+  UserSettingName.emojiset: 'emojiset',
+};
+
+const _$EmojisetEnumMap = {
+  Emojiset.google: 'google',
+  Emojiset.googleBlob: 'google-blob',
+  Emojiset.twitter: 'twitter',
+  Emojiset.text: 'text',
+};
+
 const _$MessageFlagEnumMap = {
   MessageFlag.read: 'read',
   MessageFlag.starred: 'starred',
diff --git a/lib/model/message_list.dart b/lib/model/message_list.dart
index ad10acfec2..f9f65e0aab 100644
--- a/lib/model/message_list.dart
+++ b/lib/model/message_list.dart
@@ -307,6 +307,48 @@ class MessageListView with ChangeNotifier, _MessageSequence {
   final PerAccountStore store;
   final Narrow narrow;
 
+  /// Whether [message] should actually appear in this message list,
+  /// given that it does belong to the narrow.
+  ///
+  /// This depends in particular on whether the message is muted in
+  /// one way or another.
+  ///
+  /// See also [_allMessagesVisible].
+  bool _messageVisible(Message message) {
+    switch (narrow) {
+      case AllMessagesNarrow():
+        return switch (message) {
+          StreamMessage() =>
+            store.isTopicVisible(message.streamId, message.subject),
+          DmMessage() => true,
+        };
+
+      case StreamNarrow(:final streamId):
+        assert(message is StreamMessage && message.streamId == streamId);
+        if (message is! StreamMessage) return false;
+        return store.isTopicVisibleInStream(streamId, message.subject);
+
+      case TopicNarrow():
+      case DmNarrow():
+        return true;
+    }
+  }
+
+  /// Whether [_messageVisible] is true for all possible messages.
+  ///
+  /// This is useful for an optimization.
+  bool get _allMessagesVisible {
+    switch (narrow) {
+      case AllMessagesNarrow():
+      case StreamNarrow():
+        return false;
+
+      case TopicNarrow():
+      case DmNarrow():
+        return true;
+    }
+  }
+
   /// Fetch messages, starting from scratch.
   Future<void> fetchInitial() async {
     // TODO(#80): fetch from anchor firstUnread, instead of newest
@@ -321,7 +363,9 @@ class MessageListView with ChangeNotifier, _MessageSequence {
       numAfter: 0,
     );
     for (final message in result.messages) {
-      _addMessage(message);
+      if (_messageVisible(message)) {
+        _addMessage(message);
+      }
     }
     _fetched = true;
     _haveOldest = result.foundOldest;
@@ -353,7 +397,11 @@ class MessageListView with ChangeNotifier, _MessageSequence {
         result.messages.removeLast();
       }
 
-      _insertAllMessages(0, result.messages);
+      final fetchedMessages = _allMessagesVisible
+        ? result.messages // Avoid unnecessarily copying the list.
+        : result.messages.where(_messageVisible);
+
+      _insertAllMessages(0, fetchedMessages);
       _haveOldest = result.foundOldest;
     } finally {
       _fetchingOlder = false;
@@ -366,7 +414,7 @@ class MessageListView with ChangeNotifier, _MessageSequence {
   ///
   /// Called in particular when we get a [MessageEvent].
   void maybeAddMessage(Message message) {
-    if (!narrow.containsMessage(message)) {
+    if (!narrow.containsMessage(message) || !_messageVisible(message)) {
       return;
     }
     if (!_fetched) {
diff --git a/lib/model/narrow.dart b/lib/model/narrow.dart
index 5d2784c2d0..926406bd29 100644
--- a/lib/model/narrow.dart
+++ b/lib/model/narrow.dart
@@ -11,8 +11,14 @@ sealed class Narrow {
   /// This const constructor allows subclasses to have const constructors.
   const Narrow();
 
-  // TODO implement muting; will need containsMessage to take more params
-  //   This means stream muting, topic un/muting, and user muting.
+  /// Whether this message satisfies the filters of this narrow.
+  ///
+  /// This is true just when the server would be expected to include the message
+  /// in a [getMessages] request for this narrow, given appropriate anchor etc.
+  ///
+  /// This does not necessarily mean the message list would show this message
+  /// when navigated to this narrow; in particular it does not address the
+  /// question of whether the stream or topic, or the sending user, is muted.
   bool containsMessage(Message message);
 
   /// This narrow, expressed as an [ApiNarrow].
diff --git a/lib/model/store.dart b/lib/model/store.dart
index adb6270275..ea4833ab62 100644
--- a/lib/model/store.dart
+++ b/lib/model/store.dart
@@ -18,6 +18,7 @@ import 'autocomplete.dart';
 import 'database.dart';
 import 'message_list.dart';
 import 'recent_dm_conversations.dart';
+import 'stream.dart';
 import 'unreads.dart';
 
 export 'package:drift/drift.dart' show Value;
@@ -142,35 +143,56 @@ abstract class GlobalStore extends ChangeNotifier {
 /// An instance directly of this class will not attempt to poll an event queue
 /// to keep the data up to date.  For that behavior, see the subclass
 /// [LivePerAccountStore].
-class PerAccountStore extends ChangeNotifier {
+class PerAccountStore extends ChangeNotifier with StreamStore {
   /// Create a per-account data store that does not automatically stay up to date.
   ///
   /// For a [PerAccountStore] that polls an event queue to keep itself up to
   /// date, use [LivePerAccountStore.fromInitialSnapshot].
-  PerAccountStore.fromInitialSnapshot({
+  factory PerAccountStore.fromInitialSnapshot({
+    required Account account,
+    required ApiConnection connection,
+    required InitialSnapshot initialSnapshot,
+  }) {
+    final streams = StreamStoreImpl(initialSnapshot: initialSnapshot);
+    return PerAccountStore._(
+      account: account,
+      connection: connection,
+      zulipVersion: initialSnapshot.zulipVersion,
+      maxFileUploadSizeMib: initialSnapshot.maxFileUploadSizeMib,
+      realmDefaultExternalAccounts: initialSnapshot.realmDefaultExternalAccounts,
+      realmEmoji: initialSnapshot.realmEmoji,
+      customProfileFields: _sortCustomProfileFields(initialSnapshot.customProfileFields),
+      userSettings: initialSnapshot.userSettings,
+      unreads: Unreads(
+        initial: initialSnapshot.unreadMsgs,
+        selfUserId: account.userId,
+        streamStore: streams,
+      ),
+      users: Map.fromEntries(
+        initialSnapshot.realmUsers
+        .followedBy(initialSnapshot.realmNonActiveUsers)
+        .followedBy(initialSnapshot.crossRealmBots)
+        .map((user) => MapEntry(user.userId, user))),
+      streams: streams,
+      recentDmConversationsView: RecentDmConversationsView(
+        initial: initialSnapshot.recentPrivateConversations, selfUserId: account.userId),
+    );
+  }
+
+  PerAccountStore._({
     required this.account,
     required this.connection,
-    required InitialSnapshot initialSnapshot,
-  }) : zulipVersion = initialSnapshot.zulipVersion,
-       maxFileUploadSizeMib = initialSnapshot.maxFileUploadSizeMib,
-       realmDefaultExternalAccounts = initialSnapshot.realmDefaultExternalAccounts,
-       realmEmoji = initialSnapshot.realmEmoji,
-       customProfileFields = _sortCustomProfileFields(initialSnapshot.customProfileFields),
-       userSettings = initialSnapshot.userSettings,
-       unreads = Unreads(initial: initialSnapshot.unreadMsgs, selfUserId: account.userId),
-       users = Map.fromEntries(
-         initialSnapshot.realmUsers
-         .followedBy(initialSnapshot.realmNonActiveUsers)
-         .followedBy(initialSnapshot.crossRealmBots)
-         .map((user) => MapEntry(user.userId, user))),
-       streams = Map.fromEntries(initialSnapshot.streams.map(
-         (stream) => MapEntry(stream.streamId, stream))),
-       streamsByName = Map.fromEntries(initialSnapshot.streams.map(
-         (stream) => MapEntry(stream.name, stream))),
-       subscriptions = Map.fromEntries(initialSnapshot.subscriptions.map(
-         (subscription) => MapEntry(subscription.streamId, subscription))),
-       recentDmConversationsView = RecentDmConversationsView(
-         initial: initialSnapshot.recentPrivateConversations, selfUserId: account.userId);
+    required this.zulipVersion,
+    required this.maxFileUploadSizeMib,
+    required this.realmDefaultExternalAccounts,
+    required this.realmEmoji,
+    required this.customProfileFields,
+    required this.userSettings,
+    required this.unreads,
+    required this.users,
+    required streams,
+    required this.recentDmConversationsView,
+  }) : _streams = streams;
 
   final Account account;
   final ApiConnection connection; // TODO(#135): update zulipFeatureLevel with events
@@ -192,9 +214,21 @@ class PerAccountStore extends ChangeNotifier {
   final Map<int, User> users;
 
   // Streams, topics, and stuff about them.
-  final Map<int, ZulipStream> streams;
-  final Map<String, ZulipStream> streamsByName;
-  final Map<int, Subscription> subscriptions;
+
+  @override
+  Map<int, ZulipStream> get streams => _streams.streams;
+  @override
+  Map<String, ZulipStream> get streamsByName => _streams.streamsByName;
+  @override
+  Map<int, Subscription> get subscriptions => _streams.subscriptions;
+  @override
+  UserTopicVisibilityPolicy topicVisibilityPolicy(int streamId, String topic) =>
+    _streams.topicVisibilityPolicy(streamId, topic);
+
+  final StreamStoreImpl _streams;
+
+  @visibleForTesting
+  StreamStoreImpl get debugStreamStore => _streams;
 
   // TODO lots more data.  When adding, be sure to update handleEvent too.
 
@@ -295,67 +329,18 @@ class PerAccountStore extends ChangeNotifier {
       }
       autocompleteViewManager.handleRealmUserUpdateEvent(event);
       notifyListeners();
-    } else if (event is StreamCreateEvent) {
-      assert(debugLog("server event: stream/create"));
-      streams.addEntries(event.streams.map((stream) => MapEntry(stream.streamId, stream)));
-      streamsByName.addEntries(event.streams.map((stream) => MapEntry(stream.name, stream)));
-      // (Don't touch `subscriptions`. If the user is subscribed to the stream,
-      // details will come in a later `subscription` event.)
+    } else if (event is StreamEvent) {
+      assert(debugLog("server event: stream/${event.op}"));
+      _streams.handleStreamEvent(event);
       notifyListeners();
-    } else if (event is StreamDeleteEvent) {
-      assert(debugLog("server event: stream/delete"));
-      for (final stream in event.streams) {
-        streams.remove(stream.streamId);
-        streamsByName.remove(stream.name);
-        subscriptions.remove(stream.streamId);
-      }
+    } else if (event is SubscriptionEvent) {
+      assert(debugLog("server event: subscription/${event.op}"));
+      _streams.handleSubscriptionEvent(event);
       notifyListeners();
-    } else if (event is SubscriptionAddEvent) {
-      assert(debugLog("server event: subscription/add"));
-      for (final subscription in event.subscriptions) {
-        subscriptions[subscription.streamId] = subscription;
-      }
+    } else if (event is UserTopicEvent) {
+      assert(debugLog("server event: user_topic"));
+      _streams.handleUserTopicEvent(event);
       notifyListeners();
-    } else if (event is SubscriptionRemoveEvent) {
-      assert(debugLog("server event: subscription/remove"));
-      for (final streamId in event.streamIds) {
-        subscriptions.remove(streamId);
-      }
-      notifyListeners();
-    } else if (event is SubscriptionUpdateEvent) {
-      assert(debugLog("server event: subscription/update"));
-      final subscription = subscriptions[event.streamId];
-      if (subscription == null) return; // TODO(log)
-      switch (event.property) {
-        case SubscriptionProperty.color:
-          subscription.color                  = event.value as int;
-        case SubscriptionProperty.isMuted:
-          subscription.isMuted                = event.value as bool;
-        case SubscriptionProperty.inHomeView:
-          subscription.isMuted                = !(event.value as bool);
-        case SubscriptionProperty.pinToTop:
-          subscription.pinToTop               = event.value as bool;
-        case SubscriptionProperty.desktopNotifications:
-          subscription.desktopNotifications   = event.value as bool;
-        case SubscriptionProperty.audibleNotifications:
-          subscription.audibleNotifications   = event.value as bool;
-        case SubscriptionProperty.pushNotifications:
-          subscription.pushNotifications      = event.value as bool;
-        case SubscriptionProperty.emailNotifications:
-          subscription.emailNotifications     = event.value as bool;
-        case SubscriptionProperty.wildcardMentionsNotify:
-          subscription.wildcardMentionsNotify = event.value as bool;
-        case SubscriptionProperty.unknown:
-          // unrecognized property; do nothing
-          return;
-      }
-      notifyListeners();
-    } else if (event is SubscriptionPeerAddEvent) {
-      assert(debugLog("server event: subscription/peer_add"));
-      // TODO(#374): handle event
-    } else if (event is SubscriptionPeerRemoveEvent) {
-      assert(debugLog("server event: subscription/peer_remove"));
-      // TODO(#374): handle event
     } else if (event is MessageEvent) {
       assert(debugLog("server event: message ${jsonEncode(event.message.toJson())}"));
       recentDmConversationsView.handleMessageEvent(event);
@@ -459,8 +444,9 @@ class LiveGlobalStore extends GlobalStore {
   final AppDatabase _db;
 
   @override
-  Future<PerAccountStore> loadPerAccount(Account account) {
-    return LivePerAccountStore.load(account);
+  Future<PerAccountStore> loadPerAccount(Account account) async {
+    final updateMachine = await UpdateMachine.load(account);
+    return updateMachine.store;
   }
 
   @override
@@ -477,26 +463,24 @@ class LiveGlobalStore extends GlobalStore {
   }
 }
 
-/// A [PerAccountStore] which polls an event queue to stay up to date.
+/// A [PerAccountStore] plus an event-polling loop to stay up to date.
 // TODO decouple "live"ness from polling and registerNotificationToken;
 //   the latter are made up of testable internal logic, not external integration
-class LivePerAccountStore extends PerAccountStore {
-  LivePerAccountStore.fromInitialSnapshot({
-    required super.account,
-    required super.connection,
-    required super.initialSnapshot,
+class UpdateMachine {
+  UpdateMachine.fromInitialSnapshot({
+    required this.store,
+    required InitialSnapshot initialSnapshot,
   }) : queueId = initialSnapshot.queueId ?? (() {
          // The queueId is optional in the type, but should only be missing in the
          // case of unauthenticated access to a web-public realm.  We authenticated.
          throw Exception("bad initial snapshot: missing queueId");
        })(),
-       lastEventId = initialSnapshot.lastEventId,
-       super.fromInitialSnapshot();
+       lastEventId = initialSnapshot.lastEventId;
 
   /// Load the user's data from the server, and start an event queue going.
   ///
   /// In the future this might load an old snapshot from local storage first.
-  static Future<PerAccountStore> load(Account account) async {
+  static Future<UpdateMachine> load(Account account) async {
     final connection = ApiConnection.live(
       realmUrl: account.realmUrl, zulipFeatureLevel: account.zulipFeatureLevel,
       email: account.email, apiKey: account.apiKey);
@@ -507,30 +491,33 @@ class LivePerAccountStore extends PerAccountStore {
     // TODO log the time better
     if (kDebugMode) print("initial fetch time: ${t.inMilliseconds}ms");
 
-    final store = LivePerAccountStore.fromInitialSnapshot(
+    final store = PerAccountStore.fromInitialSnapshot(
       account: account,
       connection: connection,
       initialSnapshot: initialSnapshot,
     );
-    store.poll();
+    final updateMachine = UpdateMachine.fromInitialSnapshot(
+      store: store, initialSnapshot: initialSnapshot);
+    updateMachine.poll();
     // TODO do registerNotificationToken before registerQueue:
     //   https://github.com/zulip/zulip-flutter/pull/325#discussion_r1365982807
-    store.registerNotificationToken();
-    return store;
+    updateMachine.registerNotificationToken();
+    return updateMachine;
   }
 
+  final PerAccountStore store;
   final String queueId;
   int lastEventId;
 
   void poll() async {
     while (true) {
-      final result = await getEvents(connection,
+      final result = await getEvents(store.connection,
         queueId: queueId, lastEventId: lastEventId);
       // TODO handle errors on get-events; retry with backoff
       // TODO abort long-poll and close ApiConnection on [dispose]
       final events = result.events;
       for (final event in events) {
-        handleEvent(event);
+        store.handleEvent(event);
       }
       if (events.isNotEmpty) {
         lastEventId = events.last.id;
@@ -554,6 +541,6 @@ class LivePerAccountStore extends PerAccountStore {
   Future<void> _registerNotificationToken() async {
     final token = NotificationService.instance.token.value;
     if (token == null) return;
-    await NotificationService.registerToken(connection, token: token);
+    await NotificationService.registerToken(store.connection, token: token);
   }
 }
diff --git a/lib/model/stream.dart b/lib/model/stream.dart
new file mode 100644
index 0000000000..887155988d
--- /dev/null
+++ b/lib/model/stream.dart
@@ -0,0 +1,230 @@
+import '../api/model/events.dart';
+import '../api/model/initial_snapshot.dart';
+import '../api/model/model.dart';
+
+/// The portion of [PerAccountStore] for streams, topics, and stuff about them.
+///
+/// This type is useful for expressing the needs of other parts of the
+/// implementation of [PerAccountStore], to avoid circularity.
+///
+/// The data structures described here are implemented at [StreamStoreImpl].
+mixin StreamStore {
+  Map<int, ZulipStream> get streams;
+  Map<String, ZulipStream> get streamsByName;
+  Map<int, Subscription> get subscriptions;
+  UserTopicVisibilityPolicy topicVisibilityPolicy(int streamId, String topic);
+
+  /// Whether this topic should appear when already focusing on its stream.
+  ///
+  /// This is determined purely by the user's visibility policy for the topic.
+  ///
+  /// This function is appropriate for muting calculations in UI contexts that
+  /// are already specific to a stream: for example the stream's unread count,
+  /// or the message list in the stream's narrow.
+  ///
+  /// For UI contexts that are not specific to a particular stream, see
+  /// [isTopicVisible].
+  bool isTopicVisibleInStream(int streamId, String topic) {
+    switch (topicVisibilityPolicy(streamId, topic)) {
+      case UserTopicVisibilityPolicy.none:
+        return true;
+      case UserTopicVisibilityPolicy.muted:
+        return false;
+      case UserTopicVisibilityPolicy.unmuted:
+      case UserTopicVisibilityPolicy.followed:
+        return true;
+      case UserTopicVisibilityPolicy.unknown:
+        assert(false);
+        return true;
+    }
+  }
+
+  /// Whether this topic should appear when not specifically focusing
+  /// on this stream.
+  ///
+  /// This takes into account the user's visibility policy for the stream
+  /// overall, as well as their policy for this topic.
+  ///
+  /// For UI contexts that are specific to a particular stream, see
+  /// [isTopicVisibleInStream].
+  bool isTopicVisible(int streamId, String topic) {
+    switch (topicVisibilityPolicy(streamId, topic)) {
+      case UserTopicVisibilityPolicy.none:
+        switch (subscriptions[streamId]?.isMuted) {
+          case false: return true;
+          case true:  return false;
+          case null:  return false; // not subscribed; treat like muted
+        }
+      case UserTopicVisibilityPolicy.muted:
+        return false;
+      case UserTopicVisibilityPolicy.unmuted:
+      case UserTopicVisibilityPolicy.followed:
+        return true;
+      case UserTopicVisibilityPolicy.unknown:
+        assert(false);
+        return true;
+    }
+  }
+}
+
+/// The implementation of [StreamStore] that does the work.
+///
+/// Generally the only code that should need this class is [PerAccountStore]
+/// itself.  Other code accesses this functionality through [PerAccountStore],
+/// or through the mixin [StreamStore] which describes its interface.
+class StreamStoreImpl with StreamStore {
+  factory StreamStoreImpl({required InitialSnapshot initialSnapshot}) {
+    final subscriptions = Map.fromEntries(initialSnapshot.subscriptions.map(
+      (subscription) => MapEntry(subscription.streamId, subscription)));
+
+    final streams = Map<int, ZulipStream>.of(subscriptions);
+    for (final stream in initialSnapshot.streams) {
+      streams.putIfAbsent(stream.streamId, () => stream);
+    }
+
+    final topicVisibility = <int, Map<String, UserTopicVisibilityPolicy>>{};
+    for (final item in initialSnapshot.userTopics ?? const <UserTopicItem>[]) {
+      if (_warnInvalidVisibilityPolicy(item.visibilityPolicy)) {
+        // Not a value we expect. Keep it out of our data structures. // TODO(log)
+        continue;
+      }
+      final forStream = topicVisibility.putIfAbsent(item.streamId, () => {});
+      forStream[item.topicName] = item.visibilityPolicy;
+    }
+
+    return StreamStoreImpl._(
+      streams: streams,
+      streamsByName: streams.map((_, stream) => MapEntry(stream.name, stream)),
+      subscriptions: subscriptions,
+      topicVisibility: topicVisibility,
+    );
+  }
+
+  StreamStoreImpl._({
+    required this.streams,
+    required this.streamsByName,
+    required this.subscriptions,
+    required this.topicVisibility,
+  });
+
+  @override
+  final Map<int, ZulipStream> streams;
+  @override
+  final Map<String, ZulipStream> streamsByName;
+  @override
+  final Map<int, Subscription> subscriptions;
+
+  final Map<int, Map<String, UserTopicVisibilityPolicy>> topicVisibility;
+
+  @override
+  UserTopicVisibilityPolicy topicVisibilityPolicy(int streamId, String topic) {
+    return topicVisibility[streamId]?[topic] ?? UserTopicVisibilityPolicy.none;
+  }
+
+  static bool _warnInvalidVisibilityPolicy(UserTopicVisibilityPolicy visibilityPolicy) {
+    if (visibilityPolicy == UserTopicVisibilityPolicy.unknown) {
+      // Not a value we expect. Keep it out of our data structures. // TODO(log)
+      return true;
+    }
+    return false;
+  }
+
+  void handleStreamEvent(StreamEvent event) {
+    switch (event) {
+      case StreamCreateEvent():
+        assert(event.streams.every((stream) =>
+          !streams.containsKey(stream.streamId)
+          && !streamsByName.containsKey(stream.name)));
+        streams.addEntries(event.streams.map((stream) => MapEntry(stream.streamId, stream)));
+        streamsByName.addEntries(event.streams.map((stream) => MapEntry(stream.name, stream)));
+        // (Don't touch `subscriptions`. If the user is subscribed to the stream,
+        // details will come in a later `subscription` event.)
+
+      case StreamDeleteEvent():
+        for (final stream in event.streams) {
+          assert(identical(streams[stream.streamId], streamsByName[stream.name]));
+          assert(subscriptions[stream.streamId] == null
+            || identical(subscriptions[stream.streamId], streams[stream.streamId]));
+          streams.remove(stream.streamId);
+          streamsByName.remove(stream.name);
+          subscriptions.remove(stream.streamId);
+        }
+    }
+  }
+
+  void handleSubscriptionEvent(SubscriptionEvent event) {
+    switch (event) {
+      case SubscriptionAddEvent():
+        for (final subscription in event.subscriptions) {
+          assert(streams.containsKey(subscription.streamId)
+            && streams[subscription.streamId] is! Subscription);
+          assert(streamsByName.containsKey(subscription.name)
+            && streamsByName[subscription.name] is! Subscription);
+          assert(!subscriptions.containsKey(subscription.streamId));
+          streams[subscription.streamId] = subscription;
+          streamsByName[subscription.name] = subscription;
+          subscriptions[subscription.streamId] = subscription;
+        }
+
+      case SubscriptionRemoveEvent():
+        for (final streamId in event.streamIds) {
+          subscriptions.remove(streamId);
+        }
+
+      case SubscriptionUpdateEvent():
+        final subscription = subscriptions[event.streamId];
+        if (subscription == null) return; // TODO(log)
+        assert(identical(streams[event.streamId], subscription));
+        assert(identical(streamsByName[subscription.name], subscription));
+        switch (event.property) {
+          case SubscriptionProperty.color:
+            subscription.color                  = event.value as int;
+          case SubscriptionProperty.isMuted:
+            // TODO(#421) update [MessageListView] if affected
+            subscription.isMuted                = event.value as bool;
+          case SubscriptionProperty.inHomeView:
+            subscription.isMuted                = !(event.value as bool);
+          case SubscriptionProperty.pinToTop:
+            subscription.pinToTop               = event.value as bool;
+          case SubscriptionProperty.desktopNotifications:
+            subscription.desktopNotifications   = event.value as bool;
+          case SubscriptionProperty.audibleNotifications:
+            subscription.audibleNotifications   = event.value as bool;
+          case SubscriptionProperty.pushNotifications:
+            subscription.pushNotifications      = event.value as bool;
+          case SubscriptionProperty.emailNotifications:
+            subscription.emailNotifications     = event.value as bool;
+          case SubscriptionProperty.wildcardMentionsNotify:
+            subscription.wildcardMentionsNotify = event.value as bool;
+          case SubscriptionProperty.unknown:
+            // unrecognized property; do nothing
+            return;
+        }
+
+      case SubscriptionPeerAddEvent():
+      case SubscriptionPeerRemoveEvent():
+        // We don't currently store the data these would update; that's #374.
+    }
+  }
+
+  void handleUserTopicEvent(UserTopicEvent event) {
+    UserTopicVisibilityPolicy visibilityPolicy = event.visibilityPolicy;
+    if (_warnInvalidVisibilityPolicy(visibilityPolicy)) {
+      visibilityPolicy = UserTopicVisibilityPolicy.none;
+    }
+    // TODO(#421) update [MessageListView] if affected
+    if (visibilityPolicy == UserTopicVisibilityPolicy.none) {
+      // This is the "zero value" for this type, which our data structure
+      // represents by leaving the topic out entirely.
+      final forStream = topicVisibility[event.streamId];
+      if (forStream == null) return;
+      forStream.remove(event.topicName);
+      if (forStream.isEmpty) {
+        topicVisibility.remove(event.streamId);
+      }
+    } else {
+      final forStream = topicVisibility.putIfAbsent(event.streamId, () => {});
+      forStream[event.topicName] = visibilityPolicy;
+    }
+  }
+}
diff --git a/lib/model/unreads.dart b/lib/model/unreads.dart
index 06d13fe3ff..d8485c3e0b 100644
--- a/lib/model/unreads.dart
+++ b/lib/model/unreads.dart
@@ -9,6 +9,7 @@ import '../api/model/events.dart';
 import '../log.dart';
 import 'algorithms.dart';
 import 'narrow.dart';
+import 'stream.dart';
 
 /// The view-model for unread messages.
 ///
@@ -39,7 +40,11 @@ import 'narrow.dart';
 // TODO When loading a message list with stream messages, check all the stream
 //   messages and refresh [mentions] (see [mentions] dartdoc).
 class Unreads extends ChangeNotifier {
-  factory Unreads({required UnreadMessagesSnapshot initial, required selfUserId}) {
+  factory Unreads({
+    required UnreadMessagesSnapshot initial,
+    required selfUserId,
+    required StreamStore streamStore,
+  }) {
     final streams = <int, Map<String, QueueList<int>>>{};
     final dms = <DmNarrow, QueueList<int>>{};
     final mentions = Set.of(initial.mentions);
@@ -62,6 +67,7 @@ class Unreads extends ChangeNotifier {
     }
 
     return Unreads._(
+      streamStore: streamStore,
       streams: streams,
       dms: dms,
       mentions: mentions,
@@ -71,6 +77,7 @@ class Unreads extends ChangeNotifier {
   }
 
   Unreads._({
+    required this.streamStore,
     required this.streams,
     required this.dms,
     required this.mentions,
@@ -78,6 +85,8 @@ class Unreads extends ChangeNotifier {
     required this.selfUserId,
   });
 
+  final StreamStore streamStore;
+
   // TODO excluded for now; would need to handle nuances around muting etc.
   // int count;
 
@@ -123,34 +132,64 @@ class Unreads extends ChangeNotifier {
 
   final int selfUserId;
 
-  // TODO(#346): account for muted topics and streams
   // TODO(#370): maintain this count incrementally, rather than recomputing from scratch
   int countInAllMessagesNarrow() {
     int c = 0;
     for (final messageIds in dms.values) {
       c = c + messageIds.length;
     }
-    for (final topics in streams.values) {
-      for (final messageIds in topics.values) {
-        c = c + messageIds.length;
+    for (final MapEntry(key: streamId, value: topics) in streams.entries) {
+      for (final MapEntry(key: topic, value: messageIds) in topics.entries) {
+        if (streamStore.isTopicVisible(streamId, topic)) {
+          c = c + messageIds.length;
+        }
       }
     }
     return c;
   }
 
-  // TODO(#346): account for muted topics and streams
+  /// The "strict" unread count for this stream,
+  /// using [StreamStore.isTopicVisible].
+  ///
+  /// If the stream is muted, this will count only topics that are
+  /// actively unmuted.
+  ///
+  /// For a count that's appropriate in UI contexts that are focused
+  /// specifically on this stream, see [countInStreamNarrow].
+  // TODO(#370): maintain this count incrementally, rather than recomputing from scratch
+  int countInStream(int streamId) {
+    final topics = streams[streamId];
+    if (topics == null) return 0;
+    int c = 0;
+    for (final entry in topics.entries) {
+      if (streamStore.isTopicVisible(streamId, entry.key)) {
+        c = c + entry.value.length;
+      }
+    }
+    return c;
+  }
+
+  /// The "broad" unread count for this stream,
+  /// using [StreamStore.isTopicVisibleInStream].
+  ///
+  /// This includes topics that have no visibility policy of their own,
+  /// even if the stream itself is muted.
+  ///
+  /// For a count that's appropriate in UI contexts that are not already
+  /// focused on this stream, see [countInStream].
   // TODO(#370): maintain this count incrementally, rather than recomputing from scratch
   int countInStreamNarrow(int streamId) {
     final topics = streams[streamId];
     if (topics == null) return 0;
     int c = 0;
-    for (final messageIds in topics.values) {
-      c = c + messageIds.length;
+    for (final entry in topics.entries) {
+      if (streamStore.isTopicVisibleInStream(streamId, entry.key)) {
+        c = c + entry.value.length;
+      }
     }
     return c;
   }
 
-  // TODO(#346): account for muted topics and streams
   int countInTopicNarrow(int streamId, String topic) {
     final topics = streams[streamId];
     return topics?[topic]?.length ?? 0;
diff --git a/lib/widgets/emoji_reaction.dart b/lib/widgets/emoji_reaction.dart
index ee1e22d2ad..ecd332102a 100644
--- a/lib/widgets/emoji_reaction.dart
+++ b/lib/widgets/emoji_reaction.dart
@@ -1,7 +1,6 @@
 import 'package:flutter/foundation.dart';
 import 'package:flutter/material.dart';
 
-import '../api/model/initial_snapshot.dart';
 import '../api/model/model.dart';
 import '../api/route/messages.dart';
 import '../model/content.dart';
diff --git a/lib/widgets/inbox.dart b/lib/widgets/inbox.dart
index 5d8712cb3f..e8e343aa91 100644
--- a/lib/widgets/inbox.dart
+++ b/lib/widgets/inbox.dart
@@ -135,6 +135,7 @@ class _InboxPageState extends State<InboxPage> with PerAccountStoreAwareStateMix
       final topicItems = <(String, int, int)>[];
       int countInStream = 0;
       for (final MapEntry(key: topic, value: messageIds) in topics.entries) {
+        if (!store.isTopicVisible(streamId, topic)) continue;
         final countInTopic = messageIds.length;
         topicItems.add((topic, countInTopic, messageIds.last));
         countInStream += countInTopic;
@@ -150,9 +151,6 @@ class _InboxPageState extends State<InboxPage> with PerAccountStoreAwareStateMix
       sections.add(_StreamSectionData(streamId, countInStream, topicItems));
     }
 
-    // TODO(#346) Filter out muted messages.
-    //   (Eventually let the user toggle that filtering?)
-
     return Scaffold(
       appBar: AppBar(title: const Text('Inbox')),
       body: SafeArea(
diff --git a/lib/widgets/message_list.dart b/lib/widgets/message_list.dart
index 4e14f87cef..504fe0081e 100644
--- a/lib/widgets/message_list.dart
+++ b/lib/widgets/message_list.dart
@@ -443,7 +443,7 @@ class MarkAsReadWidget extends StatelessWidget {
             ),
             onPressed: () => _handlePress(context),
             icon: const Icon(Icons.playlist_add_check),
-            label: Text(zulipLocalizations.markAsReadLabel(unreadCount))))));
+            label: Text(zulipLocalizations.markAllAsReadLabel)))));
   }
 }
 
diff --git a/lib/widgets/subscription_list.dart b/lib/widgets/subscription_list.dart
index e055c31f12..935da57280 100644
--- a/lib/widgets/subscription_list.dart
+++ b/lib/widgets/subscription_list.dart
@@ -176,7 +176,8 @@ class _SubscriptionList extends StatelessWidget {
       itemCount: subscriptions.length,
       itemBuilder: (BuildContext context, int index) {
         final subscription = subscriptions[index];
-        final unreadCount = unreadsModel!.countInStreamNarrow(subscription.streamId);
+        final unreadCount = unreadsModel!.countInStream(subscription.streamId);
+        // TODO(#346): if stream muted, show a dot for unreads
         return SubscriptionItem(subscription: subscription, unreadCount: unreadCount);
     });
   }
diff --git a/test/api/model/model_checks.dart b/test/api/model/model_checks.dart
index eb95e36ef7..ef8d583484 100644
--- a/test/api/model/model_checks.dart
+++ b/test/api/model/model_checks.dart
@@ -16,6 +16,7 @@ extension StreamColorSwatchChecks on Subject<StreamColorSwatch> {
 }
 
 extension MessageChecks on Subject<Message> {
+  Subject<int> get id => has((e) => e.id, 'id');
   Subject<String> get content => has((e) => e.content, 'content');
   Subject<bool> get isMeMessage => has((e) => e.isMeMessage, 'isMeMessage');
   Subject<int?> get lastEditTimestamp => has((e) => e.lastEditTimestamp, 'lastEditTimestamp');
diff --git a/test/example_data.dart b/test/example_data.dart
index 60fde919c7..5c66d2e871 100644
--- a/test/example_data.dart
+++ b/test/example_data.dart
@@ -431,6 +431,7 @@ InitialSnapshot initialSnapshot({
   UnreadMessagesSnapshot? unreadMsgs,
   List<ZulipStream>? streams,
   UserSettings? userSettings,
+  List<UserTopicItem>? userTopics,
   Map<String, RealmDefaultExternalAccount>? realmDefaultExternalAccounts,
   int? maxFileUploadSizeMib,
   List<User>? realmUsers,
@@ -455,6 +456,7 @@ InitialSnapshot initialSnapshot({
       displayEmojiReactionUsers: true,
       emojiset: Emojiset.google,
     ),
+    userTopics: userTopics,
     realmDefaultExternalAccounts: realmDefaultExternalAccounts ?? {},
     maxFileUploadSizeMib: maxFileUploadSizeMib ?? 25,
     realmUsers: realmUsers ?? [],
@@ -471,11 +473,11 @@ PerAccountStore store({Account? account, InitialSnapshot? initialSnapshot}) {
     initialSnapshot: initialSnapshot ?? _initialSnapshot(),
   );
 }
+const _store = store;
 
-LivePerAccountStore liveStore({Account? account, InitialSnapshot? initialSnapshot}) {
-  return LivePerAccountStore.fromInitialSnapshot(
-    account: account ?? selfAccount,
-    connection: FakeApiConnection.fromAccount(account ?? selfAccount),
-    initialSnapshot: initialSnapshot ?? _initialSnapshot(),
-  );
+UpdateMachine updateMachine({Account? account, InitialSnapshot? initialSnapshot}) {
+  initialSnapshot ??= _initialSnapshot();
+  final store = _store(account: account, initialSnapshot: initialSnapshot);
+  return UpdateMachine.fromInitialSnapshot(
+    store: store, initialSnapshot: initialSnapshot);
 }
diff --git a/test/model/message_list_test.dart b/test/model/message_list_test.dart
index 68706f0be1..4da42e3ca1 100644
--- a/test/model/message_list_test.dart
+++ b/test/model/message_list_test.dart
@@ -17,10 +17,12 @@ import '../api/model/model_checks.dart';
 import '../example_data.dart' as eg;
 import '../stdlib_checks.dart';
 import 'content_checks.dart';
+import 'test_store.dart';
 
 void main() async {
   // These variables are the common state operated on by each test.
   // Each test case calls [prepare] to initialize them.
+  late Subscription subscription;
   late PerAccountStore store;
   late FakeApiConnection connection;
   late MessageListView model;
@@ -35,7 +37,10 @@ void main() async {
 
   /// Initialize [model] and the rest of the test state.
   void prepare({Narrow narrow = const AllMessagesNarrow()}) {
-    store = eg.store();
+    final stream = eg.stream();
+    subscription = eg.subscription(stream);
+    store = eg.store()
+      ..addStream(stream)..addSubscription(subscription);
     connection = store.connection as FakeApiConnection;
     notifiedCount = 0;
     model = MessageListView.init(store: store, narrow: narrow)
@@ -548,6 +553,141 @@ void main() async {
     check(model.contents[0]).equalsNode(correctContent);
   });
 
+  group('stream/topic muting', () {
+    test('in AllMessagesNarrow', () async {
+      final stream1 = eg.stream(streamId: 1, name: 'stream 1');
+      final stream2 = eg.stream(streamId: 2, name: 'stream 2');
+      prepare(narrow: const AllMessagesNarrow());
+      store.addStreams([stream1, stream2]);
+      store.addSubscription(eg.subscription(stream1));
+      store.addUserTopic(stream1, 'B', UserTopicVisibilityPolicy.muted);
+      store.addSubscription(eg.subscription(stream2, isMuted: true));
+      store.addUserTopic(stream2, 'C', UserTopicVisibilityPolicy.unmuted);
+
+      // Check filtering on fetchInitial…
+      await prepareMessages(foundOldest: false, messages: [
+        eg.streamMessage(id: 201, stream: stream1, topic: 'A'),
+        eg.streamMessage(id: 202, stream: stream1, topic: 'B'),
+        eg.streamMessage(id: 203, stream: stream2, topic: 'C'),
+        eg.streamMessage(id: 204, stream: stream2, topic: 'D'),
+        eg.dmMessage(    id: 205, from: eg.otherUser, to: [eg.selfUser]),
+      ]);
+      final expected = <int>[];
+      check(model.messages.map((m) => m.id))
+        .deepEquals(expected..addAll([201, 203, 205]));
+
+      // … and on fetchOlder…
+      connection.prepare(json: olderResult(
+        anchor: 201, foundOldest: true, messages: [
+          eg.streamMessage(id: 101, stream: stream1, topic: 'A'),
+          eg.streamMessage(id: 102, stream: stream1, topic: 'B'),
+          eg.streamMessage(id: 103, stream: stream2, topic: 'C'),
+          eg.streamMessage(id: 104, stream: stream2, topic: 'D'),
+          eg.dmMessage(    id: 105, from: eg.otherUser, to: [eg.selfUser]),
+        ]).toJson());
+      await model.fetchOlder();
+      checkNotified(count: 2);
+      check(model.messages.map((m) => m.id))
+        .deepEquals(expected..insertAll(0, [101, 103, 105]));
+
+      // … and on maybeAddMessage.
+      model.maybeAddMessage(eg.streamMessage(id: 301, stream: stream1, topic: 'A'));
+      checkNotifiedOnce();
+      check(model.messages.map((m) => m.id)).deepEquals(expected..add(301));
+
+      model.maybeAddMessage(eg.streamMessage(id: 302, stream: stream1, topic: 'B'));
+      checkNotNotified();
+      check(model.messages.map((m) => m.id)).deepEquals(expected);
+
+      model.maybeAddMessage(eg.streamMessage(id: 303, stream: stream2, topic: 'C'));
+      checkNotifiedOnce();
+      check(model.messages.map((m) => m.id)).deepEquals(expected..add(303));
+
+      model.maybeAddMessage(eg.streamMessage(id: 304, stream: stream2, topic: 'D'));
+      checkNotNotified();
+      check(model.messages.map((m) => m.id)).deepEquals(expected);
+
+      model.maybeAddMessage(eg.dmMessage(id: 305, from: eg.otherUser, to: [eg.selfUser]));
+      checkNotifiedOnce();
+      check(model.messages.map((m) => m.id)).deepEquals(expected..add(305));
+    });
+
+    test('in StreamNarrow', () async {
+      final stream = eg.stream(streamId: 1, name: 'stream 1');
+      prepare(narrow: StreamNarrow(stream.streamId));
+      store.addStream(stream);
+      store.addSubscription(eg.subscription(stream, isMuted: true));
+      store.addUserTopic(stream, 'A', UserTopicVisibilityPolicy.unmuted);
+      store.addUserTopic(stream, 'C', UserTopicVisibilityPolicy.muted);
+
+      // Check filtering on fetchInitial…
+      await prepareMessages(foundOldest: false, messages: [
+        eg.streamMessage(id: 201, stream: stream, topic: 'A'),
+        eg.streamMessage(id: 202, stream: stream, topic: 'B'),
+        eg.streamMessage(id: 203, stream: stream, topic: 'C'),
+      ]);
+      final expected = <int>[];
+      check(model.messages.map((m) => m.id))
+        .deepEquals(expected..addAll([201, 202]));
+
+      // … and on fetchOlder…
+      connection.prepare(json: olderResult(
+        anchor: 201, foundOldest: true, messages: [
+          eg.streamMessage(id: 101, stream: stream, topic: 'A'),
+          eg.streamMessage(id: 102, stream: stream, topic: 'B'),
+          eg.streamMessage(id: 103, stream: stream, topic: 'C'),
+        ]).toJson());
+      await model.fetchOlder();
+      checkNotified(count: 2);
+      check(model.messages.map((m) => m.id))
+        .deepEquals(expected..insertAll(0, [101, 102]));
+
+      // … and on maybeAddMessage.
+      model.maybeAddMessage(eg.streamMessage(id: 301, stream: stream, topic: 'A'));
+      checkNotifiedOnce();
+      check(model.messages.map((m) => m.id)).deepEquals(expected..add(301));
+
+      model.maybeAddMessage(eg.streamMessage(id: 302, stream: stream, topic: 'B'));
+      checkNotifiedOnce();
+      check(model.messages.map((m) => m.id)).deepEquals(expected..add(302));
+
+      model.maybeAddMessage(eg.streamMessage(id: 303, stream: stream, topic: 'C'));
+      checkNotNotified();
+      check(model.messages.map((m) => m.id)).deepEquals(expected);
+    });
+
+    test('in TopicNarrow', () async {
+      final stream = eg.stream(streamId: 1, name: 'stream 1');
+      prepare(narrow: TopicNarrow(stream.streamId, 'A'));
+      store.addStream(stream);
+      store.addSubscription(eg.subscription(stream, isMuted: true));
+      store.addUserTopic(stream, 'A', UserTopicVisibilityPolicy.muted);
+
+      // Check filtering on fetchInitial…
+      await prepareMessages(foundOldest: false, messages: [
+        eg.streamMessage(id: 201, stream: stream, topic: 'A'),
+      ]);
+      final expected = <int>[];
+      check(model.messages.map((m) => m.id))
+        .deepEquals(expected..addAll([201]));
+
+      // … and on fetchOlder…
+      connection.prepare(json: olderResult(
+        anchor: 201, foundOldest: true, messages: [
+          eg.streamMessage(id: 101, stream: stream, topic: 'A'),
+        ]).toJson());
+      await model.fetchOlder();
+      checkNotified(count: 2);
+      check(model.messages.map((m) => m.id))
+        .deepEquals(expected..insertAll(0, [101]));
+
+      // … and on maybeAddMessage.
+      model.maybeAddMessage(eg.streamMessage(id: 301, stream: stream, topic: 'A'));
+      checkNotifiedOnce();
+      check(model.messages.map((m) => m.id)).deepEquals(expected..add(301));
+    });
+  });
+
   test('recipient headers are maintained consistently', () async {
     // This tests the code that maintains the invariant that recipient headers
     // are present just where [canShareRecipientHeader] requires them.
@@ -772,6 +912,22 @@ void checkInvariants(MessageListView model) {
     check(model).fetchingOlder.isFalse();
   }
 
+  for (final message in model.messages) {
+    check(model.narrow.containsMessage(message)).isTrue();
+
+    if (message is! StreamMessage) continue;
+    switch (model.narrow) {
+      case AllMessagesNarrow():
+        check(model.store.isTopicVisible(message.streamId, message.subject))
+          .isTrue();
+      case StreamNarrow():
+        check(model.store.isTopicVisibleInStream(message.streamId, message.subject))
+          .isTrue();
+      case TopicNarrow():
+      case DmNarrow():
+    }
+  }
+
   for (int i = 0; i < model.messages.length - 1; i++) {
     check(model.messages[i].id).isLessThan(model.messages[i+1].id);
   }
diff --git a/test/model/store_test.dart b/test/model/store_test.dart
index 1032eb7ebf..a13bdc9b2b 100644
--- a/test/model/store_test.dart
+++ b/test/model/store_test.dart
@@ -4,7 +4,6 @@ import 'package:checks/checks.dart';
 import 'package:flutter/foundation.dart';
 import 'package:http/http.dart' as http;
 import 'package:test/scaffolding.dart';
-import 'package:zulip/api/model/events.dart';
 import 'package:zulip/model/store.dart';
 import 'package:zulip/notifications.dart';
 
@@ -109,13 +108,13 @@ void main() {
     check(completers(1)).length.equals(1);
   });
 
-  group('PerAccountStore.registerNotificationToken', () {
-    late LivePerAccountStore store;
+  group('UpdateMachine.registerNotificationToken', () {
+    late UpdateMachine updateMachine;
     late FakeApiConnection connection;
 
     void prepareStore() {
-      store = eg.liveStore();
-      connection = store.connection as FakeApiConnection;
+      updateMachine = eg.updateMachine();
+      connection = updateMachine.store.connection as FakeApiConnection;
     }
 
     void checkLastRequestApns({required String token, required String appid}) {
@@ -144,7 +143,7 @@ void main() {
       // On store startup, send the token.
       prepareStore();
       connection.prepare(json: {});
-      await store.registerNotificationToken();
+      await updateMachine.registerNotificationToken();
       if (defaultTargetPlatform == TargetPlatform.android) {
         checkLastRequestFcm(token: '012abc');
       } else {
@@ -178,7 +177,7 @@ void main() {
 
       // On store startup, send nothing (because we have nothing to send).
       prepareStore();
-      await store.registerNotificationToken();
+      await updateMachine.registerNotificationToken();
       check(connection.lastRequest).isNull();
 
       // When the token later appears, send it.
@@ -199,52 +198,6 @@ void main() {
       }
     });
   });
-
-  group('handleEvent for SubscriptionEvent', () {
-    final stream = eg.stream();
-
-    test('SubscriptionProperty.color updates with an int value', () {
-      final store = eg.store(initialSnapshot: eg.initialSnapshot(
-        streams: [stream],
-        subscriptions: [eg.subscription(stream, color: 0xFFFF0000)],
-      ));
-      check(store.subscriptions[stream.streamId]!.color).equals(0xFFFF0000);
-
-      store.handleEvent(SubscriptionUpdateEvent(id: 1,
-        streamId: stream.streamId,
-        property: SubscriptionProperty.color,
-        value: 0xFFFF00FF));
-      check(store.subscriptions[stream.streamId]!.color).equals(0xFFFF00FF);
-    });
-
-    test('SubscriptionProperty.isMuted updates with a boolean value', () {
-      final store = eg.store(initialSnapshot: eg.initialSnapshot(
-        streams: [stream],
-        subscriptions: [eg.subscription(stream, isMuted: false)],
-      ));
-      check(store.subscriptions[stream.streamId]!.isMuted).isFalse();
-
-      store.handleEvent(SubscriptionUpdateEvent(id: 1,
-        streamId: stream.streamId,
-        property: SubscriptionProperty.isMuted,
-        value: true));
-      check(store.subscriptions[stream.streamId]!.isMuted).isTrue();
-    });
-
-    test('SubscriptionProperty.inHomeView updates isMuted instead', () {
-      final store = eg.store(initialSnapshot: eg.initialSnapshot(
-        streams: [stream],
-        subscriptions: [eg.subscription(stream, isMuted: false)],
-      ));
-      check(store.subscriptions[stream.streamId]!.isMuted).isFalse();
-
-      store.handleEvent(SubscriptionUpdateEvent(id: 1,
-        streamId: stream.streamId,
-        property: SubscriptionProperty.inHomeView,
-        value: false));
-      check(store.subscriptions[stream.streamId]!.isMuted).isTrue();
-    });
-  });
 }
 
 class LoadingTestGlobalStore extends TestGlobalStore {
diff --git a/test/model/stream_test.dart b/test/model/stream_test.dart
new file mode 100644
index 0000000000..3e6b9f16be
--- /dev/null
+++ b/test/model/stream_test.dart
@@ -0,0 +1,293 @@
+
+import 'package:checks/checks.dart';
+import 'package:test/scaffolding.dart';
+import 'package:zulip/api/model/events.dart';
+import 'package:zulip/api/model/initial_snapshot.dart';
+import 'package:zulip/api/model/model.dart';
+import 'package:zulip/model/store.dart';
+import 'package:zulip/model/stream.dart';
+
+import '../example_data.dart' as eg;
+import 'test_store.dart';
+
+void main() {
+  group('Unified stream/sub data', () {
+    /// Check that `streams`, `streamsByName`, and `subscriptions` all agree
+    /// and point to the same objects where applicable.
+    void checkUnified(StreamStore store) {
+      check(store.streamsByName).length.equals(store.streams.length);
+      for (final MapEntry(key: streamId, value: stream)
+           in store.streams.entries) {
+        check(streamId).equals(stream.streamId);
+        check(store.streamsByName[stream.name]).identicalTo(stream);
+        if (stream is Subscription) {
+          check(store.subscriptions[streamId]).identicalTo(stream);
+        } else {
+          check(store.subscriptions[streamId]).isNull();
+        }
+      }
+      for (final MapEntry(key: streamId, value: subscription)
+           in store.subscriptions.entries) {
+        check(streamId).equals(subscription.streamId);
+        check(store.streams[streamId]).identicalTo(subscription);
+      }
+    }
+
+    test('initial', () {
+      final stream1 = eg.stream(streamId: 1, name: 'stream 1');
+      final stream2 = eg.stream(streamId: 2, name: 'stream 2');
+      checkUnified(eg.store(initialSnapshot: eg.initialSnapshot(
+        streams: [stream1, stream2],
+        subscriptions: [eg.subscription(stream1)],
+      )));
+    });
+
+    test('added by events', () {
+      final stream1 = eg.stream(streamId: 1, name: 'stream 1');
+      final stream2 = eg.stream(streamId: 2, name: 'stream 2');
+      final store = eg.store();
+      checkUnified(store);
+      checkUnified(store..addStream(stream1));
+      checkUnified(store..addStream(stream2));
+      checkUnified(store..addSubscription(eg.subscription(stream1)));
+    });
+  });
+
+  group('SubscriptionEvent', () {
+    final stream = eg.stream();
+
+    test('SubscriptionProperty.color updates with an int value', () {
+      final store = eg.store(initialSnapshot: eg.initialSnapshot(
+        streams: [stream],
+        subscriptions: [eg.subscription(stream, color: 0xFFFF0000)],
+      ));
+      check(store.subscriptions[stream.streamId]!.color).equals(0xFFFF0000);
+
+      store.handleEvent(SubscriptionUpdateEvent(id: 1,
+        streamId: stream.streamId,
+        property: SubscriptionProperty.color,
+        value: 0xFFFF00FF));
+      check(store.subscriptions[stream.streamId]!.color).equals(0xFFFF00FF);
+    });
+
+    test('SubscriptionProperty.isMuted updates with a boolean value', () {
+      final store = eg.store(initialSnapshot: eg.initialSnapshot(
+        streams: [stream],
+        subscriptions: [eg.subscription(stream, isMuted: false)],
+      ));
+      check(store.subscriptions[stream.streamId]!.isMuted).isFalse();
+
+      store.handleEvent(SubscriptionUpdateEvent(id: 1,
+        streamId: stream.streamId,
+        property: SubscriptionProperty.isMuted,
+        value: true));
+      check(store.subscriptions[stream.streamId]!.isMuted).isTrue();
+    });
+
+    test('SubscriptionProperty.inHomeView updates isMuted instead', () {
+      final store = eg.store(initialSnapshot: eg.initialSnapshot(
+        streams: [stream],
+        subscriptions: [eg.subscription(stream, isMuted: false)],
+      ));
+      check(store.subscriptions[stream.streamId]!.isMuted).isFalse();
+
+      store.handleEvent(SubscriptionUpdateEvent(id: 1,
+        streamId: stream.streamId,
+        property: SubscriptionProperty.inHomeView,
+        value: false));
+      check(store.subscriptions[stream.streamId]!.isMuted).isTrue();
+    });
+  });
+
+  group('topic visibility', () {
+    final stream1 = eg.stream(streamId: 1, name: 'stream 1');
+    final stream2 = eg.stream(streamId: 2, name: 'stream 2');
+
+    group('getter topicVisibilityPolicy', () {
+      test('with nothing for stream', () {
+        final store = eg.store();
+        check(store.topicVisibilityPolicy(stream1.streamId, 'topic'))
+          .equals(UserTopicVisibilityPolicy.none);
+      });
+
+      test('with nothing for topic', () {
+        final store = eg.store()
+          ..addUserTopic(stream1, 'other topic', UserTopicVisibilityPolicy.muted);
+        check(store.topicVisibilityPolicy(stream1.streamId, 'topic'))
+          .equals(UserTopicVisibilityPolicy.none);
+      });
+
+      test('with topic present', () {
+        final store = eg.store();
+        for (final policy in [
+          UserTopicVisibilityPolicy.muted,
+          UserTopicVisibilityPolicy.unmuted,
+          UserTopicVisibilityPolicy.followed,
+        ]) {
+          store.addUserTopic(stream1, 'topic', policy);
+          check(store.topicVisibilityPolicy(stream1.streamId, 'topic'))
+            .equals(policy);
+        }
+      });
+    });
+
+    group('isTopicVisible/InStream', () {
+      test('with policy none, stream not muted', () {
+        final store = eg.store()..addStream(stream1)
+          ..addSubscription(eg.subscription(stream1));
+        check(store.isTopicVisibleInStream(stream1.streamId, 'topic')).isTrue();
+        check(store.isTopicVisible        (stream1.streamId, 'topic')).isTrue();
+      });
+
+      test('with policy none, stream muted', () {
+        final store = eg.store()..addStream(stream1)
+          ..addSubscription(eg.subscription(stream1, isMuted: true));
+        check(store.isTopicVisibleInStream(stream1.streamId, 'topic')).isTrue();
+        check(store.isTopicVisible        (stream1.streamId, 'topic')).isFalse();
+      });
+
+      test('with policy none, stream unsubscribed', () {
+        final store = eg.store()..addStream(stream1);
+        check(store.isTopicVisibleInStream(stream1.streamId, 'topic')).isTrue();
+        check(store.isTopicVisible        (stream1.streamId, 'topic')).isFalse();
+      });
+
+      test('with policy muted', () {
+        final store = eg.store()..addStream(stream1)
+          ..addSubscription(eg.subscription(stream1))
+          ..addUserTopic(stream1, 'topic', UserTopicVisibilityPolicy.muted);
+        check(store.isTopicVisibleInStream(stream1.streamId, 'topic')).isFalse();
+        check(store.isTopicVisible        (stream1.streamId, 'topic')).isFalse();
+      });
+
+      test('with policy unmuted', () {
+        final store = eg.store()..addStream(stream1)
+          ..addSubscription(eg.subscription(stream1, isMuted: true))
+          ..addUserTopic(stream1, 'topic', UserTopicVisibilityPolicy.unmuted);
+        check(store.isTopicVisibleInStream(stream1.streamId, 'topic')).isTrue();
+        check(store.isTopicVisible        (stream1.streamId, 'topic')).isTrue();
+      });
+
+      test('with policy followed', () {
+        final store = eg.store()..addStream(stream1)
+          ..addSubscription(eg.subscription(stream1, isMuted: true))
+          ..addUserTopic(stream1, 'topic', UserTopicVisibilityPolicy.followed);
+        check(store.isTopicVisibleInStream(stream1.streamId, 'topic')).isTrue();
+        check(store.isTopicVisible        (stream1.streamId, 'topic')).isTrue();
+      });
+    });
+
+    UserTopicItem makeUserTopicItem(
+        ZulipStream stream, String topic, UserTopicVisibilityPolicy policy) {
+      return UserTopicItem(
+        streamId: stream.streamId,
+        topicName: topic,
+        lastUpdated: 1234567890,
+        visibilityPolicy: policy,
+      );
+    }
+
+    void compareTopicVisibility(PerAccountStore store, List<UserTopicItem> expected) {
+      final expectedStore = eg.store(initialSnapshot: eg.initialSnapshot(
+        userTopics: expected,
+      ));
+      check(store.debugStreamStore.topicVisibility)
+        .deepEquals(expectedStore.debugStreamStore.topicVisibility);
+    }
+
+    test('data structure', () {
+      final store = eg.store(initialSnapshot: eg.initialSnapshot(
+        streams: [stream1, stream2],
+        userTopics: [
+          makeUserTopicItem(stream1, 'topic 1', UserTopicVisibilityPolicy.muted),
+          makeUserTopicItem(stream1, 'topic 2', UserTopicVisibilityPolicy.unmuted),
+          makeUserTopicItem(stream2, 'topic 3', UserTopicVisibilityPolicy.unknown),
+          makeUserTopicItem(stream2, 'topic 4', UserTopicVisibilityPolicy.followed),
+        ]));
+      check(store.debugStreamStore.topicVisibility).deepEquals({
+        stream1.streamId: {
+          'topic 1': UserTopicVisibilityPolicy.muted,
+          'topic 2': UserTopicVisibilityPolicy.unmuted,
+        },
+        stream2.streamId: {
+          // 'topic 3' -> unknown treated as no policy
+          'topic 4': UserTopicVisibilityPolicy.followed,
+        }
+      });
+    });
+
+    group('events', () {
+      test('add with new stream', () {
+        final store = eg.store()
+          ..addUserTopic(stream1, 'topic', UserTopicVisibilityPolicy.muted);
+        compareTopicVisibility(store, [
+          makeUserTopicItem(stream1, 'topic', UserTopicVisibilityPolicy.muted),
+        ]);
+      });
+
+      test('add in existing stream', () {
+        final store = eg.store()
+          ..addUserTopic(stream1, 'topic', UserTopicVisibilityPolicy.muted)
+          ..addUserTopic(stream1, 'other topic', UserTopicVisibilityPolicy.unmuted);
+        compareTopicVisibility(store, [
+          makeUserTopicItem(stream1, 'topic', UserTopicVisibilityPolicy.muted),
+          makeUserTopicItem(stream1, 'other topic', UserTopicVisibilityPolicy.unmuted),
+        ]);
+      });
+
+      test('update existing policy', () {
+        final store = eg.store()
+          ..addUserTopic(stream1, 'topic', UserTopicVisibilityPolicy.muted)
+          ..addUserTopic(stream1, 'topic', UserTopicVisibilityPolicy.unmuted);
+        compareTopicVisibility(store, [
+          makeUserTopicItem(stream1, 'topic', UserTopicVisibilityPolicy.unmuted),
+        ]);
+      });
+
+      test('remove, with others in stream', () {
+        final store = eg.store()
+          ..addUserTopic(stream1, 'topic', UserTopicVisibilityPolicy.muted)
+          ..addUserTopic(stream1, 'other topic', UserTopicVisibilityPolicy.unmuted)
+          ..addUserTopic(stream1, 'topic', UserTopicVisibilityPolicy.none);
+        compareTopicVisibility(store, [
+          makeUserTopicItem(stream1, 'other topic', UserTopicVisibilityPolicy.unmuted),
+        ]);
+      });
+
+      test('remove, as last in stream', () {
+        final store = eg.store()
+          ..addUserTopic(stream1, 'topic', UserTopicVisibilityPolicy.muted)
+          ..addUserTopic(stream1, 'topic', UserTopicVisibilityPolicy.none);
+        compareTopicVisibility(store, [
+        ]);
+      });
+
+      test('treat unknown enum value as removing', () {
+        final store = eg.store()
+          ..addUserTopic(stream1, 'topic', UserTopicVisibilityPolicy.muted)
+          ..addUserTopic(stream1, 'topic', UserTopicVisibilityPolicy.unknown);
+        compareTopicVisibility(store, [
+        ]);
+      });
+    });
+
+    test('smoke', () {
+      final stream = eg.stream();
+      final store = eg.store(initialSnapshot: eg.initialSnapshot(
+        streams: [stream],
+        userTopics: [
+          makeUserTopicItem(stream, 'topic 1', UserTopicVisibilityPolicy.muted),
+          makeUserTopicItem(stream, 'topic 2', UserTopicVisibilityPolicy.unmuted),
+          makeUserTopicItem(stream, 'topic 3', UserTopicVisibilityPolicy.followed),
+        ]));
+      check(store.topicVisibilityPolicy(stream.streamId, 'topic 1'))
+        .equals(UserTopicVisibilityPolicy.muted);
+      check(store.topicVisibilityPolicy(stream.streamId, 'topic 2'))
+        .equals(UserTopicVisibilityPolicy.unmuted);
+      check(store.topicVisibilityPolicy(stream.streamId, 'topic 3'))
+        .equals(UserTopicVisibilityPolicy.followed);
+      check(store.topicVisibilityPolicy(stream.streamId, 'topic 4'))
+        .equals(UserTopicVisibilityPolicy.none);
+    });
+  });
+}
diff --git a/test/model/test_store.dart b/test/model/test_store.dart
index 6b1d293b11..a6522b8eb1 100644
--- a/test/model/test_store.dart
+++ b/test/model/test_store.dart
@@ -84,4 +84,14 @@ extension PerAccountStoreTestExtension on PerAccountStore {
   void addSubscriptions(List<Subscription> subscriptions) {
     handleEvent(SubscriptionAddEvent(id: 1, subscriptions: subscriptions));
   }
+
+  void addUserTopic(ZulipStream stream, String topic, UserTopicVisibilityPolicy visibilityPolicy) {
+    handleEvent(UserTopicEvent(
+      id: 1,
+      streamId: stream.streamId,
+      topicName: topic,
+      lastUpdated: 1234567890,
+      visibilityPolicy: visibilityPolicy,
+    ));
+  }
 }
diff --git a/test/model/unreads_test.dart b/test/model/unreads_test.dart
index 61eb8c82ff..2725af0649 100644
--- a/test/model/unreads_test.dart
+++ b/test/model/unreads_test.dart
@@ -5,15 +5,18 @@ import 'package:zulip/api/model/events.dart';
 import 'package:zulip/api/model/initial_snapshot.dart';
 import 'package:zulip/api/model/model.dart';
 import 'package:zulip/model/narrow.dart';
+import 'package:zulip/model/store.dart';
 import 'package:zulip/model/unreads.dart';
 
 import '../example_data.dart' as eg;
+import 'test_store.dart';
 import 'unreads_checks.dart';
 
 void main() {
   // These variables are the common state operated on by each test.
   // Each test case calls [prepare] to initialize them.
   late Unreads model;
+  late PerAccountStore streamStore; // TODO reduce this to StreamStore
   late int notifiedCount;
 
   void checkNotified({required int count}) {
@@ -34,8 +37,10 @@ void main() {
       oldUnreadsMissing: false,
     ),
   }) {
+    streamStore = eg.store();
     notifiedCount = 0;
-    model = Unreads(initial: initial, selfUserId: eg.selfUser.userId)
+    model = Unreads(initial: initial,
+        selfUserId: eg.selfUser.userId, streamStore: streamStore)
       ..addListener(() {
         notifiedCount++;
       });
@@ -148,30 +153,48 @@ void main() {
 
   group('count helpers', () {
     test('countInAllMessagesNarrow', () {
-      final stream1 = eg.stream();
-      final stream2 = eg.stream();
+      final stream1 = eg.stream(streamId: 1, name: 'stream 1');
+      final stream2 = eg.stream(streamId: 2, name: 'stream 2');
+      final stream3 = eg.stream(streamId: 3, name: 'stream 3');
       prepare();
+      streamStore.addStreams([stream1, stream2, stream3]);
+      streamStore.addSubscription(eg.subscription(stream1));
+      streamStore.addSubscription(eg.subscription(stream2));
+      streamStore.addSubscription(eg.subscription(stream3, isMuted: true));
+      streamStore.addUserTopic(stream1, 'a', UserTopicVisibilityPolicy.muted);
       fillWithMessages([
         eg.streamMessage(stream: stream1, topic: 'a', flags: []),
         eg.streamMessage(stream: stream1, topic: 'b', flags: []),
         eg.streamMessage(stream: stream1, topic: 'b', flags: []),
         eg.streamMessage(stream: stream2, topic: 'c', flags: []),
+        eg.streamMessage(stream: stream3, topic: 'd', flags: []),
         eg.dmMessage(from: eg.otherUser, to: [eg.selfUser], flags: []),
         eg.dmMessage(from: eg.thirdUser, to: [eg.selfUser], flags: []),
       ]);
-      check(model.countInAllMessagesNarrow()).equals(6);
+      check(model.countInAllMessagesNarrow()).equals(5);
     });
 
-    test('countInStreamNarrow', () {
+    test('countInStream/Narrow', () {
       final stream = eg.stream();
       prepare();
+      streamStore.addStream(stream);
+      streamStore.addSubscription(eg.subscription(stream));
+      streamStore.addUserTopic(stream, 'a', UserTopicVisibilityPolicy.unmuted);
+      streamStore.addUserTopic(stream, 'c', UserTopicVisibilityPolicy.muted);
       fillWithMessages([
         eg.streamMessage(stream: stream, topic: 'a', flags: []),
         eg.streamMessage(stream: stream, topic: 'a', flags: []),
         eg.streamMessage(stream: stream, topic: 'b', flags: []),
         eg.streamMessage(stream: stream, topic: 'b', flags: []),
         eg.streamMessage(stream: stream, topic: 'b', flags: []),
+        eg.streamMessage(stream: stream, topic: 'c', flags: []),
       ]);
+      check(model.countInStream      (stream.streamId)).equals(5);
+      check(model.countInStreamNarrow(stream.streamId)).equals(5);
+
+      streamStore.handleEvent(SubscriptionUpdateEvent(id: 1, streamId: stream.streamId,
+        property: SubscriptionProperty.isMuted, value: true));
+      check(model.countInStream      (stream.streamId)).equals(2);
       check(model.countInStreamNarrow(stream.streamId)).equals(5);
     });
 
diff --git a/test/widgets/action_sheet_test.dart b/test/widgets/action_sheet_test.dart
index 992dbf87cf..e4b1df53e9 100644
--- a/test/widgets/action_sheet_test.dart
+++ b/test/widgets/action_sheet_test.dart
@@ -37,7 +37,8 @@ Future<void> setupToMessageActionSheet(WidgetTester tester, {
   final store = await testBinding.globalStore.perAccount(eg.selfAccount.id);
   store.addUser(eg.user(userId: message.senderId));
   if (message is StreamMessage) {
-    store.addStream(eg.stream(streamId: message.streamId));
+    final stream = eg.stream(streamId: message.streamId);
+    store..addStream(stream)..addSubscription(eg.subscription(stream));
   }
   final connection = store.connection as FakeApiConnection;
 
diff --git a/test/widgets/emoji_reaction_test.dart b/test/widgets/emoji_reaction_test.dart
index 27c9b92b0c..c5021b953c 100644
--- a/test/widgets/emoji_reaction_test.dart
+++ b/test/widgets/emoji_reaction_test.dart
@@ -7,7 +7,6 @@ import 'package:flutter/services.dart';
 
 import 'package:flutter_test/flutter_test.dart';
 import 'package:zulip/api/model/events.dart';
-import 'package:zulip/api/model/initial_snapshot.dart';
 import 'package:zulip/api/model/model.dart';
 import 'package:zulip/model/store.dart';
 import 'package:zulip/widgets/emoji_reaction.dart';
diff --git a/test/widgets/inbox_test.dart b/test/widgets/inbox_test.dart
index 61678527dd..46e15f9e3e 100644
--- a/test/widgets/inbox_test.dart
+++ b/test/widgets/inbox_test.dart
@@ -53,9 +53,9 @@ void main() {
 
   /// Set up an inbox view with lots of interesting content.
   Future<void> setupVarious(WidgetTester tester) async {
-    final stream1 = eg.stream(streamId: 1);
+    final stream1 = eg.stream(streamId: 1, name: 'stream 1');
     final sub1 = eg.subscription(stream1);
-    final stream2 = eg.stream(streamId: 2);
+    final stream2 = eg.stream(streamId: 2, name: 'stream 2');
     final sub2 = eg.subscription(stream2);
 
     await setupPage(tester,
@@ -140,6 +140,52 @@ void main() {
     // TODO test that tapping a conversation row opens the message list
     //   for the conversation
 
+    group('muting', () { // aka topic visibility
+      testWidgets('baseline', (tester) async {
+        final stream = eg.stream();
+        final subscription = eg.subscription(stream);
+        await setupPage(tester,
+          streams: [stream],
+          subscriptions: [subscription],
+          unreadMessages: [eg.streamMessage(stream: stream, topic: 'lunch')]);
+        check(tester.widgetList(find.text('lunch'))).length.equals(1);
+      });
+
+      testWidgets('muted topic', (tester) async {
+        final stream = eg.stream();
+        final subscription = eg.subscription(stream);
+        await setupPage(tester,
+          streams: [stream],
+          subscriptions: [subscription],
+          unreadMessages: [eg.streamMessage(stream: stream, topic: 'lunch')]);
+        store.addUserTopic(stream, 'lunch', UserTopicVisibilityPolicy.muted);
+        await tester.pump();
+        check(tester.widgetList(find.text('lunch'))).length.equals(0);
+      });
+
+      testWidgets('muted stream', (tester) async {
+        final stream = eg.stream();
+        final subscription = eg.subscription(stream, isMuted: true);
+        await setupPage(tester,
+          streams: [stream],
+          subscriptions: [subscription],
+          unreadMessages: [eg.streamMessage(stream: stream, topic: 'lunch')]);
+        check(tester.widgetList(find.text('lunch'))).length.equals(0);
+      });
+
+      testWidgets('unmuted topic in muted stream', (tester) async {
+        final stream = eg.stream();
+        final subscription = eg.subscription(stream, isMuted: true);
+        await setupPage(tester,
+          streams: [stream],
+          subscriptions: [subscription],
+          unreadMessages: [eg.streamMessage(stream: stream, topic: 'lunch')]);
+        store.addUserTopic(stream, 'lunch', UserTopicVisibilityPolicy.unmuted);
+        await tester.pump();
+        check(tester.widgetList(find.text('lunch'))).length.equals(1);
+      });
+    });
+
     group('collapsing', () {
       Icon findHeaderCollapseIcon(WidgetTester tester, Widget headerRow) {
         return tester.widget(
diff --git a/test/widgets/message_list_test.dart b/test/widgets/message_list_test.dart
index c665ba6f23..819145b4a0 100644
--- a/test/widgets/message_list_test.dart
+++ b/test/widgets/message_list_test.dart
@@ -47,7 +47,7 @@ void main() {
     UnreadMessagesSnapshot? unreadMsgs,
   }) async {
     addTearDown(testBinding.reset);
-    streams ??= subscriptions;
+    streams ??= subscriptions ??= [eg.subscription(eg.stream())];
     await testBinding.globalStore.add(eg.selfAccount, eg.initialSnapshot(
       streams: streams, subscriptions: subscriptions, unreadMsgs: unreadMsgs));
     store = await testBinding.globalStore.perAccount(eg.selfAccount.id);
@@ -214,7 +214,7 @@ void main() {
       testWidgets('show stream name in AllMessagesNarrow', (tester) async {
         await setupMessageListPage(tester,
           narrow: const AllMessagesNarrow(),
-          messages: [message], streams: [stream]);
+          messages: [message], subscriptions: [eg.subscription(stream)]);
         await tester.pump();
         check(findInMessageList('stream name')).length.equals(1);
         check(findInMessageList('topic name')).length.equals(1);
@@ -268,7 +268,7 @@ void main() {
         final stream = eg.stream(isWebPublic: false, inviteOnly: false);
         await setupMessageListPage(tester,
           messages: [eg.streamMessage(stream: stream)],
-          streams: [stream]);
+          subscriptions: [eg.subscription(stream)]);
         await tester.pump();
         check(find.descendant(
           of: find.byType(StreamMessageRecipientHeader),
@@ -280,7 +280,7 @@ void main() {
         final stream = eg.stream(isWebPublic: true);
         await setupMessageListPage(tester,
           messages: [eg.streamMessage(stream: stream)],
-          streams: [stream]);
+          subscriptions: [eg.subscription(stream)]);
         await tester.pump();
         check(find.descendant(
           of: find.byType(StreamMessageRecipientHeader),
@@ -292,7 +292,7 @@ void main() {
         final stream = eg.stream(inviteOnly: true);
         await setupMessageListPage(tester,
           messages: [eg.streamMessage(stream: stream)],
-          streams: [stream]);
+          subscriptions: [eg.subscription(stream)]);
         await tester.pump();
         check(find.descendant(
           of: find.byType(StreamMessageRecipientHeader),
@@ -304,30 +304,33 @@ void main() {
     testWidgets('show stream name from message when stream unknown', (tester) async {
       // This can perfectly well happen, because message fetches can race
       // with events.
+      // … Though not actually with AllMessagesNarrow, because that shows
+      // stream messages only in subscribed streams, hence only known streams.
+      // See skip comment below.
       final stream = eg.stream(name: 'stream name');
       await setupMessageListPage(tester,
         narrow: const AllMessagesNarrow(),
+        subscriptions: [],
         messages: [
           eg.streamMessage(stream: stream),
         ]);
       await tester.pump();
       tester.widget(find.text('stream name'));
-    });
+    }, skip: true); // TODO(#252) could repro this with search narrows, once we have those
 
     testWidgets('show stream name from stream data when known', (tester) async {
-      final stream = eg.stream(name: 'old stream name');
+      final streamBefore = eg.stream(name: 'old stream name');
+      // TODO(#182) this test would be more realistic using a StreamUpdateEvent
+      final streamAfter = ZulipStream.fromJson({
+        ...(deepToJson(streamBefore) as Map<String, dynamic>),
+        'name': 'new stream name',
+      });
       await setupMessageListPage(tester,
         narrow: const AllMessagesNarrow(),
+        subscriptions: [eg.subscription(streamAfter)],
         messages: [
-          eg.streamMessage(stream: stream),
+          eg.streamMessage(stream: streamBefore),
         ]);
-      // TODO(#182) this test would be more realistic using a StreamUpdateEvent
-      store.handleEvent(StreamCreateEvent(id: stream.streamId, streams: [
-        ZulipStream.fromJson({
-          ...(deepToJson(stream) as Map<String, dynamic>),
-          'name': 'new stream name',
-        }),
-      ]));
       await tester.pump();
       tester.widget(find.text('new stream name'));
     });
diff --git a/test/widgets/subscription_list_test.dart b/test/widgets/subscription_list_test.dart
index 1d596a0497..1259bca4a4 100644
--- a/test/widgets/subscription_list_test.dart
+++ b/test/widgets/subscription_list_test.dart
@@ -7,6 +7,7 @@ import 'package:zulip/widgets/store.dart';
 import 'package:zulip/widgets/subscription_list.dart';
 import 'package:zulip/widgets/unread_count_badge.dart';
 
+import '../flutter_checks.dart';
 import '../model/binding.dart';
 import '../example_data.dart' as eg;
 
@@ -15,12 +16,14 @@ void main() {
 
   Future<void> setupStreamListPage(WidgetTester tester, {
     required List<Subscription> subscriptions,
+    List<UserTopicItem> userTopics = const [],
     UnreadMessagesSnapshot? unreadMsgs,
   }) async {
     addTearDown(testBinding.reset);
     final initialSnapshot = eg.initialSnapshot(
       subscriptions: subscriptions,
       streams: subscriptions.toList(),
+      userTopics: userTopics,
       unreadMsgs: unreadMsgs,
     );
     await testBinding.globalStore.add(eg.selfAccount, initialSnapshot);
@@ -111,6 +114,26 @@ void main() {
     check(find.byType(UnreadCountBadge).evaluate()).length.equals(1);
   });
 
+  testWidgets('unread badge counts unmuted only', (tester) async {
+    final stream = eg.stream();
+    final unreadMsgs = eg.unreadMsgs(streams: [
+      UnreadStreamSnapshot(streamId: stream.streamId, topic: 'a', unreadMessageIds: [1, 2]),
+      UnreadStreamSnapshot(streamId: stream.streamId, topic: 'b', unreadMessageIds: [3]),
+    ]);
+    await setupStreamListPage(tester,
+      subscriptions: [eg.subscription(stream, isMuted: true)],
+        userTopics: [UserTopicItem(
+          streamId: stream.streamId,
+          topicName: 'b',
+          lastUpdated: 1234567890,
+          visibilityPolicy: UserTopicVisibilityPolicy.unmuted,
+        )],
+        unreadMsgs: unreadMsgs);
+    check(tester.widget<Text>(find.descendant(
+        of: find.byType(UnreadCountBadge), matching: find.byType(Text))))
+      .data.equals('1');
+  });
+
   testWidgets('unread badge does not show with no unreads', (tester) async {
     final stream = eg.stream();
     final unreadMsgs = eg.unreadMsgs(streams: []);