.NET Workflows - Fix "Human In Loop" sample and update "AutoSend" behavior (#991)

* Fix sample and autosend

* Specify "ConversationId"

* Namespace

* Perf fix

* Namespace

* Filter for appropriate message

dotnet/samples/GettingStarted/Workflows/Declarative/Program.cs

@@ -62,7 +62,7 @@ internal sealed class Program
         // Run the workflow, just like any other workflow
         string input = this.GetWorkflowInput();
 
-        CheckpointManager checkpointManager = CheckpointManager.Default;
+        CheckpointManager checkpointManager = CheckpointManager.CreateInMemory();
         Checkpointed<StreamingRun> run = await InProcessExecution.StreamAsync(workflow, input, checkpointManager);
 
         bool isComplete = false;
@@ -88,7 +88,7 @@ internal sealed class Program
                 // Restore the latest checkpoint.
                 Debug.WriteLine($"RESTORE #{this.LastCheckpoint.CheckpointId}");
                 Notify("\nWORKFLOW: Restore");
-                run = await InProcessExecution.ResumeStreamAsync(workflow, this.LastCheckpoint, checkpointManager);
+                run = await InProcessExecution.ResumeStreamAsync(workflow, this.LastCheckpoint, checkpointManager, run.Run.RunId);
             }
             else
             {
@@ -178,6 +178,7 @@ internal sealed class Program
                     }
                     else
                     {
+                        await run.Run.EndRunAsync().ConfigureAwait(false);
                         return requestInfo.Request;
                     }
                     break;

dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/AzureAgentProvider.cs

@@ -1,4 +1,5 @@
 // Copyright (c) Microsoft. All rights reserved.
+
 using System.Collections.Generic;
 using System.Linq;
 using System.Net.Http;

dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/AgentProviderExtensions.cs

@@ -1,7 +1,7 @@
 // Copyright (c) Microsoft. All rights reserved.
 
 using System.Collections.Generic;
