Add the Nova Sonic backend swap and AsIRealtimeClient adapter

Phase 4b of the AWS.Speech.MEAI stack. Adds the two realtime-contract adapters
that let a team move between the pipeline and native Nova Sonic, or expose
either behind MEAI's IRealtimeClient, without changing call sites.

- RealtimeMessageMapper: shared, pure translation between VoiceAgentUpdate and
  MEAI's realtime messages, both directions. Assistant audio round-trips
  through the base64 string on OutputTextAudioRealtimeServerMessage.Audio.
- NovaSonicRunner: drives VoiceAgentBackend.NovaSonic. Opens a realtime session
  on an IRealtimeClient (Bedrock Nova Sonic), pumps microphone PCM in as
  InputAudioBufferAppend messages plus a final commit, and maps the session's
  server messages back to VoiceAgentUpdates so RunAsync yields the same stream
  as the pipeline backend.
- VoiceAgentRealtimeClient: exposes a VoiceAgent through IRealtimeClient. A
  session feeds pushed audio into the pipeline via a System.IO.Pipelines.Pipe
  and maps the resulting updates to server messages. Reuses the shared
  RealtimeAudioProtocol single-enumeration guard from AWS.Bedrock.MEAI.
- VoiceAgent: Create now builds a Nova IRealtimeClient for the NovaSonic
  backend (previously threw NotSupportedException); RunAsync routes by backend;
  AsIRealtimeClient(defaultModelId) is implemented. The provider-neutral
  constructor stays pipeline-only and RunAsync throws NotSupportedException if
  asked for NovaSonic without Create.

The message types match across backends; timing and cadence differ, as the
design notes call out.

Testing: mapper round-trips (incl. audio via base64) and unknown-type skip; a
NovaSonicRunner test with a fake IRealtimeClient/session (deterministic:
server messages gated on the input-audio commit) asserting audio pump, commit,
session dispose, mapped updates, and session-option wiring; an AsIRealtimeClient
test driving a stub-backed pipeline end to end through the realtime contract;
and the constructor NovaSonic guard. All Speech tests pass on net8.0 (43) and
net472 (7, unchanged). Bedrock/Nova unchanged (243 + 173); solution builds 0/0.

Author: GarrettBeatty

src/AWS.Speech.MEAI/NovaSonicRunner.cs

@@ -0,0 +1,77 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+#if NET8_0_OR_GREATER
+using Microsoft.Extensions.AI;
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace AWS.Speech.MEAI;
+
+/// <summary>
+/// Drives the <see cref="VoiceAgentBackend.NovaSonic"/> backend: opens a realtime session on the
+/// supplied <see cref="IRealtimeClient"/> (Amazon Bedrock Nova Sonic), streams microphone PCM into it
+/// as input-audio messages, and maps the session's realtime server messages back to
+/// <see cref="VoiceAgentUpdate"/>s so the caller sees the same stream as the pipeline backend.
+/// </summary>
+internal static class NovaSonicRunner
+{
+    private const int AudioChunkBytes = 8192;
+
+    public static async IAsyncEnumerable<VoiceAgentUpdate> RunAsync(
+        IRealtimeClient client, VoiceAgentOptions options, Stream microphonePcm,
+        [EnumeratorCancellation] CancellationToken cancellationToken)
+    {
+        var sessionOptions = new RealtimeSessionOptions
+        {
+            Model = options.ModelId,
+            Instructions = options.Instructions,
+            Voice = options.Voice.Value,
+            InputAudioFormat = new RealtimeAudioFormat("audio/lpcm", options.InputSampleRateHertz),
+            OutputAudioFormat = new RealtimeAudioFormat("audio/lpcm", options.OutputSampleRateHertz),
+        };
+
+        var session = await client.CreateSessionAsync(sessionOptions, cancellationToken).ConfigureAwait(false);
+        using var pumpCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+
+        var pump = Task.Run(() => PumpAudioAsync(session, microphonePcm, pumpCts.Token), pumpCts.Token);
+        try
+        {
+            await foreach (var message in session.GetStreamingResponseAsync(cancellationToken).ConfigureAwait(false))
+            {
+                if (RealtimeMessageMapper.ToVoiceAgentUpdate(message) is { } update)
+                {
+                    yield return update;
+                }
+            }
+        }
+        finally
+        {
+            pumpCts.Cancel();
+            try { await pump.ConfigureAwait(false); }
+            catch (OperationCanceledException) { /* expected on teardown */ }
+            await session.DisposeAsync().ConfigureAwait(false);
+        }
+    }
+
+    private static async Task PumpAudioAsync(IRealtimeClientSession session, Stream microphonePcm, CancellationToken token)
+    {
+        var buffer = new byte[AudioChunkBytes];
+        int read;
+        while ((read = await microphonePcm.ReadAsync(buffer, 0, buffer.Length, token).ConfigureAwait(false)) > 0)
+        {
+            var chunk = new byte[read];
+            Array.Copy(buffer, 0, chunk, 0, read);
+            var message = new InputAudioBufferAppendRealtimeClientMessage(new DataContent(chunk, "audio/lpcm"));
+            await session.SendAsync(message, token).ConfigureAwait(false);
+        }
+
+        // End of the caller's audio: commit so the model can finalize the current utterance.
+        await session.SendAsync(new InputAudioBufferCommitRealtimeClientMessage(), token).ConfigureAwait(false);
+    }
+}
+#endif

