Add barge-in and AddVoiceAgent DI to the pipeline

Phase 4a of the AWS.Speech.MEAI stack. Completes the pipeline's usability with
interruption support and first-class dependency injection. The Nova Sonic
backend swap and AsIRealtimeClient() adapter follow in the next PR.

Barge-in:
- VoiceAgentPipeline is now an instance runner holding the current turn's
  cancellation source under a lock, so the STT consumer can interrupt the
  reasoning worker's in-flight turn without a data race between the two
  producer tasks.
- When EnableBargeIn is set, a qualifying STT partial arriving during an active
  turn cancels that turn's per-turn token. The reasoning worker observes the
  cancellation, keeps the partial reply in history so conversation state stays
  coherent, emits a Cancelled update, and moves on to the next turn. Turn
  lifecycle and audio are written with the loop token, so a cancelled turn's
  Cancelled update is still delivered.
- BargeInDetector.ShouldInterrupt is a pure noise-threshold predicate
  (minimum trimmed partial length), kept separate so the threshold is testable.

Dependency injection:
- AddVoiceAgent(IServiceCollection, Action<VoiceAgentOptions>) registers the
  three AWS clients with TryAddAWSService and the three MEAI clients plus
  VoiceAgent with TryAddSingleton, so a caller's own registration of any leg
  wins. A pre-built client on VoiceAgentOptions wins over the AWS client for
  that leg. Mirrors the AWS.AgentCore.Hosting registration pattern.
- The DI packages (AWSSDK.Extensions.NETCore.Setup,
  Microsoft.Extensions.DependencyInjection.Abstractions) are referenced only on
  net8.0, so the down-level Polly TTS surface stays lean.

Testing: 7 BargeInDetector cases, 2 barge-in pipeline tests (a deterministic
gated-stub interruption that cancels turn 1 and drives turn 2, plus a
barge-in-disabled path), and 3 DI tests (prebuilt-client composition,
pre-registered IChatClient wins, null-arg guard). All Speech tests pass on
net8.0 (36) and net472 (7, unchanged). Bedrock/Nova unchanged (243 + 173);
full solution builds 0/0.

Author: GarrettBeatty

src/AWS.Speech.MEAI/AWS.Speech.MEAI.csproj

@@ -42,9 +42,13 @@
         <PackageReference Include="Microsoft.SourceLink.GitHub" Version="10.0.300" PrivateAssets="All" />
     </ItemGroup>
 
-    <!-- Amazon Transcribe streaming requires HTTP/2, which is unavailable on net472/netstandard2.0. -->
+    <!-- Amazon Transcribe streaming requires HTTP/2, which is unavailable on net472/netstandard2.0.
+         The VoiceAgent loop and its AddVoiceAgent DI helper are net8-only for the same reason, so the
+         DI packages are referenced only there and never burden the down-level Polly TTS surface. -->
     <ItemGroup Condition="'$(TargetFramework)' == 'net8.0'">
         <PackageReference Include="AWSSDK.TranscribeStreaming" Version="4.0.101.1" />
+        <PackageReference Include="AWSSDK.Extensions.NETCore.Setup" Version="4.0.3.35" />
+        <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.0" />
     </ItemGroup>
 
     <ItemGroup>

src/AWS.Speech.MEAI/BargeInDetector.cs

@@ -0,0 +1,21 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+#if NET8_0_OR_GREATER
+namespace AWS.Speech.MEAI;
+
+/// <summary>
+/// Decides whether a speech-to-text partial arriving while the assistant is speaking is a real
+/// interruption (barge-in) rather than noise. Kept as a pure predicate so the threshold is testable
+/// without driving the whole loop.
+/// </summary>
+internal static class BargeInDetector
+{
+    /// <summary>Minimum trimmed partial length, in characters, that counts as a real interruption.</summary>
+    internal const int MinPartialChars = 3;
+
+    /// <summary>Returns <see langword="true"/> if <paramref name="partialText"/> clears the noise threshold.</summary>
+    public static bool ShouldInterrupt(string? partialText) =>
+        partialText is not null && partialText.Trim().Length >= MinPartialChars;
+}
+#endif

src/AWS.Speech.MEAI/VoiceAgentPipeline.cs

