fix: add resilient SSE streaming and HTTP keep-alive for Copilot Studio

samples/dotnet/genesys-handoff/GenesysHandoffAgent.Streaming.cs

@@ -0,0 +1,207 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+// Resilient Copilot Studio (MCS) SSE streaming with retry-with-backoff.
+// Fixes: "Unable to read data from the transport connection: An existing
+// connection was forcibly closed by the remote host." errors on long
+// generative turns where an idle-sensitive intermediary RSTs the socket.
+
+using GenesysHandoff.Services;
+using Microsoft.Agents.Builder;
+using Microsoft.Agents.Builder.State;
+using Microsoft.Agents.Core.Models;
+using Microsoft.Extensions.Logging;
+using System;
+using System.IO;
+using System.Net.Http;
+using System.Net.Sockets;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace GenesysHandoff
+{
+    public partial class GenesysHandoffAgent
+    {
+        private const int McsStreamMaxAttempts = 3;
+        private static readonly TimeSpan McsStreamBaseDelay = TimeSpan.FromMilliseconds(500);
+
+        // Friendly fallback shown when the Copilot Studio stream cannot be completed after retries.
+        private const string CopilotStreamErrorMessage =
+            "Sorry, I hit a temporary connection problem while getting that answer. Please send your message again.";
+
+        /// <summary>
+        /// Handles starting a new conversation with Copilot Studio.
+        /// StartConversation carries no user turn, so a clean retry on transient
+        /// transport faults is always safe.
+        /// </summary>
+        private async Task<string> HandleNewConversation(
+            ITurnContext turnContext,
+            ITurnState turnState,
+            Microsoft.Agents.CopilotStudio.Client.CopilotClient cpsClient,
+            CancellationToken cancellationToken)
+        {
+            ConversationReference? lastCopilotStudioRef = null;
+
+            for (var attempt = 1; attempt <= McsStreamMaxAttempts; attempt++)
+            {
+                try
+                {
+                    await foreach (IActivity activity in cpsClient.StartConversationAsync(
+                        emitStartConversationEvent: true, cancellationToken: cancellationToken))
+                    {
+                        _logger.LogInformation(
+                            "Activity from CPS (StartConversation): Id={CpsActivityId} ReplyToId={CpsReplyToId} Type={Type} Name={Name} Conversation={ConversationId}",
+                            activity.Id, activity.ReplyToId, activity.Type, activity.Name, activity.Conversation?.Id);
+
+                        lastCopilotStudioRef = activity.GetConversationReference();
+                        if (activity.IsType(ActivityTypes.Message)
+                            && !string.IsNullOrWhiteSpace(activity.Conversation?.Id))
+                        {
+                            _stateManager.SetConversationId(turnState, activity.Conversation.Id);
+                        }
+                    }
+
+                    break; // success
+                }
+                catch (Exception ex) when (IsTransientStreamFault(ex) && !cancellationToken.IsCancellationRequested)
+                {
+                    if (attempt >= McsStreamMaxAttempts)
+                    {
+                        _logger.LogError(
+                            ex,
+                            "Failed to start Copilot Studio conversation after {MaxAttempts} attempts. Surfacing fallback message.",
+                            McsStreamMaxAttempts);
+                        await turnContext.SendActivityAsync(CopilotStreamErrorMessage, cancellationToken: cancellationToken);
+                        return string.Empty;
+                    }
+
+                    var delay = TimeSpan.FromMilliseconds(McsStreamBaseDelay.TotalMilliseconds * Math.Pow(2, attempt - 1));
+                    _logger.LogWarning(
+                        ex,
+                        "Transient reset starting Copilot Studio conversation on attempt {Attempt}/{MaxAttempts}. Retrying in {DelayMs} ms.",
+                        attempt, McsStreamMaxAttempts, (int)delay.TotalMilliseconds);
+                    await Task.Delay(delay, cancellationToken);
+                }
+            }
+
+            if (lastCopilotStudioRef != null)
+            {
+                _stateManager.SetLastCopilotStudioReference(turnState, lastCopilotStudioRef);
+            }
+
+            return lastCopilotStudioRef?.Conversation.Id ?? string.Empty;
+        }
+
+        /// <summary>
+        /// Handles processing messages through Copilot Studio and checking for escalation events.
+        /// Resilient version: transient transport resets on the SSE stream are retried
+        /// (before any assistant reply is surfaced) and never leak to the end user.
+        /// </summary>
+        private async Task HandleCopilotStudioMessage(
+            ITurnContext turnContext,
+            ITurnState turnState,
+            Microsoft.Agents.CopilotStudio.Client.CopilotClient cpsClient,
+            string mcsConversationId,
+            CancellationToken cancellationToken)
+        {
+            var lastCopilotStudioRef = _stateManager.GetLastCopilotStudioReference(turnState);
+            _logger.LogInformation(
+                "Activity from Teams: Id={TeamsActivityId} ReplyToId={TeamsReplyToId} Type={Type} Conversation={ConversationId}",
+                turnContext.Activity.Id, turnContext.Activity.ReplyToId, turnContext.Activity.Type, mcsConversationId);
+
+            var activityToSend = await BuildCopilotStudioActivityAsync(
+                turnContext.Activity, lastCopilotStudioRef, mcsConversationId, cancellationToken);
+
+            // Store the Teams conversation reference so proactive messages can be sent back.
+            await _messageSender.StoreUserChannelReferenceAsync(turnContext.Activity, mcsConversationId, cancellationToken);
+
+            ConversationReference? latestCopilotStudioRef = null;
+            var assistantMessageSurfaced = false;
+            var resetDuringTurn = false;
+
+            for (var attempt = 1; attempt <= McsStreamMaxAttempts; attempt++)
+            {
+                try
+                {
+                    await foreach (IActivity activity in cpsClient.SendActivityAsync(activityToSend, cancellationToken))
+                    {
+                        latestCopilotStudioRef = activity.GetConversationReference();
+
+                        // Track whether we've already shown the user a real reply this turn.
+                        // Once we have, a later transport fault must NOT trigger a retry
+                        // (that would re-send the user's turn and double-post).
+                        if (activity.IsType(ActivityTypes.Message) || activity.IsType(ActivityTypes.InvokeResponse))
+                        {
+                            assistantMessageSurfaced = true;
+                        }
+
+                        var reset = await ProcessCopilotStudioActivityAsync(
+                            turnContext, turnState, activity, mcsConversationId, cancellationToken);
+                        if (reset)
+                        {
+                            resetDuringTurn = true;
+                            break; // conversation was reset; stop processing further CPS activities.
+                        }
+                    }
+
+                    break; // stream completed successfully
+                }
+                catch (Exception ex) when (IsTransientStreamFault(ex) && !cancellationToken.IsCancellationRequested)
+                {
+                    if (assistantMessageSurfaced || attempt >= McsStreamMaxAttempts)
+                    {
+                        _logger.LogError(
+                            ex,
+                            "Copilot Studio stream failed for conversation {ConversationId} on attempt {Attempt}/{MaxAttempts} " +
+                            "(assistantMessageSurfaced={Surfaced}). Surfacing fallback message.",
+                            mcsConversationId, attempt, McsStreamMaxAttempts, assistantMessageSurfaced);
+
+                        await turnContext.SendActivityAsync(CopilotStreamErrorMessage, cancellationToken: cancellationToken);
+                        break;
+                    }
+
+                    var delay = TimeSpan.FromMilliseconds(McsStreamBaseDelay.TotalMilliseconds * Math.Pow(2, attempt - 1));
+                    _logger.LogWarning(
+                        ex,
+                        "Transient Copilot Studio stream reset for conversation {ConversationId} on attempt {Attempt}/{MaxAttempts}. " +
+                        "Retrying in {DelayMs} ms.",
+                        mcsConversationId, attempt, McsStreamMaxAttempts, (int)delay.TotalMilliseconds);
+
+                    await Task.Delay(delay, cancellationToken);
+                }
+            }
+
+            if (latestCopilotStudioRef != null && !resetDuringTurn)
+            {
+                _stateManager.SetLastCopilotStudioReference(turnState, latestCopilotStudioRef);
+            }
+        }
+
+        /// <summary>
+        /// True when the exception represents a transient transport-level fault on the
+        /// Copilot Studio SSE stream that is safe to retry at the connection level.
+        /// </summary>
+        private static bool IsTransientStreamFault(Exception ex)
+        {
+            for (var current = ex; current is not null; current = current.InnerException)
+            {
+                switch (current)
+                {
+                    case SocketException se
+                        when se.SocketErrorCode is SocketError.ConnectionReset
+                                                or SocketError.ConnectionAborted
+                                                or SocketError.TimedOut:
+                        return true;
+                    case IOException io when io.InnerException is SocketException:
+                        return true;
+                    case HttpRequestException:
+                        return true;
+                    case OperationCanceledException oce when !oce.CancellationToken.IsCancellationRequested:
+                        return true;
+                }
+            }
+
+            return false;
+        }
+    }
+}

