element-call/src/lotus/lotusMutedSpeech.ts

/*
Copyright 2026 Lotus Guild

SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
Please see LICENSE in the repository root for full details.
*/

import {
  type LocalTrackPublication,
  type Room as LivekitRoom,
  RoomEvent,
  Track,
} from "livekit-client";
import { Observable, distinctUntilChanged, switchMap } from "rxjs";

import { type CallViewModel } from "../state/CallViewModel/CallViewModel";

/**
 * [lotus #37] "Talking while muted" detection for the LOCAL participant.
 *
 * While the mic is muted LiveKit keeps the capture alive and merely disables
 * the published MediaStreamTrack, so VAD/`speaking` is false and nothing
 * tells the user they are talking into a muted mic. This taps a CLONE of the
 * published track (the post-processor track when the in-source denoiser is
 * active, so keyboard noise doesn't count), samples RMS at ~10 Hz and emits a
 * debounced boolean. Zero cost when unmuted (tap torn down), local-only —
 * the flag rides `io.lotus.call_state` to the host and never reaches other
 * participants.
 */

export const MUTED_SPEECH_RMS = 0.015; // ≈ −36 dBFS; normal speech into a headset is 0.05–0.3
const SAMPLE_MS = 100;
const ON_SAMPLES = 3; // 300 ms of voice before we say "talking"
const OFF_SAMPLES = 8; // 800 ms of quiet before we drop it

/** Pure hysteresis gate over successive RMS samples (unit-tested). */
export class MutedSpeechGate {
  private above = 0;

  private below = 0;

  private on = false;

  public constructor(private readonly threshold = MUTED_SPEECH_RMS) {}

  public push(rms: number): boolean {
    if (rms >= this.threshold) {
      this.above += 1;
      this.below = 0;
      if (!this.on && this.above >= ON_SAMPLES) this.on = true;
    } else {
      this.below += 1;
      this.above = 0;
      if (this.on && this.below >= OFF_SAMPLES) this.on = false;
    }
    return this.on;
  }

  public get value(): boolean {
    return this.on;
  }
}

const mutedMicTrack = (room: LivekitRoom): MediaStreamTrack | null => {
  const pub: LocalTrackPublication | undefined =
    room.localParticipant.getTrackPublication(Track.Source.Microphone);
  const track = pub?.track?.mediaStreamTrack;
  return pub?.isMuted && track && track.readyState === "live" ? track : null;
};

/**
 * Emits while the local mic is muted and voice is detected on it. Emits
 * `false` whenever the mic is unmuted, unpublished or the connection changes.
 */
export function observeSpeakingWhileMuted$(
  vm: CallViewModel,
): Observable<boolean> {
  return vm.allConnections$.pipe(
    switchMap(
      (data) =>
        new Observable<boolean>((subscriber) => {
          const rooms = data.getConnections().map((c) => c.livekitRoom);
          let ctx: AudioContext | null = null;
          let clone: MediaStreamTrack | null = null;
          let timer: ReturnType<typeof setInterval> | undefined;
          let tapped: MediaStreamTrack | null = null;

          const stopTap = (): void => {
            if (timer !== undefined) clearInterval(timer);
            timer = undefined;
            clone?.stop();
            clone = null;
            void ctx?.close().catch(() => undefined);
            ctx = null;
            tapped = null;
            subscriber.next(false);
          };

          const startTap = (source: MediaStreamTrack): void => {
            try {
              clone = source.clone();
              clone.enabled = true; // the source is disabled by the mute — the clone must not be
              ctx = new AudioContext();
              const analyser = ctx.createAnalyser();
              analyser.fftSize = 1024;
              ctx
                .createMediaStreamSource(new MediaStream([clone]))
                .connect(analyser);
              const buf = new Float32Array(analyser.fftSize);
              const gate = new MutedSpeechGate();
              tapped = source;
              timer = setInterval(() => {
                analyser.getFloatTimeDomainData(buf);
                let sum = 0;
                for (let i = 0; i < buf.length; i += 1) sum += buf[i] * buf[i];
                subscriber.next(gate.push(Math.sqrt(sum / buf.length)));
              }, SAMPLE_MS);
            } catch {
              stopTap();
            }
          };

          const reconcile = (): void => {
            const track =
              rooms.map(mutedMicTrack).find((t) => t !== null) ?? null;
            if (track === tapped) return;
            if (tapped) stopTap();
            if (track) startTap(track);
          };

          const events = [
            RoomEvent.TrackMuted,
            RoomEvent.TrackUnmuted,
            RoomEvent.LocalTrackPublished,
            RoomEvent.LocalTrackUnpublished,
            RoomEvent.Reconnected,
            RoomEvent.Disconnected,
          ] as const;
          rooms.forEach((room) =>
            events.forEach((ev) => room.on(ev, reconcile)),
          );
          subscriber.next(false);
          reconcile();
          return () => {
            rooms.forEach((room) =>
              events.forEach((ev) => room.off(ev, reconcile)),
            );
            if (tapped) stopTap();
          };
        }),
    ),
    distinctUntilChanged(),
  );
}