@@ -0,0 +1,95 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+#if NET8_0_OR_GREATER
+using Amazon.BedrockRuntime;
+using Amazon.Polly;
+using Amazon.TranscribeStreaming;
+using AWS.Speech.MEAI;
+using Microsoft.Extensions.AI;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+using System;
+using System.Diagnostics.CodeAnalysis;
+
+namespace Microsoft.Extensions.DependencyInjection;
+
+/// <summary>Dependency-injection extensions for registering a <see cref="VoiceAgent"/>.</summary>
+[Experimental("MEAI001")]
+public static class VoiceAgentServiceCollectionExtensions
+{
+    /// <summary>Registers a <see cref="VoiceAgent"/> and the MEAI speech clients it composes.</summary>
+    /// <param name="services">The service collection.</param>
+    /// <param name="configure">Optional configuration for <see cref="VoiceAgentOptions"/>.</param>
+    /// <returns>The same <see cref="IServiceCollection"/> for chaining.</returns>
+    /// <remarks>
+    /// Every registration uses <c>TryAdd</c>, so a caller who has already registered an
+    /// <see cref="IAmazonPolly"/>, <see cref="IAmazonTranscribeStreaming"/>, <see cref="IAmazonBedrockRuntime"/>,
+    /// <see cref="ISpeechToTextClient"/>, <see cref="IChatClient"/>, or <see cref="ITextToSpeechClient"/> keeps
+    /// their own registration. A pre-built client on <see cref="VoiceAgentOptions"/> wins over the AWS client
+    /// for that leg. The three AWS clients come from <c>AddAWSService</c>'s default option/credential resolution;
+    /// <see cref="VoiceAgentOptions.Region"/> and <see cref="VoiceAgentOptions.Credentials"/> are honored only by
+    /// <see cref="VoiceAgent.Create"/>, not by this DI path, where AWS options are configured through the container.
+    /// </remarks>
+    /// <exception cref="ArgumentNullException"><paramref name="services"/> is <see langword="null"/>.</exception>
+    public static IServiceCollection AddVoiceAgent(this IServiceCollection services, Action<VoiceAgentOptions>? configure = null)
+    {
+        if (services is null) throw new ArgumentNullException(nameof(services));
+
+        var options = new VoiceAgentOptions();
+        configure?.Invoke(options);
+
+        RegisterSpeechToText(services, options);
+        RegisterChat(services, options);
+        RegisterTextToSpeech(services, options);
+
+        services.TryAddSingleton(sp => new VoiceAgent(
+            sp.GetRequiredService<ISpeechToTextClient>(),
+            sp.GetRequiredService<IChatClient>(),
+            sp.GetRequiredService<ITextToSpeechClient>(),
+            options));
+
+        return services;
+    }
+
+    private static void RegisterSpeechToText(IServiceCollection services, VoiceAgentOptions options)
+    {
+        if (options.SpeechToTextClient is { } stt)
+        {
+            services.TryAddSingleton(stt);
+            return;
+        }
+
+        services.TryAddAWSService<IAmazonTranscribeStreaming>();
+        services.TryAddSingleton<ISpeechToTextClient>(sp =>
+            sp.GetRequiredService<IAmazonTranscribeStreaming>()
+                .AsISpeechToTextClient(options.Language, options.InputSampleRateHertz));
+    }
+
+    private static void RegisterChat(IServiceCollection services, VoiceAgentOptions options)
+    {
+        if (options.ChatClient is { } chat)
+        {
+            services.TryAddSingleton(chat);
+            return;
+        }
+
+        services.TryAddAWSService<IAmazonBedrockRuntime>();
+        services.TryAddSingleton<IChatClient>(sp =>
+            sp.GetRequiredService<IAmazonBedrockRuntime>().AsIChatClient(options.ModelId));
+    }
+
+    private static void RegisterTextToSpeech(IServiceCollection services, VoiceAgentOptions options)
+    {
+        if (options.TextToSpeechClient is { } tts)
+        {
+            services.TryAddSingleton(tts);
+            return;
+        }
+
+        services.TryAddAWSService<IAmazonPolly>();
+        services.TryAddSingleton<ITextToSpeechClient>(sp =>
+            sp.GetRequiredService<IAmazonPolly>()
+                .AsITextToSpeechClient(options.Voice, Engine.Neural, options.OutputSampleRateHertz));
+    }
+}
+#endif

src/AWS.Speech.MEAI/VoiceAgentServiceCollectionExtensions.cs

@@ -22,6 +22,7 @@
 
     <ItemGroup>
         <PackageReference Include="Microsoft.Extensions.AI.Abstractions" Version="10.8.3" />
+        <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="10.0.0" />
         <PackageReference Include="Microsoft.NET.Test.Sdk" Version="18.5.1" />
         <PackageReference Include="Moq" Version="4.20.72" />
         <PackageReference Include="xunit.v3" Version="3.2.2" />

test/AWS.Speech.MEAI.UnitTests/AWS.Speech.MEAI.UnitTests.csproj

