Merge branch 'main' into users/rido/teams-samples

Author: rido-min

src/libraries/Builder/Microsoft.Agents.Builder/StreamingResponse.cs

@@ -3,6 +3,7 @@
 
 using Microsoft.Agents.Builder.Errors;
 using Microsoft.Agents.Core;
+using Microsoft.Agents.Core.Errors;
 using Microsoft.Agents.Core.Models;
 using System;
 using System.Collections.Generic;
@@ -40,13 +41,16 @@ internal class StreamingResponse : IStreamingResponse
     {
         public static readonly int DefaultEndStreamTimeout = (int)TimeSpan.FromMinutes(2).TotalMilliseconds;
 
+        private const string TeamsStreamCancelled = "ContentStreamNotAllowed";
+
         private readonly TurnContext _context;
         private int _nextSequence = 1;
         private bool _ended = false;
         private Timer _timer;
         private bool _messageUpdated = false;
         private bool _informativeSent = false;
         private bool _isTeamsChannel;
+        private bool _cancelled;
 
         // Queue for outgoing activities
         private readonly List<Func<IActivity>> _queue = [];
@@ -211,7 +215,7 @@ public async Task QueueInformativeUpdateAsync(string text, CancellationToken can
         /// <exception cref="System.InvalidOperationException">Throws if the stream has already ended.</exception>
         public void QueueTextChunk(string text)
         {
-            if (string.IsNullOrEmpty(text))
+            if (string.IsNullOrEmpty(text) || _cancelled)
             {
                 return;
             }
@@ -264,7 +268,7 @@ public async Task EndStreamAsync(CancellationToken cancellationToken = default)
                 }
 
                 // Timer isn't running for non-streaming channels.  Just send the Message buffer as a message.
-                if (!string.IsNullOrWhiteSpace(Message) || FinalMessage != null)
+                if (UpdatesSent() > 0 || FinalMessage != null)
                 {
                     await _context.SendActivityAsync(CreateFinalMessage(), cancellationToken).ConfigureAwait(false);
                 }
@@ -280,7 +284,7 @@ public async Task EndStreamAsync(CancellationToken cancellationToken = default)
 
                     _ended = true;
 
-                    if (UpdatesSent() == 0)
+                    if (UpdatesSent() == 0 || _cancelled)
                     {
                         // nothing was queued.  nothing to "end".
                         return;
@@ -300,7 +304,7 @@ public async Task EndStreamAsync(CancellationToken cancellationToken = default)
                     }
                 }
 
-                if (!string.IsNullOrEmpty(Message) || FinalMessage != null)
+                if (UpdatesSent() > 0 || FinalMessage != null)
                 {
                     await SendActivityAsync(CreateFinalMessage(), cancellationToken).ConfigureAwait(false);
                 }
@@ -312,7 +316,7 @@ private IActivity CreateFinalMessage()
             var activity = FinalMessage ?? new Activity();
 
             activity.Type = ActivityTypes.Message;
-            activity.Text = Message;   // Teams won't allow Activity.Text changes
+            activity.Text = Message; // !string.IsNullOrEmpty(Message) ? Message : "No text was streamed";   // Teams won't allow Activity.Text changes or empty text
             activity.Entities ??= [];
 
             // make sure the supplied Activity doesn't have a streamInfo already.
@@ -325,7 +329,7 @@ private IActivity CreateFinalMessage()
                 }
             }
 
-            activity.Entities.Add(new StreamInfo() { StreamType = StreamTypes.Final });
+            activity.Entities.Add(new StreamInfo() { StreamType = StreamTypes.Final, StreamResult = (string.IsNullOrEmpty(Message) ? StreamResults.Error : StreamResults.Success) });
 
             // Add in Generated by AI
             if (EnableGeneratedByAILabel == true)
@@ -367,6 +371,7 @@ public async Task ResetAsync(CancellationToken cancellationToken = default)
                 FinalMessage = null;
                 _nextSequence = 1;
                 StreamId = null;
+                _cancelled = false;
             }
         }
 
@@ -527,11 +532,33 @@ private async Task SendActivityAsync(IActivity activity, CancellationToken cance
                     activity.GetStreamingEntity().StreamId = StreamId;
                 }
 
-                var response = await _context.SendActivityAsync(activity, cancellationToken).ConfigureAwait(false);
-
-                if (string.IsNullOrEmpty(StreamId))
+                try
+                {
+                    var response = await _context.SendActivityAsync(activity, cancellationToken).ConfigureAwait(false);
+                    if (string.IsNullOrEmpty(StreamId))
+                    {
+                        StreamId = response.Id;
+                    }
+                }
+                catch (Exception ex)
                 {
-                    StreamId = response.Id;
+                    // We are not rethrowing from here since this is likely being called
+                    // from the Timer thread and will crash the app.  A more elegant 
+                    // solution would be to get it back to the calling thread.
+
+                    if (ex is ErrorResponseException errorResponse)
+                    {
+                        if (!TeamsStreamCancelled.Equals(errorResponse.Body.Error.Code, StringComparison.OrdinalIgnoreCase))
+                        {
+                            System.Diagnostics.Trace.WriteLine($"Exception during StreamingResponse: {ex.Message}");
+                        }
+                    }
+
+                    lock (this)
+                    {
+                        StopStream();
+                        _cancelled = true;
+                    }
                 }
             }
         }