Archive summary refinements and profile switching

MeetingAssistant/Recording/MeetingRecordingCoordinator.cs

@@ -97,9 +97,24 @@ public sealed class MeetingRecordingCoordinator
         string? launchProfileName,
         CancellationToken cancellationToken)
     {
-        return IsRecording
-            ? await StopAsync(cancellationToken)
-            : await StartAsync(launchProfileName, cancellationToken);
+        if (!IsRecording)
+        {
+            return await StartAsync(launchProfileName, cancellationToken);
+        }
+
+        if (string.IsNullOrWhiteSpace(launchProfileName))
+        {
+            return await StopAsync(cancellationToken);
+        }
+
+        var targetProfile = ResolveProfile(launchProfileName);
+        var run = currentRun;
+        if (run is not null && !run.IsLaunchProfile(targetProfile.Name))
+        {
+            return await SwitchProfileAsync(run, targetProfile, cancellationToken);
+        }
+
+        return await StopAsync(cancellationToken);
     }
 
     public async Task<RecordingStatus> StartAsync(CancellationToken cancellationToken)
@@ -119,7 +134,8 @@ public sealed class MeetingRecordingCoordinator
                 return CurrentStatus;
             }
 
-            var runOptions = ResolveOptions(launchProfileName);
+            var launchProfile = ResolveProfile(launchProfileName);
+            var runOptions = launchProfile.Options;
             currentSession = await transcriptStore.CreateSessionAsync(runOptions, cancellationToken);
             var recordedAudio = await recordedAudioStore.CreateSessionAsync(runOptions, cancellationToken);
             var startedAt = DateTimeOffset.Now;
@@ -173,6 +189,7 @@ public sealed class MeetingRecordingCoordinator
                 pipeline,
                 pipelineOptions,
                 runOptions,
+                launchProfile.Name,
                 runOptions.SpeakerIdentification.LiveSampleBufferDuration,
                 runOptions.SpeakerIdentification.MaxSnippetsPerSpeaker);
             run.Task = Task.Run(() => RecordAsync(run), CancellationToken.None);
@@ -318,12 +335,70 @@ public sealed class MeetingRecordingCoordinator
             cancellationToken);
     }
 
+    private async Task<RecordingStatus> SwitchProfileAsync(
+        RecordingRun run,
+        LaunchProfile targetProfile,
+        CancellationToken cancellationToken)
+    {
+        await gate.WaitAsync(cancellationToken);
+        try
+        {
+            if (!ReferenceEquals(currentRun, run) || run.IsCaptureStopping || run.IsLaunchProfile(targetProfile.Name))
+            {
+                return CurrentStatus;
+            }
+
+            await run.PipelineGate.WaitAsync(cancellationToken);
+            try
+            {
+                logger.LogInformation(
+                    "Switching active meeting transcription profile from {FromProfile} to {ToProfile}",
+                    run.LaunchProfileName,
+                    targetProfile.Name);
+                var pipeline = speechRecognitionPipelineFactory.Create(targetProfile.Name);
+                var pipelineOptions = await BuildSpeechRecognitionPipelineOptionsAsync(
+                    targetProfile.Options,
+                    run.MeetingNotePath,
+                    cancellationToken);
+                await pipeline.InitializeAsync(pipelineOptions, cancellationToken);
+
+                await run.Pipeline.CompleteAsync(cancellationToken);
+                await run.WaitForLiveTranscriptTasksAsync(cancellationToken);
+
+                run.ResetSpeakerIdentification();
+                run.MarkProfileSwitched();
+                var marker = new TranscriptionSegment(
+                    TimeSpan.Zero,
+                    TimeSpan.Zero,
+                    "System",
+                    $"Transcription profile changed to {targetProfile.Name}.");
+                run.AddLiveSegment(marker);
+                await transcriptStore.AppendAsync(run.Session, marker, cancellationToken);
+
+                run.ReplacePipeline(pipeline, pipelineOptions, targetProfile.Options, targetProfile.Name);
+                run.AddLiveTranscriptTask(Task.Run(
+                    () => StoreLiveTranscriptAsync(run, pipeline, run.TranscriptionCancellation),
+                    CancellationToken.None));
+            }
+            finally
+            {
+                run.PipelineGate.Release();
+            }
+
+            return CurrentStatus;
+        }
+        finally
+        {
+            gate.Release();
+        }
+    }
+
     private async Task RecordAsync(RecordingRun run)
     {
         await run.Pipeline.InitializeAsync(run.PipelineOptions, run.TranscriptionCancellation);
-        var liveTranscriptTask = Task.Run(
-            () => StoreLiveTranscriptAsync(run, run.TranscriptionCancellation),
-            CancellationToken.None);
+        run.AddLiveTranscriptTask(Task.Run(
+            () => StoreLiveTranscriptAsync(run, run.Pipeline, run.TranscriptionCancellation),
+            CancellationToken.None));
         var captureTask = Task.Run(
             () => CaptureAudioAsync(run),
             CancellationToken.None);
@@ -335,7 +410,7 @@ public sealed class MeetingRecordingCoordinator
         {
             await captureTask;
             await run.Pipeline.CompleteAsync(run.TranscriptionCancellation);
-            await liveTranscriptTask;
+            await run.WaitForLiveTranscriptTasksAsync(run.TranscriptionCancellation);
             run.CancelLiveIdentification();
             await liveIdentificationTask;
             await run.RecordedAudio.DisposeAsync();
@@ -366,7 +441,9 @@ public sealed class MeetingRecordingCoordinator
                     run.TranscriptionCancellation);
             }
 