-using System.Runtime.CompilerServices;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.AI;
@@ -10,7 +10,7 @@ namespace Microsoft.Agents.AI.Workflows.Declarative.Extensions;
 
 internal static class AgentProviderExtensions
 {
-    public static async IAsyncEnumerable<AgentRunResponseUpdate> InvokeAgentAsync(
+    public static async ValueTask<AgentRunResponse> InvokeAgentAsync(
         this WorkflowAgentProvider agentProvider,
         string executorId,
         IWorkflowContext context,
@@ -19,35 +19,68 @@ internal static class AgentProviderExtensions
         bool autoSend,
         string? additionalInstructions = null,
         IEnumerable<ChatMessage>? inputMessages = null,
-        [EnumeratorCancellation] CancellationToken cancellationToken = default)
+        CancellationToken cancellationToken = default)
     {
+        // Get the specified agent.
         AIAgent agent = await agentProvider.GetAgentAsync(agentName, cancellationToken).ConfigureAwait(false);
 
+        // Prepare the run options.
         ChatClientAgentRunOptions options =
             new(
                 new ChatOptions()
                 {
+                    ConversationId = conversationId,
                     Instructions = additionalInstructions,
                 });
 
-        AgentThread agentThread = conversationId is not null && agent is ChatClientAgent chatClientAgent ? chatClientAgent.GetNewThread(conversationId) : agent.GetNewThread();
+        // Initialize the agent thread.
         IAsyncEnumerable<AgentRunResponseUpdate> agentUpdates =
             inputMessages is not null ?
-                agent.RunStreamingAsync([.. inputMessages], agentThread, options, cancellationToken) :
-                agent.RunStreamingAsync(agentThread, options, cancellationToken);
+                agent.RunStreamingAsync([.. inputMessages], null, options, cancellationToken) :
+                agent.RunStreamingAsync(null, options, cancellationToken);
 
+        // Enable "autoSend" behavior if this is the workflow conversation.
+        bool isWorkflowConversation = context.IsWorkflowConversation(conversationId);
+        autoSend &= isWorkflowConversation;
+
+        // Process the agent response updates.
+        List<AgentRunResponseUpdate> updates = [];
         await foreach (AgentRunResponseUpdate update in agentUpdates.ConfigureAwait(false))
         {
             await AssignConversationIdAsync(((ChatResponseUpdate?)update.RawRepresentation)?.ConversationId).ConfigureAwait(false);
 
+            updates.Add(update);
+
             if (autoSend)
             {
                 await context.AddEventAsync(new AgentRunUpdateEvent(executorId, update)).ConfigureAwait(false);
             }
-
-            yield return update;
         }
 
+        AgentRunResponse response = updates.ToAgentRunResponse();
+
+        if (autoSend)
+        {
+            await context.AddEventAsync(new AgentRunResponseEvent(executorId, response)).ConfigureAwait(false);
+        }
+
+        if (autoSend && !isWorkflowConversation && conversationId is not null)
+        {
+            // Copy messages with content that aren't function calls or results.
+            IEnumerable<ChatMessage> messages =
+                response.Messages.Where(
+                    message =>
+                        !string.IsNullOrEmpty(message.Text) &&
+                        !message.Contents.OfType<FunctionCallContent>().Any() &&
+                        !message.Contents.OfType<FunctionResultContent>().Any());
+            foreach (ChatMessage message in messages)
+            {
+                await agentProvider.CreateMessageAsync(conversationId, message, cancellationToken).ConfigureAwait(false);
+            }
+        }
+
+        return response;
+
         async ValueTask AssignConversationIdAsync(string? assignValue)
         {
             if (assignValue is not null && conversationId is null)

dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/Extensions/IWorkflowContextExtensions.cs

@@ -1,5 +1,6 @@
 // Copyright (c) Microsoft. All rights reserved.
 
+using System;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Agents.AI.Workflows.Declarative.Interpreter;
@@ -46,6 +47,17 @@ internal static class IWorkflowContextExtensions
         await context.AddEventAsync(new ConversationUpdateEvent(conversationId)).ConfigureAwait(false);
     }
 
+    public static bool IsWorkflowConversation(this IWorkflowContext context, string? conversationId)
+    {
+        if (string.IsNullOrWhiteSpace(conversationId))
+        {
+            return false;
+        }
+
+        StringValue workflowId = (StringValue)context.ReadState(SystemScope.Names.ConversationId, VariableScopeNames.System);
+        return workflowId.Value.Equals(conversationId, StringComparison.Ordinal);
+    }
+
     private static DeclarativeWorkflowContext DeclarativeContext(IWorkflowContext context)
     {
         if (context is not DeclarativeWorkflowContext declarativeContext)

dotnet/src/Microsoft.Agents.AI.Workflows.Declarative/ObjectModel/InvokeAzureAgentExecutor.cs

@@ -1,7 +1,6 @@
 // Copyright (c) Microsoft. All rights reserved.
 
 using System.Collections.Generic;
-using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Agents.AI.Workflows.Declarative.Extensions;
@@ -29,12 +28,7 @@ internal sealed class InvokeAzureAgentExecutor(InvokeAzureAgent model, WorkflowA
         bool autoSend = this.GetAutoSendValue();
         IEnumerable<ChatMessage>? inputMessages = this.GetInputMessages();
 
-        AgentRunResponse agentResponse = agentProvider.InvokeAgentAsync(this.Id, context, agentName, conversationId, autoSend, additionalInstructions, inputMessages, cancellationToken).ToEnumerable().ToAgentRunResponse();
-
-        if (autoSend)
-        {
-            await context.AddEventAsync(new AgentRunResponseEvent(this.Id, agentResponse)).ConfigureAwait(false);
-        }
+        AgentRunResponse agentResponse = await agentProvider.InvokeAgentAsync(this.Id, context, agentName, conversationId, autoSend, additionalInstructions, inputMessages, cancellationToken).ConfigureAwait(false);
 
         await this.AssignAsync(this.AgentOutput?.Messages?.Path, agentResponse.Messages.ToTable(), context).ConfigureAwait(false);