@@ -0,0 +1,32 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+#if NET8_0_OR_GREATER
+using Xunit;
+
+namespace AWS.Speech.MEAI;
+
+public class BargeInDetectorTests
+{
+    [Theory]
+    [Trait("UnitTest", "Speech")]
+    [InlineData("yes please stop")]
+    [InlineData("wait")]
+    [InlineData("  hey  ")]
+    public void ShouldInterrupt_QualifyingPartial_ReturnsTrue(string partial)
+    {
+        Assert.True(BargeInDetector.ShouldInterrupt(partial));
+    }
+
+    [Theory]
+    [Trait("UnitTest", "Speech")]
+    [InlineData(null)]
+    [InlineData("")]
+    [InlineData("   ")]
+    [InlineData("hi")]
+    public void ShouldInterrupt_NoiseOrTooShort_ReturnsFalse(string? partial)
+    {
+        Assert.False(BargeInDetector.ShouldInterrupt(partial));
+    }
+}
+#endif

test/AWS.Speech.MEAI.UnitTests/BargeInDetectorTests.cs

@@ -0,0 +1,217 @@
+// 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.Linq;
+using System.Runtime.CompilerServices;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Xunit;
+
+namespace AWS.Speech.MEAI;
+
+public class VoiceAgentBargeInTests
+{
+    [Fact]
+    [Trait("UnitTest", "Speech")]
+    public async Task RunAsync_BargeInEnabled_PartialDuringTurn_CancelsAndStartsNewTurn()
+    {
+        // STT scripts turn 1's final, waits for the test to release the gate, then a qualifying partial
+        // (barge-in) followed by turn 2's final. The gate is released once the caller observes TurnStarted.
+        var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        var stt = new GatedStt(gate);
+        var chat = new BlockingFirstTurnChat();
+        var tts = new EchoTts();
+
+        var agent = new VoiceAgent(stt, chat, tts, new VoiceAgentOptions { EnableBargeIn = true });
+        using var mic = new MemoryStream(new byte[] { 0 });
+        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(15));
+
+        var updates = new List<VoiceAgentUpdate>();
+        bool released = false;
+        await foreach (var update in agent.RunAsync(mic, timeout.Token))
+        {
+            updates.Add(update);
+            if (!released && update.Kind == VoiceAgentUpdateKind.TurnStarted)
+            {
+                released = true;
+                gate.SetResult();   // let the interrupting partial flow
+            }
+        }
+
+        Assert.Contains(updates, u => u.Kind == VoiceAgentUpdateKind.Cancelled);
+
+        // Turn 1 was cancelled before it could complete; the only TurnComplete belongs to turn 2.
+        int cancelledIdx = updates.FindIndex(u => u.Kind == VoiceAgentUpdateKind.Cancelled);
+        int firstCompleteIdx = updates.FindIndex(u => u.Kind == VoiceAgentUpdateKind.TurnComplete);
+        Assert.True(firstCompleteIdx < 0 || cancelledIdx < firstCompleteIdx,
+            "The cancelled turn must not have produced a TurnComplete before the barge-in.");
+
+        // The barge-in path drove a second turn to completion.
+        Assert.Equal(2, chat.CallCount);
+        Assert.Contains(updates, u => u.Kind == VoiceAgentUpdateKind.TurnComplete);
+    }
+
+    [Fact]
+    [Trait("UnitTest", "Speech")]
+    public async Task RunAsync_BargeInDisabled_ShortNoisePartial_DoesNotCancel()
+    {
+        // Barge-in off and a below-threshold partial: the single turn completes normally.
+        var stt = new ScriptedNoGateStt(new[]
+        {
+            new SpeechToTextResponseUpdate("hi") { Kind = SpeechToTextResponseUpdateKind.TextUpdating },
+            new SpeechToTextResponseUpdate("hello") { Kind = SpeechToTextResponseUpdateKind.TextUpdated },
+        });
+        var chat = new QuickChat("Sure.");
+        var tts = new EchoTts();
+
+        var agent = new VoiceAgent(stt, chat, tts, new VoiceAgentOptions { EnableBargeIn = false });
+        using var mic = new MemoryStream(new byte[] { 0 });
+        using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(15));
+
+        var updates = new List<VoiceAgentUpdate>();
+        await foreach (var update in agent.RunAsync(mic, timeout.Token))
+        {
+            updates.Add(update);
+        }
+
+        Assert.DoesNotContain(updates, u => u.Kind == VoiceAgentUpdateKind.Cancelled);
+        Assert.Contains(updates, u => u.Kind == VoiceAgentUpdateKind.TurnComplete);
+    }
+
+    // ---- test doubles ----
+
+    private sealed class GatedStt : ISpeechToTextClient
+    {
+        private readonly TaskCompletionSource _gate;
+        public GatedStt(TaskCompletionSource gate) => _gate = gate;
+
+        public Task<SpeechToTextResponse> GetTextAsync(Stream a, SpeechToTextOptions? o, CancellationToken ct) =>
+            throw new NotImplementedException();
+
+        public async IAsyncEnumerable<SpeechToTextResponseUpdate> GetStreamingTextAsync(
+            Stream audioSpeechStream, SpeechToTextOptions? options,
+            [EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            yield return new SpeechToTextResponseUpdate("first question") { Kind = SpeechToTextResponseUpdateKind.TextUpdated };
+
+            // Wait until the caller has seen turn 1 start, then interrupt it.
+            await _gate.Task.WaitAsync(cancellationToken).ConfigureAwait(false);
+
+            yield return new SpeechToTextResponseUpdate("actually wait") { Kind = SpeechToTextResponseUpdateKind.TextUpdating };
+            yield return new SpeechToTextResponseUpdate("actually wait") { Kind = SpeechToTextResponseUpdateKind.TextUpdated };
+        }
+
+        public object? GetService(System.Type serviceType, object? serviceKey = null) =>
+            serviceType.IsInstanceOfType(this) ? this : null;
+
+        public void Dispose() { }
+    }
+
+    private sealed class ScriptedNoGateStt : ISpeechToTextClient
+    {
+        private readonly SpeechToTextResponseUpdate[] _updates;
+        public ScriptedNoGateStt(SpeechToTextResponseUpdate[] updates) => _updates = updates;
+
+        public Task<SpeechToTextResponse> GetTextAsync(Stream a, SpeechToTextOptions? o, CancellationToken ct) =>
+            throw new NotImplementedException();
+
+        public async IAsyncEnumerable<SpeechToTextResponseUpdate> GetStreamingTextAsync(
+            Stream audioSpeechStream, SpeechToTextOptions? options,
+            [EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            foreach (var u in _updates)
+            {
+                cancellationToken.ThrowIfCancellationRequested();
+                yield return u;
+                await Task.Yield();
+            }
+        }
+
+        public object? GetService(System.Type serviceType, object? serviceKey = null) =>
+            serviceType.IsInstanceOfType(this) ? this : null;
+
+        public void Dispose() { }
+    }
+
+    private sealed class BlockingFirstTurnChat : IChatClient
+    {
+        public int CallCount { get; private set; }
+
+        public Task<ChatResponse> GetResponseAsync(IEnumerable<ChatMessage> m, ChatOptions? o, CancellationToken ct) =>
+            throw new NotImplementedException();
+
+        public async IAsyncEnumerable<ChatResponseUpdate> GetStreamingResponseAsync(
+            IEnumerable<ChatMessage> messages, ChatOptions? options,
+            [EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            CallCount++;
+            if (CallCount == 1)
+            {
+                yield return new ChatResponseUpdate(ChatRole.Assistant, "Working on it. ");
+                // Block until the turn's token is cancelled by barge-in.
+                await Task.Delay(Timeout.Infinite, cancellationToken).ConfigureAwait(false);
+                yield break;
+            }
+
+            yield return new ChatResponseUpdate(ChatRole.Assistant, "Second answer.");
+            await Task.Yield();
+        }
+
+        public object? GetService(System.Type serviceType, object? serviceKey = null) =>
+            serviceType.IsInstanceOfType(this) ? this : null;
+
+        public void Dispose() { }
+    }
+
+    private sealed class QuickChat : IChatClient
+    {
+        private readonly string _reply;
+        public QuickChat(string reply) => _reply = reply;
+
+        public Task<ChatResponse> GetResponseAsync(IEnumerable<ChatMessage> m, ChatOptions? o, CancellationToken ct) =>
+            throw new NotImplementedException();
+
+        public async IAsyncEnumerable<ChatResponseUpdate> GetStreamingResponseAsync(
+            IEnumerable<ChatMessage> messages, ChatOptions? options,
+            [EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            yield return new ChatResponseUpdate(ChatRole.Assistant, _reply);
+            await Task.Yield();
+        }
+
+        public object? GetService(System.Type serviceType, object? serviceKey = null) =>
+            serviceType.IsInstanceOfType(this) ? this : null;
+
+        public void Dispose() { }
+    }
+
+    private sealed class EchoTts : ITextToSpeechClient
+    {
+        public Task<TextToSpeechResponse> GetAudioAsync(string t, TextToSpeechOptions? o, CancellationToken ct) =>
+            throw new NotImplementedException();
+
+        public async IAsyncEnumerable<TextToSpeechResponseUpdate> GetStreamingAudioAsync(
+            string text, TextToSpeechOptions? options,
+            [EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            var bytes = Encoding.UTF8.GetBytes(text);
+            yield return new TextToSpeechResponseUpdate(new List<AIContent> { new DataContent(bytes, "audio/lpcm") })
+            {
+                Kind = TextToSpeechResponseUpdateKind.AudioUpdating,
+            };
+            await Task.Yield();
+        }
+
+        public object? GetService(System.Type serviceType, object? serviceKey = null) =>
+            serviceType.IsInstanceOfType(this) ? this : null;
+
+        public void Dispose() { }
+    }
+}
+#endif