samples/dotnet/genesys-handoff/GenesysHandoffAgent.cs

@@ -20,7 +20,7 @@ namespace GenesysHandoff
     /// <summary>
     /// An AgentApplication that integrates with Genesys for human handoff.
     /// </summary>
-    public class GenesysHandoffAgent : AgentApplication
+    public partial class GenesysHandoffAgent : AgentApplication
     {
         private const string McsHandlerName = "mcs";
 
@@ -190,71 +190,8 @@ private async Task HandleEscalatedMessageAsync(
             }
         }
 
-        /// <summary>
-        /// Handles starting a new conversation with Copilot Studio.
-        /// The last activity received from CPS is stored in conversation state so that
-        /// subsequent turns can be stitched to this conversation.
-        /// </summary>
-        private async Task<string> HandleNewConversation(ITurnContext turnContext, ITurnState turnState, Microsoft.Agents.CopilotStudio.Client.CopilotClient cpsClient, CancellationToken cancellationToken)
-        {
-            ConversationReference? lastCopilotStudioRef = null;
-
-            await foreach (IActivity activity in cpsClient.StartConversationAsync(emitStartConversationEvent: true, cancellationToken: cancellationToken))
-            {
-                _logger.LogInformation(
-                    "Activity from CPS (StartConversation): Id={CpsActivityId} ReplyToId={CpsReplyToId} Type={Type} Name={Name} Conversation={ConversationId}",
-                    activity.Id, activity.ReplyToId, activity.Type, activity.Name, activity.Conversation?.Id);
-                lastCopilotStudioRef = activity.GetConversationReference();
-                if (activity.IsType(ActivityTypes.Message))
-                {
-                    //var responseActivity = _responseProcessor.CreateResponseActivity(activity, "StartConversation");
-                  //await turnContext.SendActivityAsync(responseActivity, cancellationToken);
-                    if (!string.IsNullOrWhiteSpace(activity.Conversation?.Id))
-                    {
-                        _stateManager.SetConversationId(turnState, activity.Conversation.Id);
-                    }
-                }
-            }
-
-            if (lastCopilotStudioRef != null)
-            {
-                _stateManager.SetLastCopilotStudioReference(turnState, lastCopilotStudioRef);
-            }
-
-            return lastCopilotStudioRef?.Conversation.Id ?? string.Empty;
-        }
-
-        /// <summary>
-        /// Handles processing messages through Copilot Studio and checking for escalation events.
-        /// The last activity received from CPS is stored in conversation state so that
-        /// subsequent turns can be stitched to this conversation.
-        /// </summary>
-        private async Task HandleCopilotStudioMessage(ITurnContext turnContext, ITurnState turnState, Microsoft.Agents.CopilotStudio.Client.CopilotClient cpsClient, string mcsConversationId, CancellationToken cancellationToken)
-        {
-            // When a message is received from the user, it is forwarded to Copilot Studio using the conversation ID stored in state.
-            // The agent then listens for responses from Copilot Studio. If a message activity is received, it is sent back to the user.
-            // If an event activity with the name "GenesysHandoff" is received, it indicates that the conversation should be escalated to a human agent through Genesys.
-            var lastCopilotStudioRef = _stateManager.GetLastCopilotStudioReference(turnState);
-            _logger.LogInformation(
-                "Activity from Teams: Id={TeamsActivityId} ReplyToId={TeamsReplyToId} Type={Type} Conversation={ConversationId}",
-                turnContext.Activity.Id, turnContext.Activity.ReplyToId, turnContext.Activity.Type, mcsConversationId);
-            var activityToSend = await BuildCopilotStudioActivityAsync(turnContext.Activity, lastCopilotStudioRef, mcsConversationId, cancellationToken);
-
-            // Store the Teams conversation reference so proactive messages (e.g. from the reset API) can be sent back.
-            await _messageSender.StoreUserChannelReferenceAsync(turnContext.Activity, mcsConversationId, cancellationToken);
-            ConversationReference? latestCopilotStudioRef = null;
-            await foreach (IActivity activity in cpsClient.SendActivityAsync(activityToSend, cancellationToken))
-            {
-                latestCopilotStudioRef = activity.GetConversationReference();
-                var result = await ProcessCopilotStudioActivityAsync(turnContext, turnState, activity, mcsConversationId, cancellationToken);
-                if (result) break; // If true is returned, it indicates the conversation has been reset and we should stop processing further CPS activities for this turn.
-            }
-
-            if (latestCopilotStudioRef != null)
-            {
-                _stateManager.SetLastCopilotStudioReference(turnState, latestCopilotStudioRef);
-            }
-        }
+        // HandleNewConversation and HandleCopilotStudioMessage are in
+        // GenesysHandoffAgent.Streaming.cs (resilient SSE streaming with retry).
 
         private async Task<bool> ProcessCopilotStudioActivityAsync(
             ITurnContext turnContext,

samples/dotnet/genesys-handoff/Program.cs

@@ -11,11 +11,12 @@
 using Microsoft.AspNetCore.Http;
 using Microsoft.Extensions.DependencyInjection;
 using System;
+using System.Net.Http;
 using System.Threading;
 
 var builder = WebApplication.CreateBuilder(args);
 
-builder.Services.AddHttpClient();
+builder.Services.AddResilientMcsHttpClient();
 
 // Register IStorage.  For development, MemoryStorage is suitable.
 // For production Agents, persisted storage should be used so

samples/dotnet/genesys-handoff/README.md

@@ -682,4 +682,48 @@ With the basics in place, you can use this foundation to further integrate and f
 7. **State persistence in Cosmos DB:** Throughout the flow, the Agent SDK reads and writes conversation metadata (for example, mappings between Teams and Genesys conversations, handoff flags) in persistent storage such as Azure Cosmos DB, so state survives restarts and scales beyond a single instance.
 8. **Agent disconnect detection (optional):** When `EnableNotifications` is enabled, the Agent SDK maintains a WebSocket connection to the Genesys Cloud Notification Service. Upon escalation, it subscribes to the `v2.detail.events.conversation.{id}.user.end` topic. When a Genesys agent disconnects, the Agent SDK proactively notifies the Teams user and clears the escalation flag on the next user message, returning the conversation to Copilot Studio.
 
-This architecture lets the user stay in a single Teams conversation while the Agent SDK, Copilot Studio runtime, Genesys Cloud, and persistent storage coordinate the escalation and message exchange behind the scenes. During escalation, Copilot Studio’s role is limited to raising the `GenesysHandoff` event; the actual Genesys conversation is managed directly between the Agent SDK and Genesys Cloud.
+This architecture lets the user stay in a single Teams conversation while the Agent SDK, Copilot Studio runtime, Genesys Cloud, and persistent storage coordinate the escalation and message exchange behind the scenes. During escalation, Copilot Studio’s role is limited to raising the `GenesysHandoff` event; the actual Genesys conversation is managed directly between the Agent SDK and Genesys Cloud.
+
+---
+
+## Production Hardening
+
+This sample is a demonstration starting point. Before deploying to production, address the following:
+
+### SSE Streaming Resilience
+
+The Copilot Studio SSE stream can be reset by idle-sensitive intermediaries (SNAT, nginx, island-gateway) during long generative turns (~10s+ silence on the wire). The `GenesysHandoffAgent.Streaming.cs` partial class wraps the SSE enumeration with retry-with-backoff and a friendly fallback message. The `McsHttpClientRegistration` configures the `"mcs"` HttpClient with HTTP/2 keep-alive pings to prevent most resets.
+
+### Azure Blob Storage RBAC
+
+If using Azure Blob Storage for `ConversationMappingStore`, the app's managed identity **must** have the **Storage Blob Data Contributor** role on the storage account. Without this, you will see recurring `Azure.RequestFailedException 403 AuthorizationFailure` errors on `ConversationMappingStore.LoadAsync`, which can trigger a Genesys notification WebSocket reconnect storm.
+
+To fix:
+```bash
+az role assignment create \
+  --assignee <app-managed-identity-object-id> \
+  --role "Storage Blob Data Contributor" \
+  --scope /subscriptions/<sub>/resourceGroups/<rg>/providers/Microsoft.Storage/storageAccounts/<account>
+```
+
+### MSAL Token Cache
+
+The default configuration uses an in-memory MSAL token cache. For production multi-instance deployments, configure a distributed token cache (e.g., Redis):
+
+```csharp
+builder.Services.AddDistributedMemoryCache(); // Replace with Redis in production
+// Or: builder.Services.AddStackExchangeRedisCache(options => { ... });
+```
+
+Without this, each instance independently acquires tokens, increasing latency and AAD throttling risk.
+
+### Persistent Storage
+
+Replace `MemoryStorage` with a persistent `IStorage` implementation (e.g., Azure Cosmos DB, Azure Blob Storage) so conversation state survives app restarts and works correctly across multiple instances:
+
+```csharp
+// Replace:
+builder.Services.AddSingleton<IStorage, MemoryStorage>();
+// With (example):
+builder.Services.AddSingleton<IStorage>(new CosmosDbPartitionedStorage(cosmosOptions));
+```