perf(lotus): call_state field-wise dedupe, 500 ms trailing throttle (max 2/s)

Fixes #20

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PPmy3tPq869XDW4njjVaKA

src/lotus/lotusCallState.test.ts

@@ -0,0 +1,128 @@
+/*
+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 { afterEach, beforeEach, expect, test, vi } from "vitest";
+import { BehaviorSubject, of } from "rxjs";
+
+import { type CallViewModel } from "../state/CallViewModel/CallViewModel";
+import { startLotusCallState } from "./lotusCallState";
+
+const send = vi.fn().mockResolvedValue({});
+
+vi.mock("../widget", () => ({
+  widget: { api: { transport: { send: (...a: unknown[]) => send(...a) } } },
+}));
+
+// `lotusFlag`/`lotusParam` (lotusWidget.ts) read `window.location` directly
+// and memoize per module load, so drive the flag through the URL once, at
+// import time (NOT per-test/in a hook: re-navigating with
+// `window.history.pushState` between tests was observed to corrupt rxjs's
+// shared `asyncScheduler` under `vi.useFakeTimers()`, silently starving a
+// later test's `throttleTime` of any emission).
+window.history.pushState({}, "", "/?lotusCallState=1");
+
+beforeEach(() => {
+  send.mockClear();
+  vi.useFakeTimers();
+});
+
+afterEach(() => {
+  vi.useRealTimers();
+});
+
+interface Member {
+  id: string;
+  userId: string;
+  speaking$: BehaviorSubject<boolean>;
+  audioEnabled$: BehaviorSubject<boolean>;
+  videoEnabled$: BehaviorSubject<boolean>;
+}
+
+function mockMember(id: string, userId: string): Member {
+  return {
+    id,
+    userId,
+    speaking$: new BehaviorSubject(false),
+    audioEnabled$: new BehaviorSubject(true),
+    videoEnabled$: new BehaviorSubject(true),
+  };
+}
+
+function mockVm(members: Member[]): CallViewModel {
+  return { userMedia$: of(members) } as unknown as CallViewModel;
+}
+
+function participantsOf(call: number): unknown[] {
+  return (send.mock.calls[call][1] as { participants: unknown[] }).participants;
+}
+
+test("[lotus #20] rapid speaking toggles in a busy call are capped well under the old 8/sec rate", () => {
+  const members = Array.from({ length: 15 }, (_, i) =>
+    mockMember(`@u${i}:example.org:DEV`, `@u${i}:example.org`),
+  );
+  const stop = startLotusCallState(mockVm(members));
+
+  // Simulate a noisy multi-person conversation: flip `speaking` on every
+  // member every 20ms (50Hz of raw churn) for 2 seconds.
+  for (let t = 0; t < 2000; t += 20) {
+    for (const m of members) m.speaking$.next(!m.speaking$.value);
+    vi.advanceTimersByTime(20);
+  }
+
+  // The old leading+trailing 250ms throttle allowed ~8 sends/sec => up to 16
+  // over 2s. Trailing-only 500ms must cap this to at most 4 (one per window).
+  expect(send.mock.calls.length).toBeLessThanOrEqual(5);
+  expect(send.mock.calls.length).toBeGreaterThan(0);
+
+  // Drain any still-pending trailing-edge throttle action before tearing
+  // down: `rxjs`'s default `asyncScheduler` is a process-wide singleton, and
+  // leaving a scheduled action dangling across a test/timer-implementation
+  // boundary can wedge its queue for every later test in this file.
+  vi.advanceTimersByTime(500);
+  stop();
+});
+
+test("[lotus #20] a mute/camera change and a same-tick no-op speaking flip are deduped field-wise", () => {
+  const alice = mockMember("@alice:example.org:DEV", "@alice:example.org");
+  const stop = startLotusCallState(mockVm([alice]));
+
+  vi.advanceTimersByTime(500);
+  send.mockClear();
+
+  // No actual change: re-emitting the same speaking value must not count as
+  // a change (distinctUntilChanged happens before the throttle).
+  alice.speaking$.next(false);
+  vi.advanceTimersByTime(500);
+  expect(send).not.toHaveBeenCalled();
+
+  alice.audioEnabled$.next(false);
+  vi.advanceTimersByTime(500);
+  expect(send).toHaveBeenCalledTimes(1);
+  expect(participantsOf(0)).toEqual([
+    {
+      id: "@alice:example.org:DEV",
+      userId: "@alice:example.org",
+      speaking: false,
+      audioEnabled: false,
+      videoEnabled: true,
+    },
+  ]);
+
+  stop();
+});
+
+test("does nothing (and sends nothing) once torn down", () => {
+  const alice = mockMember("@alice:example.org:DEV", "@alice:example.org");
+  const stop = startLotusCallState(mockVm([alice]));
+  vi.advanceTimersByTime(500);
+  send.mockClear();
+
+  stop();
+  alice.speaking$.next(true);
+  vi.advanceTimersByTime(1000);
+  expect(send).not.toHaveBeenCalled();
+});

src/lotus/lotusCallState.ts

@@ -27,6 +27,29 @@ interface ParticipantState {
   videoEnabled: boolean;
 }
 
+/**
+ * [lotus #20] Field-wise equality for `ParticipantState[]`, used in place of
+ * `JSON.stringify` comparison: cheaper (no serialisation of every
+ * participant on every emission) and just as correct, since array order here
+ * is stable (it mirrors `members` from `userMedia$`).
+ */
+function participantsEqual(
+  a: ParticipantState[],
+  b: ParticipantState[],
+): boolean {
+  return (
+    a.length === b.length &&
+    a.every(
+      (p, i) =>
+        p.id === b[i].id &&
+        p.userId === b[i].userId &&
+        p.speaking === b[i].speaking &&
+        p.audioEnabled === b[i].audioEnabled &&
+        p.videoEnabled === b[i].videoEnabled,
+    )
+  );
+}
+
 /**
  * Stream per-participant speaking / mute / camera state to the Lotus host
  * (cinny) over the widget API, so the host can drive speaking rings, mute
@@ -73,11 +96,19 @@ export function startLotusCallState(vm: CallViewModel): () => void {
             ),
       ),
       // `speaking` flips rapidly; drop no-op repeats BEFORE throttling so
-      // the throttle window isn't spent re-emitting an unchanged value, then
+      // the throttle window isn't spent re-emitting an unchanged value.
-      // cap the send rate. 250ms is plenty for speaking rings / mute badges
+      // Field-wise (cheaper than JSON.stringify, and correct: array order is
-      // and keeps the request/response widget traffic modest.
+      // stable since it mirrors `members` from userMedia$).
-      distinctUntilChanged((a, b) => JSON.stringify(a) === JSON.stringify(b)),
+      distinctUntilChanged(participantsEqual),
-      throttleTime(250, undefined, { leading: true, trailing: true }),
+      // [lotus #20] `speaking` is the field that flips constantly in an active
+      // conversation; mute/camera toggles are rare and user-intentional and
+      // would ideally stay prompt, but a single combined stream is far
+      // simpler than splitting it, and the leading+trailing 250ms window
+      // previously allowed ~8 sends/sec (each re-serialising every
+      // participant) in a busy call. Trailing-only + a longer window caps
+      // that to 2/sec while still reflecting mute/camera changes within
+      // 500ms.
+      throttleTime(500, undefined, { leading: false, trailing: true }),
     )
     .subscribe((participants) => {
       lotusSendToHost(LotusWidgetActions.CallState, { participants });