Carry speaker assignments into the live transcript engine

Naming a speaker mid-meeting only relabeled the main transcript view through
a JavaScript overlay; the live engine kept emitting unlabeled segments, so
the meeting float and plugin snapshot still showed "Speaker N", and a
listener refresh started a fresh engine that knew nothing about the pick.

The desktop config sync now follows the captured session's newest transcript
and pushes its resolved identity assignments with the participant list. The
engine folds them into segment building the same way the settled render does
(hints first, participant channel defaults after), re-renders the retained
window on an identity update so on-screen segments relabel without waiting
for the next stream response, and seeds a refreshed listener with the same
assignments. Assignment changes never trigger a listener refresh.

Author: ComputelessComputer

apps/desktop/src/services/event-listeners.test.tsx

@@ -106,6 +106,17 @@ vi.mock("~/store/zustand/listener/instance", () => ({
   },
 }));
 
+function findLiveQueryHandlers(sqlFragment: string) {
+  const call = liveQuerySubscribeMock.mock.calls.find(([sql]) =>
+    String(sql).includes(sqlFragment),
+  );
+  expect(call).toBeDefined();
+  return call![2] as {
+    onData: (rows: unknown[]) => void;
+    onError: (error: unknown) => void;
+  };
+}
+
 describe("EventListeners notification events", () => {
   beforeEach(() => {
     cancelAutoStopEndedNotification("session-1");
@@ -275,9 +286,9 @@ describe("EventListeners notification events", () => {
     render(<EventListeners />);
 
     await vi.waitFor(() =>
-      expect(liveQuerySubscribeMock).toHaveBeenCalledTimes(1),
+      expect(liveQuerySubscribeMock).toHaveBeenCalledTimes(2),
     );
-    const handlers = liveQuerySubscribeMock.mock.calls[0]?.[2];
+    const handlers = findLiveQueryHandlers("session_participants");
     handlers.onData([
       {
         session_id: "session-1",
@@ -292,6 +303,148 @@ describe("EventListeners notification events", () => {
       languages: ["ko"],
       participant_human_ids: ["human-remote"],
       self_human_id: "human-self",
+      speaker_assignments: [],
+    });
+  });
+
+  test("live capture config sync waits for the transcript snapshot before pushing", async () => {
+    vi.useFakeTimers();
+    useConfigValuesMock.mockReturnValue({
+      ai_language: "ko",
+      spoken_languages: ["ko"],
+      current_stt_provider: "soniox",
+      current_stt_model: "stt-v4",
+    });
+    // The transcript query answers later than the participant query here.
+    liveQuerySubscribeMock.mockImplementation(
+      async (sql, _params, handlers) => {
+        if (!String(sql).includes("FROM transcripts")) {
+          handlers.onData([]);
+        }
+        return async () => {};
+      },
+    );
+
+    render(<EventListeners />);
+
+    await vi.waitFor(() =>
+      expect(liveQuerySubscribeMock).toHaveBeenCalledTimes(2),
+    );
+    findLiveQueryHandlers("session_participants").onData([
+      {
+        session_id: "session-1",
+        owner_user_id: "human-self",
+        human_id: "human-remote",
+      },
+    ]);
+    await vi.runOnlyPendingTimersAsync();
+
+    expect(updateCaptureConfigMock).not.toHaveBeenCalled();
+
+    findLiveQueryHandlers("FROM transcripts").onData([]);
+    await vi.runOnlyPendingTimersAsync();
+
+    expect(updateCaptureConfigMock).toHaveBeenCalledTimes(1);
+    expect(updateCaptureConfigMock).toHaveBeenCalledWith(
+      expect.objectContaining({
+        session_id: "session-1",
+        speaker_assignments: [],
+      }),
+    );
+  });
+
+  test("live capture config sync pushes the active transcript's speaker assignments", async () => {
+    vi.useFakeTimers();
+    useConfigValuesMock.mockReturnValue({
+      ai_language: "en",
+      spoken_languages: ["en"],
+      current_stt_provider: "soniox",
+      current_stt_model: "stt-v4",
+    });
+
+    render(<EventListeners />);
+
+    await vi.waitFor(() =>
+      expect(liveQuerySubscribeMock).toHaveBeenCalledTimes(2),
+    );
+    const transcriptCall = liveQuerySubscribeMock.mock.calls.find(([sql]) =>
+      String(sql).includes("FROM transcripts"),
+    );
+    expect(transcriptCall?.[1]).toEqual(["session-1"]);
+
+    findLiveQueryHandlers("session_participants").onData([
+      {
+        session_id: "session-1",
+        owner_user_id: "human-self",
+        human_id: "human-artem",
+      },
+      {
+        session_id: "session-1",
+        owner_user_id: "human-self",
+        human_id: "human-guest",
+      },
+    ]);
+    findLiveQueryHandlers("FROM transcripts").onData([
+      {
+        id: "transcript-1",
+        started_at_ms: 1_000,
+        words_json: JSON.stringify([
+          { id: "w1", text: " hello", start_ms: 0, end_ms: 100, channel: 1 },
+          { id: "w2", text: " there", start_ms: 100, end_ms: 200, channel: 1 },
+        ]),
+        speaker_hints_json: JSON.stringify([
+          {
+            id: "w1:provider_speaker_index",
+            word_id: "w1",
+            type: "provider_speaker_index",
+            value: JSON.stringify({ channel: 1, speaker_index: 0 }),
+          },
+          {
+            id: "w1:user_speaker_assignment",
+            word_id: "w1",
+            type: "user_speaker_assignment",
+            value: JSON.stringify({
+              human_id: "human-artem",
+              scope: "speaker",
+              channel: 1,
+              speaker_index: 0,
+            }),
+          },
+          {
+            id: "w2:user_speaker_assignment:segment",
+            word_id: "w2",
+            type: "user_speaker_assignment",
+            value: JSON.stringify({
+              human_id: "human-guest",
+              scope: "segment",
+              word_ids: ["w2"],
+            }),
+          },
+        ]),
+      },
+    ]);
+    await vi.runOnlyPendingTimersAsync();
+
+    expect(updateCaptureConfigMock).toHaveBeenCalledTimes(1);
+    expect(updateCaptureConfigMock).toHaveBeenCalledWith({
+      session_id: "session-1",
+      languages: ["en"],
+      participant_human_ids: ["human-artem", "human-guest"],
+      self_human_id: "human-self",
+      speaker_assignments: [
+        {
+          human_id: "human-artem",
+          scope: {
+            kind: "channel_speaker",
+            channel: "RemoteParty",
+            speaker_index: 0,
+          },
+        },
+        {
+          human_id: "human-guest",
+          scope: { kind: "words", word_ids: ["w2"] },
+        },
+      ],
     });
   });
 

apps/desktop/src/services/event-listeners.tsx

@@ -1,6 +1,10 @@
 import { type UnlistenFn } from "@tauri-apps/api/event";
 
 import { events as notificationEvents } from "@anlg/plugin-notification";
+import type {
+  CaptureConfigUpdate,
+  IdentityAssignment,
+} from "@anlg/plugin-transcription";
 import {
   commands as updaterCommands,
   events as updaterEvents,
@@ -26,13 +30,21 @@ import {
   getLiveTranscriptionConfig,
   getTranscriptionLanguages,
 } from "~/stt/capabilities";
+import { buildRenderTranscriptRequestFromRows } from "~/stt/render-transcript";
 
 type CaptureIdentitySqlRow = {
   session_id: string;
   owner_user_id: string;
   human_id: string | null;
 };
 
+type LiveTranscriptIdentitySqlRow = {
+  id: string;
+  started_at_ms: number | string;
+  words_json: string;
+  speaker_hints_json: string;
+};
+
 const CAPTURE_IDENTITY_SQL = `
   SELECT
     session.id AS session_id,
@@ -48,6 +60,21 @@ const CAPTURE_IDENTITY_SQL = `
   ORDER BY session.id, participant.human_id
 `;
 
+// The capture writes into the session's newest transcript. Speaker hints for
+// live words are only materialized into speaker_hints_json by transcript
+// mutations (assignments, flushes), so this does not need the delta journal.
+const LIVE_TRANSCRIPT_IDENTITY_SQL = `
+  SELECT
+    transcript.id,
+    transcript.started_at_ms,
+    transcript.words_json,
+    transcript.speaker_hints_json
+  FROM transcripts AS transcript
+  WHERE transcript.session_id = ? AND transcript.deleted_at IS NULL
+  ORDER BY transcript.started_at_ms DESC, transcript.created_at DESC
+  LIMIT 1
+`;
+
 const LIVE_CAPTURE_CONFIG_DEBOUNCE_MS = 750;
 
 async function shouldAutoStartNotificationSession(
@@ -138,12 +165,34 @@ function getSessionParticipantHumanIds(
   return participantHumanIds;
 }
 
-function createCaptureConfigSignature(config: {
-  session_id: string;
-  languages: string[];
-  participant_human_ids: string[];
-  self_human_id: string | null;
-}) {
+function getLiveSpeakerAssignments(
+  rows: LiveTranscriptIdentitySqlRow[],
+): IdentityAssignment[] {
+  const row = rows[0];
+  if (!row) {
+    return [];
+  }
+
+  const request = buildRenderTranscriptRequestFromRows([
+    {
+      started_at: Number(row.started_at_ms),
+      words: parseJsonArray(row.words_json),
+      speaker_hints: parseJsonArray(row.speaker_hints_json),
+    },
+  ]);
+  return request?.transcripts[0]?.assignments ?? [];
+}
+
+function parseJsonArray<T>(value: string): T[] {
+  try {
+    const parsed = JSON.parse(value);
+    return Array.isArray(parsed) ? (parsed as T[]) : [];
+  } catch {
+    return [];
+  }
+}
+
+function createCaptureConfigSignature(config: CaptureConfigUpdate) {
   return JSON.stringify(config);
 }
 
@@ -205,8 +254,12 @@ function LiveCaptureConfigSyncReady({
     let lastSignature: string | null = null;
     let rows: CaptureIdentitySqlRow[] = [];
     let hasSnapshot = false;
+    let transcriptRows: LiveTranscriptIdentitySqlRow[] = [];
+    let hasTranscriptSnapshot = false;
+    let transcriptSessionId: string | null = null;
     let cancelled = false;
     let unsubscribeDatabase: (() => Promise<void>) | null = null;
+    let unsubscribeTranscript: (() => Promise<void>) | null = null;
 
     const pushConfig = async () => {
       if (!hasSnapshot) {
@@ -218,6 +271,12 @@ function LiveCaptureConfigSyncReady({
         return;
       }
 
+      // An empty assignment list is not a no-op: the engine drops the names it
+      // holds. Wait for the transcript query's first rows before pushing.
+      if (transcriptSessionId === live.sessionId && !hasTranscriptSnapshot) {
+        return;
+      }
+
       const languages = getLiveConfigLanguages(
         settingsValues.ai_language,
         settingsValues.spoken_languages,
@@ -233,14 +292,18 @@ function LiveCaptureConfigSyncReady({
       }
 
       const session = rows.find((row) => row.session_id === live.sessionId);
-      const nextConfig = {
+      const nextConfig: CaptureConfigUpdate = {
         session_id: live.sessionId,
         languages: liveConfig.languages,
         participant_human_ids: getSessionParticipantHumanIds(
           rows,
           live.sessionId,
         ),
         self_human_id: session?.owner_user_id || null,
+        speaker_assignments:
+          transcriptSessionId === live.sessionId
+            ? getLiveSpeakerAssignments(transcriptRows)
+            : [],
       };
       const signature = createCaptureConfigSignature(nextConfig);
       if (signature === lastSignature) {
@@ -266,7 +329,65 @@ function LiveCaptureConfigSyncReady({
       }, LIVE_CAPTURE_CONFIG_DEBOUNCE_MS);
     };
 
-    const unsubscribeListener = listenerStore.subscribe(schedulePush);
+    // Speaker hints live on the transcript row, so follow whichever session is
+    // being captured instead of watching every transcript in the database.
+    const syncTranscriptSubscription = () => {
+      const live = listenerStore.getState().live;
+      const sessionId = live.status === "active" ? live.sessionId : null;
+      if (sessionId === transcriptSessionId) {
+        return;
+      }
+
+      transcriptSessionId = sessionId;
+      transcriptRows = [];
+      hasTranscriptSnapshot = false;
+      void unsubscribeTranscript?.();
+      unsubscribeTranscript = null;
+      if (!sessionId) {
+        return;
+      }
+
+      void liveQueryClient
+        .subscribe<LiveTranscriptIdentitySqlRow>(
+          LIVE_TRANSCRIPT_IDENTITY_SQL,
+          [sessionId],
+          {
+            onData: (nextRows) => {
+              if (transcriptSessionId !== sessionId) {
+                return;
+              }
+              transcriptRows = nextRows;
+              hasTranscriptSnapshot = true;
+              schedulePush();
+            },
+            onError: (error) => {
+              console.error(
+                "[listener] failed to read live transcript speakers",
+                error,
+              );
+            },
+          },
+        )
+        .then((unsubscribe) => {
+          if (cancelled || transcriptSessionId !== sessionId) {
+            void unsubscribe();
+          } else {
+            unsubscribeTranscript = unsubscribe;
+          }
+        })
+        .catch((error) => {
+          console.error(
+            "[listener] failed to subscribe to live transcript speakers",
+            error,
+          );
+        });
+    };
+
+    const unsubscribeListener = listenerStore.subscribe(() => {
+      syncTranscriptSubscription();
+      schedulePush();
+    });
+    syncTranscriptSubscription();
     void liveQueryClient
       .subscribe<CaptureIdentitySqlRow>(CAPTURE_IDENTITY_SQL, [], {
         onData: (nextRows) => {
@@ -302,6 +423,7 @@ function LiveCaptureConfigSyncReady({
       }
       unsubscribeListener();
       void unsubscribeDatabase?.();
+      void unsubscribeTranscript?.();
     };
   });