From 002b33924935ed954c0fc3a1cc6924b704b5662e Mon Sep 17 00:00:00 2001 From: John Jeong Date: Thu, 3 Sep 2026 18:17:47 +0900 Subject: [PATCH 1/3] 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. --- .../src/services/event-listeners.test.tsx | 157 +++++++++++++++++- apps/desktop/src/services/event-listeners.tsx | 138 ++++++++++++++- .../src/actors/listener/adapters.rs | 1 + .../listener-core/src/actors/listener/mod.rs | 20 ++- .../src/actors/session/supervisor.rs | 25 +++ .../src/actors/session/supervisor/children.rs | 1 + .../listener-core/src/actors/session/types.rs | 8 + crates/listener-core/src/live_transcript.rs | 44 ++++- .../src/live_transcript/segments.rs | 152 ++++++++++++++++- .../src/live_transcript/tests.rs | 72 ++++++++ plugins/transcription/js/bindings.gen.ts | 4 +- plugins/transcription/src/api.rs | 6 + 12 files changed, 598 insertions(+), 30 deletions(-) diff --git a/apps/desktop/src/services/event-listeners.test.tsx b/apps/desktop/src/services/event-listeners.test.tsx index c036f640e8b..4c5c98f7ea8 100644 --- a/apps/desktop/src/services/event-listeners.test.tsx +++ b/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(); 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(); + + 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(); + + 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"] }, + }, + ], }); }); diff --git a/apps/desktop/src/services/event-listeners.tsx b/apps/desktop/src/services/event-listeners.tsx index b9b2bf0a5d9..05d7db7982f 100644 --- a/apps/desktop/src/services/event-listeners.tsx +++ b/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,6 +30,7 @@ import { getLiveTranscriptionConfig, getTranscriptionLanguages, } from "~/stt/capabilities"; +import { buildRenderTranscriptRequestFromRows } from "~/stt/render-transcript"; type CaptureIdentitySqlRow = { session_id: string; @@ -33,6 +38,13 @@ type CaptureIdentitySqlRow = { 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(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) | null = null; + let unsubscribeTranscript: (() => Promise) | 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,7 +292,7 @@ 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( @@ -241,6 +300,10 @@ function LiveCaptureConfigSyncReady({ 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( + 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(CAPTURE_IDENTITY_SQL, [], { onData: (nextRows) => { @@ -302,6 +423,7 @@ function LiveCaptureConfigSyncReady({ } unsubscribeListener(); void unsubscribeDatabase?.(); + void unsubscribeTranscript?.(); }; }); diff --git a/crates/listener-core/src/actors/listener/adapters.rs b/crates/listener-core/src/actors/listener/adapters.rs index e63bed8eca3..266feeb1cba 100644 --- a/crates/listener-core/src/actors/listener/adapters.rs +++ b/crates/listener-core/src/actors/listener/adapters.rs @@ -670,6 +670,7 @@ mod tests { session_id: "session".to_string(), participant_human_ids: vec![], self_human_id: None, + speaker_assignments: vec![], } } diff --git a/crates/listener-core/src/actors/listener/mod.rs b/crates/listener-core/src/actors/listener/mod.rs index 9d4ab92d383..85a599f88c2 100644 --- a/crates/listener-core/src/actors/listener/mod.rs +++ b/crates/listener-core/src/actors/listener/mod.rs @@ -9,6 +9,7 @@ use ractor::{Actor, ActorName, ActorProcessingErr, ActorRef, RpcReplyPort, Super use tokio::time::error::Elapsed; use tracing::Instrument; +use anlg_transcript::IdentityAssignment; use owhisper_interface::stream::StreamResponse; use owhisper_interface::{ControlMessage, MixedMessage}; @@ -48,6 +49,7 @@ pub struct ListenerConfigUpdate { pub languages: Vec, pub participant_human_ids: Vec, pub self_human_id: Option, + pub speaker_assignments: Vec, } #[derive(Clone)] @@ -67,6 +69,7 @@ pub struct ListenerArgs { pub session_id: String, pub participant_human_ids: Vec, pub self_human_id: Option, + pub speaker_assignments: Vec, } pub struct ListenerState { @@ -160,10 +163,11 @@ impl Actor for ListenerActor { adapter: adapter_name.clone(), }); - let transcript = LiveTranscriptEngine::new( + let transcript = LiveTranscriptEngine::with_speaker_assignments( &adapter_name, &args.participant_human_ids, args.self_human_id.as_deref(), + args.speaker_assignments.clone(), ); let state = ListenerState { @@ -288,10 +292,20 @@ impl Actor for ListenerActor { state.args.languages = update.languages; state.args.participant_human_ids = update.participant_human_ids; state.args.self_human_id = update.self_human_id; - state.transcript.update_participants( + state.args.speaker_assignments = update.speaker_assignments; + if let Some(segment_delta) = state.transcript.update_identities( &state.args.participant_human_ids, state.args.self_human_id.as_deref(), - ); + state.args.speaker_assignments.clone(), + ) { + state + .args + .runtime + .emit_data(SessionDataEvent::TranscriptSegmentDelta { + session_id: state.args.session_id.clone(), + delta: Box::new(segment_delta), + }); + } } ListenerMsg::StreamResponse(response, reply) => { diff --git a/crates/listener-core/src/actors/session/supervisor.rs b/crates/listener-core/src/actors/session/supervisor.rs index 0388a40e735..8e2314aded9 100644 --- a/crates/listener-core/src/actors/session/supervisor.rs +++ b/crates/listener-core/src/actors/session/supervisor.rs @@ -299,6 +299,7 @@ async fn update_config( state.ctx.params.languages = update.languages; state.ctx.params.participant_human_ids = update.participant_human_ids; state.ctx.params.self_human_id = update.self_human_id; + state.ctx.params.speaker_assignments = update.speaker_assignments; if should_refresh_listener { refresh_listener(myself, state).await; @@ -311,6 +312,7 @@ async fn update_config( languages: state.ctx.params.languages.clone(), participant_human_ids: state.ctx.params.participant_human_ids.clone(), self_human_id: state.ctx.params.self_human_id.clone(), + speaker_assignments: state.ctx.params.speaker_assignments.clone(), })) { tracing::warn!(?error, "failed_to_cast_listener_config_update"); } @@ -687,6 +689,7 @@ mod tests { mic_device: None, participant_human_ids: vec![], self_human_id: None, + speaker_assignments: vec![], }, app_dir: std::env::temp_dir(), started_at_instant: Instant::now(), @@ -721,9 +724,31 @@ mod tests { .map(ToString::to_string) .collect(), self_human_id: self_human_id.map(ToString::to_string), + speaker_assignments: vec![], } } + #[test] + fn config_update_does_not_refresh_for_speaker_assignments() { + let mut ctx = test_ctx(); + ctx.params.participant_human_ids = vec!["self".to_string(), "remote-a".to_string()]; + ctx.params.self_human_id = Some("self".to_string()); + let state = test_state(ctx); + let mut update = test_update(vec![], vec!["self", "remote-a"], Some("self")); + update.speaker_assignments = vec![anlg_transcript::IdentityAssignment { + human_id: "remote-a".to_string(), + scope: anlg_transcript::IdentityScope::ChannelSpeaker { + channel: anlg_transcript::ChannelProfile::RemoteParty, + speaker_index: 0, + }, + }]; + + assert!(!update_requires_listener_refresh( + &state.ctx.params, + &update + )); + } + #[test] fn config_update_refreshes_when_languages_change() { let mut ctx = test_ctx(); diff --git a/crates/listener-core/src/actors/session/supervisor/children.rs b/crates/listener-core/src/actors/session/supervisor/children.rs index 718ea31d3ef..bfdb9e954c8 100644 --- a/crates/listener-core/src/actors/session/supervisor/children.rs +++ b/crates/listener-core/src/actors/session/supervisor/children.rs @@ -138,6 +138,7 @@ pub(super) async fn spawn_listener( session_id: ctx.params.session_id.clone(), participant_human_ids: ctx.params.participant_human_ids.clone(), self_human_id: ctx.params.self_human_id.clone(), + speaker_assignments: ctx.params.speaker_assignments.clone(), }, supervisor_cell, ) diff --git a/crates/listener-core/src/actors/session/types.rs b/crates/listener-core/src/actors/session/types.rs index 580ccc48007..f16a27536f1 100644 --- a/crates/listener-core/src/actors/session/types.rs +++ b/crates/listener-core/src/actors/session/types.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use std::time::{Instant, SystemTime}; use anlg_audio::AudioProvider; +use anlg_transcript::IdentityAssignment; use crate::{ListenerRuntime, TranscriptionMode}; @@ -30,6 +31,10 @@ pub struct SessionParams { pub participant_human_ids: Vec, #[serde(default)] pub self_human_id: Option, + /// Persisted speaker identities of the transcript being captured, so the + /// live segments name speakers the same way the settled render does. + #[serde(default)] + pub speaker_assignments: Vec, } #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -41,6 +46,8 @@ pub struct SessionConfigUpdate { pub participant_human_ids: Vec, #[serde(default)] pub self_human_id: Option, + #[serde(default)] + pub speaker_assignments: Vec, } // The single requested-to-effective transcription mode policy: every capture @@ -159,6 +166,7 @@ mod tests { mic_device: None, participant_human_ids: vec![], self_human_id: None, + speaker_assignments: vec![], } } diff --git a/crates/listener-core/src/live_transcript.rs b/crates/listener-core/src/live_transcript.rs index 5fe98340c39..865a32f1287 100644 --- a/crates/listener-core/src/live_transcript.rs +++ b/crates/listener-core/src/live_transcript.rs @@ -1,6 +1,6 @@ use anlg_transcript::{ - FinalizedWord, PartialWord, SegmentKey, SegmentWord, TranscriptDelta, TranscriptProcessor, - channel_assignments_for_participants, segment_options_for_participants, + FinalizedWord, IdentityAssignment, PartialWord, SegmentKey, SegmentWord, TranscriptDelta, + TranscriptProcessor, channel_assignments_for_participants, segment_options_for_participants, }; use owhisper_interface::stream::StreamResponse; @@ -74,6 +74,23 @@ impl LiveTranscriptEngine { provider_name: &str, participant_human_ids: &[String], self_human_id: Option<&str>, + ) -> Self { + Self::with_speaker_assignments( + provider_name, + participant_human_ids, + self_human_id, + Vec::new(), + ) + } + + /// `speaker_assignments` are the persisted identity hints of the transcript + /// being captured (user picks and automatic matches), so segments the + /// engine emits carry the same names the settled render will. + pub fn with_speaker_assignments( + provider_name: &str, + participant_human_ids: &[String], + self_human_id: Option<&str>, + speaker_assignments: Vec, ) -> Self { let channel_assignments = channel_assignments_for_participants(participant_human_ids, self_human_id); @@ -89,7 +106,11 @@ impl LiveTranscriptEngine { .with_partial_finalization(normalizer.finalize_partials()) .with_flush_partial_finalization(normalizer.flush_partials()), normalizer, - rendered_segments: RenderedSegmentState::new(channel_assignments, segment_options), + rendered_segments: RenderedSegmentState::new( + channel_assignments, + speaker_assignments, + segment_options, + ), max_speaker_index, } } @@ -106,17 +127,22 @@ impl LiveTranscriptEngine { }) } - pub fn update_participants( + /// Returns the segments whose labels changed so they can be pushed to + /// listeners right away; a user naming a speaker mid-meeting should not + /// wait for the next stream response to see it. + pub fn update_identities( &mut self, participant_human_ids: &[String], self_human_id: Option<&str>, - ) { - self.rendered_segments.update_participants( - channel_assignments_for_participants(participant_human_ids, self_human_id), - segment_options_for_participants(participant_human_ids, self_human_id), - ); + speaker_assignments: Vec, + ) -> Option { self.max_speaker_index = max_speaker_index_for_participants(participant_human_ids, self_human_id); + self.rendered_segments.update_identities( + channel_assignments_for_participants(participant_human_ids, self_human_id), + speaker_assignments, + segment_options_for_participants(participant_human_ids, self_human_id), + ) } pub fn flush(&mut self) -> Option { diff --git a/crates/listener-core/src/live_transcript/segments.rs b/crates/listener-core/src/live_transcript/segments.rs index cd92ec8ad48..ea67fd6200a 100644 --- a/crates/listener-core/src/live_transcript/segments.rs +++ b/crates/listener-core/src/live_transcript/segments.rs @@ -15,30 +15,40 @@ const MAX_RENDERED_WINDOW_MS: i64 = 30 * 60 * 1_000; #[derive(Default)] pub(super) struct RenderedSegmentState { words: Vec, + partials: Vec, segment_revisions: BTreeMap, channel_assignments: Vec, + speaker_assignments: Vec, segment_options: Option, } impl RenderedSegmentState { pub(super) fn new( channel_assignments: Vec, + speaker_assignments: Vec, segment_options: SegmentBuilderOptions, ) -> Self { Self { channel_assignments, + speaker_assignments, segment_options: Some(segment_options), ..Default::default() } } - pub(super) fn update_participants( + /// Replaces the identity inputs and re-renders the retained window so + /// segments already on screen pick up the new names without waiting for + /// the next stream response. + pub(super) fn update_identities( &mut self, channel_assignments: Vec, + speaker_assignments: Vec, segment_options: SegmentBuilderOptions, - ) { + ) -> Option { self.channel_assignments = channel_assignments; + self.speaker_assignments = speaker_assignments; self.segment_options = Some(segment_options); + self.render() } pub(super) fn apply_delta( @@ -72,17 +82,30 @@ impl RenderedSegmentState { .max() .unwrap_or_default(); prune_rendered_words(&mut self.words, latest_end_ms); - let partials = bounded_partials( + self.partials = bounded_partials( &delta.partials, latest_end_ms, MAX_RENDERED_WORDS.saturating_sub(self.words.len()), MAX_RENDERED_TEXT_BYTES.saturating_sub(rendered_text_bytes(&self.words)), ); + self.render() + } + + fn render(&mut self) -> Option { + // Mirrors `render_transcript_segments`: participant channel defaults + // are appended after the persisted hints so both paths resolve the + // same identity for the same word. + let assignments = self + .speaker_assignments + .iter() + .chain(self.channel_assignments.iter()) + .cloned() + .collect::>(); let next_segments = build_live_segments( &self.words, - &partials, - &self.channel_assignments, + &self.partials, + &assignments, self.segment_options.as_ref(), ); let mut next_revisions = BTreeMap::new(); @@ -290,7 +313,10 @@ fn build_live_segments( #[cfg(test)] mod tests { - use anlg_transcript::{FinalizedWord, PartialWord, SegmentBuilderOptions, WordState}; + use anlg_transcript::{ + ChannelProfile, FinalizedWord, IdentityAssignment, IdentityScope, PartialWord, + SegmentBuilderOptions, WordState, + }; use super::{ LiveTranscriptDelta, MAX_RENDERED_TEXT_BYTES, MAX_RENDERED_WINDOW_MS, MAX_RENDERED_WORDS, @@ -309,8 +335,122 @@ mod tests { } } + fn remote_word(id: &str, text: &str, start_ms: i64, speaker_index: i32) -> FinalizedWord { + FinalizedWord { + channel: 1, + speaker_index: Some(speaker_index), + ..finalized_word(id, text.to_string(), start_ms) + } + } + + fn speaker_assignment(human_id: &str, speaker_index: i32) -> IdentityAssignment { + IdentityAssignment { + human_id: human_id.to_string(), + scope: IdentityScope::ChannelSpeaker { + channel: ChannelProfile::RemoteParty, + speaker_index, + }, + } + } + + #[test] + fn identity_update_relabels_rendered_segments_and_keeps_partials() { + let mut state = state(); + let partial = PartialWord { + text: " again".to_string(), + start_ms: 5_100, + end_ms: 5_200, + channel: 1, + speaker_index: Some(1), + }; + state.apply_delta(&LiveTranscriptDelta { + new_words: vec![ + remote_word("a", " hello", 0, 0), + remote_word("b", " there", 100, 0), + remote_word("c", " hi", 5_000, 1), + ], + replaced_ids: Vec::new(), + partials: vec![partial.clone()], + }); + + let update = state + .update_identities( + Vec::new(), + vec![speaker_assignment("artem", 0)], + SegmentBuilderOptions { + max_gap_ms: Some(500), + complete_channels: None, + min_segment_words: None, + min_segment_ms: None, + }, + ) + .expect("assignment should change the rendered segments"); + + let named = update + .upserts + .iter() + .filter(|segment| segment.key.speaker_human_id.as_deref() == Some("artem")) + .collect::>(); + assert_eq!(named.len(), 1); + assert_eq!(named[0].text, "hello there"); + assert!( + update + .upserts + .iter() + .all(|segment| segment.key.speaker_index != Some(1)), + "the unassigned speaker's segment must not be re-emitted" + ); + assert_eq!( + update.removed_ids.len(), + 1, + "the old unnamed segment goes away" + ); + + let next = state + .apply_delta(&LiveTranscriptDelta { + new_words: vec![remote_word("d", " later", 9_000, 0)], + replaced_ids: Vec::new(), + partials: vec![partial], + }) + .expect("new words should render"); + assert_eq!( + next.upserts.len(), + 1, + "unchanged segments are not re-emitted" + ); + assert_eq!(next.upserts[0].text, "later"); + assert_eq!( + next.upserts[0].key.speaker_human_id.as_deref(), + Some("artem") + ); + } + + #[test] + fn identity_update_without_changes_emits_nothing() { + let mut state = state(); + state.apply_delta(&LiveTranscriptDelta { + new_words: vec![remote_word("a", " hello", 0, 0)], + replaced_ids: Vec::new(), + partials: Vec::new(), + }); + + let update = state.update_identities( + Vec::new(), + vec![speaker_assignment("artem", 3)], + SegmentBuilderOptions { + max_gap_ms: Some(500), + complete_channels: None, + min_segment_words: None, + min_segment_ms: None, + }, + ); + + assert!(update.is_none()); + } + fn state() -> RenderedSegmentState { RenderedSegmentState::new( + Vec::new(), Vec::new(), SegmentBuilderOptions { max_gap_ms: Some(500), diff --git a/crates/listener-core/src/live_transcript/tests.rs b/crates/listener-core/src/live_transcript/tests.rs index 9c04289aa29..5c991ce176f 100644 --- a/crates/listener-core/src/live_transcript/tests.rs +++ b/crates/listener-core/src/live_transcript/tests.rs @@ -663,6 +663,78 @@ fn clamps_single_remote_speaker_to_zero() { assert_eq!(channel.alternatives[0].words[0].speaker, Some(0)); } +#[test] +fn speaker_assignment_names_later_segments_from_the_same_speaker() { + let participants = ["self".to_string(), "artem".to_string(), "guest".to_string()]; + let mut engine = LiveTranscriptEngine::new("deepgram", &participants, Some("self")); + let spoken = |text: &str, start: f64, speaker: i32| { + let mut words = words_from_text(text, start, 1.0); + for word in &mut words { + word.speaker = Some(speaker); + } + transcript_response_at(text, words, true, 1, start, 1.0) + }; + + let first = engine + .process(&spoken("ah okay it is this week", 0.0, 0)) + .expect("first update"); + let first_segments = first.segment_delta.expect("first segments").upserts; + assert_eq!(first_segments.len(), 1); + assert_eq!(first_segments[0].key.speaker_index, Some(0)); + assert_eq!(first_segments[0].key.speaker_human_id, None); + + let relabeled = engine + .update_identities( + &participants, + Some("self"), + vec![IdentityAssignment { + human_id: "artem".to_string(), + scope: anlg_transcript::IdentityScope::ChannelSpeaker { + channel: anlg_transcript::ChannelProfile::RemoteParty, + speaker_index: 0, + }, + }], + ) + .expect("assignment relabels the segment already on screen"); + assert_eq!(relabeled.upserts.len(), 1); + assert_eq!( + relabeled.upserts[0].key.speaker_human_id.as_deref(), + Some("artem") + ); + assert_eq!(relabeled.removed_ids, vec![first_segments[0].id.clone()]); + + let later = engine + .process(&spoken("yeah let us discuss it", 10.0, 0)) + .expect("later update"); + let later_segments = later.segment_delta.expect("later segments").upserts; + assert_eq!(later_segments.len(), 1); + assert_eq!( + later_segments[0].key.speaker_human_id.as_deref(), + Some("artem") + ); + assert!(later_segments[0].text.ends_with("yeah let us discuss it")); + + // The processor holds the newest words as partials until the stream moves + // on, so flush to settle the other speaker's turn. + engine + .process(&spoken("nice", 20.0, 1)) + .expect("other speaker update"); + let settled = engine.flush().expect("flush update"); + let settled_segments = settled.segment_delta.expect("settled segments").upserts; + let other = settled_segments + .iter() + .find(|segment| segment.text == "nice") + .expect("the other speaker gets a segment of their own"); + assert_eq!(other.key.speaker_index, Some(1)); + assert_eq!(other.key.speaker_human_id, None); + assert!( + settled_segments + .iter() + .filter(|segment| segment.key.speaker_human_id.as_deref() == Some("artem")) + .all(|segment| !segment.text.contains("nice")) + ); +} + #[test] fn live_transcript_delta_keeps_speaker_index_on_words() { let delta = TranscriptDelta { diff --git a/plugins/transcription/js/bindings.gen.ts b/plugins/transcription/js/bindings.gen.ts index b658430736f..913722ab4c0 100644 --- a/plugins/transcription/js/bindings.gen.ts +++ b/plugins/transcription/js/bindings.gen.ts @@ -222,8 +222,8 @@ export type BatchResults = { channels: BatchChannel[] } export type BatchRunMode = "direct" | "streamed" export type BatchStreamEvent = { type: "progress"; percentage: number; partial_text?: string | null } | { type: "segment"; response: StreamResponse; percentage: number } | { type: "terminal"; request_id: string; created: string; duration: number; channels: number } | { type: "result"; response: BatchResponse } | { type: "error"; error_code: number | null; error_message: string; provider: string } export type BatchWord = { word: string; start: number; end: number; confidence: number; channel?: number; speaker: number | null; punctuated_word: string | null } -export type CaptureConfigUpdate = { session_id: string; languages: string[]; participant_human_ids?: string[]; self_human_id?: string | null } -export type CaptureDataEvent = { type: "audio_amplitude"; session_id: string; mic: number; speaker: number } | { type: "mic_muted"; session_id: string; value: boolean } | { type: "mic_isolated"; session_id: string; value: boolean } | { type: "transcript_delta"; session_id: string; delta: LiveTranscriptDelta } | { type: "transcript_segment_delta"; session_id: string; delta: LiveTranscriptSegmentDelta } +export type CaptureConfigUpdate = { session_id: string; languages: string[]; participant_human_ids?: string[]; self_human_id?: string | null; speaker_assignments?: IdentityAssignment[] } +export type CaptureDataEvent = { type: "audio_amplitude"; session_id: string; mic: number; speaker: number } | { type: "mic_muted"; session_id: string; value: boolean } | { type: "mic_isolated"; session_id: string; value: boolean } | { type: "mic_dropouts"; session_id: string; ratio: number } | { type: "transcript_delta"; session_id: string; delta: LiveTranscriptDelta } | { type: "transcript_segment_delta"; session_id: string; delta: LiveTranscriptSegmentDelta } export type CaptureLifecycleEvent = { type: "started"; session_id: string; requested_live_transcription: boolean; live_transcription_active: boolean; degraded: DegradedError | null } | { type: "finalizing"; session_id: string } | { type: "stopped"; session_id: string; audio_path: string | null; requested_live_transcription: boolean; live_transcription_active: boolean; error: string | null } export type CaptureParams = { session_id: string; languages: string[]; onboarding: boolean; model: string; base_url: string; api_key: string; keywords: string[]; mic_device?: string | null; transcription_mode?: TranscriptionMode | null; participant_human_ids?: string[]; self_human_id?: string | null } export type CaptureSnapshot = { state: CaptureState; activeSessionId: string | null; finalizingSessionIds: string[]; requestedLiveTranscription: boolean | null; liveTranscriptionActive: boolean | null; liveSegmentsSessionId?: string | null; liveSegments?: LiveTranscriptSegment[] | null } diff --git a/plugins/transcription/src/api.rs b/plugins/transcription/src/api.rs index 22c26529f97..8d6ba399438 100644 --- a/plugins/transcription/src/api.rs +++ b/plugins/transcription/src/api.rs @@ -49,6 +49,8 @@ pub struct CaptureConfigUpdate { pub participant_human_ids: Vec, #[serde(default)] pub self_human_id: Option, + #[serde(default)] + pub speaker_assignments: Vec, } #[derive(serde::Serialize, serde::Deserialize, Clone, specta::Type, tauri_specta::Event)] @@ -202,6 +204,9 @@ impl From for listener::actors::SessionParams { mic_device: value.mic_device, participant_human_ids: value.participant_human_ids, self_human_id: value.self_human_id, + // The desktop config sync pushes the transcript's persisted hints + // once the capture is active; a new transcript has none yet. + speaker_assignments: Vec::new(), } } } @@ -213,6 +218,7 @@ impl From for listener::actors::SessionConfigUpdate { languages: value.languages, participant_human_ids: value.participant_human_ids, self_human_id: value.self_human_id, + speaker_assignments: value.speaker_assignments, } } } From 75df9ba68467110fea3cadcf514b7e2a32ac2381 Mon Sep 17 00:00:00 2001 From: John Jeong Date: Thu, 3 Sep 2026 20:40:50 +0900 Subject: [PATCH 2/3] Re-push speaker assignments when a capture restarts on the same session The config signature that dedupes live capture pushes outlived the capture: restarting on the same session recomputed the same config, the push was skipped, and the fresh engine, which always starts with no speaker assignments, never got the names. Scope the signature to the store's capture generation so a new capture always pushes once. --- .../src/services/event-listeners.test.tsx | 123 +++++++++++++++++- apps/desktop/src/services/event-listeners.tsx | 10 ++ 2 files changed, 132 insertions(+), 1 deletion(-) diff --git a/apps/desktop/src/services/event-listeners.test.tsx b/apps/desktop/src/services/event-listeners.test.tsx index 4c5c98f7ea8..d0c09feb4da 100644 --- a/apps/desktop/src/services/event-listeners.test.tsx +++ b/apps/desktop/src/services/event-listeners.test.tsx @@ -164,7 +164,11 @@ describe("EventListeners notification events", () => { setTriggerAppIds: setTriggerAppIdsMock, stop: stopMock, updateCaptureConfig: updateCaptureConfigMock, - live: { status: "active", sessionId: "session-1" }, + live: { + status: "active", + sessionId: "session-1", + captureGenerationBySession: { "session-1": 1 }, + }, }); }); @@ -448,6 +452,123 @@ describe("EventListeners notification events", () => { }); }); + test("live capture config sync pushes again after a restart on the same session", async () => { + vi.useFakeTimers(); + useConfigValuesMock.mockReturnValue({ + ai_language: "en", + spoken_languages: ["en"], + current_stt_provider: "soniox", + current_stt_model: "stt-v4", + }); + liveQuerySubscribeMock.mockImplementation( + async (sql, _params, handlers) => { + if (!String(sql).includes("FROM transcripts")) { + handlers.onData([]); + } + return async () => {}; + }, + ); + const setLive = (live: Record) => + getListenerStateMock.mockReturnValue({ + setTriggerAppIds: setTriggerAppIdsMock, + stop: stopMock, + updateCaptureConfig: updateCaptureConfigMock, + live, + }); + const latestTranscriptHandlers = () => { + const calls = liveQuerySubscribeMock.mock.calls.filter(([sql]) => + String(sql).includes("FROM transcripts"), + ); + const call = calls[calls.length - 1]; + expect(call).toBeDefined(); + return call![2] as { onData: (rows: unknown[]) => void }; + }; + const transcriptRows = [ + { + id: "transcript-1", + started_at_ms: 1_000, + words_json: JSON.stringify([ + { id: "w1", text: " hello", start_ms: 0, end_ms: 100, 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, + }), + }, + ]), + }, + ]; + + render(); + + await vi.waitFor(() => + expect(liveQuerySubscribeMock).toHaveBeenCalledTimes(2), + ); + const storeListener = listenerSubscribeMock.mock.calls[0]?.[0]; + expect(storeListener).toBeTypeOf("function"); + + findLiveQueryHandlers("session_participants").onData([ + { + session_id: "session-1", + owner_user_id: "human-self", + human_id: "human-artem", + }, + ]); + latestTranscriptHandlers().onData(transcriptRows); + await vi.runOnlyPendingTimersAsync(); + expect(updateCaptureConfigMock).toHaveBeenCalledTimes(1); + + setLive({ + status: "inactive", + sessionId: null, + captureGenerationBySession: {}, + }); + storeListener(); + await vi.runOnlyPendingTimersAsync(); + + setLive({ + status: "active", + sessionId: "session-1", + captureGenerationBySession: { "session-1": 2 }, + }); + storeListener(); + await vi.waitFor(() => + expect(liveQuerySubscribeMock).toHaveBeenCalledTimes(3), + ); + latestTranscriptHandlers().onData(transcriptRows); + await vi.runOnlyPendingTimersAsync(); + + expect(updateCaptureConfigMock).toHaveBeenCalledTimes(2); + expect(updateCaptureConfigMock.mock.calls[1]?.[0]).toEqual( + updateCaptureConfigMock.mock.calls[0]?.[0], + ); + expect(updateCaptureConfigMock.mock.calls[1]?.[0]).toMatchObject({ + speaker_assignments: [ + { + human_id: "human-artem", + scope: { + kind: "channel_speaker", + channel: "RemoteParty", + speaker_index: 0, + }, + }, + ], + }); + }); + test("notification_confirm with auto-stop prompt ignores collapsed body click", async () => { render(); diff --git a/apps/desktop/src/services/event-listeners.tsx b/apps/desktop/src/services/event-listeners.tsx index 05d7db7982f..ad5b4ece3eb 100644 --- a/apps/desktop/src/services/event-listeners.tsx +++ b/apps/desktop/src/services/event-listeners.tsx @@ -252,6 +252,7 @@ function LiveCaptureConfigSyncReady({ useMountEffect(() => { let timeoutId: ReturnType | null = null; let lastSignature: string | null = null; + let lastCaptureGeneration: number | null = null; let rows: CaptureIdentitySqlRow[] = []; let hasSnapshot = false; let transcriptRows: LiveTranscriptIdentitySqlRow[] = []; @@ -305,6 +306,15 @@ function LiveCaptureConfigSyncReady({ ? getLiveSpeakerAssignments(transcriptRows) : [], }; + // Every capture starts its engine without speaker assignments, so a + // config identical to the previous capture's still has to be pushed. + const captureGeneration = + live.captureGenerationBySession[live.sessionId] ?? null; + if (captureGeneration !== lastCaptureGeneration) { + lastCaptureGeneration = captureGeneration; + lastSignature = null; + } + const signature = createCaptureConfigSignature(nextConfig); if (signature === lastSignature) { return; From 7f4b649f7a0b459c4a6cfd04f0cf07fcdb7422cb Mon Sep 17 00:00:00 2001 From: John Jeong Date: Thu, 3 Sep 2026 20:57:06 +0900 Subject: [PATCH 3/3] Keep live capture config flowing when the transcript read fails The wait-for-snapshot gate held every push until the transcript query answered, and a query error or a rejected subscription never opened it, so a failed transcript read also froze participant and language updates for the rest of the session. A failure before the first rows now counts as an empty snapshot and the config runs without names; a failure after them keeps the last known assignments. --- .../src/services/event-listeners.test.tsx | 87 +++++++++++++++++++ apps/desktop/src/services/event-listeners.tsx | 18 +++- 2 files changed, 103 insertions(+), 2 deletions(-) diff --git a/apps/desktop/src/services/event-listeners.test.tsx b/apps/desktop/src/services/event-listeners.test.tsx index d0c09feb4da..fdd84675535 100644 --- a/apps/desktop/src/services/event-listeners.test.tsx +++ b/apps/desktop/src/services/event-listeners.test.tsx @@ -357,6 +357,93 @@ describe("EventListeners notification events", () => { ); }); + test("live capture config sync runs without names when the transcript read fails", async () => { + vi.useFakeTimers(); + vi.spyOn(console, "error").mockImplementation(() => {}); + useConfigValuesMock.mockReturnValue({ + ai_language: "ko", + spoken_languages: ["ko"], + current_stt_provider: "soniox", + current_stt_model: "stt-v4", + }); + liveQuerySubscribeMock.mockImplementation( + async (sql, _params, handlers) => { + if (String(sql).includes("FROM transcripts")) { + handlers.onError("no such table: transcripts"); + } else { + handlers.onData([]); + } + return async () => {}; + }, + ); + + render(); + + 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).toHaveBeenCalledTimes(1); + expect(updateCaptureConfigMock).toHaveBeenCalledWith({ + session_id: "session-1", + languages: ["ko"], + participant_human_ids: ["human-remote"], + self_human_id: "human-self", + speaker_assignments: [], + }); + }); + + test("live capture config sync runs without names when the transcript subscription rejects", async () => { + vi.useFakeTimers(); + vi.spyOn(console, "error").mockImplementation(() => {}); + useConfigValuesMock.mockReturnValue({ + ai_language: "ko", + spoken_languages: ["ko"], + current_stt_provider: "soniox", + current_stt_model: "stt-v4", + }); + liveQuerySubscribeMock.mockImplementation( + async (sql, _params, handlers) => { + if (String(sql).includes("FROM transcripts")) { + throw new Error("subscribe failed"); + } + handlers.onData([]); + return async () => {}; + }, + ); + + render(); + + 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).toHaveBeenCalledTimes(1); + expect(updateCaptureConfigMock).toHaveBeenCalledWith( + expect.objectContaining({ + session_id: "session-1", + participant_human_ids: ["human-remote"], + speaker_assignments: [], + }), + ); + }); + test("live capture config sync pushes the active transcript's speaker assignments", async () => { vi.useFakeTimers(); useConfigValuesMock.mockReturnValue({ diff --git a/apps/desktop/src/services/event-listeners.tsx b/apps/desktop/src/services/event-listeners.tsx index ad5b4ece3eb..57db0ea85cb 100644 --- a/apps/desktop/src/services/event-listeners.tsx +++ b/apps/desktop/src/services/event-listeners.tsx @@ -357,6 +357,20 @@ function LiveCaptureConfigSyncReady({ return; } + // A read that fails before the first rows would otherwise hold every + // participant and language update for the session. Run without names + // instead; a failure after the first rows keeps the last known ones. + const onTranscriptUnavailable = (message: string, error: unknown) => { + console.error(message, error); + if (cancelled || transcriptSessionId !== sessionId) { + return; + } + if (!hasTranscriptSnapshot) { + hasTranscriptSnapshot = true; + schedulePush(); + } + }; + void liveQueryClient .subscribe( LIVE_TRANSCRIPT_IDENTITY_SQL, @@ -371,7 +385,7 @@ function LiveCaptureConfigSyncReady({ schedulePush(); }, onError: (error) => { - console.error( + onTranscriptUnavailable( "[listener] failed to read live transcript speakers", error, ); @@ -386,7 +400,7 @@ function LiveCaptureConfigSyncReady({ } }) .catch((error) => { - console.error( + onTranscriptUnavailable( "[listener] failed to subscribe to live transcript speakers", error, );