-            await RunSummaryAsync(run, run.TranscriptionCancellation);
+            var summaryState = await RunSummaryAsync(run, run.TranscriptionCancellation);
+            await CompletePendingFinalSpeakerIdentificationAsync(run, run.TranscriptionCancellation);
+            await TransitionMeetingAsync(run, summaryState, CancellationToken.None);
         }
         catch (OperationCanceledException) when (run.TranscriptionCancellation.IsCancellationRequested)
         {
@@ -383,7 +460,7 @@ public sealed class MeetingRecordingCoordinator
         {
             run.CancelLiveIdentification();
             await run.RecordedAudio.DeleteAsync(CancellationToken.None);
-            await run.Pipeline.DisposeAsync();
+            await run.DisposePipelinesAsync();
             await CompleteRunAsync(run);
         }
     }
@@ -407,9 +484,8 @@ public sealed class MeetingRecordingCoordinator
                         chunk.Channels);
                 }
 
-                run.AppendAudio(chunk);
                 await run.RecordedAudio.AppendAsync(chunk, run.CaptureCancellation);
-                await run.Pipeline.WriteAsync(chunk, run.CaptureCancellation);
+                await run.WriteAudioAsync(chunk, run.CaptureCancellation);
             }
 
             logger.LogInformation(
@@ -432,9 +508,10 @@ public sealed class MeetingRecordingCoordinator
 
     private async Task StoreLiveTranscriptAsync(
         RecordingRun run,
+        ISpeechRecognitionPipeline pipeline,
         CancellationToken cancellationToken)
     {
-        await foreach (var segment in run.Pipeline.ReadLiveTranscriptAsync(cancellationToken))
+        await foreach (var segment in pipeline.ReadLiveTranscriptAsync(cancellationToken))
         {
             run.TryAddSpeakerSample(segment);
             var relabeledSegment = run.Relabel(segment);
@@ -716,6 +793,18 @@ public sealed class MeetingRecordingCoordinator
         RecordingRun run,
         CancellationToken cancellationToken)
     {
+        if (run.HasSwitchedProfile)
+        {
+            var liveSegments = run.GetLiveSegmentsSnapshot();
+            run.SetPendingFinalSpeakerIdentification(
+                run.RecordedAudio.AudioPath,
+                liveSegments,
+                run.GetSpeakerSamplesSnapshot(),
+                run.GetSpeakerMappingsSnapshot());
+            await transcriptStore.ReplaceAsync(run.Session, liveSegments, cancellationToken);
+            return;
+        }
+
         var finishedSegments = await run.Pipeline.ReadFinishedTranscriptAsync(
             run.RecordedAudio.AudioPath,
             await BuildSpeechRecognitionPipelineOptionsAsync(run.Options, run.MeetingNotePath, cancellationToken),
@@ -728,19 +817,34 @@ public sealed class MeetingRecordingCoordinator
         finishedSegments = finishedSegments
             .Select(run.Relabel)
             .ToList();
-        finishedSegments = await IdentifySpeakersAsync(
+        var samples = run.GetSpeakerSamplesSnapshot();
+        var knownSpeakerMappings = run.GetSpeakerMappingsSnapshot().ToDictionary(
+            pair => pair.Key,
+            pair => pair.Value,
+            StringComparer.OrdinalIgnoreCase);
+        var finishedSpeakerResult = await IdentifyFinishedSpeakersForTranscriptAsync(
             run.RecordedAudio.AudioPath,
             finishedSegments,
-            run.GetSpeakerSamplesSnapshot(),
-            run.GetSpeakerMappingsSnapshot(),
+            samples,
+            knownSpeakerMappings,
             run.Artifacts,
             run.MeetingNotePath,
             run.Options,
             cancellationToken);
-        await transcriptStore.ReplaceAsync(run.Session, finishedSegments, cancellationToken);
+        foreach (var (sourceSpeaker, targetSpeaker) in finishedSpeakerResult.SpeakerMappings)
+        {
+            knownSpeakerMappings[sourceSpeaker] = targetSpeaker;
+        }
+
+        run.SetPendingFinalSpeakerIdentification(
+            run.RecordedAudio.AudioPath,
+            finishedSpeakerResult.Segments,
+            samples,
+            knownSpeakerMappings);
+        await transcriptStore.ReplaceAsync(run.Session, finishedSpeakerResult.Segments, cancellationToken);
     }
 
-    private async Task<IReadOnlyList<TranscriptionSegment>> IdentifySpeakersAsync(
+    private async Task<SpeakerIdentificationResult> IdentifyFinishedSpeakersForTranscriptAsync(
         string audioPath,
         IReadOnlyList<TranscriptionSegment> finishedSegments,
         IReadOnlyList<SpeakerAudioSample> samples,
@@ -752,13 +856,13 @@ public sealed class MeetingRecordingCoordinator
     {
         if (speakerIdentificationService is null || string.IsNullOrWhiteSpace(meetingNotePath))
         {
-            return finishedSegments;
+            return new SpeakerIdentificationResult(finishedSegments, new Dictionary<string, string>());
         }
 
         try
         {
             var meetingNote = await meetingNoteStore.ReadAsync(meetingNotePath, cancellationToken);
-            var result = await speakerIdentificationService.ProcessFinishedTranscriptAsync(
+            var result = await speakerIdentificationService.IdentifyFinishedSpeakersAsync(
                 new SpeakerIdentificationRequest(
                     audioPath,
                     meetingNote,
@@ -772,7 +876,7 @@ public sealed class MeetingRecordingCoordinator
                 meetingNotePath,
                 runOptions,
                 cancellationToken);
-            return result.Segments;
+            return result;
         }
         catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
         {
@@ -780,8 +884,92 @@ public sealed class MeetingRecordingCoordinator
         }
         catch (Exception exception)
         {
-            logger.LogWarning(exception, "Speaker identity learning failed; keeping ASR speaker labels");
-            return finishedSegments;
+            logger.LogWarning(exception, "Finished speaker identity matching failed; keeping ASR speaker labels");
+            return new SpeakerIdentificationResult(finishedSegments, new Dictionary<string, string>());
+        }
+    }
+
+    private async Task CompletePendingFinalSpeakerIdentificationAsync(
+        RecordingRun run,
+        CancellationToken cancellationToken)
+    {
+        var pending = run.TakePendingFinalSpeakerIdentification();
+        if (pending is null ||
+            speakerIdentificationService is null ||
+            string.IsNullOrWhiteSpace(run.MeetingNotePath))
+        {
+            return;
+        }
+
+        try
+        {
+            var meetingNote = await meetingNoteStore.ReadAsync(run.MeetingNotePath, cancellationToken);
+            var segments = pending.Segments.ToList();
+            var knownSpeakerMappings = pending.KnownSpeakerMappings.ToDictionary(
+                pair => pair.Key,
+                pair => pair.Value,
+                StringComparer.OrdinalIgnoreCase);
+            var speakerOverrides = await SpeakerOverrideArtifacts.ReadFromAssistantContextAsync(
+                run.Artifacts.AssistantContextPath,
+                cancellationToken);
+            foreach (var speakerOverride in speakerOverrides)
+            {
+                await speakerIdentificationService.ApplySpeakerOverrideAsync(
+                    new SpeakerIdentificationRequest(
+                        pending.AudioPath,
+                        meetingNote,
+                        segments,
+                        pending.Samples,
+                        knownSpeakerMappings),
+                    speakerOverride.From,
+                    speakerOverride.To,
+                    cancellationToken);
+                segments = segments
+                    .Select(segment => string.Equals(segment.Speaker, speakerOverride.From, StringComparison.OrdinalIgnoreCase)
+                        ? segment with { Speaker = speakerOverride.To }
+                        : segment)
+                    .ToList();
+                knownSpeakerMappings[speakerOverride.From] = speakerOverride.To;
+            }
+
+            var identityDeletions = await SpeakerOverrideArtifacts.ReadIdentityDeletionsFromAssistantContextAsync(
+                run.Artifacts.AssistantContextPath,
+                cancellationToken);
+            foreach (var deletion in identityDeletions)
+            {
+                await speakerIdentificationService.DeleteSpeakerIdentityAsync(
+                    deletion.Identity,
+                    cancellationToken);
+                segments = segments
+                    .Select(segment => string.Equals(segment.Speaker, deletion.Identity, StringComparison.OrdinalIgnoreCase)
+                        ? segment with { Speaker = deletion.Replacement }
+                        : segment)
+                    .ToList();
+            }
+
+            var result = await speakerIdentificationService.ProcessFinishedTranscriptAsync(
+                new SpeakerIdentificationRequest(
+                    pending.AudioPath,
+                    meetingNote,
+                    segments,
+                    pending.Samples,
+                    knownSpeakerMappings),
+                cancellationToken);
+            await AddIdentifiedSpeakersToMeetingAttendeesAsync(
+                result.AttendeeMatches,
+                run.Artifacts,
+                run.MeetingNotePath,
+                run.Options,
+                cancellationToken);
+            await transcriptStore.ReplaceAsync(run.Session, result.Segments, cancellationToken);
+        }
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+        {
+            throw;
+        }
+        catch (Exception exception)
+        {
+            logger.LogWarning(exception, "Final speaker identity learning failed after summary");
         }
     }
 
@@ -943,13 +1131,13 @@ public sealed class MeetingRecordingCoordinator
         return true;
     }
 
-    private async Task RunSummaryAsync(
+    private async Task<AssistantContextState> RunSummaryAsync(
         RecordingRun run,
         CancellationToken cancellationToken)
     {
         if (run.IsAborted)
         {
-            return;
+            return AssistantContextState.Finished;
         }
 
         try
@@ -963,10 +1151,7 @@ public sealed class MeetingRecordingCoordinator
                 AssistantContextState.Summarizing,
                 cancellationToken);
             var result = await summaryPipeline.RunAsync(run.Artifacts, run.Options, cancellationToken);
-            await TransitionMeetingAsync(
-                run,
-                result.Succeeded ? AssistantContextState.Finished : AssistantContextState.Error,
-                CancellationToken.None);
+            return result.Succeeded ? AssistantContextState.Finished : AssistantContextState.Error;
         }
         catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
         {
@@ -975,10 +1160,7 @@ public sealed class MeetingRecordingCoordinator
         catch (Exception exception)
         {
             logger.LogError(exception, "Meeting summary pipeline failed unexpectedly");
-            await TransitionMeetingAsync(
-                run,
-                AssistantContextState.Error,
-                CancellationToken.None);
+            return AssistantContextState.Error;
         }
     }
 
@@ -1035,17 +1217,17 @@ public sealed class MeetingRecordingCoordinator
             : new SpeechRecognitionPipelineOptions(DictationWords: dictationWords);
     }
 
-    private MeetingAssistantOptions ResolveOptions(string? launchProfileName)
+    private LaunchProfile ResolveProfile(string? launchProfileName)
     {
         if (launchProfiles is not null)
         {
-            return launchProfiles.GetRequiredProfile(launchProfileName).Options;
+            return launchProfiles.GetRequiredProfile(launchProfileName);
         }
 
         if (string.IsNullOrWhiteSpace(launchProfileName) ||
             launchProfileName.Equals(ConfigurationLaunchProfileOptionsProvider.DefaultProfileName, StringComparison.OrdinalIgnoreCase))
         {
-            return options;
+            return new LaunchProfile(ConfigurationLaunchProfileOptionsProvider.DefaultProfileName, options);
         }
 
         throw new InvalidOperationException(
@@ -1102,11 +1284,15 @@ public sealed class MeetingRecordingCoordinator
         private int completed;
         private readonly object stateGate = new();
         private readonly object liveSegmentsGate = new();
+        private readonly object liveTranscriptTasksGate = new();
         private readonly object liveIdentificationGate = new();
         private readonly List<TranscriptionSegment> liveSegments = [];
+        private readonly List<Task> liveTranscriptTasks = [];
+        private readonly List<ISpeechRecognitionPipeline> pipelines = [];
         private readonly ConcurrentDictionary<string, string> speakerMappings = new(StringComparer.OrdinalIgnoreCase);
         private readonly SpeakerAudioSampleCollector speakerSampleCollector;
         private LiveIdentificationCheckpoint? lastLiveIdentificationCheckpoint;
+        private FinalSpeakerIdentificationWork? pendingFinalSpeakerIdentification;
 
         public RecordingRun(
             CancellationTokenSource captureCancellation,
@@ -1118,6 +1304,7 @@ public sealed class MeetingRecordingCoordinator
             ISpeechRecognitionPipeline pipeline,
             SpeechRecognitionPipelineOptions pipelineOptions,
             MeetingAssistantOptions options,
+            string launchProfileName,
             TimeSpan liveSampleBufferDuration,
             int maxSpeakerSamples)
         {
@@ -1131,6 +1318,8 @@ public sealed class MeetingRecordingCoordinator
             Pipeline = pipeline;
             PipelineOptions = pipelineOptions;
             Options = options;
+            LaunchProfileName = launchProfileName;
+            pipelines.Add(pipeline);
             speakerSampleCollector = new SpeakerAudioSampleCollector(liveSampleBufferDuration, maxSpeakerSamples);
         }
 
@@ -1148,11 +1337,15 @@ public sealed class MeetingRecordingCoordinator
 
         public MeetingSessionArtifacts Artifacts { get; }
 
-        public ISpeechRecognitionPipeline Pipeline { get; }
+        public SemaphoreSlim PipelineGate { get; } = new(1, 1);
 
-        public SpeechRecognitionPipelineOptions PipelineOptions { get; }
+        public ISpeechRecognitionPipeline Pipeline { get; private set; }
 
-        public MeetingAssistantOptions Options { get; }
+        public SpeechRecognitionPipelineOptions PipelineOptions { get; private set; }
+
+        public MeetingAssistantOptions Options { get; private set; }
+
+        public string LaunchProfileName { get; private set; }
 
         public CancellationToken CaptureCancellation => CaptureCancellationSource.Token;
 
@@ -1166,6 +1359,8 @@ public sealed class MeetingRecordingCoordinator
 
         public bool IsAborted { get; private set; }
 
+        public bool HasSwitchedProfile { get; private set; }
+
         public AssistantContextState ContextState { get; private set; } = AssistantContextState.CollectingMetadata;
 
         public void StopCapture()
@@ -1191,6 +1386,103 @@ public sealed class MeetingRecordingCoordinator
             LiveIdentificationCancellationSource.Cancel();
         }
 
+        public bool IsLaunchProfile(string launchProfileName)
+        {
+            return string.Equals(LaunchProfileName, launchProfileName, StringComparison.OrdinalIgnoreCase);
+        }
+
+        public async ValueTask WriteAudioAsync(AudioChunk chunk, CancellationToken cancellationToken)
+        {
+            await PipelineGate.WaitAsync(cancellationToken);
+            try
+            {
+                AppendAudio(chunk);
+                await Pipeline.WriteAsync(chunk, cancellationToken);
+            }
+            finally
+            {
+                PipelineGate.Release();
+            }
+        }
+
+        public void ReplacePipeline(
+            ISpeechRecognitionPipeline pipeline,
+            SpeechRecognitionPipelineOptions pipelineOptions,
+            MeetingAssistantOptions options,
+            string launchProfileName)
+        {
+            Pipeline = pipeline;
+            PipelineOptions = pipelineOptions;
+            Options = options;
+            LaunchProfileName = launchProfileName;
+            pipelines.Add(pipeline);
+        }
+
+        public void AddLiveTranscriptTask(Task task)
+        {
+            lock (liveTranscriptTasksGate)
+            {
+                liveTranscriptTasks.Add(task);
+            }
+        }
+
+        public Task WaitForLiveTranscriptTasksAsync(CancellationToken cancellationToken)
+        {
+            Task[] tasks;
+            lock (liveTranscriptTasksGate)
+            {
+                tasks = liveTranscriptTasks.ToArray();
+            }
+
+            return Task.WhenAll(tasks).WaitAsync(cancellationToken);
+        }
+
+        public async Task DisposePipelinesAsync()
+        {
+            foreach (var pipeline in pipelines.Distinct())
+            {
+                await pipeline.DisposeAsync();
+            }
+        }
+
+        public void MarkProfileSwitched()
+        {
+            HasSwitchedProfile = true;
+        }
+
+        public void ResetSpeakerIdentification()
+        {
+            speakerMappings.Clear();
+            speakerSampleCollector.Reset();
+            lock (liveIdentificationGate)
+            {
+                lastLiveIdentificationCheckpoint = null;
+            }
+        }
+
+        public void SetPendingFinalSpeakerIdentification(
+            string audioPath,
+            IReadOnlyList<TranscriptionSegment> segments,
+            IReadOnlyList<SpeakerAudioSample> samples,
+            IReadOnlyDictionary<string, string> knownSpeakerMappings)
+        {
+            pendingFinalSpeakerIdentification = new FinalSpeakerIdentificationWork(
+                audioPath,
+                segments.ToList(),
+                samples.ToList(),
+                knownSpeakerMappings.ToDictionary(
+                    pair => pair.Key,
+                    pair => pair.Value,
+                    StringComparer.OrdinalIgnoreCase));
+        }
+
+        public FinalSpeakerIdentificationWork? TakePendingFinalSpeakerIdentification()
+        {
+            var pending = pendingFinalSpeakerIdentification;
+            pendingFinalSpeakerIdentification = null;
+            return pending;
+        }
+
         public bool TryTransitionTo(
             AssistantContextState state,
             out AssistantContextState fromState)
@@ -1340,6 +1632,7 @@ public sealed class MeetingRecordingCoordinator
             CaptureCancellationSource.Dispose();
             TranscriptionCancellationSource.Dispose();
             LiveIdentificationCancellationSource.Dispose();
+            PipelineGate.Dispose();
         }
 
         private static int StateRank(AssistantContextState state)
@@ -1367,6 +1660,12 @@ public sealed class MeetingRecordingCoordinator
                     Speakers.SequenceEqual(other.Speakers, StringComparer.OrdinalIgnoreCase);
             }
         }
+
+        public sealed record FinalSpeakerIdentificationWork(
+            string AudioPath,
+            IReadOnlyList<TranscriptionSegment> Segments,
+            IReadOnlyList<SpeakerAudioSample> Samples,
+            IReadOnlyDictionary<string, string> KnownSpeakerMappings);
     }
 
     private sealed class EmptyDictationWordStore : IDictationWordStore

MeetingAssistant/Recording/RollingAudioBuffer.cs

@@ -31,6 +31,15 @@ internal sealed class RollingAudioBuffer
         }
     }
 
+    public void Reset()
+    {
+        lock (gate)
+        {
+            chunks.Clear();
+            position = TimeSpan.Zero;
+        }
+    }
+
     public byte[] TryExtractWav(TimeSpan start, TimeSpan end)
     {
         if (end <= start)

MeetingAssistant/Recording/SpeakerAudioSampleCollector.cs

@@ -21,6 +21,15 @@ internal sealed class SpeakerAudioSampleCollector
         audioBuffer.Append(chunk);
     }
 
+    public void Reset()
+    {
+        lock (gate)
+        {
+            samplesBySpeaker.Clear();
+            audioBuffer.Reset();
+        }
+    }
+
     public void TryAdd(TranscriptionSegment segment)
     {
         if (!IsDiarizedSpeaker(segment.Speaker))