src/AWS.Speech.MEAI/RealtimeMessageMapper.cs

@@ -0,0 +1,121 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+#if NET8_0_OR_GREATER
+using Microsoft.Extensions.AI;
+using System;
+
+namespace AWS.Speech.MEAI;
+
+/// <summary>
+/// Translates between the <see cref="VoiceAgentUpdate"/> stream and MEAI's realtime message vocabulary,
+/// so the pipeline can be exposed as an <see cref="IRealtimeClient"/> and a Nova Sonic session can be
+/// consumed as a <see cref="VoiceAgent"/>. The message types line up; timing and cadence do not.
+/// </summary>
+internal static class RealtimeMessageMapper
+{
+    /// <summary>Maps a realtime server message to a <see cref="VoiceAgentUpdate"/>, or <see langword="null"/> to skip it.</summary>
+    public static VoiceAgentUpdate? ToVoiceAgentUpdate(RealtimeServerMessage message)
+    {
+        var type = message.Type.Value;
+
+        if (type == RealtimeServerMessageType.InputAudioTranscriptionDelta.Value &&
+            message is InputAudioTranscriptionRealtimeServerMessage partial)
+        {
+            return new VoiceAgentUpdate { Kind = VoiceAgentUpdateKind.UserTranscriptPartial, Text = partial.Transcription };
+        }
+
+        if (type == RealtimeServerMessageType.InputAudioTranscriptionCompleted.Value &&
+            message is InputAudioTranscriptionRealtimeServerMessage final)
+        {
+            return new VoiceAgentUpdate { Kind = VoiceAgentUpdateKind.UserTranscriptFinal, Text = final.Transcription, IsFinal = true };
+        }
+
+        if (type == RealtimeServerMessageType.ResponseCreated.Value && message is ResponseCreatedRealtimeServerMessage created)
+        {
+            return new VoiceAgentUpdate { Kind = VoiceAgentUpdateKind.TurnStarted, ResponseId = created.ResponseId };
+        }
+
+        if (type == RealtimeServerMessageType.OutputTextDelta.Value && message is OutputTextAudioRealtimeServerMessage text)
+        {
+            return new VoiceAgentUpdate { Kind = VoiceAgentUpdateKind.AssistantText, Text = text.Text, ResponseId = text.ResponseId };
+        }
+
+        if (type == RealtimeServerMessageType.OutputAudioDelta.Value && message is OutputTextAudioRealtimeServerMessage audio)
+        {
+            var bytes = string.IsNullOrEmpty(audio.Audio) ? Array.Empty<byte>() : Convert.FromBase64String(audio.Audio);
+            return new VoiceAgentUpdate { Kind = VoiceAgentUpdateKind.AssistantAudio, Audio = bytes, ResponseId = audio.ResponseId };
+        }
+
+        if (type == RealtimeServerMessageType.ResponseDone.Value && message is ResponseCreatedRealtimeServerMessage done)
+        {
+            var cancelled = string.Equals(done.Status, RealtimeResponseStatus.Cancelled, StringComparison.Ordinal);
+            return new VoiceAgentUpdate
+            {
+                Kind = cancelled ? VoiceAgentUpdateKind.Cancelled : VoiceAgentUpdateKind.TurnComplete,
+                Usage = done.Usage,
+                ResponseId = done.ResponseId,
+            };
+        }
+
+        return null;
+    }
+
+    /// <summary>Maps a <see cref="VoiceAgentUpdate"/> to a realtime server message, or <see langword="null"/> to skip it.</summary>
+    public static RealtimeServerMessage? ToServerMessage(VoiceAgentUpdate update)
+    {
+        switch (update.Kind)
+        {
+            case VoiceAgentUpdateKind.UserTranscriptPartial:
+                return new InputAudioTranscriptionRealtimeServerMessage(RealtimeServerMessageType.InputAudioTranscriptionDelta)
+                {
+                    Transcription = update.Text,
+                };
+
+            case VoiceAgentUpdateKind.UserTranscriptFinal:
+                return new InputAudioTranscriptionRealtimeServerMessage(RealtimeServerMessageType.InputAudioTranscriptionCompleted)
+                {
+                    Transcription = update.Text,
+                };
+
+            case VoiceAgentUpdateKind.TurnStarted:
+                return new ResponseCreatedRealtimeServerMessage(RealtimeServerMessageType.ResponseCreated)
+                {
+                    ResponseId = update.ResponseId,
+                };
+
+            case VoiceAgentUpdateKind.AssistantText:
+                return new OutputTextAudioRealtimeServerMessage(RealtimeServerMessageType.OutputTextDelta)
+                {
+                    Text = update.Text,
+                    ResponseId = update.ResponseId,
+                };
+
+            case VoiceAgentUpdateKind.AssistantAudio:
+                return new OutputTextAudioRealtimeServerMessage(RealtimeServerMessageType.OutputAudioDelta)
+                {
+                    Audio = update.Audio is { } pcm ? Convert.ToBase64String(pcm.ToArray()) : string.Empty,
+                    ResponseId = update.ResponseId,
+                };
+
+            case VoiceAgentUpdateKind.TurnComplete:
+                return new ResponseCreatedRealtimeServerMessage(RealtimeServerMessageType.ResponseDone)
+                {
+                    Status = RealtimeResponseStatus.Completed,
+                    Usage = update.Usage,
+                    ResponseId = update.ResponseId,
+                };
+
+            case VoiceAgentUpdateKind.Cancelled:
+                return new ResponseCreatedRealtimeServerMessage(RealtimeServerMessageType.ResponseDone)
+                {
+                    Status = RealtimeResponseStatus.Cancelled,
+                    ResponseId = update.ResponseId,
+                };
+
+            default:
+                return null;
+        }
+    }
+}
+#endif

