Add resilient Azure offline transcription backlog
PR and Push Build/Test / build-and-test (push) Successful in 18m41s

This commit is contained in:
2026-06-15 17:26:34 +02:00
parent b22754ce5d
commit b774ccc375
35 changed files with 2456 additions and 198 deletions
@@ -0,0 +1,134 @@
using System.Text.Json;
using Microsoft.Extensions.Options;
namespace MeetingAssistant.Recording;
public sealed class FileOfflineTranscriptionBacklog : IOfflineTranscriptionBacklog
{
private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web)
{
WriteIndented = true
};
private readonly MeetingAssistantOptions options;
private readonly ILogger<FileOfflineTranscriptionBacklog> logger;
public FileOfflineTranscriptionBacklog(
IOptions<MeetingAssistantOptions> options,
ILogger<FileOfflineTranscriptionBacklog> logger)
{
this.options = options.Value;
this.logger = logger;
}
public async Task EnqueueAsync(OfflineTranscriptionBacklogItem item, CancellationToken cancellationToken)
{
var folder = GetBacklogFolder(options);
Directory.CreateDirectory(folder);
var path = GetItemPath(folder, item.Id);
await File.WriteAllTextAsync(path, JsonSerializer.Serialize(item, JsonOptions), cancellationToken);
logger.LogInformation(
"Queued offline transcription backlog item {BacklogItemId} for recording {RecordingPath}",
item.Id,
item.AudioPath);
}
public async Task<IReadOnlyList<OfflineTranscriptionBacklogItem>> ListAsync(CancellationToken cancellationToken)
{
var folder = GetBacklogFolder(options);
if (!Directory.Exists(folder))
{
return [];
}
var items = new List<OfflineTranscriptionBacklogItem>();
foreach (var path in Directory.EnumerateFiles(folder, "*.json", SearchOption.TopDirectoryOnly))
{
cancellationToken.ThrowIfCancellationRequested();
try
{
var content = await File.ReadAllTextAsync(path, cancellationToken);
if (JsonSerializer.Deserialize<OfflineTranscriptionBacklogItem>(content, JsonOptions) is { } item)
{
items.Add(item);
}
}
catch (JsonException exception)
{
logger.LogWarning(exception, "Could not read offline transcription backlog item {BacklogItemPath}", path);
}
catch (IOException exception)
{
logger.LogWarning(exception, "Could not read offline transcription backlog item {BacklogItemPath}", path);
}
}
return items;
}
public async Task CompleteAsync(string id, CancellationToken cancellationToken)
{
var folder = GetBacklogFolder(options);
var path = GetItemPath(folder, id);
if (!File.Exists(path))
{
return;
}
OfflineTranscriptionBacklogItem? item = null;
try
{
item = JsonSerializer.Deserialize<OfflineTranscriptionBacklogItem>(
await File.ReadAllTextAsync(path, cancellationToken),
JsonOptions);
}
catch (JsonException exception)
{
logger.LogWarning(exception, "Could not read completed offline transcription backlog item {BacklogItemPath}", path);
}
File.Delete(path);
if (item is not null && File.Exists(item.AudioPath))
{
DeleteCompletedAudio(item.AudioPath);
}
}
private static string GetBacklogFolder(MeetingAssistantOptions options)
{
return Path.Combine(VaultPath.Resolve(options.Recording.TemporaryRecordingsFolder), "offline-transcription-backlog");
}
private static string GetItemPath(string folder, string id)
{
return Path.Combine(folder, $"{id}.json");
}
private void DeleteCompletedAudio(string audioPath)
{
try
{
File.Delete(audioPath);
}
catch (FileNotFoundException)
{
}
catch (DirectoryNotFoundException)
{
}
catch (IOException exception)
{
logger.LogWarning(
exception,
"Could not delete completed offline transcription recording file {RecordingPath}",
audioPath);
}
catch (UnauthorizedAccessException exception)
{
logger.LogWarning(
exception,
"Could not delete completed offline transcription recording file {RecordingPath}",
audioPath);
}
}
}
+26 -2
View File
@@ -14,13 +14,30 @@ public interface ITranscriptStore
Task AppendAsync(TranscriptSession session, TranscriptionSegment segment, CancellationToken cancellationToken)
{
return AppendLineAsync(
return AppendLineAndDiscardReferenceAsync(
session,
TranscriptLineFormatter.Format(segment).Line,
cancellationToken);
}
Task AppendLineAsync(TranscriptSession session, string line, CancellationToken cancellationToken);
private async Task AppendLineAndDiscardReferenceAsync(
TranscriptSession session,
string line,
CancellationToken cancellationToken)
{
_ = await AppendLineAsync(session, line, cancellationToken);
}
Task<TranscriptLineReference> AppendLineAsync(
TranscriptSession session,
string line,
CancellationToken cancellationToken);
Task ReplaceLineAsync(
TranscriptSession session,
TranscriptLineReference lineReference,
string replacementLine,
CancellationToken cancellationToken);
Task ReplaceAsync(
TranscriptSession session,
@@ -47,11 +64,18 @@ public interface ITranscriptStore
public sealed record TranscriptSession(string TranscriptPath);
public sealed record TranscriptLineReference(string TranscriptPath, int BodyLineIndex, string OriginalLine);
public static class TranscriptLineFormatter
{
public static FormattedTranscriptLine Format(TranscriptionSegment segment)
{
var speaker = NormalizeSpeaker(segment.Speaker);
if (segment.Kind == TranscriptionSegmentKind.Marker)
{
return new FormattedTranscriptLine(segment.Text, speaker);
}
return new FormattedTranscriptLine(
$"[{segment.Start:hh\\:mm\\:ss}] {speaker}: {segment.Text}",
speaker);
@@ -31,6 +31,7 @@ public sealed class MeetingRecordingCoordinator
private readonly IMeetingRunArtifactCleaner artifactCleaner;
private readonly IMeetingInactivityPromptService inactivityPromptService;
private readonly IMeetingInactivityClock inactivityClock;
private readonly IOfflineTranscriptionBacklog offlineTranscriptionBacklog;
private readonly MeetingAssistantOptions options;
private readonly ILogger<MeetingRecordingCoordinator> logger;
private readonly SemaphoreSlim gate = new(1, 1);
@@ -59,7 +60,8 @@ public sealed class MeetingRecordingCoordinator
IMeetingScreenshotService? screenshotService = null,
IMeetingRunArtifactCleaner? artifactCleaner = null,
IMeetingInactivityPromptService? inactivityPromptService = null,
IMeetingInactivityClock? inactivityClock = null)
IMeetingInactivityClock? inactivityClock = null,
IOfflineTranscriptionBacklog? offlineTranscriptionBacklog = null)
{
this.audioSource = audioSource;
this.speechRecognitionPipelineFactory = speechRecognitionPipelineFactory;
@@ -79,6 +81,7 @@ public sealed class MeetingRecordingCoordinator
this.artifactCleaner = artifactCleaner ?? new MeetingRunArtifactCleaner();
this.inactivityPromptService = inactivityPromptService ?? new NoopMeetingInactivityPromptService();
this.inactivityClock = inactivityClock ?? new SystemMeetingInactivityClock();
this.offlineTranscriptionBacklog = offlineTranscriptionBacklog ?? NoopOfflineTranscriptionBacklog.Instance;
this.options = options.Value;
this.logger = logger;
}
@@ -301,7 +304,17 @@ public sealed class MeetingRecordingCoordinator
}
catch (TimeoutException)
{
logger.LogWarning("Timed out while draining transcription after recording stop; forcing transcription cancellation");
if (IsAzureSpeechRun(run))
{
logger.LogWarning("Timed out while draining Azure Speech transcription after recording stop; queueing offline transcription backlog item");
run.MarkQueuedForOfflineTranscription();
await offlineTranscriptionBacklog.EnqueueAsync(CreateOfflineBacklogItem(run), cancellationToken);
}
else
{
logger.LogWarning("Timed out while draining transcription after recording stop; forcing transcription cancellation");
}
run.CancelTranscription();
await run.Task.WaitAsync(cancellationToken);
}
@@ -424,6 +437,7 @@ public sealed class MeetingRecordingCoordinator
await run.Pipeline.CompleteAsync(cancellationToken);
await run.WaitForLiveTranscriptTasksAsync(cancellationToken);
await run.WaitForTranscriptWorkflowTasksAsync(cancellationToken);
run.ResetSpeakerIdentification();
run.MarkProfileSwitched();
@@ -471,6 +485,7 @@ public sealed class MeetingRecordingCoordinator
await captureTask;
await run.Pipeline.CompleteAsync(run.TranscriptionCancellation);
await run.WaitForLiveTranscriptTasksAsync(run.TranscriptionCancellation);
await run.WaitForTranscriptWorkflowTasksAsync(run.TranscriptionCancellation);
run.CancelLiveIdentification();
await liveIdentificationTask;
await run.RecordedAudio.DisposeAsync();
@@ -528,7 +543,15 @@ public sealed class MeetingRecordingCoordinator
finally
{
run.CancelLiveIdentification();
await run.RecordedAudio.DeleteAsync(CancellationToken.None);
if (run.IsQueuedForOfflineTranscription)
{
await run.RecordedAudio.DisposeAsync();
}
else
{
await run.RecordedAudio.DeleteAsync(CancellationToken.None);
}
await run.DisposePipelinesAsync();
await CompleteRunAsync(run);
}
@@ -582,6 +605,17 @@ public sealed class MeetingRecordingCoordinator
{
await foreach (var segment in pipeline.ReadLiveTranscriptAsync(cancellationToken))
{
if (segment.Kind == TranscriptionSegmentKind.Marker)
{
await AppendTranscriptSegmentAsync(run, segment, cancellationToken);
continue;
}
if (segment.ReplacesMarkerId is not null)
{
run.ResetSpeakerIdentification();
}
run.RecordTranscriptActivity(segment, inactivityClock.Now);
var sample = run.TryAddSpeakerSample(segment);
if (sample is not null)
@@ -607,8 +641,69 @@ public sealed class MeetingRecordingCoordinator
TranscriptionSegment segment,
CancellationToken cancellationToken)
{
var line = await TransformTranscriptLineAsync(run, segment, cancellationToken);
await transcriptStore.AppendLineAsync(run.Session, line, cancellationToken);
var formatted = TranscriptLineFormatter.Format(segment);
var lineReference = segment.ReplacesMarkerId is { } replacesMarkerId &&
run.TryTakeTranscriptMarker(replacesMarkerId, out var markerReference)
? await ReplaceTranscriptMarkerAsync(run, markerReference, formatted, cancellationToken)
: segment.Kind == TranscriptionSegmentKind.Marker &&
segment.MarkerId is { } existingMarkerId &&
run.TryTakeTranscriptMarker(existingMarkerId, out var existingMarkerReference)
? await ReplaceTranscriptMarkerAsync(run, existingMarkerReference, formatted, cancellationToken)
: await transcriptStore.AppendLineAsync(run.Session, formatted.Line, cancellationToken);
if (segment.Kind == TranscriptionSegmentKind.Marker && segment.MarkerId is { } markerId)
{
run.SetTranscriptMarker(markerId, lineReference);
return;
}
run.AddTranscriptWorkflowTask(
() => ApplyTranscriptLineWorkflowAsync(run, lineReference, formatted, run.TranscriptionCancellation));
}
private async Task<TranscriptLineReference> ReplaceTranscriptMarkerAsync(
RecordingRun run,
TranscriptLineReference markerReference,
FormattedTranscriptLine formatted,
CancellationToken cancellationToken)
{
await transcriptStore.ReplaceLineAsync(run.Session, markerReference, formatted.Line, cancellationToken);
return markerReference with { OriginalLine = formatted.Line };
}
private async Task ApplyTranscriptLineWorkflowAsync(
RecordingRun run,
TranscriptLineReference lineReference,
FormattedTranscriptLine formatted,
CancellationToken cancellationToken)
{
var transformed = await TryTransformTranscriptLineAsync(
run,
formatted.Line,
formatted.Speaker,
cancellationToken);
if (!string.Equals(formatted.Line, transformed, StringComparison.Ordinal))
{
try
{
await transcriptStore.ReplaceLineAsync(
run.Session,
lineReference,
transformed,
cancellationToken);
run.RecordTranscriptLineRewrite(formatted.Line, transformed);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw;
}
catch (Exception exception)
{
logger.LogError(
exception,
"Transcript line rewrite failed for speaker {Speaker}; keeping original transcript line",
formatted.Speaker);
}
}
}
private async Task ReplaceTranscriptSegmentsAsync(
@@ -616,28 +711,41 @@ public sealed class MeetingRecordingCoordinator
IReadOnlyList<TranscriptionSegment> segments,
CancellationToken cancellationToken)
{
var lines = new List<string>(segments.Count);
foreach (var segment in segments)
{
lines.Add(await TransformTranscriptLineAsync(run, segment, cancellationToken));
}
var lines = segments
.Select(segment => run.ApplyTranscriptLineRewrite(TranscriptLineFormatter.Format(segment).Line))
.ToList();
await transcriptStore.ReplaceLinesAsync(run.Session, lines, cancellationToken);
}
private async Task<string> TransformTranscriptLineAsync(
private async Task<string> TryTransformTranscriptLineAsync(
RecordingRun run,
TranscriptionSegment segment,
string formattedLine,
string speaker,
CancellationToken cancellationToken)
{
var formatted = TranscriptLineFormatter.Format(segment);
return await meetingWorkflowEngine.TransformTranscriptLineAsync(
MeetingWorkflowEvent.TranscriptLine(
run.Artifacts,
formatted.Line,
formatted.Speaker),
run.Options,
cancellationToken);
try
{
return await meetingWorkflowEngine.TransformTranscriptLineAsync(
MeetingWorkflowEvent.TranscriptLine(
run.Artifacts,
formattedLine,
speaker),
run.Options,
cancellationToken);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw;
}
catch (Exception exception)
{
logger.LogError(
exception,
"Transcript line workflow failed for speaker {Speaker}; keeping original transcript line",
speaker);
return formattedLine;
}
}
private async Task RunInactivitySafeguardAsync(RecordingRun run)
@@ -1459,17 +1567,14 @@ public sealed class MeetingRecordingCoordinator
string? meetingNotePath,
CancellationToken cancellationToken)
{
var dictationWords = await dictationWordProvider.ReadOptionalWordsAsync(runOptions, cancellationToken);
if (string.IsNullOrWhiteSpace(meetingNotePath))
{
return new SpeechRecognitionPipelineOptions(DictationWords: dictationWords);
}
var meetingNote = await meetingNoteStore.ReadAsync(meetingNotePath, cancellationToken);
var attendeeCount = meetingNote.Frontmatter.Attendees.Count(attendee => !string.IsNullOrWhiteSpace(attendee));
return attendeeCount > 1
? new SpeechRecognitionPipelineOptions(attendeeCount, dictationWords)
: new SpeechRecognitionPipelineOptions(DictationWords: dictationWords);
var meetingNote = string.IsNullOrWhiteSpace(meetingNotePath)
? null
: await meetingNoteStore.ReadAsync(meetingNotePath, cancellationToken);
return await RecordingSpeechRecognitionPipelineOptions.BuildAsync(
dictationWordProvider,
runOptions,
meetingNote,
cancellationToken);
}
private LaunchProfile ResolveProfile(string? launchProfileName)
@@ -1515,6 +1620,25 @@ public sealed class MeetingRecordingCoordinator
return minimumDuration > TimeSpan.Zero && stoppedAt - run.StartedAt < minimumDuration;
}
private static bool IsAzureSpeechRun(RecordingRun run)
{
return run.Options.Recording.TranscriptionProvider.Equals("azure-speech", StringComparison.OrdinalIgnoreCase);
}
private static OfflineTranscriptionBacklogItem CreateOfflineBacklogItem(RecordingRun run)
{
return new OfflineTranscriptionBacklogItem(
Id: Guid.NewGuid().ToString("N"),
AudioPath: run.RecordedAudio.AudioPath,
TranscriptPath: run.Session.TranscriptPath,
MeetingNotePath: run.MeetingNotePath,
AssistantContextPath: run.Artifacts.AssistantContextPath,
SummaryPath: run.Artifacts.SummaryPath,
StartedAt: run.StartedAt,
InferredEndTime: run.InferredEndTime,
LaunchProfileName: run.LaunchProfileName);
}
private static IReadOnlyList<TimeSpan> GetInactivityPromptThresholds(
RecordingInactivitySafeguardOptions options)
{
@@ -1566,10 +1690,14 @@ public sealed class MeetingRecordingCoordinator
private readonly object stateGate = new();
private readonly object liveSegmentsGate = new();
private readonly object liveTranscriptTasksGate = new();
private readonly object transcriptWorkflowTasksGate = new();
private readonly object liveIdentificationGate = new();
private readonly object transcriptActivityGate = new();
private readonly List<TranscriptionSegment> liveSegments = [];
private readonly List<Task> liveTranscriptTasks = [];
private Task transcriptWorkflowTail = Task.CompletedTask;
private readonly ConcurrentDictionary<string, TranscriptLineReference> transcriptMarkers = new(StringComparer.OrdinalIgnoreCase);
private readonly ConcurrentDictionary<string, string> transcriptLineRewrites = new(StringComparer.Ordinal);
private readonly List<ISpeechRecognitionPipeline> pipelines = [];
private readonly ConcurrentDictionary<string, string> speakerMappings = new(StringComparer.OrdinalIgnoreCase);
private readonly SpeakerAudioSampleCollector speakerSampleCollector;
@@ -1656,6 +1784,8 @@ public sealed class MeetingRecordingCoordinator
public bool IsAborted { get; private set; }
public bool IsQueuedForOfflineTranscription { get; private set; }
public bool HasSwitchedProfile { get; private set; }
public AssistantContextState ContextState { get; private set; } = AssistantContextState.CollectingMetadata;
@@ -1689,6 +1819,11 @@ public sealed class MeetingRecordingCoordinator
TranscriptionCancellationSource.Cancel();
}
public void MarkQueuedForOfflineTranscription()
{
IsQueuedForOfflineTranscription = true;
}
public void CancelLiveIdentification()
{
LiveIdentificationCancellationSource.Cancel();
@@ -1745,6 +1880,41 @@ public sealed class MeetingRecordingCoordinator
return Task.WhenAll(tasks).WaitAsync(cancellationToken);
}
public void AddTranscriptWorkflowTask(Func<Task> taskFactory)
{
lock (transcriptWorkflowTasksGate)
{
transcriptWorkflowTail = RunAfterAsync(transcriptWorkflowTail, taskFactory);
}
}
public Task WaitForTranscriptWorkflowTasksAsync(CancellationToken cancellationToken)
{
Task task;
lock (transcriptWorkflowTasksGate)
{
task = transcriptWorkflowTail;
}
return task.WaitAsync(cancellationToken);
}
private static async Task RunAfterAsync(Task previous, Func<Task> next)
{
await previous.ConfigureAwait(false);
await next().ConfigureAwait(false);
}
public void SetTranscriptMarker(string markerId, TranscriptLineReference lineReference)
{
transcriptMarkers[markerId] = lineReference;
}
public bool TryTakeTranscriptMarker(string markerId, out TranscriptLineReference lineReference)
{
return transcriptMarkers.TryRemove(markerId, out lineReference!);
}
public async Task DisposePipelinesAsync()
{
foreach (var pipeline in pipelines.Distinct())
@@ -1817,6 +1987,18 @@ public sealed class MeetingRecordingCoordinator
}
}
public void RecordTranscriptLineRewrite(string originalLine, string replacementLine)
{
transcriptLineRewrites[originalLine] = replacementLine;
}
public string ApplyTranscriptLineRewrite(string line)
{
return transcriptLineRewrites.TryGetValue(line, out var replacement)
? replacement
: line;
}
public void RecordTranscriptActivity(TranscriptionSegment segment, DateTimeOffset arrivedAt)
{
if (string.IsNullOrWhiteSpace(segment.Text))
@@ -0,0 +1,45 @@
namespace MeetingAssistant.Recording;
public interface IOfflineTranscriptionBacklog
{
Task EnqueueAsync(OfflineTranscriptionBacklogItem item, CancellationToken cancellationToken);
Task<IReadOnlyList<OfflineTranscriptionBacklogItem>> ListAsync(CancellationToken cancellationToken);
Task CompleteAsync(string id, CancellationToken cancellationToken);
}
public sealed record OfflineTranscriptionBacklogItem(
string Id,
string AudioPath,
string TranscriptPath,
string MeetingNotePath,
string AssistantContextPath,
string SummaryPath,
DateTimeOffset StartedAt,
DateTimeOffset? InferredEndTime,
string LaunchProfileName);
public sealed class NoopOfflineTranscriptionBacklog : IOfflineTranscriptionBacklog
{
public static readonly NoopOfflineTranscriptionBacklog Instance = new();
private NoopOfflineTranscriptionBacklog()
{
}
public Task EnqueueAsync(OfflineTranscriptionBacklogItem item, CancellationToken cancellationToken)
{
return Task.CompletedTask;
}
public Task<IReadOnlyList<OfflineTranscriptionBacklogItem>> ListAsync(CancellationToken cancellationToken)
{
return Task.FromResult<IReadOnlyList<OfflineTranscriptionBacklogItem>>([]);
}
public Task CompleteAsync(string id, CancellationToken cancellationToken)
{
return Task.CompletedTask;
}
}
@@ -0,0 +1,53 @@
namespace MeetingAssistant.Recording;
public sealed class OfflineTranscriptionBacklogHostedService : BackgroundService
{
private static readonly TimeSpan RetryDelay = TimeSpan.FromMinutes(1);
private readonly OfflineTranscriptionBacklogProcessor processor;
private readonly ILogger<OfflineTranscriptionBacklogHostedService> logger;
public OfflineTranscriptionBacklogHostedService(
OfflineTranscriptionBacklogProcessor processor,
ILogger<OfflineTranscriptionBacklogHostedService> logger)
{
this.processor = processor;
this.logger = logger;
}
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
while (!stoppingToken.IsCancellationRequested)
{
try
{
var processed = await processor.ProcessPendingAsync(stoppingToken);
if (processed > 0)
{
logger.LogInformation(
"Processed {ProcessedCount} offline transcription backlog item(s)",
processed);
}
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
return;
}
catch (Exception exception)
{
logger.LogWarning(
exception,
"Offline transcription backlog processing failed; it will be retried later");
}
try
{
await Task.Delay(RetryDelay, stoppingToken);
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
return;
}
}
}
}
@@ -0,0 +1,243 @@
using MeetingAssistant.LaunchProfiles;
using MeetingAssistant.MeetingNotes;
using MeetingAssistant.Summary;
using MeetingAssistant.Transcription;
using MeetingAssistant.Workflow;
using Microsoft.Extensions.Options;
using NAudio.Wave;
namespace MeetingAssistant.Recording;
public sealed class OfflineTranscriptionBacklogProcessor
{
private const int MaxChunkDurationMilliseconds = 1000;
private readonly IOfflineTranscriptionBacklog backlog;
private readonly ISpeechRecognitionPipelineFactory pipelineFactory;
private readonly ITranscriptStore transcriptStore;
private readonly IMeetingNoteStore meetingNoteStore;
private readonly IMeetingArtifactStore meetingArtifactStore;
private readonly IMeetingSummaryPipeline summaryPipeline;
private readonly IMeetingWorkflowEngine workflowEngine;
private readonly IRecordingDictationWordProvider dictationWordProvider;
private readonly MeetingAssistantOptions options;
private readonly ILogger<OfflineTranscriptionBacklogProcessor> logger;
private readonly ILaunchProfileOptionsProvider? launchProfiles;
public OfflineTranscriptionBacklogProcessor(
IOfflineTranscriptionBacklog backlog,
ISpeechRecognitionPipelineFactory pipelineFactory,
ITranscriptStore transcriptStore,
IMeetingNoteStore meetingNoteStore,
IMeetingArtifactStore meetingArtifactStore,
IMeetingSummaryPipeline summaryPipeline,
IMeetingWorkflowEngine workflowEngine,
IRecordingDictationWordProvider dictationWordProvider,
IOptions<MeetingAssistantOptions> options,
ILogger<OfflineTranscriptionBacklogProcessor> logger,
ILaunchProfileOptionsProvider? launchProfiles = null)
{
this.backlog = backlog;
this.pipelineFactory = pipelineFactory;
this.transcriptStore = transcriptStore;
this.meetingNoteStore = meetingNoteStore;
this.meetingArtifactStore = meetingArtifactStore;
this.summaryPipeline = summaryPipeline;
this.workflowEngine = workflowEngine;
this.dictationWordProvider = dictationWordProvider;
this.options = options.Value;
this.logger = logger;
this.launchProfiles = launchProfiles;
}
public async Task<int> ProcessPendingAsync(CancellationToken cancellationToken)
{
var processed = 0;
foreach (var item in await backlog.ListAsync(cancellationToken))
{
cancellationToken.ThrowIfCancellationRequested();
if (await TryProcessAsync(item, cancellationToken))
{
processed++;
}
}
return processed;
}
private async Task<bool> TryProcessAsync(
OfflineTranscriptionBacklogItem item,
CancellationToken cancellationToken)
{
try
{
await ProcessAsync(item, cancellationToken);
await backlog.CompleteAsync(item.Id, cancellationToken);
logger.LogInformation(
"Completed offline transcription backlog item {BacklogItemId} for recording {RecordingPath}",
item.Id,
item.AudioPath);
return true;
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw;
}
catch (Exception exception)
{
logger.LogWarning(
exception,
"Could not process offline transcription backlog item {BacklogItemId}; it will be retried later",
item.Id);
return false;
}
}
private async Task ProcessAsync(
OfflineTranscriptionBacklogItem item,
CancellationToken cancellationToken)
{
var runOptions = ResolveOptions(item.LaunchProfileName);
var artifacts = new MeetingSessionArtifacts(
item.MeetingNotePath,
item.TranscriptPath,
item.AssistantContextPath,
item.SummaryPath);
var session = new TranscriptSession(item.TranscriptPath);
var meetingNote = await meetingNoteStore.ReadAsync(item.MeetingNotePath, cancellationToken);
var pipelineOptions = await RecordingSpeechRecognitionPipelineOptions.BuildAsync(
dictationWordProvider,
runOptions,
meetingNote,
cancellationToken);
await using var pipeline = pipelineFactory.Create(item.LaunchProfileName);
await pipeline.InitializeAsync(pipelineOptions, cancellationToken);
await pipeline.WaitUntilReadyAsync(cancellationToken);
await WriteRecordingToPipelineAsync(item.AudioPath, pipeline, cancellationToken);
var finishedSegments = await pipeline.ReadFinishedTranscriptAsync(
item.AudioPath,
pipelineOptions,
cancellationToken);
var lines = await TransformTranscriptLinesAsync(artifacts, runOptions, finishedSegments, cancellationToken);
await transcriptStore.ReplaceLinesAsync(session, lines, cancellationToken);
meetingNote.Frontmatter.EndTime = item.InferredEndTime ?? DateTimeOffset.Now;
var completedMeetingNote = await meetingNoteStore.SaveAsync(meetingNote, runOptions, cancellationToken);
await transcriptStore.UpdateMetadataAsync(session, artifacts, completedMeetingNote, cancellationToken);
await meetingArtifactStore.UpdateAssistantContextMeetingAsync(artifacts, completedMeetingNote, cancellationToken);
await TransitionMeetingAsync(
artifacts,
runOptions,
AssistantContextState.Transcribing,
AssistantContextState.Summarizing,
cancellationToken);
var summaryResult = await summaryPipeline.RunAsync(artifacts, runOptions, cancellationToken);
await TransitionMeetingAsync(
artifacts,
runOptions,
AssistantContextState.Summarizing,
summaryResult.Succeeded ? AssistantContextState.Finished : AssistantContextState.Error,
cancellationToken);
}
private MeetingAssistantOptions ResolveOptions(string? launchProfileName)
{
if (launchProfiles is not null)
{
return launchProfiles.GetRequiredProfile(launchProfileName).Options;
}
return options;
}
private async Task<List<string>> TransformTranscriptLinesAsync(
MeetingSessionArtifacts artifacts,
MeetingAssistantOptions runOptions,
IReadOnlyList<TranscriptionSegment> segments,
CancellationToken cancellationToken)
{
var lines = new List<string>(segments.Count);
foreach (var segment in segments)
{
var formatted = TranscriptLineFormatter.Format(segment);
lines.Add(await TransformTranscriptLineAsync(artifacts, runOptions, formatted, cancellationToken));
}
return lines;
}
private async Task<string> TransformTranscriptLineAsync(
MeetingSessionArtifacts artifacts,
MeetingAssistantOptions runOptions,
FormattedTranscriptLine formatted,
CancellationToken cancellationToken)
{
try
{
return await workflowEngine.TransformTranscriptLineAsync(
MeetingWorkflowEvent.TranscriptLine(artifacts, formatted.Line, formatted.Speaker),
runOptions,
cancellationToken);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw;
}
catch (Exception exception)
{
logger.LogError(
exception,
"Offline transcript line workflow failed for speaker {Speaker}; keeping original transcript line",
formatted.Speaker);
return formatted.Line;
}
}
private async Task TransitionMeetingAsync(
MeetingSessionArtifacts artifacts,
MeetingAssistantOptions runOptions,
AssistantContextState from,
AssistantContextState to,
CancellationToken cancellationToken)
{
await meetingArtifactStore.UpdateAssistantContextStateAsync(artifacts, to, cancellationToken);
await workflowEngine.RunAsync(
MeetingWorkflowEvent.StateTransition(artifacts, from, to),
runOptions,
cancellationToken);
}
private static async Task WriteRecordingToPipelineAsync(
string audioPath,
ISpeechRecognitionPipeline pipeline,
CancellationToken cancellationToken)
{
using var reader = new WaveFileReader(audioPath);
var bytesPerSecond = reader.WaveFormat.AverageBytesPerSecond;
var chunkSize = Math.Max(
reader.WaveFormat.BlockAlign,
bytesPerSecond * MaxChunkDurationMilliseconds / 1000);
chunkSize -= chunkSize % reader.WaveFormat.BlockAlign;
var buffer = new byte[chunkSize];
while (true)
{
cancellationToken.ThrowIfCancellationRequested();
var read = await reader.ReadAsync(buffer.AsMemory(0, buffer.Length), cancellationToken);
if (read == 0)
{
break;
}
await pipeline.WriteAsync(
new AudioChunk(
buffer[..read],
reader.WaveFormat.SampleRate,
reader.WaveFormat.Channels),
cancellationToken);
}
}
}
@@ -0,0 +1,20 @@
using MeetingAssistant.MeetingNotes;
using MeetingAssistant.Transcription;
namespace MeetingAssistant.Recording;
public static class RecordingSpeechRecognitionPipelineOptions
{
public static async Task<SpeechRecognitionPipelineOptions> BuildAsync(
IRecordingDictationWordProvider dictationWordProvider,
MeetingAssistantOptions runOptions,
MeetingNote? meetingNote,
CancellationToken cancellationToken)
{
var dictationWords = await dictationWordProvider.ReadOptionalWordsAsync(runOptions, cancellationToken);
var attendeeCount = meetingNote?.Frontmatter.Attendees.Count(attendee => !string.IsNullOrWhiteSpace(attendee)) ?? 0;
return attendeeCount > 1
? new SpeechRecognitionPipelineOptions(attendeeCount, dictationWords)
: new SpeechRecognitionPipelineOptions(DictationWords: dictationWords);
}
}
@@ -7,13 +7,16 @@ public sealed class TemporaryRecordedAudioStore : IRecordedAudioStore
{
private readonly MeetingAssistantOptions options;
private readonly ILogger<TemporaryRecordedAudioStore> logger;
private readonly IOfflineTranscriptionBacklog offlineTranscriptionBacklog;
public TemporaryRecordedAudioStore(
IOptions<MeetingAssistantOptions> options,
ILogger<TemporaryRecordedAudioStore> logger)
ILogger<TemporaryRecordedAudioStore> logger,
IOfflineTranscriptionBacklog? offlineTranscriptionBacklog = null)
{
this.options = options.Value;
this.logger = logger;
this.offlineTranscriptionBacklog = offlineTranscriptionBacklog ?? NoopOfflineTranscriptionBacklog.Instance;
}
public Task<IRecordedAudioSink> CreateSessionAsync(CancellationToken cancellationToken)
@@ -35,21 +38,28 @@ public sealed class TemporaryRecordedAudioStore : IRecordedAudioStore
return Task.FromResult<IRecordedAudioSink>(new TemporaryRecordedAudioSink(path, format, logger));
}
public Task DeleteStaleRecordingsAsync(CancellationToken cancellationToken)
public async Task DeleteStaleRecordingsAsync(CancellationToken cancellationToken)
{
var folder = GetTemporaryRecordingsFolder(options);
if (!Directory.Exists(folder))
{
return Task.CompletedTask;
return;
}
var queuedAudioPaths = (await offlineTranscriptionBacklog.ListAsync(cancellationToken))
.Select(item => item.AudioPath)
.ToHashSet(StringComparer.OrdinalIgnoreCase);
foreach (var path in Directory.EnumerateFiles(folder, "*.wav", SearchOption.TopDirectoryOnly))
{
cancellationToken.ThrowIfCancellationRequested();
if (queuedAudioPaths.Contains(path))
{
logger.LogInformation("Keeping queued offline transcription recording file {RecordingPath}", path);
continue;
}
DeleteFile(path);
}
return Task.CompletedTask;
}
private static string GetTemporaryRecordingsFolder(MeetingAssistantOptions options)
@@ -1,6 +1,7 @@
using MeetingAssistant.Transcription;
using MeetingAssistant.MeetingNotes;
using Microsoft.Extensions.Options;
using System.Collections.Concurrent;
namespace MeetingAssistant.Recording;
@@ -8,6 +9,7 @@ public sealed class VaultTranscriptStore : ITranscriptStore
{
private readonly MeetingAssistantOptions options;
private readonly ILogger<VaultTranscriptStore> logger;
private readonly ConcurrentDictionary<string, SemaphoreSlim> fileGates = new(StringComparer.OrdinalIgnoreCase);
public VaultTranscriptStore(IOptions<MeetingAssistantOptions> options, ILogger<VaultTranscriptStore> logger)
{
@@ -36,9 +38,50 @@ public sealed class VaultTranscriptStore : ITranscriptStore
return new TranscriptSession(path);
}
public Task AppendLineAsync(TranscriptSession session, string line, CancellationToken cancellationToken)
public async Task<TranscriptLineReference> AppendLineAsync(
TranscriptSession session,
string line,
CancellationToken cancellationToken)
{
return AppendToBodyAsync(session.TranscriptPath, line + Environment.NewLine, cancellationToken);
return await WithFileGateAsync(
session.TranscriptPath,
() => AppendToBodyAsync(session.TranscriptPath, line, cancellationToken),
cancellationToken);
}
public async Task ReplaceLineAsync(
TranscriptSession session,
TranscriptLineReference lineReference,
string replacementLine,
CancellationToken cancellationToken)
{
await WithFileGateAsync(
session.TranscriptPath,
async () =>
{
if (!File.Exists(session.TranscriptPath))
{
return;
}
var content = await File.ReadAllTextAsync(session.TranscriptPath, cancellationToken);
var (frontmatter, body) = MeetingArtifactFrontmatterRenderer.Split(content);
var lines = SplitLines(body);
if (lineReference.BodyLineIndex < 0 ||
lineReference.BodyLineIndex >= lines.Count ||
!string.Equals(lines[lineReference.BodyLineIndex], lineReference.OriginalLine, StringComparison.Ordinal))
{
logger.LogWarning(
"Could not replace transcript line in {TranscriptPath} at body line {BodyLineIndex} because the original line no longer matched",
session.TranscriptPath,
lineReference.BodyLineIndex);
return;
}
lines[lineReference.BodyLineIndex] = replacementLine;
await WriteBodyAsync(session.TranscriptPath, frontmatter, string.Join(Environment.NewLine, lines), cancellationToken);
},
cancellationToken);
}
public Task ReplaceLinesAsync(
@@ -46,18 +89,21 @@ public sealed class VaultTranscriptStore : ITranscriptStore
IReadOnlyList<string> replacementLines,
CancellationToken cancellationToken)
{
var existing = File.Exists(session.TranscriptPath)
? File.ReadAllText(session.TranscriptPath)
: "";
var (frontmatter, _) = MeetingArtifactFrontmatterRenderer.Split(existing);
var body = "# Meeting Transcript"
+ Environment.NewLine
+ Environment.NewLine
+ string.Concat(replacementLines.Select(line => line + Environment.NewLine));
var content = string.IsNullOrWhiteSpace(frontmatter)
? body
: "---" + Environment.NewLine + frontmatter + Environment.NewLine + "---" + Environment.NewLine + Environment.NewLine + body;
return File.WriteAllTextAsync(session.TranscriptPath, content, cancellationToken);
return WithFileGateAsync(
session.TranscriptPath,
async () =>
{
var existing = File.Exists(session.TranscriptPath)
? File.ReadAllText(session.TranscriptPath)
: "";
var (frontmatter, _) = MeetingArtifactFrontmatterRenderer.Split(existing);
var body = "# Meeting Transcript"
+ Environment.NewLine
+ Environment.NewLine
+ string.Concat(replacementLines.Select(line => line + Environment.NewLine));
await WriteBodyAsync(session.TranscriptPath, frontmatter, body, cancellationToken);
},
cancellationToken);
}
public async Task UpdateMetadataAsync(
@@ -66,50 +112,108 @@ public sealed class VaultTranscriptStore : ITranscriptStore
MeetingNote meetingNote,
CancellationToken cancellationToken)
{
var body = File.Exists(session.TranscriptPath)
? MeetingArtifactFrontmatterRenderer.Split(await File.ReadAllTextAsync(session.TranscriptPath, cancellationToken)).Body
: "# Meeting Transcript" + Environment.NewLine + Environment.NewLine;
var frontmatter = MeetingArtifactFrontmatterRenderer.Create(
artifacts,
meetingNote,
MeetingArtifactFrontmatterRenderer.DefaultTitle(meetingNote, "Meeting Transcript"),
session.TranscriptPath);
await File.WriteAllTextAsync(
await WithFileGateAsync(
session.TranscriptPath,
MeetingArtifactFrontmatterRenderer.Render(frontmatter, body),
async () =>
{
var body = File.Exists(session.TranscriptPath)
? MeetingArtifactFrontmatterRenderer.Split(await File.ReadAllTextAsync(session.TranscriptPath, cancellationToken)).Body
: "# Meeting Transcript" + Environment.NewLine + Environment.NewLine;
var frontmatter = MeetingArtifactFrontmatterRenderer.Create(
artifacts,
meetingNote,
MeetingArtifactFrontmatterRenderer.DefaultTitle(meetingNote, "Meeting Transcript"),
session.TranscriptPath);
await File.WriteAllTextAsync(
session.TranscriptPath,
MeetingArtifactFrontmatterRenderer.Render(frontmatter, body),
cancellationToken);
},
cancellationToken);
}
private static async Task AppendToBodyAsync(
private static async Task<TranscriptLineReference> AppendToBodyAsync(
string path,
string text,
string line,
CancellationToken cancellationToken)
{
if (!File.Exists(path))
{
await File.AppendAllTextAsync(path, text, cancellationToken);
return;
await File.AppendAllTextAsync(path, line + Environment.NewLine, cancellationToken);
return new TranscriptLineReference(path, 0, line);
}
var content = await File.ReadAllTextAsync(path, cancellationToken);
var (frontmatter, body) = MeetingArtifactFrontmatterRenderer.Split(content);
var bodyLineIndex = GetAppendBodyLineIndex(body);
if (string.IsNullOrWhiteSpace(frontmatter))
{
await File.AppendAllTextAsync(path, text, cancellationToken);
return;
await File.AppendAllTextAsync(path, line + Environment.NewLine, cancellationToken);
return new TranscriptLineReference(path, bodyLineIndex, line);
}
var updated = "---"
+ Environment.NewLine
+ frontmatter
+ Environment.NewLine
+ "---"
+ Environment.NewLine
+ Environment.NewLine
+ body
+ text;
await File.WriteAllTextAsync(path, updated, cancellationToken);
await WriteBodyAsync(path, frontmatter, body + line + Environment.NewLine, cancellationToken);
return new TranscriptLineReference(path, bodyLineIndex, line);
}
private async Task WithFileGateAsync(
string path,
Func<Task> action,
CancellationToken cancellationToken)
{
await WithFileGateAsync(
path,
async () =>
{
await action();
return true;
},
cancellationToken);
}
private async Task<T> WithFileGateAsync<T>(
string path,
Func<Task<T>> action,
CancellationToken cancellationToken)
{
var gate = fileGates.GetOrAdd(path, static _ => new SemaphoreSlim(1, 1));
await gate.WaitAsync(cancellationToken);
try
{
return await action();
}
finally
{
gate.Release();
}
}
private static async Task WriteBodyAsync(
string path,
string frontmatter,
string body,
CancellationToken cancellationToken)
{
var content = string.IsNullOrWhiteSpace(frontmatter)
? body
: "---" + Environment.NewLine + frontmatter + Environment.NewLine + "---" + Environment.NewLine + Environment.NewLine + body;
await File.WriteAllTextAsync(path, content, cancellationToken);
}
private static int GetAppendBodyLineIndex(string body)
{
var lines = SplitLines(body);
return lines.Count > 0 && lines[^1].Length == 0
? lines.Count - 1
: lines.Count;
}
private static List<string> SplitLines(string text)
{
return text
.Replace("\r\n", "\n", StringComparison.Ordinal)
.Split('\n')
.ToList();
}
}