Start every camera track with the pipeline on

A camera track made without the processor lets the room see the camera
until the processor is built. So the processor travels in the publish
request: the lobby opens the camera with it, the join publishes with it,
the room's capture defaults follow the requests made since the join, and
a camera turned off before the processor was chosen is replaced on turning
it on rather than unmuted, so that the new track starts with it.

Ported from spec/background-effects/2-pipeline (#4282): the publish half
of a91dab8f. The replacement lives in the mute switch's track supply,
which the call view model shares with the lobby.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

sdk/src/media-backend/livekit/Publisher.test.ts

@@ -120,6 +120,24 @@ describe("Publisher", () => {
       expect(participant.createTracks).not.toHaveBeenCalled();
     });
 
+    it("creates both with the options of requests made since the join", async () => {
+      const processor: VideoProcessor = {
+        name: "effect",
+        init: vi.fn(),
+        restart: vi.fn(),
+        destroy: vi.fn(),
+      };
+      want(microphone, { source: "camera", deviceId: "front", processor });
+
+      publisher.start();
+      await flushPromises();
+
+      expect(room.options.videoCaptureDefaults).toMatchObject({
+        deviceId: "front",
+        processor: expect.objectContaining({ source: processor }),
+      });
+    });
+
     it("asks for the camera and the microphone in one call", async () => {
       const enableBoth = vi.spyOn(participant, "enableCameraAndMicrophone");
       want(microphone, camera);

sdk/src/media-backend/livekit/Publisher.ts

@@ -120,8 +120,10 @@ export class Publisher {
     // resolves these once the track is published, which may block on the
     // connection; LocalTrackPublished is what tells us a track exists.
     const captureBoth = toCapture.has("microphone") && toCapture.has("camera");
-    if (captureBoth)
+    if (captureBoth) {
+      this.takeCaptureDefaults(requests);
       void this.room.localParticipant.enableCameraAndMicrophone();
+    }
     for (const request of requests)
       if (
         request.source === "screenShare" ||
@@ -133,6 +135,27 @@ export class Publisher {
         });
   }
 
+  /**
+   * Makes the requests the room's capture defaults. The room took its defaults
+   * from the requests at the join, but one made since, such as a camera turned
+   * on before the start, has options of its own: a video processor dropped
+   * here would let the room see the camera before it is applied.
+   */
+  private takeCaptureDefaults(requests: PublishRequest[]): void {
+    const options = this.room.options;
+    for (const request of requests)
+      if (request.source === "microphone")
+        options.audioCaptureDefaults = {
+          ...options.audioCaptureDefaults,
+          ...audioCaptureOptions(request),
+        };
+      else if (request.source === "camera")
+        options.videoCaptureDefaults = {
+          ...options.videoCaptureDefaults,
+          ...videoCaptureOptions(request),
+        };
+  }
+
   private hasCaptured(source: MediaSource): boolean {
     return (
       (source === "microphone" || source === "camera") &&

src/room/InCallView.tsx

@@ -177,7 +177,12 @@ export const ActiveCall: FC<ActiveCallProps> = (props) => {
       publish:
         props.media.requests.length > 0
           ? props.media
-          : initialPublishRequests(props.muteStates, mediaDevices, capture),
+          : initialPublishRequests(
+              props.muteStates,
+              mediaDevices,
+              capture,
+              backgroundEffectsState$.value.processor,
+            ),
     });
 
     const vm = createCallViewModel$(

src/state/CallViewModel/CallViewModel.test.ts

@@ -28,6 +28,8 @@ import { type Layout } from "../layout-types.ts";
 import {
   mockMatrixRoomMember,
   mockMediaDevices,
+  mockMuteStates,
+  flushPromises,
   mockRemoteParticipant,
   withTestScheduler,
   mockRtcMembership,
@@ -62,7 +64,15 @@ import {
 import { MatrixRTCMode } from "../../config/ConfigOptions.ts";
 import { initializeWidget } from "../../widget.ts";
 import { computeUrlParams } from "../../UrlParams.ts";
-import { callViewModelOptionsFromParams } from "./CallViewModel.ts";
+import {
+  callViewModelOptionsFromParams,
+  initialPublishRequests,
+} from "./CallViewModel.ts";
+import { type Handler, MuteState } from "../MuteStates.ts";
+import {
+  type RTCParticipation,
+  type VideoProcessor,
+} from "@element-hq/matrixrtc-sdk";
 
 initializeWidget();
 
@@ -1664,3 +1674,73 @@ describe("callViewModelOptionsFromParams", () => {
     expect(callViewModelOptionsFromParams(params).hideScreensharing).toBe(true);
   });
 });
+
+describe("a camera with a video processor", () => {
+  const withCallViewModel = withCallViewModelInMode(MatrixRTCMode.Matrix_2_0);
+  const processor = { name: "effect" } as VideoProcessor;
+
+  /** The handler the view model gives the camera's mute switch. */
+  const cameraHandler = (setHandler: {
+    mock: { calls: [Handler][] };
+  }): Handler => setHandler.mock.calls[1][0];
+
+  it("starts each camera track with the processor", () => {
+    const muteStates = mockMuteStates();
+    vi.spyOn(muteStates.video.enabled$, "value", "get").mockReturnValue(true);
+    expect(
+      initialPublishRequests(
+        muteStates,
+        mockMediaDevices({}),
+        {},
+        processor,
+      ).find((request) => request.source === "camera"),
+    ).toMatchObject({ processor });
+  });
+
+  it("replaces a camera turned off before the processor was chosen", async () => {
+    const setHandler = vi.spyOn(MuteState.prototype, "setHandler");
+    onTestFinished(() => setHandler.mockRestore());
+    const videoProcessor$ = new BehaviorSubject<VideoProcessor | undefined>(
+      undefined,
+    );
+    let participation: RTCParticipation | undefined;
+    withCallViewModel(
+      {},
+      (_vm, _session, { rtcParticipation }) => {
+        participation = rtcParticipation;
+      },
+      { videoProcessor$ },
+    );
+    // The camera is off, so it is stopped and can't take the processor
+    videoProcessor$.next(processor);
+    await cameraHandler(setHandler)(true);
+    await flushPromises();
+
+    expect(participation!.unpublish).toHaveBeenCalledWith("camera");
+    expect(participation!.publish).toHaveBeenCalledWith(
+      expect.objectContaining({ source: "camera", processor }),
+    );
+  });
+
+  it("turns on a camera that took the processor while running", async () => {
+    const setHandler = vi.spyOn(MuteState.prototype, "setHandler");
+    onTestFinished(() => setHandler.mockRestore());
+    const videoProcessor$ = new BehaviorSubject<VideoProcessor | undefined>(
+      undefined,
+    );
+    let participation: RTCParticipation | undefined;
+    withCallViewModel(
+      { videoEnabled: new Map([[localParticipant, constant(true)]]) },
+      (_vm, _session, { rtcParticipation }) => {
+        participation = rtcParticipation;
+      },
+      { videoProcessor$ },
+    );
+    videoProcessor$.next(processor);
+    await cameraHandler(setHandler)(true);
+    await flushPromises();
+
+    expect(participation!.unpublish).not.toHaveBeenCalled();
+    expect(participation!.publish).not.toHaveBeenCalled();
+  });
+});

src/state/CallViewModel/CallViewModel.ts

@@ -14,6 +14,7 @@ import {
   type EncryptionSystem,
   generateItems,
   type LocalAudioMediaTrack,
+  type LocalMediaTrack,
   type LocalRTCMember,
   type LocalVideoMediaTrack,
   type RTCParticipation,
@@ -140,14 +141,16 @@ export interface CaptureSettings {
 }
 
 /**
- * What to publish for a microphone or camera: the device the user picked and
- * the host's capture settings. The video processor is not in here; the call
- * attaches it to the camera track once that exists.
+ * What to publish for a microphone or camera: the device the user picked, the
+ * host's capture settings and, for the camera, the video processor, so that
+ * the track starts with it rather than letting the room through while it is
+ * built.
  */
 export function publishRequest(
   source: "microphone" | "camera",
   mediaDevices: MediaDevices,
   capture: CaptureSettings,
+  processor?: VideoProcessor,
 ): PublishRequest {
   return source === "microphone"
     ? {
@@ -159,6 +162,7 @@ export function publishRequest(
         source,
         deviceId: mediaDevices.videoInput.selected$.value?.id,
         capture: capture.camera,
+        ...(processor && { processor }),
       };
 }
 
@@ -167,12 +171,13 @@ export function initialPublishRequests(
   muteStates: MuteStates,
   mediaDevices: MediaDevices,
   capture: CaptureSettings,
+  processor?: VideoProcessor,
 ): PublishRequest[] {
   const requests: PublishRequest[] = [];
   if (muteStates.audio.enabled$.value)
     requests.push(publishRequest("microphone", mediaDevices, capture));
   if (muteStates.video.enabled$.value)
-    requests.push(publishRequest("camera", mediaDevices, capture));
+    requests.push(publishRequest("camera", mediaDevices, capture, processor));
   return requests;
 }
 
@@ -508,9 +513,36 @@ export function createCallViewModel$(
     "screenShare",
   ) as Behavior<LocalVideoMediaTrack | undefined>;
 
+  // Camera tracks the processor reached while they ran. A camera turned off is
+  // stopped, and a stopped track can't take one.
+  const tookProcessor = new WeakMap<LocalVideoMediaTrack, VideoProcessor>();
+  const missesProcessor = (track: LocalMediaTrack): boolean => {
+    const processor = videoProcessor$.value;
+    return (
+      processor !== undefined &&
+      tookProcessor.get(track as LocalVideoMediaTrack) !== processor
+    );
+  };
   const supply = (source: "microphone" | "camera"): TrackSupply => ({
     publish: async () =>
-      rtcParticipation.publish(publishRequest(source, mediaDevices, capture)),
+      rtcParticipation.publish(
+        publishRequest(
+          source,
+          mediaDevices,
+          capture,
+          source === "camera" ? videoProcessor$.value : undefined,
+        ),
+      ),
+    // A camera turned off before the processor was chosen is replaced, so
+    // that the new one starts with it
+    replace:
+      source === "camera"
+        ? async (track): Promise<boolean> => {
+            if (!missesProcessor(track)) return false;
+            await rtcParticipation.unpublish("camera");
+            return true;
+          }
+        : undefined,
   });
   followMuteState(
     scope,
@@ -550,7 +582,13 @@ export function createCallViewModel$(
         });
       });
 
-  followProcessor(scope, camera$, videoProcessor$, logger);
+  followProcessor(
+    scope,
+    camera$,
+    videoProcessor$,
+    logger,
+    (camera, processor) => tookProcessor.set(camera, processor),
+  );
 
   const members$ = scope.behavior<(LocalRTCMember | RemoteRTCMember)[]>(
     combineLatest(

src/state/LobbyMediaViewModel.test.ts

@@ -29,7 +29,8 @@ describe("LobbyMediaViewModel", () => {
 
     expect(media.capture).toHaveBeenCalledExactlyOnceWith([
       expect.objectContaining({ source: "microphone", deviceId: "mic" }),
-      expect.objectContaining({ source: "camera", deviceId: "cam" }),
+      // With the processor, so that the camera starts with it
+      expect.objectContaining({ source: "camera", deviceId: "cam", processor }),
     ]);
     const camera = vm.videoTrack$.value!;
     expect(camera.source).toBe("camera");

src/state/LobbyMediaViewModel.ts

@@ -71,7 +71,14 @@ export function createLobbyMediaViewModel(
       if (audio && microphone === undefined && !opening.has("microphone"))
         open.push(publishRequest("microphone", mediaDevices, capture));
       if (video && camera === undefined && !opening.has("camera"))
-        open.push(publishRequest("camera", mediaDevices, capture));
+        open.push(
+          publishRequest(
+            "camera",
+            mediaDevices,
+            capture,
+            videoProcessor$.value,
+          ),
+        );
       if (open.length > 0) {
         for (const { source } of open) opening.add(source);
         media

src/state/localMediaControls.ts

@@ -18,9 +18,14 @@ import { combineLatest } from "rxjs";
 import { type MediaDevice } from "./MediaDevices";
 import { type Handler } from "./MuteStates";
 
-/** How a mute switch gets a track where there is none. */
+/**
+ * How a mute switch gets a track where there is none and, where given,
+ * whether a track it has should be let go and published anew instead of
+ * unmuted: resolves true once it is gone.
+ */
 export interface TrackSupply {
   publish: () => Promise<unknown>;
+  replace?: (track: LocalMediaTrack) => Promise<boolean>;
 }
 
 /**
@@ -46,7 +51,9 @@ export function followMuteState(
   muteState.setHandler(async (enabled) => {
     desired = enabled;
     const track = track$.value;
-    if (track !== undefined) {
+    const replaced =
+      enabled && track !== undefined && (await supply.replace?.(track));
+    if (track !== undefined && !replaced) {
       desired = await track.setEnabled(enabled);
       return desired;
     }
@@ -100,11 +107,15 @@ export function followProcessor(
   camera$: Behavior<LocalVideoMediaTrack | undefined>,
   videoProcessor$: Behavior<VideoProcessor | undefined>,
   logger: Logger,
+  /** Told of each camera a processor reaches while it runs. */
+  applied?: (camera: LocalVideoMediaTrack, processor: VideoProcessor) => void,
 ): void {
   combineLatest([camera$, videoProcessor$])
     .pipe(scope.bind())
     .subscribe(([camera, processor]) => {
-      camera?.setProcessor(processor).catch((e) => {
+      if (camera === undefined) return;
+      if (processor && !camera.muted$.value) applied?.(camera, processor);
+      camera.setProcessor(processor).catch((e) => {
         logger.error("Failed to set the video processor", e);
       });
     });