diff --git a/apps/desktop/src/services/event-listeners.test.tsx b/apps/desktop/src/services/event-listeners.test.tsx index c036f640e8b..fdd84675535 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"); @@ -153,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 }, + }, }); }); @@ -275,9 +290,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 +307,352 @@ 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 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({ + 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"] }, + }, + ], + }); + }); + + 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, + }, + }, + ], }); }); diff --git a/apps/desktop/src/services/event-listeners.tsx b/apps/desktop/src/services/event-listeners.tsx index b9b2bf0a5d9..57db0ea85cb 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); } @@ -203,10 +252,15 @@ 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[] = []; + 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 +272,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 +293,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,7 +301,20 @@ function LiveCaptureConfigSyncReady({ live.sessionId, ), self_human_id: session?.owner_user_id || null, + speaker_assignments: + transcriptSessionId === live.sessionId + ? 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; @@ -266,7 +339,79 @@ 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; + } + + // 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, + [sessionId], + { + onData: (nextRows) => { + if (transcriptSessionId !== sessionId) { + return; + } + transcriptRows = nextRows; + hasTranscriptSnapshot = true; + schedulePush(); + }, + onError: (error) => { + onTranscriptUnavailable( + "[listener] failed to read live transcript speakers", + error, + ); + }, + }, + ) + .then((unsubscribe) => { + if (cancelled || transcriptSessionId !== sessionId) { + void unsubscribe(); + } else { + unsubscribeTranscript = unsubscribe; + } + }) + .catch((error) => { + onTranscriptUnavailable( + "[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 +447,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, } } }