src/AWS.Speech.MEAI/VoiceAgent.cs

@@ -12,6 +12,7 @@
 using System.Collections.Generic;
 using System.Diagnostics.CodeAnalysis;
 using System.IO;
+using System.Runtime.ExceptionServices;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -23,61 +24,75 @@ namespace AWS.Speech.MEAI;
 /// <see cref="VoiceAgentUpdate"/>s.
 /// </summary>
 /// <remarks>
-/// The default backend is <see cref="VoiceAgentBackend.Pipeline"/>: Amazon Transcribe streaming for
-/// STT, Amazon Bedrock via <c>AWS.Bedrock.MEAI</c> for reasoning, and Amazon Polly for TTS. Call
-/// <see cref="Create"/> for the one-line default-chain factory, or use the constructor to compose any
-/// MEAI clients you already have. Barge-in, the Nova Sonic backend swap, the <c>AsIRealtimeClient()</c>
-/// adapter, and DI registration are wired in a later phase.
+/// The default backend is <see cref="VoiceAgentBackend.Pipeline"/>: Amazon Transcribe streaming for STT,
+/// Amazon Bedrock via <c>AWS.Bedrock.MEAI</c> for reasoning, and Amazon Polly for TTS. Set
+/// <see cref="VoiceAgentOptions.Backend"/> to <see cref="VoiceAgentBackend.NovaSonic"/> on
+/// <see cref="Create"/> to route the same <c>RunAsync</c> stream through Amazon Bedrock Nova Sonic
+/// instead. <see cref="AsIRealtimeClient"/> exposes either backend through MEAI's
+/// <see cref="IRealtimeClient"/> contract.
 /// </remarks>
 [Experimental("MEAI001")]
 public sealed class VoiceAgent : IAsyncDisposable
 {
-    private readonly ISpeechToTextClient _stt;
-    private readonly IChatClient _chat;
-    private readonly ITextToSpeechClient _tts;
+    private readonly ISpeechToTextClient? _stt;
+    private readonly IChatClient? _chat;
+    private readonly ITextToSpeechClient? _tts;
+    private readonly IRealtimeClient? _novaClient;
     private readonly VoiceAgentOptions _options;
     private readonly List<IDisposable> _ownedResources;
     private int _disposed;
 
-    /// <summary>Initializes a provider-neutral <see cref="VoiceAgent"/> around any MEAI clients.</summary>
+    /// <summary>Initializes a provider-neutral pipeline <see cref="VoiceAgent"/> around any MEAI clients.</summary>
     /// <exception cref="ArgumentNullException">A client is <see langword="null"/>.</exception>
     public VoiceAgent(ISpeechToTextClient stt, IChatClient chat, ITextToSpeechClient tts, VoiceAgentOptions? options = null)
-        : this(stt, chat, tts, options ?? new VoiceAgentOptions(), ownedResources: null)
+        : this(
+            stt ?? throw new ArgumentNullException(nameof(stt)),
+            chat ?? throw new ArgumentNullException(nameof(chat)),
+            tts ?? throw new ArgumentNullException(nameof(tts)),
+            novaClient: null,
+            options ?? new VoiceAgentOptions(),
+            ownedResources: null)
     {
     }
 
-    private VoiceAgent(ISpeechToTextClient stt, IChatClient chat, ITextToSpeechClient tts,
+    private VoiceAgent(
+        ISpeechToTextClient? stt, IChatClient? chat, ITextToSpeechClient? tts, IRealtimeClient? novaClient,
         VoiceAgentOptions options, List<IDisposable>? ownedResources)
     {
-        _stt = stt ?? throw new ArgumentNullException(nameof(stt));
-        _chat = chat ?? throw new ArgumentNullException(nameof(chat));
-        _tts = tts ?? throw new ArgumentNullException(nameof(tts));
+        _stt = stt;
+        _chat = chat;
+        _tts = tts;
+        _novaClient = novaClient;
         _options = options;
         _ownedResources = ownedResources ?? new List<IDisposable>();
     }
 
     /// <summary>Creates a <see cref="VoiceAgent"/> with the default AWS credential and region chains.</summary>
     /// <remarks>
-    /// Constructs an Amazon Transcribe streaming client, an Amazon Bedrock runtime client (adapted via
-    /// <c>AmazonBedrockRuntimeExtensions.AsIChatClient</c>), and an Amazon Polly client, honoring
-    /// <see cref="VoiceAgentOptions.Credentials"/> and <see cref="VoiceAgentOptions.Region"/> when set.
-    /// A pre-built client on the options object wins over the default AWS client for that leg. The
-    /// returned agent owns any clients it constructed and disposes them on <see cref="DisposeAsync"/>.
+    /// For <see cref="VoiceAgentBackend.Pipeline"/> this constructs an Amazon Transcribe streaming client,
+    /// an Amazon Bedrock runtime client adapted to an <see cref="IChatClient"/>, and an Amazon Polly client.
+    /// For <see cref="VoiceAgentBackend.NovaSonic"/> it constructs an Amazon Bedrock runtime client adapted
+    /// to an <see cref="IRealtimeClient"/>. Both honor <see cref="VoiceAgentOptions.Credentials"/> and
+    /// <see cref="VoiceAgentOptions.Region"/>, and a pre-built client on the options wins over the default
+    /// AWS client for its leg. The returned agent owns any clients it constructed and disposes them on
+    /// <see cref="DisposeAsync"/>.
     /// </remarks>
-    /// <exception cref="NotSupportedException">The requested backend is not yet available in this preview.</exception>
     public static VoiceAgent Create(Action<VoiceAgentOptions>? configure = null)
     {
         var options = new VoiceAgentOptions();
         configure?.Invoke(options);
 
-        if (options.Backend != VoiceAgentBackend.Pipeline)
+        var owned = new List<IDisposable>();
+
+        if (options.Backend == VoiceAgentBackend.NovaSonic)
         {
-            throw new NotSupportedException(
-                $"The {options.Backend} backend is not available yet in this preview. Use VoiceAgentBackend.Pipeline.");
+            var bedrock = CreateBedrockClient(options.Credentials, options.Region);
+            owned.Add(bedrock);
+            var nova = bedrock.AsIRealtimeClient(options.ModelId);
+            if (nova is IDisposable disposableNova) owned.Add(disposableNova);
+            return new VoiceAgent(stt: null, chat: null, tts: null, nova, options, owned);
         }
 
-        var owned = new List<IDisposable>();
-
         var stt = options.SpeechToTextClient;
         if (stt is null)
         {
@@ -102,24 +117,48 @@ public static VoiceAgent Create(Action<VoiceAgentOptions>? configure = null)
             tts = polly.AsITextToSpeechClient(options.Voice, Engine.Neural, options.OutputSampleRateHertz);
         }
 
-        return new VoiceAgent(stt, chat, tts, options, owned);
+        return new VoiceAgent(stt, chat, tts, novaClient: null, options, owned);
     }
 
     /// <summary>Runs the voice loop over the caller's microphone PCM stream.</summary>
     /// <param name="microphonePcm">
-    /// Input audio at <see cref="VoiceAgentOptions.InputSampleRateHertz"/>, 16-bit signed
-    /// little-endian mono PCM. The agent never disposes this stream.
+    /// Input audio at <see cref="VoiceAgentOptions.InputSampleRateHertz"/>, 16-bit signed little-endian
+    /// mono PCM. The agent never disposes this stream.
     /// </param>
     /// <param name="cancellationToken">Stops the loop; the returned enumerable then completes.</param>
     /// <returns>One ordered stream of <see cref="VoiceAgentUpdate"/>s.</returns>
     /// <exception cref="ArgumentNullException"><paramref name="microphonePcm"/> is <see langword="null"/>.</exception>
     /// <exception cref="ObjectDisposedException">The agent has been disposed.</exception>
+    /// <exception cref="NotSupportedException">
+    /// The agent is configured for <see cref="VoiceAgentBackend.NovaSonic"/> but was built with the
+    /// provider-neutral constructor, which is pipeline-only. Use <see cref="Create"/> for Nova Sonic.
+    /// </exception>
     public IAsyncEnumerable<VoiceAgentUpdate> RunAsync(Stream microphonePcm, CancellationToken cancellationToken = default)
     {
         if (microphonePcm is null) throw new ArgumentNullException(nameof(microphonePcm));
         ThrowIfDisposed();
 
-        return VoiceAgentPipeline.RunAsync(_stt, _chat, _tts, _options, microphonePcm, cancellationToken);
+        if (_options.Backend == VoiceAgentBackend.NovaSonic)
+        {
+            if (_novaClient is null)
+            {
+                throw new NotSupportedException(
+                    "The NovaSonic backend requires VoiceAgent.Create; the provider-neutral constructor is pipeline-only.");
+            }
+            return NovaSonicRunner.RunAsync(_novaClient, _options, microphonePcm, cancellationToken);
+        }
+
+        return VoiceAgentPipeline.RunAsync(_stt!, _chat!, _tts!, _options, microphonePcm, cancellationToken);
+    }
+
+    /// <summary>Exposes this agent through MEAI's <see cref="IRealtimeClient"/> contract.</summary>
+    /// <param name="defaultModelId">An optional default model ID recorded on the adapter.</param>
+    /// <returns>An <see cref="IRealtimeClient"/> whose sessions drive this agent's loop.</returns>
+    /// <exception cref="ObjectDisposedException">The agent has been disposed.</exception>
+    public IRealtimeClient AsIRealtimeClient(string? defaultModelId = null)
+    {
+        ThrowIfDisposed();
+        return new VoiceAgentRealtimeClient(this, defaultModelId);
     }
 
     /// <summary>Returns the underlying MEAI client for the requested service type, or <see langword="null"/>.</summary>
@@ -131,6 +170,7 @@ public IAsyncEnumerable<VoiceAgentUpdate> RunAsync(Stream microphonePcm, Cancell
         if (serviceType == typeof(ISpeechToTextClient)) return _stt;
         if (serviceType == typeof(IChatClient)) return _chat;
         if (serviceType == typeof(ITextToSpeechClient)) return _tts;
+        if (serviceType == typeof(IRealtimeClient)) return _novaClient;
         return serviceType.IsInstanceOfType(this) ? this : null;
     }
 
@@ -149,7 +189,7 @@ public ValueTask DisposeAsync()
 
         if (first is not null)
         {
-            System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(first).Throw();
+            ExceptionDispatchInfo.Capture(first).Throw();
         }
         return default;
     }