Merge pull request #3346 from element-hq/robin/behaviors

Create a type-level distinction between raw Observables and Behaviors

.eslintrc.cjs

@@ -44,7 +44,7 @@ module.exports = {
     ],
     // To encourage good usage of RxJS:
     "rxjs/no-exposed-subjects": "error",
-    "rxjs/finnish": "error",
+    "rxjs/finnish": ["error", { names: { "^this$": false } }],
   },
   settings: {
     react: {

src/button/ReactionToggleButton.tsx

@@ -24,8 +24,6 @@ import {
 import { useTranslation } from "react-i18next";
 import { logger } from "matrix-js-sdk/lib/logger";
 import classNames from "classnames";
-import { useObservableState } from "observable-hooks";
-import { map } from "rxjs";
 
 import { useReactionsSender } from "../reactions/useReactionsSender";
 import styles from "./ReactionToggleButton.module.css";
@@ -36,6 +34,7 @@ import {
 } from "../reactions";
 import { Modal } from "../Modal";
 import { type CallViewModel } from "../state/CallViewModel";
+import { useBehavior } from "../useBehavior";
 
 interface InnerButtonProps extends ComponentPropsWithoutRef<"button"> {
   raised: boolean;
@@ -180,12 +179,8 @@ export function ReactionToggleButton({
   const [showReactionsMenu, setShowReactionsMenu] = useState(false);
   const [errorText, setErrorText] = useState<string>();
 
-  const isHandRaised = useObservableState(
+  const isHandRaised = !!useBehavior(vm.handsRaised$)[identifier];
-    vm.handsRaised$.pipe(map((v) => !!v[identifier])),
+  const canReact = !useBehavior(vm.reactions$)[identifier];
-  );
-  const canReact = useObservableState(
-    vm.reactions$.pipe(map((v) => !v[identifier])),
-  );
 
   useEffect(() => {
     // Clear whenever the reactions menu state changes.

src/grid/Grid.tsx

@@ -24,16 +24,16 @@ import {
   createContext,
   memo,
   use,
+  useCallback,
   useEffect,
   useMemo,
   useRef,
   useState,
+  useSyncExternalStore,
 } from "react";
 import useMeasure from "react-use-measure";
 import classNames from "classnames";
 import { logger } from "matrix-js-sdk/lib/logger";
-import { useObservableEagerState } from "observable-hooks";
-import { fromEvent, map, startWith } from "rxjs";
 
 import styles from "./Grid.module.css";
 import { useMergedRefs } from "../useMergedRefs";
@@ -155,11 +155,6 @@ export function useVisibleTiles(callback: VisibleTilesCallback): void {
   );
 }
 
-const windowHeightObservable$ = fromEvent(window, "resize").pipe(
-  startWith(null),
-  map(() => window.innerHeight),
-);
-
 export interface LayoutProps<LayoutModel, TileModel, R extends HTMLElement> {
   ref?: Ref<R>;
   model: LayoutModel;
@@ -261,7 +256,13 @@ export function Grid<
   const [gridRoot, gridRef2] = useState<HTMLElement | null>(null);
   const gridRef = useMergedRefs<HTMLElement>(gridRef1, gridRef2);
 
-  const windowHeight = useObservableEagerState(windowHeightObservable$);
+  const windowHeight = useSyncExternalStore(
+    useCallback((onChange) => {
+      window.addEventListener("resize", onChange);
+      return (): void => window.removeEventListener("resize", onChange);
+    }, []),
+    useCallback(() => window.innerHeight, []),
+  );
   const [layoutRoot, setLayoutRoot] = useState<HTMLElement | null>(null);
   const [generation, setGeneration] = useState<number | null>(null);
   const [visibleTilesCallback, setVisibleTilesCallback] =

src/grid/OneOnOneLayout.tsx

@@ -13,6 +13,7 @@ import { type OneOnOneLayout as OneOnOneLayoutModel } from "../state/CallViewMod
 import { type CallLayout, arrangeTiles } from "./CallLayout";
 import styles from "./OneOnOneLayout.module.css";
 import { type DragCallback, useUpdateLayout } from "./Grid";
+import { useBehavior } from "../useBehavior";
 
 /**
  * An implementation of the "one-on-one" layout, in which the remote participant
@@ -32,7 +33,7 @@ export const makeOneOnOneLayout: CallLayout<OneOnOneLayoutModel> = ({
   scrolling: function OneOnOneLayoutScrolling({ ref, model, Slot }): ReactNode {
     useUpdateLayout();
     const { width, height } = useObservableEagerState(minBounds$);
-    const pipAlignmentValue = useObservableEagerState(pipAlignment$);
+    const pipAlignmentValue = useBehavior(pipAlignment$);
     const { tileWidth, tileHeight } = useMemo(
       () => arrangeTiles(width, height, 1),
       [width, height],

src/grid/SpotlightExpandedLayout.tsx

@@ -6,12 +6,12 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { type ReactNode, useCallback } from "react";
-import { useObservableEagerState } from "observable-hooks";
 
 import { type SpotlightExpandedLayout as SpotlightExpandedLayoutModel } from "../state/CallViewModel";
 import { type CallLayout } from "./CallLayout";
 import { type DragCallback, useUpdateLayout } from "./Grid";
 import styles from "./SpotlightExpandedLayout.module.css";
+import { useBehavior } from "../useBehavior";
 
 /**
  * An implementation of the "expanded spotlight" layout, in which the spotlight
@@ -46,7 +46,7 @@ export const makeSpotlightExpandedLayout: CallLayout<
     Slot,
   }): ReactNode {
     useUpdateLayout();
-    const pipAlignmentValue = useObservableEagerState(pipAlignment$);
+    const pipAlignmentValue = useBehavior(pipAlignment$);
 
     const onDragPip: DragCallback = useCallback(
       ({ xRatio, yRatio }) =>

src/grid/SpotlightPortraitLayout.tsx

@@ -13,6 +13,7 @@ import { type CallLayout, arrangeTiles } from "./CallLayout";
 import { type SpotlightPortraitLayout as SpotlightPortraitLayoutModel } from "../state/CallViewModel";
 import styles from "./SpotlightPortraitLayout.module.css";
 import { useUpdateLayout, useVisibleTiles } from "./Grid";
+import { useBehavior } from "../useBehavior";
 
 interface GridCSSProperties extends CSSProperties {
   "--grid-gap": string;
@@ -65,8 +66,7 @@ export const makeSpotlightPortraitLayout: CallLayout<
       width,
       model.grid.length,
     );
-    const withIndicators =
+    const withIndicators = useBehavior(model.spotlight.media$).length > 1;
-      useObservableEagerState(model.spotlight.media$).length > 1;
 
     return (
       <div

src/livekit/useLivekit.ts

@@ -157,10 +157,13 @@ export function useLivekit(
     useObservableEagerState(
       useObservable(
         (room$) =>
+          room$.pipe(
+            switchMap(([room]) =>
               observeTrackReference$(
-            room$.pipe(map(([room]) => room.localParticipant)),
+                room.localParticipant,
                 Track.Source.Camera,
-          ).pipe(
+              ),
+            ),
             map((trackRef) => {
               const track = trackRef?.publication?.track;
               return track instanceof LocalVideoTrack ? track : null;

src/reactions/useReactionsSender.tsx

@@ -16,12 +16,12 @@ import {
 } from "react";
 import { type MatrixRTCSession } from "matrix-js-sdk/lib/matrixrtc";
 import { logger } from "matrix-js-sdk/lib/logger";
-import { useObservableEagerState } from "observable-hooks";
 
 import { useMatrixRTCSessionMemberships } from "../useMatrixRTCSessionMemberships";
 import { useClientState } from "../ClientContext";
 import { ElementCallReactionEventType, type ReactionOption } from ".";
 import { type CallViewModel } from "../state/CallViewModel";
+import { useBehavior } from "../useBehavior";
 
 interface ReactionsSenderContextType {
   supportsReactions: boolean;
@@ -70,7 +70,7 @@ export const ReactionsSenderProvider = ({
     [memberships, myUserId, myDeviceId],
   );
 
-  const reactions = useObservableEagerState(vm.reactions$);
+  const reactions = useBehavior(vm.reactions$);
   const myReaction = useMemo(
     () =>
       myMembershipIdentifier !== undefined
@@ -79,7 +79,7 @@ export const ReactionsSenderProvider = ({
     [myMembershipIdentifier, reactions],
   );
 
-  const handsRaised = useObservableEagerState(vm.handsRaised$);
+  const handsRaised = useBehavior(vm.handsRaised$);
   const myRaisedHand = useMemo(
     () =>
       myMembershipIdentifier !== undefined

src/room/GroupCallView.test.tsx

@@ -16,7 +16,6 @@ import {
 import { render, waitFor, screen } from "@testing-library/react";
 import { type MatrixClient, JoinRule, type RoomState } from "matrix-js-sdk";
 import { type MatrixRTCSession } from "matrix-js-sdk/lib/matrixrtc";
-import { of } from "rxjs";
 import { BrowserRouter } from "react-router-dom";
 import userEvent from "@testing-library/user-event";
 import { type RelationsContainer } from "matrix-js-sdk/lib/models/relations-container";
@@ -43,6 +42,7 @@ import { MatrixRTCFocusMissingError } from "../utils/errors";
 import { ProcessorProvider } from "../livekit/TrackProcessorContext";
 import { MediaDevicesContext } from "../MediaDevicesContext";
 import { HeaderStyle } from "../UrlParams";
+import { constant } from "../state/Behavior";
 
 vi.mock("../soundUtils");
 vi.mock("../useAudioContext");
@@ -141,7 +141,7 @@ function createGroupCallView(
     room,
     localRtcMember,
     [],
-  ).withMemberships(of([]));
+  ).withMemberships(constant([]));
   rtcSession.joined = joined;
   const muteState = {
     audio: { enabled: false },

src/room/GroupCallView.tsx

@@ -24,7 +24,6 @@ import {
   type MatrixRTCSession,
 } from "matrix-js-sdk/lib/matrixrtc";
 import { useNavigate } from "react-router-dom";
-import { useObservableEagerState } from "observable-hooks";
 
 import type { IWidgetApiRequest } from "matrix-widget-api";
 import {
@@ -72,6 +71,7 @@ import {
 import { useTypedEventEmitter } from "../useEvents";
 import { muteAllAudio$ } from "../state/MuteAllAudioModel.ts";
 import { useAppBarTitle } from "../AppBar.tsx";
+import { useBehavior } from "../useBehavior.ts";
 
 declare global {
   interface Window {
@@ -110,7 +110,7 @@ export const GroupCallView: FC<Props> = ({
   );
   const memberships = useMatrixRTCSessionMemberships(rtcSession);
 
-  const muteAllAudio = useObservableEagerState(muteAllAudio$);
+  const muteAllAudio = useBehavior(muteAllAudio$);
   const leaveSoundContext = useLatest(
     useAudioContext({
       sounds: callEventAudioSounds,

src/room/InCallView.tsx

@@ -25,7 +25,7 @@ import useMeasure from "react-use-measure";
 import { type MatrixRTCSession } from "matrix-js-sdk/lib/matrixrtc";
 import classNames from "classnames";
 import { BehaviorSubject, map } from "rxjs";
-import { useObservable, useObservableEagerState } from "observable-hooks";
+import { useObservable } from "observable-hooks";
 import { logger } from "matrix-js-sdk/lib/logger";
 import { RoomAndToDeviceEvents } from "matrix-js-sdk/lib/matrixrtc/RoomAndToDeviceKeyTransport";
 import {
@@ -112,6 +112,7 @@ import { useMatrixRTCSessionMemberships } from "../useMatrixRTCSessionMembership
 import { useMediaDevices } from "../MediaDevicesContext.ts";
 import { EarpieceOverlay } from "./EarpieceOverlay.tsx";
 import { useAppBarHidden, useAppBarSecondaryButton } from "../AppBar.tsx";
+import { useBehavior } from "../useBehavior.ts";
 
 const canScreenshare = "getDisplayMedia" in (navigator.mediaDevices ?? {});
 
@@ -251,7 +252,7 @@ export const InCallView: FC<InCallViewProps> = ({
     room: livekitRoom,
   });
 
-  const muteAllAudio = useObservableEagerState(muteAllAudio$);
+  const muteAllAudio = useBehavior(muteAllAudio$);
 
   // This seems like it might be enough logic to use move it into the call view model?
   const [didFallbackToRoomKey, setDidFallbackToRoomKey] = useState(false);
@@ -302,15 +303,15 @@ export const InCallView: FC<InCallViewProps> = ({
     () => void toggleRaisedHand(),
   );
 
-  const windowMode = useObservableEagerState(vm.windowMode$);
+  const windowMode = useBehavior(vm.windowMode$);
-  const layout = useObservableEagerState(vm.layout$);
+  const layout = useBehavior(vm.layout$);
-  const tileStoreGeneration = useObservableEagerState(vm.tileStoreGeneration$);
+  const tileStoreGeneration = useBehavior(vm.tileStoreGeneration$);
   const [debugTileLayout] = useSetting(debugTileLayoutSetting);
-  const gridMode = useObservableEagerState(vm.gridMode$);
+  const gridMode = useBehavior(vm.gridMode$);
-  const showHeader = useObservableEagerState(vm.showHeader$);
+  const showHeader = useBehavior(vm.showHeader$);
-  const showFooter = useObservableEagerState(vm.showFooter$);
+  const showFooter = useBehavior(vm.showFooter$);
-  const earpieceMode = useObservableEagerState(vm.earpieceMode$);
+  const earpieceMode = useBehavior(vm.earpieceMode$);
-  const audioOutputSwitcher = useObservableEagerState(vm.audioOutputSwitcher$);
+  const audioOutputSwitcher = useBehavior(vm.audioOutputSwitcher$);
   const switchCamera = useSwitchCamera(vm.localVideo$);
 
   // Ideally we could detect taps by listening for click events and checking
@@ -527,16 +528,12 @@ export const InCallView: FC<InCallViewProps> = ({
         targetHeight,
         model,
       }: TileProps<TileViewModel, HTMLDivElement>): ReactNode {
-        const spotlightExpanded = useObservableEagerState(
+        const spotlightExpanded = useBehavior(vm.spotlightExpanded$);
-          vm.spotlightExpanded$,
+        const onToggleExpanded = useBehavior(vm.toggleSpotlightExpanded$);
-        );
+        const showSpeakingIndicatorsValue = useBehavior(
-        const onToggleExpanded = useObservableEagerState(
-          vm.toggleSpotlightExpanded$,
-        );
-        const showSpeakingIndicatorsValue = useObservableEagerState(
           vm.showSpeakingIndicators$,
         );
-        const showSpotlightIndicatorsValue = useObservableEagerState(
+        const showSpotlightIndicatorsValue = useBehavior(
           vm.showSpotlightIndicators$,
         );
 

src/room/ReactionsOverlay.tsx

@@ -6,16 +6,16 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { type ReactNode } from "react";
-import { useObservableState } from "observable-hooks";
 
 import styles from "./ReactionsOverlay.module.css";
 import { type CallViewModel } from "../state/CallViewModel";
+import { useBehavior } from "../useBehavior";
 
 export function ReactionsOverlay({ vm }: { vm: CallViewModel }): ReactNode {
-  const reactionsIcons = useObservableState(vm.visibleReactions$);
+  const reactionsIcons = useBehavior(vm.visibleReactions$);
   return (
     <div className={styles.container}>
-      {reactionsIcons?.map(({ sender, emoji, startX }) => (
+      {reactionsIcons.map(({ sender, emoji, startX }) => (
         <span
           // Reactions effects are considered presentation elements. The reaction
           // is also present on the sender's tile, which assistive technology can

src/settings/SettingsModal.tsx

@@ -10,7 +10,6 @@ import { useTranslation } from "react-i18next";
 import { type MatrixClient } from "matrix-js-sdk";
 import { Button, Root as Form, Separator } from "@vector-im/compound-web";
 import { type Room as LivekitRoom } from "livekit-client";
-import { useObservableEagerState } from "observable-hooks";
 
 import { Modal } from "../Modal";
 import styles from "./SettingsModal.module.css";
@@ -34,6 +33,7 @@ import { DeveloperSettingsTab } from "./DeveloperSettingsTab";
 import { FieldRow, InputField } from "../input/Input";
 import { useSubmitRageshake } from "./submit-rageshake";
 import { useUrlParams } from "../UrlParams";
+import { useBehavior } from "../useBehavior";
 
 type SettingsTab =
   | "audio"
@@ -112,7 +112,7 @@ export const SettingsModal: FC<Props> = ({
   // rather than the input section.
   const { controlledAudioDevices } = useUrlParams();
   // If we are on iOS we will show a button to open the native audio device picker.
-  const iosDeviceMenu = useObservableEagerState(iosDeviceMenu$);
+  const iosDeviceMenu = useBehavior(iosDeviceMenu$);
 
   const audioTab: Tab<SettingsTab> = {
     key: "audio",

src/settings/settings.ts

@@ -6,10 +6,11 @@ Please see LICENSE in the repository root for full details.
 */
 
 import { logger } from "matrix-js-sdk/lib/logger";
-import { BehaviorSubject, type Observable } from "rxjs";
+import { BehaviorSubject } from "rxjs";
-import { useObservableEagerState } from "observable-hooks";
 
 import { PosthogAnalytics } from "../analytics/PosthogAnalytics";
+import { type Behavior } from "../state/Behavior";
+import { useBehavior } from "../useBehavior";
 
 export class Setting<T> {
   public constructor(
@@ -38,7 +39,7 @@ export class Setting<T> {
   private readonly key: string;
 
   private readonly _value$: BehaviorSubject<T>;
-  public readonly value$: Observable<T>;
+  public readonly value$: Behavior<T>;
 
   public readonly setValue = (value: T): void => {
     this._value$.next(value);
@@ -53,7 +54,7 @@ export class Setting<T> {
  * React hook that returns a settings's current value and a setter.
  */
 export function useSetting<T>(setting: Setting<T>): [T, (value: T) => void] {
-  return [useObservableEagerState(setting.value$), setting.setValue];
+  return [useBehavior(setting.value$), setting.setValue];
 }
 
 // null = undecided

src/state/Behavior.ts

@@ -0,0 +1,26 @@
+/*
+Copyright 2025 New Vector Ltd.
+
+SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+Please see LICENSE in the repository root for full details.
+*/
+
+import { BehaviorSubject } from "rxjs";
+
+/**
+ * A stateful, read-only reactive value. As an Observable, it is "hot" and
+ * always replays the current value upon subscription.
+ *
+ * A Behavior is to BehaviorSubject what Observable is to Subject; it does not
+ * provide a way to imperatively set new values. For more info on the
+ * distinction between Behaviors and Observables, see
+ * https://monoid.dk/post/behaviors-and-streams-why-both/.
+ */
+export type Behavior<T> = Omit<BehaviorSubject<T>, "next" | "observers">;
+
+/**
+ * Creates a Behavior which never changes in value.
+ */
+export function constant<T>(value: T): Behavior<T> {
+  return new BehaviorSubject(value);
+}

src/state/CallViewModel.test.ts

@@ -12,9 +12,9 @@ import {
   debounceTime,
   distinctUntilChanged,
   map,
+  NEVER,
   type Observable,
   of,
-  skip,
   switchMap,
 } from "rxjs";
 import { type MatrixClient } from "matrix-js-sdk";
@@ -75,10 +75,18 @@ import {
 import { ObservableScope } from "./ObservableScope";
 import { MediaDevices } from "./MediaDevices";
 import { getValue } from "../utils/observable";
+import { type Behavior, constant } from "./Behavior";
 
 const getUrlParams = vi.hoisted(() => vi.fn(() => ({})));
 vi.mock("../UrlParams", () => ({ getUrlParams }));
 
+vi.mock("rxjs", async (importOriginal) => ({
+  ...(await importOriginal()),
+  // Disable interval Observables for the following tests since the test
+  // scheduler will loop on them forever and never call the test 'done'
+  interval: (): Observable<number> => NEVER,
+}));
+
 vi.mock("@livekit/components-core");
 
 const daveRtcMember = mockRtcMembership("@dave:example.org", "DDDD");
@@ -157,9 +165,10 @@ function summarizeLayout$(l$: Observable<Layout>): Observable<LayoutSummary> {
         case "grid":
           return combineLatest(
             [
-              l.spotlight?.media$ ?? of(undefined),
+              l.spotlight?.media$ ?? constant(undefined),
               ...l.grid.map((vm) => vm.media$),
             ],
+            // eslint-disable-next-line rxjs/finnish -- false positive
             (spotlight, ...grid) => ({
               type: l.type,
               spotlight: spotlight?.map((vm) => vm.id),
@@ -178,7 +187,8 @@ function summarizeLayout$(l$: Observable<Layout>): Observable<LayoutSummary> {
           );
         case "spotlight-expanded":
           return combineLatest(
-            [l.spotlight.media$, l.pip?.media$ ?? of(undefined)],
+            [l.spotlight.media$, l.pip?.media$ ?? constant(undefined)],
+            // eslint-disable-next-line rxjs/finnish -- false positive
             (spotlight, pip) => ({
               type: l.type,
               spotlight: spotlight.map((vm) => vm.id),
@@ -212,8 +222,8 @@ function summarizeLayout$(l$: Observable<Layout>): Observable<LayoutSummary> {
 }
 
 function withCallViewModel(
-  remoteParticipants$: Observable<RemoteParticipant[]>,
+  remoteParticipants$: Behavior<RemoteParticipant[]>,
-  rtcMembers$: Observable<Partial<CallMembership>[]>,
+  rtcMembers$: Behavior<Partial<CallMembership>[]>,
   connectionState$: Observable<ECConnectionState>,
   speaking: Map<Participant, Observable<boolean>>,
   mediaDevices: MediaDevices,
@@ -291,7 +301,7 @@ function withCallViewModel(
 }
 
 test("participants are retained during a focus switch", () => {
-  withTestScheduler(({ hot, expectObservable }) => {
+  withTestScheduler(({ behavior, expectObservable }) => {
     // Participants disappear on frame 2 and come back on frame 3
     const participantInputMarbles = "a-ba";
     // Start switching focus on frame 1 and reconnect on frame 3
@@ -300,12 +310,12 @@ test("participants are retained during a focus switch", () => {
     const expectedLayoutMarbles = "  a";
 
     withCallViewModel(
-      hot(participantInputMarbles, {
+      behavior(participantInputMarbles, {
         a: [aliceParticipant, bobParticipant],
         b: [],
       }),
-      of([aliceRtcMember, bobRtcMember]),
+      constant([aliceRtcMember, bobRtcMember]),
-      hot(connectionInputMarbles, {
+      behavior(connectionInputMarbles, {
         c: ConnectionState.Connected,
         s: ECAddonConnectionState.ECSwitchingFocus,
       }),
@@ -328,7 +338,7 @@ test("participants are retained during a focus switch", () => {
 });
 
 test("screen sharing activates spotlight layout", () => {
-  withTestScheduler(({ hot, schedule, expectObservable }) => {
+  withTestScheduler(({ behavior, schedule, expectObservable }) => {
     // Start with no screen shares, then have Alice and Bob share their screens,
     // then return to no screen shares, then have just Alice share for a bit
     const participantInputMarbles = "    abcda-ba";
@@ -341,13 +351,13 @@ test("screen sharing activates spotlight layout", () => {
     const expectedLayoutMarbles = "      abcdaefeg";
     const expectedShowSpeakingMarbles = "y----nyny";
     withCallViewModel(
-      hot(participantInputMarbles, {
+      behavior(participantInputMarbles, {
         a: [aliceParticipant, bobParticipant],
         b: [aliceSharingScreen, bobParticipant],
         c: [aliceSharingScreen, bobSharingScreen],
         d: [aliceParticipant, bobSharingScreen],
       }),
-      of([aliceRtcMember, bobRtcMember]),
+      constant([aliceRtcMember, bobRtcMember]),
       of(ConnectionState.Connected),
       new Map(),
       mockMediaDevices({}),
@@ -413,7 +423,7 @@ test("screen sharing activates spotlight layout", () => {
 });
 
 test("participants stay in the same order unless to appear/disappear", () => {
-  withTestScheduler(({ hot, schedule, expectObservable }) => {
+  withTestScheduler(({ behavior, schedule, expectObservable }) => {
     const visibilityInputMarbles = "a";
     // First Bob speaks, then Dave, then Alice
     const aSpeakingInputMarbles = " n- 1998ms - 1999ms y";
@@ -426,13 +436,22 @@ test("participants stay in the same order unless to appear/disappear", () => {
     const expectedLayoutMarbles = " a  1999ms b 1999ms a 57999ms c 1999ms a";
 
     withCallViewModel(
-      of([aliceParticipant, bobParticipant, daveParticipant]),
+      constant([aliceParticipant, bobParticipant, daveParticipant]),
-      of([aliceRtcMember, bobRtcMember, daveRtcMember]),
+      constant([aliceRtcMember, bobRtcMember, daveRtcMember]),
       of(ConnectionState.Connected),
       new Map([
-        [aliceParticipant, hot(aSpeakingInputMarbles, { y: true, n: false })],
+        [
-        [bobParticipant, hot(bSpeakingInputMarbles, { y: true, n: false })],
+          aliceParticipant,
-        [daveParticipant, hot(dSpeakingInputMarbles, { y: true, n: false })],
+          behavior(aSpeakingInputMarbles, { y: true, n: false }),
+        ],
+        [
+          bobParticipant,
+          behavior(bSpeakingInputMarbles, { y: true, n: false }),
+        ],
+        [
+          daveParticipant,
+          behavior(dSpeakingInputMarbles, { y: true, n: false }),
+        ],
       ]),
       mockMediaDevices({}),
       (vm) => {
@@ -472,7 +491,7 @@ test("participants stay in the same order unless to appear/disappear", () => {
 });
 
 test("participants adjust order when space becomes constrained", () => {
-  withTestScheduler(({ hot, schedule, expectObservable }) => {
+  withTestScheduler(({ behavior, schedule, expectObservable }) => {
     // Start with all tiles on screen then shrink to 3
     const visibilityInputMarbles = "a-b";
     // Bob and Dave speak
@@ -484,12 +503,18 @@ test("participants adjust order when space becomes constrained", () => {
     const expectedLayoutMarbles = " a-b";
 
     withCallViewModel(
-      of([aliceParticipant, bobParticipant, daveParticipant]),
+      constant([aliceParticipant, bobParticipant, daveParticipant]),
-      of([aliceRtcMember, bobRtcMember, daveRtcMember]),
+      constant([aliceRtcMember, bobRtcMember, daveRtcMember]),
       of(ConnectionState.Connected),
       new Map([
-        [bobParticipant, hot(bSpeakingInputMarbles, { y: true, n: false })],
+        [
-        [daveParticipant, hot(dSpeakingInputMarbles, { y: true, n: false })],
+          bobParticipant,
+          behavior(bSpeakingInputMarbles, { y: true, n: false }),
+        ],
+        [
+          daveParticipant,
+          behavior(dSpeakingInputMarbles, { y: true, n: false }),
+        ],
       ]),
       mockMediaDevices({}),
       (vm) => {
@@ -523,7 +548,7 @@ test("participants adjust order when space becomes constrained", () => {
 });
 
 test("spotlight speakers swap places", () => {
-  withTestScheduler(({ hot, schedule, expectObservable }) => {
+  withTestScheduler(({ behavior, schedule, expectObservable }) => {
     // Go immediately into spotlight mode for the test
     const modeInputMarbles = "     s";
     // First Bob speaks, then Dave, then Alice
@@ -537,13 +562,22 @@ test("spotlight speakers swap places", () => {
     const expectedLayoutMarbles = "abcd";
 
     withCallViewModel(
-      of([aliceParticipant, bobParticipant, daveParticipant]),
+      constant([aliceParticipant, bobParticipant, daveParticipant]),
-      of([aliceRtcMember, bobRtcMember, daveRtcMember]),
+      constant([aliceRtcMember, bobRtcMember, daveRtcMember]),
       of(ConnectionState.Connected),
       new Map([
-        [aliceParticipant, hot(aSpeakingInputMarbles, { y: true, n: false })],
+        [
-        [bobParticipant, hot(bSpeakingInputMarbles, { y: true, n: false })],
+          aliceParticipant,
-        [daveParticipant, hot(dSpeakingInputMarbles, { y: true, n: false })],
+          behavior(aSpeakingInputMarbles, { y: true, n: false }),
+        ],
+        [
+          bobParticipant,
+          behavior(bSpeakingInputMarbles, { y: true, n: false }),
+        ],
+        [
+          daveParticipant,
+          behavior(dSpeakingInputMarbles, { y: true, n: false }),
+        ],
       ]),
       mockMediaDevices({}),
       (vm) => {
@@ -587,8 +621,8 @@ test("layout enters picture-in-picture mode when requested", () => {
     const expectedLayoutMarbles = " aba";
 
     withCallViewModel(
-      of([aliceParticipant, bobParticipant]),
+      constant([aliceParticipant, bobParticipant]),
-      of([aliceRtcMember, bobRtcMember]),
+      constant([aliceRtcMember, bobRtcMember]),
       of(ConnectionState.Connected),
       new Map(),
       mockMediaDevices({}),
@@ -629,8 +663,8 @@ test("spotlight remembers whether it's expanded", () => {
     const expectedLayoutMarbles = "abcbada";
 
     withCallViewModel(
-      of([aliceParticipant, bobParticipant]),
+      constant([aliceParticipant, bobParticipant]),
-      of([aliceRtcMember, bobRtcMember]),
+      constant([aliceRtcMember, bobRtcMember]),
       of(ConnectionState.Connected),
       new Map(),
       mockMediaDevices({}),
@@ -678,7 +712,7 @@ test("spotlight remembers whether it's expanded", () => {
 });
 
 test("participants must have a MatrixRTCSession to be visible", () => {
-  withTestScheduler(({ hot, expectObservable }) => {
+  withTestScheduler(({ behavior, expectObservable }) => {
     // iterate through a number of combinations of participants and MatrixRTC memberships
     // Bob never has an MatrixRTC membership
     const scenarioInputMarbles = " abcdec";
@@ -686,14 +720,14 @@ test("participants must have a MatrixRTCSession to be visible", () => {
     const expectedLayoutMarbles = "a-bc-b";
 
     withCallViewModel(
-      hot(scenarioInputMarbles, {
+      behavior(scenarioInputMarbles, {
         a: [],
         b: [bobParticipant],
         c: [aliceParticipant, bobParticipant],
         d: [aliceParticipant, daveParticipant, bobParticipant],
         e: [aliceParticipant, daveParticipant, bobSharingScreen],
       }),
-      hot(scenarioInputMarbles, {
+      behavior(scenarioInputMarbles, {
         a: [],
         b: [],
         c: [aliceRtcMember],
@@ -734,17 +768,17 @@ test("shows participants without MatrixRTCSession when enabled in settings", ()
   try {
     // enable the setting:
     showNonMemberTiles.setValue(true);
-    withTestScheduler(({ hot, expectObservable }) => {
+    withTestScheduler(({ behavior, expectObservable }) => {
       const scenarioInputMarbles = " abc";
       const expectedLayoutMarbles = "abc";
 
       withCallViewModel(
-        hot(scenarioInputMarbles, {
+        behavior(scenarioInputMarbles, {
           a: [],
           b: [aliceParticipant],
           c: [aliceParticipant, bobParticipant],
         }),
-        of([]), // No one joins the MatrixRTC session
+        constant([]), // No one joins the MatrixRTC session
         of(ConnectionState.Connected),
         new Map(),
         mockMediaDevices({}),
@@ -779,15 +813,15 @@ test("shows participants without MatrixRTCSession when enabled in settings", ()
 });
 
 it("should show at least one tile per MatrixRTCSession", () => {
-  withTestScheduler(({ hot, expectObservable }) => {
+  withTestScheduler(({ behavior, expectObservable }) => {
     // iterate through some combinations of MatrixRTC memberships
     const scenarioInputMarbles = " abcd";
     // There should always be one tile for each MatrixRTCSession
     const expectedLayoutMarbles = "abcd";
 
     withCallViewModel(
-      of([]),
+      constant([]),
-      hot(scenarioInputMarbles, {
+      behavior(scenarioInputMarbles, {
         a: [],
         b: [aliceRtcMember],
         c: [aliceRtcMember, daveRtcMember],
@@ -829,13 +863,13 @@ it("should show at least one tile per MatrixRTCSession", () => {
 });
 
 test("should disambiguate users with the same displayname", () => {
-  withTestScheduler(({ hot, expectObservable }) => {
+  withTestScheduler(({ behavior, expectObservable }) => {
     const scenarioInputMarbles = "abcde";
     const expectedLayoutMarbles = "abcde";
 
     withCallViewModel(
-      of([]),
+      constant([]),
-      hot(scenarioInputMarbles, {
+      behavior(scenarioInputMarbles, {
         a: [],
         b: [aliceRtcMember],
         c: [aliceRtcMember, aliceDoppelgangerRtcMember],
@@ -846,10 +880,7 @@ test("should disambiguate users with the same displayname", () => {
       new Map(),
       mockMediaDevices({}),
       (vm) => {
-        // Skip the null state.
+        expectObservable(vm.memberDisplaynames$).toBe(expectedLayoutMarbles, {
-        expectObservable(vm.memberDisplaynames$.pipe(skip(1))).toBe(
-          expectedLayoutMarbles,
-          {
           // Carol has no displayname - So userId is used.
           a: new Map([[carolId, carol.userId]]),
           b: new Map([
@@ -875,21 +906,20 @@ test("should disambiguate users with the same displayname", () => {
             [aliceDoppelgangerId, "Alice"],
             [bobId, bob.rawDisplayName],
           ]),
-          },
+        });
-        );
       },
     );
   });
 });
 
 test("should disambiguate users with invisible characters", () => {
-  withTestScheduler(({ hot, expectObservable }) => {
+  withTestScheduler(({ behavior, expectObservable }) => {
     const scenarioInputMarbles = "ab";
     const expectedLayoutMarbles = "ab";
 
     withCallViewModel(
-      of([]),
+      constant([]),
-      hot(scenarioInputMarbles, {
+      behavior(scenarioInputMarbles, {
         a: [],
         b: [bobRtcMember, bobZeroWidthSpaceRtcMember],
       }),
@@ -897,10 +927,7 @@ test("should disambiguate users with invisible characters", () => {
       new Map(),
       mockMediaDevices({}),
       (vm) => {
-        // Skip the null state.
+        expectObservable(vm.memberDisplaynames$).toBe(expectedLayoutMarbles, {
-        expectObservable(vm.memberDisplaynames$.pipe(skip(1))).toBe(
-          expectedLayoutMarbles,
-          {
           // Carol has no displayname - So userId is used.
           a: new Map([[carolId, carol.userId]]),
           // Both Bobs join, and should handle zero width hacks.
@@ -912,21 +939,20 @@ test("should disambiguate users with invisible characters", () => {
               `${bobZeroWidthSpace.rawDisplayName} (${bobZeroWidthSpace.userId})`,
             ],
           ]),
-          },
+        });
-        );
       },
     );
   });
 });
 
 test("should strip RTL characters from displayname", () => {
-  withTestScheduler(({ hot, expectObservable }) => {
+  withTestScheduler(({ behavior, expectObservable }) => {
     const scenarioInputMarbles = "ab";
     const expectedLayoutMarbles = "ab";
 
     withCallViewModel(
-      of([]),
+      constant([]),
-      hot(scenarioInputMarbles, {
+      behavior(scenarioInputMarbles, {
         a: [],
         b: [daveRtcMember, daveRTLRtcMember],
       }),
@@ -934,10 +960,7 @@ test("should strip RTL characters from displayname", () => {
       new Map(),
       mockMediaDevices({}),
       (vm) => {
-        // Skip the null state.
+        expectObservable(vm.memberDisplaynames$).toBe(expectedLayoutMarbles, {
-        expectObservable(vm.memberDisplaynames$.pipe(skip(1))).toBe(
-          expectedLayoutMarbles,
-          {
           // Carol has no displayname - So userId is used.
           a: new Map([[carolId, carol.userId]]),
           // Both Dave's join. Since after stripping
@@ -948,8 +971,7 @@ test("should strip RTL characters from displayname", () => {
             // This one is, since it's using RTL.
             [daveRTLId, `evaD (${daveRTL.userId})`],
           ]),
-          },
+        });
-        );
       },
     );
   });
@@ -961,8 +983,8 @@ it("should rank raised hands above video feeds and below speakers and presenters
     const expectedLayoutMarbles = "ab";
 
     withCallViewModel(
-      of([aliceParticipant, bobParticipant]),
+      constant([aliceParticipant, bobParticipant]),
-      of([aliceRtcMember, bobRtcMember]),
+      constant([aliceRtcMember, bobRtcMember]),
       of(ConnectionState.Connected),
       new Map(),
       mockMediaDevices({}),
@@ -1036,8 +1058,8 @@ test("audio output changes when toggling earpiece mode", () => {
     const expectedTargetStateMarbles = " sese";
 
     withCallViewModel(
-      of([]),
+      constant([]),
-      of([]),
+      constant([]),
       of(ConnectionState.Connected),
       new Map(),
       devices,

src/state/CallViewModel.ts

@@ -94,6 +94,7 @@ import { observeSpeaker$ } from "./observeSpeaker";
 import { shallowEquals } from "../utils/array";
 import { calculateDisplayName, shouldDisambiguate } from "../utils/displayname";
 import { type MediaDevices } from "./MediaDevices";
+import { type Behavior } from "./Behavior";
 
 // How long we wait after a focus switch before showing the real participant
 // list again
@@ -250,8 +251,8 @@ class UserMedia {
     LocalParticipant | RemoteParticipant | undefined
   >;
 
-  public readonly speaker$: Observable<boolean>;
+  public readonly speaker$: Behavior<boolean>;
-  public readonly presenter$: Observable<boolean>;
+  public readonly presenter$: Behavior<boolean>;
   public constructor(
     public readonly id: string,
     member: RoomMember | undefined,
@@ -268,12 +269,12 @@ class UserMedia {
       this.vm = new LocalUserMediaViewModel(
         this.id,
         member,
-        this.participant$.asObservable() as Observable<LocalParticipant>,
+        this.participant$ as Behavior<LocalParticipant>,
         encryptionSystem,
         livekitRoom,
-        displayname$,
+        this.scope.behavior(displayname$),
-        handRaised$,
+        this.scope.behavior(handRaised$),
-        reaction$,
+        this.scope.behavior(reaction$),
       );
     } else {
       this.vm = new RemoteUserMediaViewModel(
@@ -284,15 +285,16 @@ class UserMedia {
         >,
         encryptionSystem,
         livekitRoom,
-        displayname$,
+        this.scope.behavior(displayname$),
-        handRaised$,
+        this.scope.behavior(handRaised$),
-        reaction$,
+        this.scope.behavior(reaction$),
       );
     }
 
-    this.speaker$ = observeSpeaker$(this.vm.speaking$).pipe(this.scope.state());
+    this.speaker$ = this.scope.behavior(observeSpeaker$(this.vm.speaking$));
 
-    this.presenter$ = this.participant$.pipe(
+    this.presenter$ = this.scope.behavior(
+      this.participant$.pipe(
         switchMap(
           (p) =>
             (p &&
@@ -305,7 +307,7 @@ class UserMedia {
               ).pipe(map((p) => p.isScreenShareEnabled))) ??
             of(false),
         ),
-      this.scope.state(),
+      ),
     );
   }
 
@@ -325,6 +327,7 @@ class UserMedia {
 }
 
 class ScreenShare {
+  private readonly scope = new ObservableScope();
   public readonly vm: ScreenShareViewModel;
   private readonly participant$: BehaviorSubject<
     LocalParticipant | RemoteParticipant
@@ -336,7 +339,7 @@ class ScreenShare {
     participant: LocalParticipant | RemoteParticipant,
     encryptionSystem: EncryptionSystem,
     liveKitRoom: LivekitRoom,
-    displayname$: Observable<string>,
+    displayName$: Observable<string>,
   ) {
     this.participant$ = new BehaviorSubject(participant);
 
@@ -346,12 +349,13 @@ class ScreenShare {
       this.participant$.asObservable(),
       encryptionSystem,
       liveKitRoom,
-      displayname$,
+      this.scope.behavior(displayName$),
       participant.isLocal,
     );
   }
 
   public destroy(): void {
+    this.scope.end();
     this.vm.destroy();
   }
 }
@@ -382,28 +386,32 @@ function getRoomMemberFromRtcMember(
 
 // TODO: Move wayyyy more business logic from the call and lobby views into here
 export class CallViewModel extends ViewModel {
-  public readonly localVideo$: Observable<LocalVideoTrack | null> =
+  public readonly localVideo$ = this.scope.behavior<LocalVideoTrack | null>(
     observeTrackReference$(
-      of(this.livekitRoom.localParticipant),
+      this.livekitRoom.localParticipant,
       Track.Source.Camera,
     ).pipe(
       map((trackRef) => {
         const track = trackRef?.publication?.track;
         return track instanceof LocalVideoTrack ? track : null;
       }),
+    ),
   );
 
   /**
    * The raw list of RemoteParticipants as reported by LiveKit
    */
-  private readonly rawRemoteParticipants$: Observable<RemoteParticipant[]> =
+  private readonly rawRemoteParticipants$ = this.scope.behavior<
-    connectedParticipantsObserver(this.livekitRoom).pipe(this.scope.state());
+    RemoteParticipant[]
+  >(connectedParticipantsObserver(this.livekitRoom), []);
 
   /**
    * Lists of RemoteParticipants to "hold" on display, even if LiveKit claims that
    * they've left
    */
-  private readonly remoteParticipantHolds$: Observable<RemoteParticipant[][]> =
+  private readonly remoteParticipantHolds$ = this.scope.behavior<
+    RemoteParticipant[][]
+  >(
     this.connectionState$.pipe(
       withLatestFrom(this.rawRemoteParticipants$),
       mergeMap(([s, ps]) => {
@@ -435,12 +443,15 @@ export class CallViewModel extends ViewModel {
           ? [instruction.hold, ...holds]
           : holds.filter((h) => h !== instruction.unhold),
       ),
+    ),
   );
 
   /**
    * The RemoteParticipants including those that are being "held" on the screen
    */
-  private readonly remoteParticipants$: Observable<RemoteParticipant[]> =
+  private readonly remoteParticipants$ = this.scope.behavior<
+    RemoteParticipant[]
+  >(
     combineLatest(
       [this.rawRemoteParticipants$, this.remoteParticipantHolds$],
       (raw, holds) => {
@@ -459,6 +470,7 @@ export class CallViewModel extends ViewModel {
 
         return result;
       },
+    ),
   );
 
   /**
@@ -466,9 +478,13 @@ export class CallViewModel extends ViewModel {
    * any displaynames that clashes with another member. Only members
    * joined to the call are considered here.
    */
-  public readonly memberDisplaynames$ = merge(
+  public readonly memberDisplaynames$ = this.scope.behavior(
+    merge(
       // Handle call membership changes.
-    fromEvent(this.matrixRTCSession, MatrixRTCSessionEvent.MembershipsChanged),
+      fromEvent(
+        this.matrixRTCSession,
+        MatrixRTCSessionEvent.MembershipsChanged,
+      ),
       // Handle room membership changes (and displayname updates)
       fromEvent(this.matrixRTCSession.room, RoomStateEvent.Members),
     ).pipe(
@@ -482,7 +498,10 @@ export class CallViewModel extends ViewModel {
           const matrixIdentifier = `${rtcMember.sender}:${rtcMember.deviceId}`;
           const { member } = getRoomMemberFromRtcMember(rtcMember, room);
           if (!member) {
-          logger.error("Could not find member for media id:", matrixIdentifier);
+            logger.error(
+              "Could not find member for media id:",
+              matrixIdentifier,
+            );
             continue;
           }
           const disambiguate = shouldDisambiguate(member, memberships, room);
@@ -494,15 +513,31 @@ export class CallViewModel extends ViewModel {
         return displaynameMap;
       }),
       // It turns out that doing the disambiguation above is rather expensive on Safari (10x slower
-    // than on Chrome/Firefox). This means it is important that we share() the result so that we
+      // than on Chrome/Firefox). This means it is important that we multicast the result so that we
-    // don't do this work more times than we need to. This is achieve through the state() operator:
+      // don't do this work more times than we need to. This is achieved by converting to a behavior:
-    this.scope.state(),
+    ),
+  );
+
+  public readonly handsRaised$ = this.scope.behavior(this.handsRaisedSubject$);
+
+  public readonly reactions$ = this.scope.behavior(
+    this.reactionsSubject$.pipe(
+      map((v) =>
+        Object.fromEntries(
+          Object.entries(v).map(([a, { reactionOption }]) => [
+            a,
+            reactionOption,
+          ]),
+        ),
+      ),
+    ),
   );
 
   /**
    * List of MediaItems that we want to display
    */
-  private readonly mediaItems$: Observable<MediaItem[]> = combineLatest([
+  private readonly mediaItems$ = this.scope.behavior<MediaItem[]>(
+    combineLatest([
       this.remoteParticipants$,
       observeParticipantMedia(this.livekitRoom.localParticipant),
       duplicateTiles.value$,
@@ -644,7 +679,9 @@ export class CallViewModel extends ViewModel {
                               this.encryptionSystem,
                               this.livekitRoom,
                               this.memberDisplaynames$.pipe(
-                              map((m) => m.get(participant.identity) ?? "[👻]"),
+                                map(
+                                  (m) => m.get(participant.identity) ?? "[👻]",
+                                ),
                               ),
                               of(null),
                               of(null),
@@ -665,7 +702,8 @@ export class CallViewModel extends ViewModel {
             ...newItems.entries(),
           ]);
 
-        for (const [id, t] of prevItems) if (!combinedNew.has(id)) t.destroy();
+          for (const [id, t] of prevItems)
+            if (!combinedNew.has(id)) t.destroy();
           return combinedNew;
         },
         new Map<string, MediaItem>(),
@@ -674,16 +712,18 @@ export class CallViewModel extends ViewModel {
       finalizeValue((ts) => {
         for (const t of ts) t.destroy();
       }),
-    this.scope.state(),
+    ),
   );
 
   /**
    * List of MediaItems that we want to display, that are of type UserMedia
    */
-  private readonly userMedia$: Observable<UserMedia[]> = this.mediaItems$.pipe(
+  private readonly userMedia$ = this.scope.behavior<UserMedia[]>(
+    this.mediaItems$.pipe(
       map((mediaItems) =>
         mediaItems.filter((m): m is UserMedia => m instanceof UserMedia),
       ),
+    ),
   );
 
   public readonly memberChanges$ = this.userMedia$
@@ -702,15 +742,16 @@ export class CallViewModel extends ViewModel {
   /**
    * List of MediaItems that we want to display, that are of type ScreenShare
    */
-  private readonly screenShares$: Observable<ScreenShare[]> =
+  private readonly screenShares$ = this.scope.behavior<ScreenShare[]>(
     this.mediaItems$.pipe(
       map((mediaItems) =>
         mediaItems.filter((m): m is ScreenShare => m instanceof ScreenShare),
       ),
-      this.scope.state(),
+    ),
   );
 
-  private readonly spotlightSpeaker$: Observable<UserMediaViewModel | null> =
+  private readonly spotlightSpeaker$ =
+    this.scope.behavior<UserMediaViewModel | null>(
       this.userMedia$.pipe(
         switchMap((mediaItems) =>
           mediaItems.length === 0
@@ -743,10 +784,10 @@ export class CallViewModel extends ViewModel {
           null,
         ),
         map((speaker) => speaker?.vm ?? null),
-      this.scope.state(),
+      ),
     );
 
-  private readonly grid$: Observable<UserMediaViewModel[]> =
+  private readonly grid$ = this.scope.behavior<UserMediaViewModel[]>(
     this.userMedia$.pipe(
       switchMap((mediaItems) => {
         const bins = mediaItems.map((m) =>
@@ -784,10 +825,10 @@ export class CallViewModel extends ViewModel {
             );
       }),
       distinctUntilChanged(shallowEquals),
-      this.scope.state(),
+    ),
   );
 
-  private readonly spotlight$: Observable<MediaViewModel[]> =
+  private readonly spotlight$ = this.scope.behavior<MediaViewModel[]>(
     this.screenShares$.pipe(
       switchMap((screenShares) => {
         if (screenShares.length > 0) {
@@ -799,10 +840,11 @@ export class CallViewModel extends ViewModel {
         );
       }),
       distinctUntilChanged(shallowEquals),
-      this.scope.state(),
+    ),
   );
 
-  private readonly pip$: Observable<UserMediaViewModel | null> = combineLatest([
+  private readonly pip$ = this.scope.behavior<UserMediaViewModel | null>(
+    combineLatest([
       this.screenShares$,
       this.spotlightSpeaker$,
       this.mediaItems$,
@@ -836,7 +878,7 @@ export class CallViewModel extends ViewModel {
           }),
         );
       }),
-    this.scope.state(),
+    ),
   );
 
   private readonly hasRemoteScreenShares$: Observable<boolean> =
@@ -847,14 +889,10 @@ export class CallViewModel extends ViewModel {
       distinctUntilChanged(),
     );
 
-  private readonly pipEnabled$: Observable<boolean> = setPipEnabled$.pipe(
+  private readonly pipEnabled$ = this.scope.behavior(setPipEnabled$, false);
-    startWith(false),
-  );
 
-  private readonly naturalWindowMode$: Observable<WindowMode> = fromEvent(
+  private readonly naturalWindowMode$ = this.scope.behavior<WindowMode>(
-    window,
+    fromEvent(window, "resize").pipe(
-    "resize",
-  ).pipe(
       startWith(null),
       map(() => {
         const height = window.innerHeight;
@@ -867,36 +905,44 @@ export class CallViewModel extends ViewModel {
         if (width <= 600) return "narrow";
         return "normal";
       }),
-    this.scope.state(),
+    ),
   );
 
   /**
    * The general shape of the window.
    */
-  public readonly windowMode$: Observable<WindowMode> = this.pipEnabled$.pipe(
+  public readonly windowMode$ = this.scope.behavior<WindowMode>(
-    switchMap((pip) => (pip ? of<WindowMode>("pip") : this.naturalWindowMode$)),
+    this.pipEnabled$.pipe(
+      switchMap((pip) =>
+        pip ? of<WindowMode>("pip") : this.naturalWindowMode$,
+      ),
+    ),
   );
 
   private readonly spotlightExpandedToggle$ = new Subject<void>();
-  public readonly spotlightExpanded$: Observable<boolean> =
+  public readonly spotlightExpanded$ = this.scope.behavior<boolean>(
     this.spotlightExpandedToggle$.pipe(
       accumulate(false, (expanded) => !expanded),
-      this.scope.state(),
+    ),
   );
 
   private readonly gridModeUserSelection$ = new Subject<GridMode>();
   /**
    * The layout mode of the media tile grid.
    */
-  public readonly gridMode$: Observable<GridMode> =
+  public readonly gridMode$ =
     // If the user hasn't selected spotlight and somebody starts screen sharing,
     // automatically switch to spotlight mode and reset when screen sharing ends
+    this.scope.behavior<GridMode>(
       this.gridModeUserSelection$.pipe(
         startWith(null),
         switchMap((userSelection) =>
           (userSelection === "spotlight"
             ? EMPTY
-          : combineLatest([this.hasRemoteScreenShares$, this.windowMode$]).pipe(
+            : combineLatest([
+                this.hasRemoteScreenShares$,
+                this.windowMode$,
+              ]).pipe(
                 skip(userSelection === null ? 0 : 1),
                 map(
                   ([hasScreenShares, windowMode]): GridMode =>
@@ -907,7 +953,7 @@ export class CallViewModel extends ViewModel {
               )
           ).pipe(startWith(userSelection ?? "grid")),
         ),
-      this.scope.state(),
+      ),
     );
 
   public setGridMode(value: GridMode): void {
@@ -969,7 +1015,7 @@ export class CallViewModel extends ViewModel {
   /**
    * The media to be used to produce a layout.
    */
-  private readonly layoutMedia$: Observable<LayoutMedia> =
+  private readonly layoutMedia$ = this.scope.behavior<LayoutMedia>(
     this.windowMode$.pipe(
       switchMap((windowMode) => {
         switch (windowMode) {
@@ -1032,7 +1078,7 @@ export class CallViewModel extends ViewModel {
             return this.pipLayoutMedia$;
         }
       }),
-      this.scope.state(),
+    ),
   );
 
   // There is a cyclical dependency here: the layout algorithms want to know
@@ -1043,9 +1089,10 @@ export class CallViewModel extends ViewModel {
   private readonly setVisibleTiles = (value: number): void =>
     this.visibleTiles$.next(value);
 
-  public readonly layoutInternals$: Observable<
+  private readonly layoutInternals$ = this.scope.behavior<
     LayoutScanState & { layout: Layout }
-  > = combineLatest([
+  >(
+    combineLatest([
       this.layoutMedia$,
       this.visibleTiles$.pipe(startWith(0), distinctUntilChanged()),
     ]).pipe(
@@ -1083,32 +1130,29 @@ export class CallViewModel extends ViewModel {
         },
         { layout: null, tiles: TileStore.empty() },
       ),
-    this.scope.state(),
+    ),
   );
 
   /**
    * The layout of tiles in the call interface.
    */
-  public readonly layout$: Observable<Layout> = this.layoutInternals$.pipe(
+  public readonly layout$ = this.scope.behavior<Layout>(
-    map(({ layout }) => layout),
+    this.layoutInternals$.pipe(map(({ layout }) => layout)),
-    this.scope.state(),
   );
 
   /**
    * The current generation of the tile store, exposed for debugging purposes.
    */
-  public readonly tileStoreGeneration$: Observable<number> =
+  public readonly tileStoreGeneration$ = this.scope.behavior<number>(
-    this.layoutInternals$.pipe(
+    this.layoutInternals$.pipe(map(({ tiles }) => tiles.generation)),
-      map(({ tiles }) => tiles.generation),
-      this.scope.state(),
   );
 
-  public showSpotlightIndicators$: Observable<boolean> = this.layout$.pipe(
+  public showSpotlightIndicators$ = this.scope.behavior<boolean>(
-    map((l) => l.type !== "grid"),
+    this.layout$.pipe(map((l) => l.type !== "grid")),
-    this.scope.state(),
   );
 
-  public showSpeakingIndicators$: Observable<boolean> = this.layout$.pipe(
+  public showSpeakingIndicators$ = this.scope.behavior<boolean>(
+    this.layout$.pipe(
       switchMap((l) => {
         switch (l.type) {
           case "spotlight-landscape":
@@ -1132,10 +1176,12 @@ export class CallViewModel extends ViewModel {
             return of(true);
         }
       }),
-    this.scope.state(),
+    ),
   );
 
-  public readonly toggleSpotlightExpanded$: Observable<(() => void) | null> =
+  public readonly toggleSpotlightExpanded$ = this.scope.behavior<
+    (() => void) | null
+  >(
     this.windowMode$.pipe(
       switchMap((mode) =>
         mode === "normal"
@@ -1152,7 +1198,7 @@ export class CallViewModel extends ViewModel {
       map((enabled) =>
         enabled ? (): void => this.spotlightExpandedToggle$.next() : null,
       ),
-      this.scope.state(),
+    ),
   );
 
   private readonly screenTap$ = new Subject<void>();
@@ -1188,12 +1234,12 @@ export class CallViewModel extends ViewModel {
     this.screenUnhover$.next();
   }
 
-  public readonly showHeader$: Observable<boolean> = this.windowMode$.pipe(
+  public readonly showHeader$ = this.scope.behavior<boolean>(
-    map((mode) => mode !== "pip" && mode !== "flat"),
+    this.windowMode$.pipe(map((mode) => mode !== "pip" && mode !== "flat")),
-    this.scope.state(),
   );
 
-  public readonly showFooter$: Observable<boolean> = this.windowMode$.pipe(
+  public readonly showFooter$ = this.scope.behavior<boolean>(
+    this.windowMode$.pipe(
       switchMap((mode) => {
         switch (mode) {
           case "pip":
@@ -1244,20 +1290,23 @@ export class CallViewModel extends ViewModel {
             );
         }
       }),
-    this.scope.state(),
+    ),
   );
 
   /**
    * Whether audio is currently being output through the earpiece.
    */
-  public readonly earpieceMode$: Observable<boolean> = combineLatest(
+  public readonly earpieceMode$ = this.scope.behavior<boolean>(
+    combineLatest(
       [
         this.mediaDevices.audioOutput.available$,
         this.mediaDevices.audioOutput.selected$,
       ],
       (available, selected) =>
-      selected !== undefined && available.get(selected.id)?.type === "earpiece",
+        selected !== undefined &&
-  ).pipe(this.scope.state());
+        available.get(selected.id)?.type === "earpiece",
+    ),
+  );
 
   /**
    * Callback to toggle between the earpiece and the loudspeaker.
@@ -1265,10 +1314,11 @@ export class CallViewModel extends ViewModel {
    * This will be `null` in case the target does not exist in the list
    * of available audio outputs.
    */
-  public readonly audioOutputSwitcher$: Observable<{
+  public readonly audioOutputSwitcher$ = this.scope.behavior<{
     targetOutput: "earpiece" | "speaker";
     switch: () => void;
-  } | null> = combineLatest(
+  } | null>(
+    combineLatest(
       [
         this.mediaDevices.audioOutput.available$,
         this.mediaDevices.audioOutput.selected$,
@@ -1277,7 +1327,7 @@ export class CallViewModel extends ViewModel {
         const selectionType = selected && available.get(selected.id)?.type;
 
         // If we are in any output mode other than speaker switch to speaker.
-      const newSelectionType =
+        const newSelectionType: "earpiece" | "speaker" =
           selectionType === "speaker" ? "earpiece" : "speaker";
         const newSelection = [...available].find(
           ([, d]) => d.type === newSelectionType,
@@ -1287,25 +1337,17 @@ export class CallViewModel extends ViewModel {
         const [id] = newSelection;
         return {
           targetOutput: newSelectionType,
-        switch: () => this.mediaDevices.audioOutput.select(id),
+          switch: (): void => this.mediaDevices.audioOutput.select(id),
         };
       },
-  );
-
-  public readonly reactions$ = this.reactionsSubject$.pipe(
-    map((v) =>
-      Object.fromEntries(
-        Object.entries(v).map(([a, { reactionOption }]) => [a, reactionOption]),
-      ),
     ),
   );
 
-  public readonly handsRaised$ = this.handsRaisedSubject$.pipe();
-
   /**
    * Emits an array of reactions that should be visible on the screen.
    */
-  public readonly visibleReactions$ = showReactions.value$.pipe(
+  public readonly visibleReactions$ = this.scope.behavior(
+    showReactions.value$.pipe(
       switchMap((show) => (show ? this.reactions$ : of({}))),
       scan<
         Record<string, ReactionOption>,
@@ -1320,6 +1362,7 @@ export class CallViewModel extends ViewModel {
         }
         return newSet;
       }, []),
+    ),
   );
 
   /**

src/state/MediaDevices.ts

@@ -10,7 +10,6 @@ import {
   filter,
   map,
   merge,
-  of,
   pairwise,
   startWith,
   Subject,
@@ -34,6 +33,7 @@ import {
 import { getUrlParams } from "../UrlParams";
 import { platform } from "../Platform";
 import { switchWhen } from "../utils/observable";
+import { type Behavior, constant } from "./Behavior";
 
 // This hardcoded id is used in EX ios! It can only be changed in coordination with
 // the ios swift team.
@@ -74,11 +74,11 @@ export interface MediaDevice<Label, Selected> {
   /**
    * A map from available device IDs to labels.
    */
-  available$: Observable<Map<string, Label>>;
+  available$: Behavior<Map<string, Label>>;
   /**
    * The selected device.
    */
-  selected$: Observable<Selected | undefined>;
+  selected$: Behavior<Selected | undefined>;
   /**
    * Selects a new device.
    */
@@ -94,19 +94,20 @@ export interface MediaDevice<Label, Selected> {
  *   `availableOutputDevices$.includes((d)=>d.forEarpiece)`
  */
 export const iosDeviceMenu$ =
-  platform === "ios" ? of(true) : alwaysShowIphoneEarpieceSetting.value$;
+  platform === "ios" ? constant(true) : alwaysShowIphoneEarpieceSetting.value$;
 
 function availableRawDevices$(
   kind: MediaDeviceKind,
-  usingNames$: Observable<boolean>,
+  usingNames$: Behavior<boolean>,
   scope: ObservableScope,
-): Observable<MediaDeviceInfo[]> {
+): Behavior<MediaDeviceInfo[]> {
   const logError = (e: Error): void =>
     logger.error("Error creating MediaDeviceObserver", e);
   const devices$ = createMediaDeviceObserver(kind, logError, false);
   const devicesWithNames$ = createMediaDeviceObserver(kind, logError, true);
 
-  return usingNames$.pipe(
+  return scope.behavior(
+    usingNames$.pipe(
       switchMap((withNames) =>
         withNames
           ? // It might be that there is already a media stream running somewhere,
@@ -121,8 +122,8 @@ function availableRawDevices$(
             )
           : devices$,
       ),
-    startWith([]),
+    ),
-    scope.state(),
+    [],
   );
 }
 
@@ -161,18 +162,15 @@ function selectDevice$<Label>(
 }
 
 class AudioInput implements MediaDevice<DeviceLabel, SelectedAudioInputDevice> {
-  private readonly availableRaw$: Observable<MediaDeviceInfo[]> =
+  private readonly availableRaw$: Behavior<MediaDeviceInfo[]> =
     availableRawDevices$("audioinput", this.usingNames$, this.scope);
 
-  public readonly available$ = this.availableRaw$.pipe(
+  public readonly available$ = this.scope.behavior(
-    map(buildDeviceMap),
+    this.availableRaw$.pipe(map(buildDeviceMap)),
-    this.scope.state(),
   );
 
-  public readonly selected$ = selectDevice$(
+  public readonly selected$ = this.scope.behavior(
-    this.available$,
+    selectDevice$(this.available$, audioInputSetting.value$).pipe(
-    audioInputSetting.value$,
-  ).pipe(
       map((id) =>
         id === undefined
           ? undefined
@@ -181,14 +179,16 @@ class AudioInput implements MediaDevice<DeviceLabel, SelectedAudioInputDevice> {
               // We can identify when the hardware device has changed by watching for
               // changes in the group ID
               hardwareDeviceChange$: this.availableRaw$.pipe(
-              map((devices) => devices.find((d) => d.deviceId === id)?.groupId),
+                map(
+                  (devices) => devices.find((d) => d.deviceId === id)?.groupId,
+                ),
                 pairwise(),
                 filter(([before, after]) => before !== after),
                 map(() => undefined),
               ),
             },
       ),
-    this.scope.state(),
+    ),
   );
 
   public select(id: string): void {
@@ -196,7 +196,7 @@ class AudioInput implements MediaDevice<DeviceLabel, SelectedAudioInputDevice> {
   }
 
   public constructor(
-    private readonly usingNames$: Observable<boolean>,
+    private readonly usingNames$: Behavior<boolean>,
     private readonly scope: ObservableScope,
   ) {
     this.available$.subscribe((available) => {
@@ -208,11 +208,8 @@ class AudioInput implements MediaDevice<DeviceLabel, SelectedAudioInputDevice> {
 class AudioOutput
   implements MediaDevice<AudioOutputDeviceLabel, SelectedAudioOutputDevice>
 {
-  public readonly available$ = availableRawDevices$(
+  public readonly available$ = this.scope.behavior(
-    "audiooutput",
+    availableRawDevices$("audiooutput", this.usingNames$, this.scope).pipe(
-    this.usingNames$,
-    this.scope,
-  ).pipe(
       map((availableRaw) => {
         const available: Map<string, AudioOutputDeviceLabel> =
           buildDeviceMap(availableRaw);
@@ -229,13 +226,11 @@ class AudioOutput
         // automatically track the default device.
         return available;
       }),
-    this.scope.state(),
+    ),
   );
 
-  public readonly selected$ = selectDevice$(
+  public readonly selected$ = this.scope.behavior(
-    this.available$,
+    selectDevice$(this.available$, audioOutputSetting.value$).pipe(
-    audioOutputSetting.value$,
-  ).pipe(
       map((id) =>
         id === undefined
           ? undefined
@@ -244,15 +239,14 @@ class AudioOutput
               virtualEarpiece: false,
             },
       ),
-    this.scope.state(),
+    ),
   );
-
   public select(id: string): void {
     audioOutputSetting.setValue(id);
   }
 
   public constructor(
-    private readonly usingNames$: Observable<boolean>,
+    private readonly usingNames$: Behavior<boolean>,
     private readonly scope: ObservableScope,
   ) {
     this.available$.subscribe((available) => {
@@ -273,7 +267,8 @@ class ControlledAudioOutput
     this.scope,
   );
 
-  public readonly available$ = combineLatest(
+  public readonly available$ = this.scope.behavior(
+    combineLatest(
       [controlledAvailableOutputDevices$.pipe(startWith([])), iosDeviceMenu$],
       (availableRaw, iosDeviceMenu) => {
         const available = new Map<string, AudioOutputDeviceLabel>(
@@ -296,7 +291,8 @@ class ControlledAudioOutput
 
         return available;
       },
-  ).pipe(this.scope.state());
+    ),
+  );
 
   private readonly deviceSelection$ = new Subject<string>();
 
@@ -304,7 +300,8 @@ class ControlledAudioOutput
     this.deviceSelection$.next(id);
   }
 
-  public readonly selected$ = combineLatest(
+  public readonly selected$ = this.scope.behavior(
+    combineLatest(
       [
         this.available$,
         merge(
@@ -318,10 +315,11 @@ class ControlledAudioOutput
           ? undefined
           : { id, virtualEarpiece: id === EARPIECE_CONFIG_ID };
       },
-  ).pipe(this.scope.state());
+    ),
+  );
 
   public constructor(
-    private readonly usingNames$: Observable<boolean>,
+    private readonly usingNames$: Behavior<boolean>,
     private readonly scope: ObservableScope,
   ) {
     this.selected$.subscribe((device) => {
@@ -346,26 +344,22 @@ class ControlledAudioOutput
 }
 
 class VideoInput implements MediaDevice<DeviceLabel, SelectedDevice> {
-  public readonly available$ = availableRawDevices$(
+  public readonly available$ = this.scope.behavior(
-    "videoinput",
+    availableRawDevices$("videoinput", this.usingNames$, this.scope).pipe(
-    this.usingNames$,
+      map(buildDeviceMap),
-    this.scope,
+    ),
-  ).pipe(map(buildDeviceMap));
+  );
-
+  public readonly selected$ = this.scope.behavior(
-  public readonly selected$ = selectDevice$(
+    selectDevice$(this.available$, videoInputSetting.value$).pipe(
-    this.available$,
+      map((id) => (id === undefined ? undefined : { id })),
-    videoInputSetting.value$,
+    ),
-  ).pipe(
-    map((id) => (id === undefined ? undefined : { id })),
-    this.scope.state(),
   );
-
   public select(id: string): void {
     videoInputSetting.setValue(id);
   }
 
   public constructor(
-    private readonly usingNames$: Observable<boolean>,
+    private readonly usingNames$: Behavior<boolean>,
     private readonly scope: ObservableScope,
   ) {
     // This also has the purpose of subscribing to the available devices
@@ -393,12 +387,10 @@ export class MediaDevices {
   // you to do to receive device names in lieu of a more explicit permissions
   // API. This flag never resets to false, because once permissions are granted
   // the first time, the user won't be prompted again until reload of the page.
-  private readonly usingNames$ = this.deviceNamesRequest$.pipe(
+  private readonly usingNames$ = this.scope.behavior(
-    map(() => true),
+    this.deviceNamesRequest$.pipe(map(() => true)),
-    startWith(false),
+    false,
-    this.scope.state(),
   );
-
   public readonly audioInput: MediaDevice<
     DeviceLabel,
     SelectedAudioInputDevice

src/state/MediaViewModel.ts

@@ -51,27 +51,20 @@ import { accumulate } from "../utils/observable";
 import { type EncryptionSystem } from "../e2ee/sharedKeyManagement";
 import { E2eeType } from "../e2ee/e2eeType";
 import { type ReactionOption } from "../reactions";
+import { type Behavior } from "./Behavior";
 
 export function observeTrackReference$(
-  participant$: Observable<Participant | undefined>,
+  participant: Participant,
   source: Track.Source,
-): Observable<TrackReferenceOrPlaceholder | undefined> {
+): Observable<TrackReferenceOrPlaceholder> {
-  return participant$.pipe(
+  return observeParticipantMedia(participant).pipe(
-    switchMap((p) => {
-      if (p) {
-        return observeParticipantMedia(p).pipe(
     map(() => ({
-            participant: p,
+      participant: participant,
-            publication: p.getTrackPublication(source),
+      publication: participant.getTrackPublication(source),
       source,
     })),
     distinctUntilKeyChanged("publication"),
   );
-      } else {
-        return of(undefined);
-      }
-    }),
-  );
 }
 
 export function observeRtpStreamStats$(
@@ -82,7 +75,7 @@ export function observeRtpStreamStats$(
   RTCInboundRtpStreamStats | RTCOutboundRtpStreamStats | undefined
 > {
   return combineLatest([
-    observeTrackReference$(of(participant), source),
+    observeTrackReference$(participant, source),
     interval(1000).pipe(startWith(0)),
   ]).pipe(
     switchMap(async ([trackReference]) => {
@@ -223,19 +216,31 @@ abstract class BaseMediaViewModel extends ViewModel {
   /**
    * The LiveKit video track for this media.
    */
-  public readonly video$: Observable<TrackReferenceOrPlaceholder | undefined>;
+  public readonly video$: Behavior<TrackReferenceOrPlaceholder | undefined>;
   /**
    * Whether there should be a warning that this media is unencrypted.
    */
-  public readonly unencryptedWarning$: Observable<boolean>;
+  public readonly unencryptedWarning$: Behavior<boolean>;
 
-  public readonly encryptionStatus$: Observable<EncryptionStatus>;
+  public readonly encryptionStatus$: Behavior<EncryptionStatus>;
 
   /**
    * Whether this media corresponds to the local participant.
    */
   public abstract readonly local: boolean;
 
+  private observeTrackReference$(
+    source: Track.Source,
+  ): Behavior<TrackReferenceOrPlaceholder | undefined> {
+    return this.scope.behavior(
+      this.participant$.pipe(
+        switchMap((p) =>
+          p === undefined ? of(undefined) : observeTrackReference$(p, source),
+        ),
+      ),
+    );
+  }
+
   public constructor(
     /**
      * An opaque identifier for this media.
@@ -257,24 +262,25 @@ abstract class BaseMediaViewModel extends ViewModel {
     audioSource: AudioSource,
     videoSource: VideoSource,
     livekitRoom: LivekitRoom,
-    public readonly displayname$: Observable<string>,
+    public readonly displayName$: Behavior<string>,
   ) {
     super();
-    const audio$ = observeTrackReference$(participant$, audioSource).pipe(
+
-      this.scope.state(),
+    const audio$ = this.observeTrackReference$(audioSource);
-    );
+    this.video$ = this.observeTrackReference$(videoSource);
-    this.video$ = observeTrackReference$(participant$, videoSource).pipe(
+
-      this.scope.state(),
+    this.unencryptedWarning$ = this.scope.behavior(
-    );
+      combineLatest(
-    this.unencryptedWarning$ = combineLatest(
         [audio$, this.video$],
         (a, v) =>
           encryptionSystem.kind !== E2eeType.NONE &&
           (a?.publication?.isEncrypted === false ||
             v?.publication?.isEncrypted === false),
-    ).pipe(this.scope.state());
+      ),
+    );
 
-    this.encryptionStatus$ = this.participant$.pipe(
+    this.encryptionStatus$ = this.scope.behavior(
+      this.participant$.pipe(
         switchMap((participant): Observable<EncryptionStatus> => {
           if (!participant) {
             return of(EncryptionStatus.Connecting);
@@ -334,7 +340,7 @@ abstract class BaseMediaViewModel extends ViewModel {
             );
           }
         }),
-      this.scope.state(),
+      ),
     );
   }
 }
@@ -354,31 +360,33 @@ abstract class BaseUserMediaViewModel extends BaseMediaViewModel {
   /**
    * Whether the participant is speaking.
    */
-  public readonly speaking$ = this.participant$.pipe(
+  public readonly speaking$ = this.scope.behavior(
+    this.participant$.pipe(
       switchMap((p) =>
         p
-        ? observeParticipantEvents(p, ParticipantEvent.IsSpeakingChanged).pipe(
+          ? observeParticipantEvents(
-            map((p) => p.isSpeaking),
+              p,
-          )
+              ParticipantEvent.IsSpeakingChanged,
+            ).pipe(map((p) => p.isSpeaking))
           : of(false),
       ),
-    this.scope.state(),
+    ),
   );
 
   /**
    * Whether this participant is sending audio (i.e. is unmuted on their side).
    */
-  public readonly audioEnabled$: Observable<boolean>;
+  public readonly audioEnabled$: Behavior<boolean>;
   /**
    * Whether this participant is sending video.
    */
-  public readonly videoEnabled$: Observable<boolean>;
+  public readonly videoEnabled$: Behavior<boolean>;
 
   private readonly _cropVideo$ = new BehaviorSubject(true);
   /**
    * Whether the tile video should be contained inside the tile or be cropped to fit.
    */
-  public readonly cropVideo$: Observable<boolean> = this._cropVideo$;
+  public readonly cropVideo$: Behavior<boolean> = this._cropVideo$;
 
   public constructor(
     id: string,
@@ -386,9 +394,9 @@ abstract class BaseUserMediaViewModel extends BaseMediaViewModel {
     participant$: Observable<LocalParticipant | RemoteParticipant | undefined>,
     encryptionSystem: EncryptionSystem,
     livekitRoom: LivekitRoom,
-    displayname$: Observable<string>,
+    displayName$: Behavior<string>,
-    public readonly handRaised$: Observable<Date | null>,
+    public readonly handRaised$: Behavior<Date | null>,
-    public readonly reaction$: Observable<ReactionOption | null>,
+    public readonly reaction$: Behavior<ReactionOption | null>,
   ) {
     super(
       id,
@@ -398,18 +406,19 @@ abstract class BaseUserMediaViewModel extends BaseMediaViewModel {
       Track.Source.Microphone,
       Track.Source.Camera,
       livekitRoom,
-      displayname$,
+      displayName$,
     );
 
-    const media$ = participant$.pipe(
+    const media$ = this.scope.behavior(
+      participant$.pipe(
         switchMap((p) => (p && observeParticipantMedia(p)) ?? of(undefined)),
-      this.scope.state(),
+      ),
     );
-    this.audioEnabled$ = media$.pipe(
+    this.audioEnabled$ = this.scope.behavior(
-      map((m) => m?.microphoneTrack?.isMuted === false),
+      media$.pipe(map((m) => m?.microphoneTrack?.isMuted === false)),
     );
-    this.videoEnabled$ = media$.pipe(
+    this.videoEnabled$ = this.scope.behavior(
-      map((m) => m?.cameraTrack?.isMuted === false),
+      media$.pipe(map((m) => m?.cameraTrack?.isMuted === false)),
     );
   }
 
@@ -436,7 +445,8 @@ export class LocalUserMediaViewModel extends BaseUserMediaViewModel {
   /**
    * Whether the video should be mirrored.
    */
-  public readonly mirror$ = this.video$.pipe(
+  public readonly mirror$ = this.scope.behavior(
+    this.video$.pipe(
       switchMap((v) => {
         const track = v?.publication?.track;
         if (!(track instanceof LocalTrack)) return of(false);
@@ -447,7 +457,7 @@ export class LocalUserMediaViewModel extends BaseUserMediaViewModel {
           map(() => facingModeFromLocalTrack(track).facingMode === "user"),
         );
       }),
-    this.scope.state(),
+    ),
   );
 
   /**
@@ -460,12 +470,12 @@ export class LocalUserMediaViewModel extends BaseUserMediaViewModel {
   public constructor(
     id: string,
     member: RoomMember | undefined,
-    participant$: Observable<LocalParticipant | undefined>,
+    participant$: Behavior<LocalParticipant | undefined>,
     encryptionSystem: EncryptionSystem,
     livekitRoom: LivekitRoom,
-    displayname$: Observable<string>,
+    displayName$: Behavior<string>,
-    handRaised$: Observable<Date | null>,
+    handRaised$: Behavior<Date | null>,
-    reaction$: Observable<ReactionOption | null>,
+    reaction$: Behavior<ReactionOption | null>,
   ) {
     super(
       id,
@@ -473,7 +483,7 @@ export class LocalUserMediaViewModel extends BaseUserMediaViewModel {
       participant$,
       encryptionSystem,
       livekitRoom,
-      displayname$,
+      displayName$,
       handRaised$,
       reaction$,
     );
@@ -512,7 +522,8 @@ export class RemoteUserMediaViewModel extends BaseUserMediaViewModel {
    * The volume to which this participant's audio is set, as a scalar
    * multiplier.
    */
-  public readonly localVolume$: Observable<number> = merge(
+  public readonly localVolume$ = this.scope.behavior<number>(
+    merge(
       this.locallyMutedToggle$.pipe(map(() => "toggle mute" as const)),
       this.localVolumeAdjustment$,
       this.localVolumeCommit$.pipe(map(() => "commit" as const)),
@@ -539,15 +550,14 @@ export class RemoteUserMediaViewModel extends BaseUserMediaViewModel {
         }
       }),
       map(({ volume }) => volume),
-    this.scope.state(),
+    ),
   );
 
   /**
    * Whether this participant's audio is disabled.
    */
-  public readonly locallyMuted$: Observable<boolean> = this.localVolume$.pipe(
+  public readonly locallyMuted$ = this.scope.behavior<boolean>(
-    map((volume) => volume === 0),
+    this.localVolume$.pipe(map((volume) => volume === 0)),
-    this.scope.state(),
   );
 
   public constructor(
@@ -556,9 +566,9 @@ export class RemoteUserMediaViewModel extends BaseUserMediaViewModel {
     participant$: Observable<RemoteParticipant | undefined>,
     encryptionSystem: EncryptionSystem,
     livekitRoom: LivekitRoom,
-    displayname$: Observable<string>,
+    displayname$: Behavior<string>,
-    handRaised$: Observable<Date | null>,
+    handRaised$: Behavior<Date | null>,
-    reaction$: Observable<ReactionOption | null>,
+    reaction$: Behavior<ReactionOption | null>,
   ) {
     super(
       id,
@@ -621,7 +631,7 @@ export class ScreenShareViewModel extends BaseMediaViewModel {
     participant$: Observable<LocalParticipant | RemoteParticipant>,
     encryptionSystem: EncryptionSystem,
     livekitRoom: LivekitRoom,
-    displayname$: Observable<string>,
+    displayname$: Behavior<string>,
     public readonly local: boolean,
   ) {
     super(

src/state/MuteAllAudioModel.test.ts

@@ -26,11 +26,9 @@ test("muteAllAudio$", () => {
 
   muteAllAudio.unsubscribe();
 
-  expect(valueMock).toHaveBeenCalledTimes(6);
+  expect(valueMock).toHaveBeenCalledTimes(4);
   expect(valueMock).toHaveBeenNthCalledWith(1, false); // startWith([false, muteAllAudioSetting.getValue()]);
   expect(valueMock).toHaveBeenNthCalledWith(2, true); // setAudioEnabled$.next(false);
   expect(valueMock).toHaveBeenNthCalledWith(3, false); // setAudioEnabled$.next(true);
-  expect(valueMock).toHaveBeenNthCalledWith(4, false); // muteAllAudioSetting.setValue(false);
+  expect(valueMock).toHaveBeenNthCalledWith(4, true); // muteAllAudioSetting.setValue(true);
-  expect(valueMock).toHaveBeenNthCalledWith(5, true); // muteAllAudioSetting.setValue(true);
-  expect(valueMock).toHaveBeenNthCalledWith(6, true); // setAudioEnabled$.next(false);
 });

src/state/MuteAllAudioModel.ts

@@ -9,11 +9,14 @@ import { combineLatest, startWith } from "rxjs";
 
 import { setAudioEnabled$ } from "../controls";
 import { muteAllAudio as muteAllAudioSetting } from "../settings/settings";
+import { globalScope } from "./ObservableScope";
 
 /**
  * This can transition into sth more complete: `GroupCallViewModel.ts`
  */
-export const muteAllAudio$ = combineLatest(
+export const muteAllAudio$ = globalScope.behavior(
+  combineLatest(
     [setAudioEnabled$.pipe(startWith(true)), muteAllAudioSetting.value$],
     (outputEnabled, settingsMute) => !outputEnabled || settingsMute,
+  ),
 );

src/state/ObservableScope.ts

@@ -6,15 +6,19 @@ Please see LICENSE in the repository root for full details.
 */
 
 import {
+  BehaviorSubject,
   distinctUntilChanged,
   type Observable,
-  shareReplay,
   Subject,
   takeUntil,
 } from "rxjs";
 
+import { type Behavior } from "./Behavior";
+
 type MonoTypeOperator = <T>(o: Observable<T>) => Observable<T>;
 
+const nothing = Symbol("nothing");
+
 /**
  * A scope which limits the execution lifetime of its bound Observables.
  */
@@ -31,20 +35,31 @@ export class ObservableScope {
     return this.bindImpl;
   }
 
-  private readonly stateImpl: MonoTypeOperator = (o$) =>
-    o$.pipe(
-      this.bind(),
-      distinctUntilChanged(),
-      shareReplay({ bufferSize: 1, refCount: false }),
-    );
-
   /**
-   * Transforms an Observable into a hot state Observable which replays its
+   * Converts an Observable to a Behavior. If no initial value is specified, the
-   * latest value upon subscription, skips updates with identical values, and
+   * Observable must synchronously emit an initial value.
-   * is bound to this scope.
    */
-  public state(): MonoTypeOperator {
+  public behavior<T>(
-    return this.stateImpl;
+    setValue$: Observable<T>,
+    initialValue: T | typeof nothing = nothing,
+  ): Behavior<T> {
+    const subject$ = new BehaviorSubject(initialValue);
+    // Push values from the Observable into the BehaviorSubject.
+    // BehaviorSubjects have an undesirable feature where if you call 'complete',
+    // they will no longer re-emit their current value upon subscription. We want
+    // to support Observables that complete (for example `of({})`), so we have to
+    // take care to not propagate the completion event.
+    setValue$.pipe(this.bind(), distinctUntilChanged()).subscribe({
+      next(value) {
+        subject$.next(value);
+      },
+      error(err: unknown) {
+        subject$.error(err);
+      },
+    });
+    if (subject$.value === nothing)
+      throw new Error("Behavior failed to synchronously emit an initial value");
+    return subject$ as Behavior<T>;
   }
 
   /**
@@ -55,3 +70,8 @@ export class ObservableScope {
     this.ended$.complete();
   }
 }
+
+/**
+ * The global scope, a scope which never ends.
+ */
+export const globalScope = new ObservableScope();

src/state/TileViewModel.ts

@@ -5,10 +5,9 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 Please see LICENSE in the repository root for full details.
 */
 
-import { type Observable } from "rxjs";
-
 import { ViewModel } from "./ViewModel";
 import { type MediaViewModel, type UserMediaViewModel } from "./MediaViewModel";
+import { type Behavior } from "./Behavior";
 
 let nextId = 0;
 function createId(): string {
@@ -18,15 +17,15 @@ function createId(): string {
 export class GridTileViewModel extends ViewModel {
   public readonly id = createId();
 
-  public constructor(public readonly media$: Observable<UserMediaViewModel>) {
+  public constructor(public readonly media$: Behavior<UserMediaViewModel>) {
     super();
   }
 }
 
 export class SpotlightTileViewModel extends ViewModel {
   public constructor(
-    public readonly media$: Observable<MediaViewModel[]>,
+    public readonly media$: Behavior<MediaViewModel[]>,
-    public readonly maximised$: Observable<boolean>,
+    public readonly maximised$: Behavior<boolean>,
   ) {
     super();
   }

src/tile/GridTile.test.tsx

@@ -9,7 +9,6 @@ import { type RemoteTrackPublication } from "livekit-client";
 import { test, expect } from "vitest";
 import { render, screen } from "@testing-library/react";
 import { axe } from "vitest-axe";
-import { of } from "rxjs";
 import { type MatrixRTCSession } from "matrix-js-sdk/lib/matrixrtc";
 
 import { GridTile } from "./GridTile";
@@ -17,6 +16,7 @@ import { mockRtcMembership, withRemoteMedia } from "../utils/test";
 import { GridTileViewModel } from "../state/TileViewModel";
 import { ReactionsSenderProvider } from "../reactions/useReactionsSender";
 import type { CallViewModel } from "../state/CallViewModel";
+import { constant } from "../state/Behavior";
 
 global.IntersectionObserver = class MockIntersectionObserver {
   public observe(): void {}
@@ -53,13 +53,13 @@ test("GridTile is accessible", async () => {
         memberships: [],
       } as unknown as MatrixRTCSession;
       const cVm = {
-        reactions$: of({}),
+        reactions$: constant({}),
-        handsRaised$: of({}),
+        handsRaised$: constant({}),
       } as Partial<CallViewModel> as CallViewModel;
       const { container } = render(
         <ReactionsSenderProvider vm={cVm} rtcSession={fakeRtcSession}>
           <GridTile
-            vm={new GridTileViewModel(of(vm))}
+            vm={new GridTileViewModel(constant(vm))}
             onOpenProfile={() => {}}
             targetWidth={300}
             targetHeight={200}

src/tile/GridTile.tsx

@@ -35,7 +35,7 @@ import {
   ToggleMenuItem,
   Menu,
 } from "@vector-im/compound-web";
-import { useObservableEagerState, useObservableState } from "observable-hooks";
+import { useObservableEagerState } from "observable-hooks";
 
 import styles from "./GridTile.module.css";
 import {
@@ -49,6 +49,7 @@ import { useLatest } from "../useLatest";
 import { type GridTileViewModel } from "../state/TileViewModel";
 import { useMergedRefs } from "../useMergedRefs";
 import { useReactionsSender } from "../reactions/useReactionsSender";
+import { useBehavior } from "../useBehavior";
 
 interface TileProps {
   ref?: Ref<HTMLDivElement>;
@@ -81,19 +82,19 @@ const UserMediaTile: FC<UserMediaTileProps> = ({
 }) => {
   const { toggleRaisedHand } = useReactionsSender();
   const { t } = useTranslation();
-  const video = useObservableEagerState(vm.video$);
+  const video = useBehavior(vm.video$);
-  const unencryptedWarning = useObservableEagerState(vm.unencryptedWarning$);
+  const unencryptedWarning = useBehavior(vm.unencryptedWarning$);
-  const encryptionStatus = useObservableEagerState(vm.encryptionStatus$);
+  const encryptionStatus = useBehavior(vm.encryptionStatus$);
   const audioStreamStats = useObservableEagerState<
     RTCInboundRtpStreamStats | RTCOutboundRtpStreamStats | undefined
   >(vm.audioStreamStats$);
   const videoStreamStats = useObservableEagerState<
     RTCInboundRtpStreamStats | RTCOutboundRtpStreamStats | undefined
   >(vm.videoStreamStats$);
-  const audioEnabled = useObservableEagerState(vm.audioEnabled$);
+  const audioEnabled = useBehavior(vm.audioEnabled$);
-  const videoEnabled = useObservableEagerState(vm.videoEnabled$);
+  const videoEnabled = useBehavior(vm.videoEnabled$);
-  const speaking = useObservableEagerState(vm.speaking$);
+  const speaking = useBehavior(vm.speaking$);
-  const cropVideo = useObservableEagerState(vm.cropVideo$);
+  const cropVideo = useBehavior(vm.cropVideo$);
   const onSelectFitContain = useCallback(
     (e: Event) => {
       e.preventDefault();
@@ -101,8 +102,8 @@ const UserMediaTile: FC<UserMediaTileProps> = ({
     },
     [vm],
   );
-  const handRaised = useObservableState(vm.handRaised$);
+  const handRaised = useBehavior(vm.handRaised$);
-  const reaction = useObservableState(vm.reaction$);
+  const reaction = useBehavior(vm.reaction$);
 
   const AudioIcon = locallyMuted
     ? VolumeOffSolidIcon
@@ -205,8 +206,8 @@ const LocalUserMediaTile: FC<LocalUserMediaTileProps> = ({
   ...props
 }) => {
   const { t } = useTranslation();
-  const mirror = useObservableEagerState(vm.mirror$);
+  const mirror = useBehavior(vm.mirror$);
-  const alwaysShow = useObservableEagerState(vm.alwaysShow$);
+  const alwaysShow = useBehavior(vm.alwaysShow$);
   const latestAlwaysShow = useLatest(alwaysShow);
   const onSelectAlwaysShow = useCallback(
     (e: Event) => {
@@ -256,8 +257,8 @@ const RemoteUserMediaTile: FC<RemoteUserMediaTileProps> = ({
   ...props
 }) => {
   const { t } = useTranslation();
-  const locallyMuted = useObservableEagerState(vm.locallyMuted$);
+  const locallyMuted = useBehavior(vm.locallyMuted$);
-  const localVolume = useObservableEagerState(vm.localVolume$);
+  const localVolume = useBehavior(vm.localVolume$);
   const onSelectMute = useCallback(
     (e: Event) => {
       e.preventDefault();
@@ -328,8 +329,8 @@ export const GridTile: FC<GridTileProps> = ({
 }) => {
   const ourRef = useRef<HTMLDivElement | null>(null);
   const ref = useMergedRefs(ourRef, theirRef);
-  const media = useObservableEagerState(vm.media$);
+  const media = useBehavior(vm.media$);
-  const displayName = useObservableEagerState(media.displayname$);
+  const displayName = useBehavior(media.displayName$);
 
   if (media instanceof LocalUserMediaViewModel) {
     return (

src/tile/SpotlightTile.test.tsx

@@ -9,7 +9,6 @@ import { test, expect, vi } from "vitest";
 import { isInaccessible, render, screen } from "@testing-library/react";
 import { axe } from "vitest-axe";
 import userEvent from "@testing-library/user-event";
-import { of } from "rxjs";
 
 import { SpotlightTile } from "./SpotlightTile";
 import {
@@ -18,6 +17,7 @@ import {
   withRemoteMedia,
 } from "../utils/test";
 import { SpotlightTileViewModel } from "../state/TileViewModel";
+import { constant } from "../state/Behavior";
 
 global.IntersectionObserver = class MockIntersectionObserver {
   public observe(): void {}
@@ -44,7 +44,12 @@ test("SpotlightTile is accessible", async () => {
           const toggleExpanded = vi.fn();
           const { container } = render(
             <SpotlightTile
-              vm={new SpotlightTileViewModel(of([vm1, vm2]), of(false))}
+              vm={
+                new SpotlightTileViewModel(
+                  constant([vm1, vm2]),
+                  constant(false),
+                )
+              }
               targetWidth={300}
               targetHeight={200}
               expanded={false}

src/tile/SpotlightTile.tsx

@@ -23,7 +23,7 @@ import {
 } from "@vector-im/compound-design-tokens/assets/web/icons";
 import { animated } from "@react-spring/web";
 import { type Observable, map } from "rxjs";
-import { useObservableEagerState, useObservableRef } from "observable-hooks";
+import { useObservableRef } from "observable-hooks";
 import { useTranslation } from "react-i18next";
 import classNames from "classnames";
 import { type TrackReferenceOrPlaceholder } from "@livekit/components-core";
@@ -43,6 +43,7 @@ import { useMergedRefs } from "../useMergedRefs";
 import { useReactiveState } from "../useReactiveState";
 import { useLatest } from "../useLatest";
 import { type SpotlightTileViewModel } from "../state/TileViewModel";
+import { useBehavior } from "../useBehavior";
 
 interface SpotlightItemBaseProps {
   ref?: Ref<HTMLDivElement>;
@@ -73,7 +74,7 @@ const SpotlightLocalUserMediaItem: FC<SpotlightLocalUserMediaItemProps> = ({
   vm,
   ...props
 }) => {
-  const mirror = useObservableEagerState(vm.mirror$);
+  const mirror = useBehavior(vm.mirror$);
   return <MediaView mirror={mirror} {...props} />;
 };
 
@@ -87,8 +88,8 @@ const SpotlightUserMediaItem: FC<SpotlightUserMediaItemProps> = ({
   vm,
   ...props
 }) => {
-  const videoEnabled = useObservableEagerState(vm.videoEnabled$);
+  const videoEnabled = useBehavior(vm.videoEnabled$);
-  const cropVideo = useObservableEagerState(vm.cropVideo$);
+  const cropVideo = useBehavior(vm.cropVideo$);
 
   const baseProps: SpotlightUserMediaItemBaseProps &
     RefAttributes<HTMLDivElement> = {
@@ -130,10 +131,10 @@ const SpotlightItem: FC<SpotlightItemProps> = ({
 }) => {
   const ourRef = useRef<HTMLDivElement | null>(null);
   const ref = useMergedRefs(ourRef, theirRef);
-  const displayName = useObservableEagerState(vm.displayname$);
+  const displayName = useBehavior(vm.displayName$);
-  const video = useObservableEagerState(vm.video$);
+  const video = useBehavior(vm.video$);
-  const unencryptedWarning = useObservableEagerState(vm.unencryptedWarning$);
+  const unencryptedWarning = useBehavior(vm.unencryptedWarning$);
-  const encryptionStatus = useObservableEagerState(vm.encryptionStatus$);
+  const encryptionStatus = useBehavior(vm.encryptionStatus$);
 
   // Hook this item up to the intersection observer
   useEffect(() => {
@@ -200,8 +201,8 @@ export const SpotlightTile: FC<Props> = ({
   const { t } = useTranslation();
   const [ourRef, root$] = useObservableRef<HTMLDivElement | null>(null);
   const ref = useMergedRefs(ourRef, theirRef);
-  const maximised = useObservableEagerState(vm.maximised$);
+  const maximised = useBehavior(vm.maximised$);
-  const media = useObservableEagerState(vm.media$);
+  const media = useBehavior(vm.media$);
   const [visibleId, setVisibleId] = useState<string | undefined>(media[0]?.id);
   const latestMedia = useLatest(media);
   const latestVisibleId = useLatest(visibleId);

src/useAudioContext.test.tsx

@@ -10,12 +10,12 @@ import { type FC } from "react";
 import { render } from "@testing-library/react";
 import userEvent, { type UserEvent } from "@testing-library/user-event";
 import { BrowserRouter } from "react-router-dom";
-import { of } from "rxjs";
 
 import { MediaDevicesContext } from "./MediaDevicesContext";
 import { useAudioContext } from "./useAudioContext";
 import { soundEffectVolume as soundEffectVolumeSetting } from "./settings/settings";
 import { mockMediaDevices } from "./utils/test";
+import { constant } from "./state/Behavior";
 
 const staticSounds = Promise.resolve({
   aSound: new ArrayBuffer(0),
@@ -128,8 +128,8 @@ test("will use the correct device", () => {
     <MediaDevicesContext
       value={mockMediaDevices({
         audioOutput: {
-          available$: of(new Map<never, never>()),
+          available$: constant(new Map<never, never>()),
-          selected$: of({ id: "chosen-device", virtualEarpiece: false }),
+          selected$: constant({ id: "chosen-device", virtualEarpiece: false }),
           select: () => {},
         },
       })}
@@ -161,8 +161,8 @@ test("will use the pan if earpiece is selected", async () => {
     <MediaDevicesContext
       value={mockMediaDevices({
         audioOutput: {
-          available$: of(new Map<never, never>()),
+          available$: constant(new Map<never, never>()),
-          selected$: of({ id: "chosen-device", virtualEarpiece: true }),
+          selected$: constant({ id: "chosen-device", virtualEarpiece: true }),
           select: () => {},
         },
       })}

src/useBehavior.ts

@@ -0,0 +1,25 @@
+/*
+Copyright 2025 New Vector Ltd.
+
+SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
+Please see LICENSE in the repository root for full details.
+*/
+
+import { useCallback, useSyncExternalStore } from "react";
+
+import { type Behavior } from "./state/Behavior";
+
+/**
+ * React hook which reactively reads the value of a behavior.
+ */
+export function useBehavior<T>(behavior: Behavior<T>): T {
+  const subscribe = useCallback(
+    (onChange: () => void) => {
+      const s = behavior.subscribe(onChange);
+      return (): void => s.unsubscribe();
+    },
+    [behavior],
+  );
+  const getValue = useCallback(() => behavior.value, [behavior]);
+  return useSyncExternalStore(subscribe, getValue);
+}

src/utils/test.ts

@@ -47,6 +47,8 @@ import {
 } from "../config/ConfigOptions";
 import { Config } from "../config/Config";
 import { type MediaDevices } from "../state/MediaDevices";
+import { type Behavior, constant } from "../state/Behavior";
+import { ObservableScope } from "../state/ObservableScope";
 
 export function withFakeTimers(continuation: () => void): void {
   vi.useFakeTimers();
@@ -67,6 +69,11 @@ export interface OurRunHelpers extends RunHelpers {
    * diagram.
    */
   schedule: (marbles: string, actions: Record<string, () => void>) => void;
+  behavior<T = string>(
+    marbles: string,
+    values?: { [marble: string]: T },
+    error?: unknown,
+  ): Behavior<T>;
 }
 
 interface TestRunnerGlobal {
@@ -82,6 +89,7 @@ export function withTestScheduler(
   const scheduler = new TestScheduler((actual, expected) => {
     expect(actual).deep.equals(expected);
   });
+  const scope = new ObservableScope();
   // we set the test scheduler as a global so that you can watch it in a debugger
   // and get the frame number. e.g. `rxjsTestScheduler?.now()`
   (global as unknown as TestRunnerGlobal).rxjsTestScheduler = scheduler;
@@ -98,8 +106,36 @@ export function withTestScheduler(
         // Run the actions and verify that none of them error
         helpers.expectObservable(actionsObservable$).toBe(marbles, results);
       },
+      behavior<T>(
+        marbles: string,
+        values?: { [marble: string]: T },
+        error?: unknown,
+      ) {
+        // Generate a hot Observable with helpers.hot and use it as a Behavior.
+        // To do this, we need to ensure that the initial value emits
+        // synchronously upon subscription. The issue is that helpers.hot emits
+        // frame 0 of the marble diagram *asynchronously*, only once we return
+        // from the continuation, so we need to splice out the initial marble
+        // and turn it into a proper initial value.
+        const initialMarbleIndex = marbles.search(/[^ ]/);
+        if (initialMarbleIndex === -1)
+          throw new Error("Behavior must have an initial value");
+        const initialMarble = marbles[initialMarbleIndex];
+        const initialValue =
+          values === undefined ? (initialMarble as T) : values[initialMarble];
+        // The remainder of the marble diagram should start on frame 1
+        return scope.behavior(
+          helpers.hot(
+            `-${marbles.slice(initialMarbleIndex + 1)}`,
+            values,
+            error,
+          ),
+          initialValue,
+        );
+      },
     }),
   );
+  scope.end();
 }
 
 interface EmitterMock<T> {
@@ -211,14 +247,14 @@ export async function withLocalMedia(
   const vm = new LocalUserMediaViewModel(
     "local",
     mockMatrixRoomMember(localRtcMember, roomMember),
-    of(localParticipant),
+    constant(localParticipant),
     {
       kind: E2eeType.PER_PARTICIPANT,
     },
     mockLivekitRoom({ localParticipant }),
-    of(roomMember.rawDisplayName ?? "nodisplayname"),
+    constant(roomMember.rawDisplayName ?? "nodisplayname"),
-    of(null),
+    constant(null),
-    of(null),
+    constant(null),
   );
   try {
     await continuation(vm);
@@ -255,9 +291,9 @@ export async function withRemoteMedia(
       kind: E2eeType.PER_PARTICIPANT,
     },
     mockLivekitRoom({}, { remoteParticipants$: of([remoteParticipant]) }),
-    of(roomMember.rawDisplayName ?? "nodisplayname"),
+    constant(roomMember.rawDisplayName ?? "nodisplayname"),
-    of(null),
+    constant(null),
-    of(null),
+    constant(null),
   );
   try {
     await continuation(vm);
@@ -299,7 +335,7 @@ export class MockRTCSession extends TypedEventEmitter<
   }
 
   public withMemberships(
-    rtcMembers$: Observable<Partial<CallMembership>[]>,
+    rtcMembers$: Behavior<Partial<CallMembership>[]>,
   ): MockRTCSession {
     rtcMembers$.subscribe((m) => {
       const old = this.memberships;