.NET: feat: Host Workflow as AIAgent (#469)

* feat: Host Workflow as AIAgent

* Also changes AIAgent-as-Executor to use streaming runs and streaming
  events
* Also enables default setting for yielding events

* fix: Infinite loop in GenerateNewId()

* docs: Spelling

* test: Add Workflow-as-Agent sample and test

dotnet/src/Microsoft.Agents.Workflows/AIAgentsAbstractionsExtensions.cs

@@ -0,0 +1,42 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using System.Diagnostics;
+using Microsoft.Extensions.AI;
+using Microsoft.Extensions.AI.Agents;
+
+namespace Microsoft.Agents.Workflows;
+
+internal static class AIAgentsAbstractionsExtensions
+{
+    public static ChatMessage ToChatMessage(this AgentRunResponseUpdate update)
+    {
+        return new ChatMessage
+        {
+            AuthorName = update.AuthorName,
+            Contents = update.Contents,
+            Role = update.Role ?? ChatRole.User,
+            CreatedAt = update.CreatedAt,
+            MessageId = update.MessageId,
+            RawRepresentation = update.RawRepresentation,
+        };
+    }
+
+    public static ChatMessage UpdateWith(this ChatMessage baseMessage, AgentRunResponseUpdate update)
+    {
+        Debug.Assert(update.MessageId == null || baseMessage.MessageId == update.MessageId);
+
+        List<AIContent> mergedContent = new(baseMessage.Contents);
+        mergedContent.AddRange(update.Contents);
+
+        return new ChatMessage
+        {
+            AuthorName = update.AuthorName ?? baseMessage.AuthorName,
+            Contents = mergedContent,
+            Role = update.Role ?? baseMessage.Role,
+            CreatedAt = update.CreatedAt ?? baseMessage.CreatedAt,
+            MessageId = baseMessage.MessageId,
+            RawRepresentation = update.RawRepresentation ?? baseMessage.RawRepresentation,
+        };
+    }
+}

dotnet/src/Microsoft.Agents.Workflows/AgentRunEvent.cs

@@ -1,26 +0,0 @@
-// Copyright (c) Microsoft. All rights reserved.
-
-using Microsoft.Extensions.AI.Agents;
-
-namespace Microsoft.Agents.Workflows;
-
-/// <summary>
-/// Event triggered when an agent run is completed.
-/// </summary>
-public class AgentRunEvent : ExecutorEvent
-{
-    /// <summary>
-    /// Initializes a new instance of the <see cref="AgentRunEvent"/> class.
-    /// </summary>
-    /// <param name="executorId">The identifier of the executor that generated this event.</param>
-    /// <param name="response"></param>
-    public AgentRunEvent(string executorId, AgentRunResponse? response = null) : base(executorId, data: response)
-    {
-        this.Response = response;
-    }
-
-    /// <summary>
-    /// Gets the content of the agent response.
-    /// </summary>
-    public AgentRunResponse? Response { get; }
-}

dotnet/src/Microsoft.Agents.Workflows/AgentRunUpdateEvent.cs

@@ -0,0 +1,37 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using Microsoft.Extensions.AI.Agents;
+
+namespace Microsoft.Agents.Workflows;
+
+/// <summary>
+/// Event triggered when an agent run produces an update.
+/// </summary>
+public class AgentRunUpdateEvent : ExecutorEvent
+{
+    /// <summary>
+    /// Initializes a new instance of the <see cref="AgentRunUpdateEvent"/> class.
+    /// </summary>
+    /// <param name="executorId">The identifier of the executor that generated this event.</param>
+    /// <param name="update"></param>
+    public AgentRunUpdateEvent(string executorId, AgentRunResponseUpdate update) : base(executorId, data: update)
+    {
+        this.Update = update;
+    }
+
+    /// <summary>
+    /// Gets the content of the agent response.
+    /// </summary>
+    public AgentRunResponseUpdate Update { get; }
+
+    /// <summary>
+    /// Converts this event to an <see cref="AgentRunResponse"/> containing just this update.
+    /// </summary>
+    /// <returns></returns>
+    public AgentRunResponse AsResponse()
+    {
+        IEnumerable<AgentRunResponseUpdate> updates = [this.Update];
+        return updates.ToAgentRunResponse();
+    }
+}

dotnet/src/Microsoft.Agents.Workflows/MessageMerger.cs

@@ -0,0 +1,217 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Microsoft.Extensions.AI;
+using Microsoft.Extensions.AI.Agents;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.Workflows;
+
+internal class MessageMerger
+{
+    private class ResponseMergeState(string? responseId)
+    {
+        public string? ResponseId { get; } = responseId;
+
+        public Dictionary<string, List<AgentRunResponseUpdate>> UpdatesByMessageId { get; } = new();
+        public List<AgentRunResponseUpdate> DanglingUpdates { get; } = new();
+
+        public void AddUpdate(AgentRunResponseUpdate update)
+        {
+            if (update.MessageId is null)
+            {
+                this.DanglingUpdates.Add(update);
+            }
+            else
+            {
+                if (!this.UpdatesByMessageId.TryGetValue(update.MessageId, out List<AgentRunResponseUpdate>? updates))
+                {
+                    this.UpdatesByMessageId[update.MessageId] = updates = new List<AgentRunResponseUpdate>();
+                }
+
+                updates.Add(update);
+            }
+        }
+
+        private AgentRunResponse ComputeResponse(List<AgentRunResponseUpdate> updates) => updates.ToAgentRunResponse();
+
+        public AgentRunResponse ComputeMerged(string messageId)
+        {
+            if (this.UpdatesByMessageId.TryGetValue(Throw.IfNull(messageId), out List<AgentRunResponseUpdate>? updates))
+            {
+                return updates.ToAgentRunResponse();
+            }
+
+            throw new KeyNotFoundException($"No updates found for message ID '{messageId}' in response '{this.ResponseId}'.");
+        }
+
+        public AgentRunResponse ComputeDangling()
+        {
+            if (this.DanglingUpdates.Count == 0)
+            {
+                throw new InvalidOperationException("No dangling updates to compute a response from.");
+            }
+
+            return this.DanglingUpdates.ToAgentRunResponse();
+        }
+
+        public List<ChatMessage> ComputeFlattened()
+        {
+            List<ChatMessage> result = this.UpdatesByMessageId.Keys.Select(AggregateUpdatesToMessage)
+                                                                   .ToList();
+
+            result.AddRange(this.ComputeDangling().Messages);
+            return result;
+
+            ChatMessage AggregateUpdatesToMessage(string messageId)
+            {
+                List<AgentRunResponseUpdate> updates = this.UpdatesByMessageId[messageId];
+                if (updates.Count == 0)
+                {
+                    throw new InvalidOperationException($"No updates found for message ID '{messageId}' in response '{this.ResponseId}'.");
+                }
+
+                return updates.Aggregate(null,
+                    (ChatMessage? previous, AgentRunResponseUpdate current) =>
+                    {
+                        return previous == null
+                             ? current.ToChatMessage()
+                             : previous.UpdateWith(current);
+                    })!;
+            }
+        }
+    }
+
+    private readonly Dictionary<string, ResponseMergeState> _mergeStates = new();
+    private readonly ResponseMergeState _danglingState = new(null);
+
+    public void AddUpdate(AgentRunResponseUpdate update)
+    {
+        if (update.ResponseId is null)
+        {
+            this._danglingState.DanglingUpdates.Add(update);
+        }
+        else
+        {
+            if (!this._mergeStates.TryGetValue(update.ResponseId, out ResponseMergeState? state))
+            {
+                this._mergeStates[update.ResponseId] = state = new ResponseMergeState(update.ResponseId);
+            }
+
+            state.AddUpdate(update);
+        }
+    }
+
+    private int CompareByDateTimeOffset(AgentRunResponse left, AgentRunResponse right)
+    {
+        const int LESS = -1, EQ = 0, GREATER = 1;
+
+        if (left.CreatedAt == right.CreatedAt)
+        {
+            return EQ;
+        }
+
+        if (!left.CreatedAt.HasValue)
+        {
+            return GREATER;
+        }
+
+        if (!right.CreatedAt.HasValue)
+        {
+            return LESS;
+        }
+
+        return left.CreatedAt.Value.CompareTo(right.CreatedAt.Value);
+    }
+
+    public AgentRunResponse ComputeMerged(string primaryResponseId)
+    {
+        List<ChatMessage> messages = [];
+        Dictionary<string, AgentRunResponse> responses = new();
+
+        foreach (string responseId in this._mergeStates.Keys)
+        {
+            ResponseMergeState mergeState = this._mergeStates[responseId];
+
+            List<AgentRunResponse> responseList = mergeState.UpdatesByMessageId.Keys.Select(messageId => mergeState.ComputeMerged(messageId)).ToList();
+            if (mergeState.DanglingUpdates.Count > 0)
+            {
+                responseList.Add(mergeState.ComputeDangling());
+            }
+
+            responseList.Sort(this.CompareByDateTimeOffset);
+            responses[responseId] = responseList.Aggregate(MergeResponses);
+            messages.AddRange(responses[responseId].Messages);
+        }
+
+        messages.AddRange(this._danglingState.ComputeFlattened());
+        return new AgentRunResponse(messages)
+        {
+            ResponseId = primaryResponseId,
+        };
+
+        AgentRunResponse MergeResponses(AgentRunResponse? current, AgentRunResponse incoming)
+        {
+            if (current is null)
+            {
+                return incoming;
+            }
+
+            if (current.ResponseId != incoming.ResponseId)
+            {
+                throw new InvalidOperationException($"Cannot merge responses with different IDs: '{current.ResponseId}' and '{incoming.ResponseId}'.");
+            }
+
+            List<object?> rawRepresentation = current.RawRepresentation as List<object?> ?? [];
+            rawRepresentation.Add(incoming.RawRepresentation);
+
+            return new()
+            {
+                AgentId = incoming.AgentId ?? current.AgentId,
+                AdditionalProperties = incoming.AdditionalProperties ?? current.AdditionalProperties,
+                CreatedAt = incoming.CreatedAt ?? current.CreatedAt,
+                Messages = current.Messages.Concat(incoming.Messages).ToList(),
+                ResponseId = current.ResponseId,
+                RawRepresentation = rawRepresentation,
+                Usage = Merge(current.Usage, incoming.Usage),
+            };
+        }
+
+        static UsageDetails? Merge(UsageDetails? current, UsageDetails? incoming)
+        {
+            if (current == null)
+            {
+                return incoming;
+            }
+
+            AdditionalPropertiesDictionary<long>? additionalCounts = current.AdditionalCounts;
+            if (incoming == null)
+            {
+                return current;
+            }
+
+            if (additionalCounts == null)
+            {
+                additionalCounts = incoming.AdditionalCounts;
+            }
+            else if (incoming.AdditionalCounts != null)
+            {
+                foreach (string key in incoming.AdditionalCounts.Keys)
+                {
+                    additionalCounts[key] = incoming.AdditionalCounts[key] +
+                                            (additionalCounts.TryGetValue(key, out long? existingCount) ? existingCount.Value : 0);
+                }
+            }
+
+            return new UsageDetails
+            {
+                InputTokenCount = current.InputTokenCount + incoming.InputTokenCount,
+                OutputTokenCount = current.OutputTokenCount + incoming.OutputTokenCount,
+                TotalTokenCount = current.TotalTokenCount + incoming.TotalTokenCount,
+                AdditionalCounts = additionalCounts,
+            };
+        }
+    }
+}

dotnet/src/Microsoft.Agents.Workflows/Specialized/AIAgentHostExecutor.cs

@@ -1,7 +1,6 @@
 // Copyright (c) Microsoft. All rights reserved.
 
 using System.Collections.Generic;
-using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.Extensions.AI;
 using Microsoft.Extensions.AI.Agents;
@@ -10,13 +9,15 @@ namespace Microsoft.Agents.Workflows.Specialized;
 
 internal class AIAgentHostExecutor : Executor
 {
+    private readonly bool _emitEvents;
     private readonly AIAgent _agent;
     private readonly List<ChatMessage> _pendingMessages = new();
     private AgentThread? _thread = null;
 
-    public AIAgentHostExecutor(AIAgent agent) : base(id: agent.Id)
+    public AIAgentHostExecutor(AIAgent agent, bool emitEvents = false) : base(id: agent.Id)
     {
         this._agent = agent;
+        this._emitEvents = emitEvents;
     }
 
     private AgentThread EnsureThread()
@@ -50,17 +51,35 @@ internal class AIAgentHostExecutor : Executor
 
     public async ValueTask TakeTurnAsync(TurnToken token, IWorkflowContext context)
     {
-        // TODO: Ideally we want to be able to split the Run across multiple super-steps so that we can stream out
-        // incremental updates from the chat model. 
-        AgentRunResponse runResponse = await this._agent.RunAsync(this._pendingMessages, this.EnsureThread())
-                                                        .ConfigureAwait(false);
+        bool emitEvents = token.EmitEvents.HasValue ? token.EmitEvents.Value : this._emitEvents;
+        IAsyncEnumerable<AgentRunResponseUpdate> agentStream = this._agent.RunStreamingAsync(this._pendingMessages, this.EnsureThread());
 
-        if (token.EmitEvents)
+        List<AgentRunResponseUpdate> updates = new();
+        await foreach (AgentRunResponseUpdate update in agentStream.ConfigureAwait(false))
         {
-            await context.AddEventAsync(new AgentRunEvent(this.Id, runResponse)).ConfigureAwait(false);
+            if (emitEvents)
+            {
+                await context.AddEventAsync(new AgentRunUpdateEvent(this.Id, update)).ConfigureAwait(false);
+            }
+
+            // TODO: FunctionCall request handling, and user info request handling.
+            // In some sense: We should just let it be handled as a ChatMessage, though we should consider
+            // providing some mechanisms to help the user complete the request, or route it out of the
+            // workflow.
+
+            updates.Add(update);
+            ChatMessage message = new(update.Role ?? ChatRole.Assistant, update.Contents)
+            {
+                AuthorName = update.AuthorName,
+                CreatedAt = update.CreatedAt,
+                MessageId = update.MessageId,
+                RawRepresentation = update.RawRepresentation,
+                AdditionalProperties = update.AdditionalProperties
+            };
+
+            await context.SendMessageAsync(message).ConfigureAwait(false);
         }
 
-        await context.SendMessageAsync(runResponse.Messages.ToList()).ConfigureAwait(false);
         await context.SendMessageAsync(token).ConfigureAwait(false);
     }
 }

dotnet/src/Microsoft.Agents.Workflows/TurnToken.cs

@@ -9,10 +9,11 @@ namespace Microsoft.Agents.Workflows;
 /// a response to accumulated <see cref="Microsoft.Extensions.AI.ChatMessage"/>.
 /// </summary>
 /// <param name="emitEvents">Whether to raise AgentRunEvents for this executor.</param>
-public class TurnToken(bool emitEvents = false)
+public class TurnToken(bool? emitEvents = null)
 {
     /// <summary>
-    /// Gets a value indicating whether events are emitted by the receiving executor.
+    /// Gets a value indicating whether events are emitted by the receiving executor. If the
+    /// value is not set, defaults to the configuration in the executor.
     /// </summary>
-    public bool EmitEvents => emitEvents;
+    public bool? EmitEvents => emitEvents;
 }

dotnet/src/Microsoft.Agents.Workflows/WorkflowHostAgent.cs

@@ -0,0 +1,150 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.AI;
+using Microsoft.Extensions.AI.Agents;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.Workflows;
+
+internal class WorkflowHostAgent : AIAgent
+{
+    private readonly Workflow<List<ChatMessage>> _workflow;
+    private readonly string? _id, _name;
+
+    private readonly ConcurrentDictionary<string, string> _assignedRunIds = new();
+    private readonly Dictionary<string, StreamingRun> _runningWorkflows = new();
+
+    public WorkflowHostAgent(Workflow<List<ChatMessage>> workflow, string? id = null, string? name = null)
+    {
+        this._workflow = Throw.IfNull(workflow, nameof(workflow));
+
+        this._id = id;
+        this._name = name;
+    }
+
+    public override string? Name => this._name;
+    public override string Id => this._id ?? base.Id;
+
+    private string GenerateNewId()
+    {
+        string result;
+
+        do
+        {
+            result = Guid.NewGuid().ToString("N");
+        } while (!this._assignedRunIds.TryAdd(result, result));
+
+        return result;
+    }
+
+    public override AgentThread GetNewThread() => new WorkflowThread(this.Id, this.Name, this.GenerateNewId());
+
+    private async
+    IAsyncEnumerable<AgentRunResponseUpdate> InvokeStageAsync(
+        WorkflowThread conversation,
+        [EnumeratorCancellation] CancellationToken cancellation = default)
+    {
+        string runId = conversation.RunId;
+        List<ChatMessage> messages = conversation.MessageStore.GetFromBookmark().ToList();
+
+        try
+        {
+            // technically there is a race condition here between assigning the ID, and checking if it exists
+            // in the case of new threads.
+            if (!this._runningWorkflows.TryGetValue(runId, out StreamingRun? run))
+            {
+                run = await InProcessExecution.StreamAsync<List<ChatMessage>>(this._workflow, messages, cancellation)
+                                                       .ConfigureAwait(false);
+                this._runningWorkflows[runId] = run;
+            }
+            else
+            {
+                bool sentMessages = await run.TrySendMessageAsync(messages).ConfigureAwait(false);
+                Debug.Assert(sentMessages, "Hosted workflow is required to take List<ChatMessage> as input.");
+            }
+
+            await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false);
+            await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false, cancellation)
+                                               .ConfigureAwait(false)
+                                               .WithCancellation(cancellation))
+            {
+                switch (evt)
+                {
+                    case AgentRunUpdateEvent agentUpdate:
+                        yield return agentUpdate.Update;
+                        break;
+                    case RequestInfoEvent requestInfo:
+                        FunctionCallContent fcContent = requestInfo.Request.ToFunctionCall();
+                        AgentRunResponseUpdate update = conversation.CreateUpdate(fcContent);
+                        yield return update;
+                        break;
+                }
+            }
+        }
+        finally
+        {
+            // Do we want to try to undo the step, and not update the bookmark?
+            conversation.MessageStore.UpdateBookmark();
+        }
+    }
+
+    private async ValueTask<WorkflowThread> UpdateThreadAsync(IReadOnlyCollection<ChatMessage> messages, AgentThread? thread = null, CancellationToken cancellation = default)
+    {
+        if (thread is null)
+        {
+            thread = this.GetNewThread();
+        }
+
+        if (thread is not WorkflowThread workflowThread)
+        {
+            throw new ArgumentException($"Incompatible thread type: {thread.GetType()} (expecting {typeof(WorkflowThread)})", nameof(thread));
+        }
+
+        await workflowThread.MessageStore.AddMessagesAsync(messages, cancellation).ConfigureAwait(false);
+        return workflowThread;
+    }
+
+    public override async
+    Task<AgentRunResponse> RunAsync(
+        IReadOnlyCollection<ChatMessage> messages,
+        AgentThread? thread = null,
+        AgentRunOptions? options = null,
+        CancellationToken cancellationToken = default)
+    {
+        WorkflowThread workflowThread = await this.UpdateThreadAsync(messages, thread, cancellationToken).ConfigureAwait(false);
+        MessageMerger merger = new();
+
+        await foreach (AgentRunResponseUpdate update in this.InvokeStageAsync(workflowThread, cancellationToken)
+                                                            .ConfigureAwait(false)
+                                                            .WithCancellation(cancellationToken))
+        {
+            merger.AddUpdate(update);
+        }
+
+        return merger.ComputeMerged(workflowThread.ResponseId);
+    }
+
+    public override async
+    IAsyncEnumerable<AgentRunResponseUpdate> RunStreamingAsync(
+        IReadOnlyCollection<ChatMessage> messages,
+        AgentThread? thread = null,
+        AgentRunOptions? options = null,
+        [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        WorkflowThread workflowThread = await this.UpdateThreadAsync(messages, thread, cancellationToken).ConfigureAwait(false);
+        await foreach (AgentRunResponseUpdate update in this.InvokeStageAsync(workflowThread, cancellationToken)
+                                                            .ConfigureAwait(false)
+                                                            .WithCancellation(cancellationToken))
+        {
+            yield return update;
+        }
+    }
+}

dotnet/src/Microsoft.Agents.Workflows/WorkflowHostingExtensions.cs

@@ -0,0 +1,35 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using Microsoft.Extensions.AI;
+using Microsoft.Extensions.AI.Agents;
+
+namespace Microsoft.Agents.Workflows;
+
+/// <summary>
+/// Provides extension methods for treating workflows as <see cref="AIAgent"/>
+/// </summary>
+public static class WorkflowHostingExtensions
+{
+    /// <summary>
+    /// Convert a workflow with the appropriate primary input type to an <see cref="AIAgent"/>.
+    /// </summary>
+    /// <param name="workflow"></param>
+    /// <param name="id"></param>
+    /// <param name="name"></param>
+    /// <returns></returns>
+    public static AIAgent AsAgent(this Workflow<List<ChatMessage>> workflow, string? id = null, string? name = null)
+    {
+        return new WorkflowHostAgent(workflow, id, name);
+    }
+
+    internal static FunctionCallContent ToFunctionCall(this ExternalRequest request)
+    {
+        Dictionary<string, object?> parameters = new()
+        {
+            { "data", request.Data}
+        };
+
+        return new FunctionCallContent(request.RequestId, request.Port.Id, parameters);
+    }
+}

dotnet/src/Microsoft.Agents.Workflows/WorkflowMessageStore.cs

@@ -0,0 +1,90 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Collections.Generic;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.AI;
+using Microsoft.Extensions.AI.Agents;
+
+namespace Microsoft.Agents.Workflows;
+
+internal class WorkflowMessageStore : IChatMessageStore
+{
+    private int _bookmark = 0;
+    private readonly List<ChatMessage> _chatMessages = new();
+
+    internal class StoreState
+    {
+        public int Bookmark { get; set; }
+        public IList<ChatMessage> Messages { get; set; } = new List<ChatMessage>();
+    }
+
+    internal void AddMessages(params ChatMessage[] messages)
+    {
+        this._chatMessages.AddRange(messages);
+    }
+
+    public Task AddMessagesAsync(IReadOnlyCollection<ChatMessage> messages, CancellationToken cancellationToken)
+    {
+        this._chatMessages.AddRange(messages);
+
+        return Task.CompletedTask;
+    }
+
+    public Task<IEnumerable<ChatMessage>> GetMessagesAsync(CancellationToken cancellationToken)
+    {
+        return Task.FromResult<IEnumerable<ChatMessage>>(this._chatMessages.AsReadOnly());
+    }
+
+    public IEnumerable<ChatMessage> GetFromBookmark()
+    {
+        for (int i = this._bookmark; i < this._chatMessages.Count; i++)
+        {
+            yield return this._chatMessages[i];
+        }
+    }
+
+    public void UpdateBookmark()
+    {
+        this._bookmark = this._chatMessages.Count;
+    }
+
+    public ValueTask DeserializeStateAsync(JsonElement? serializedStoreState, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
+    {
+        if (serializedStoreState is null)
+        {
+            return default;
+        }
+
+        object? maybeState =
+            JsonSerializer.Deserialize(
+                serializedStoreState.Value,
+                AgentAbstractionsJsonUtilities.DefaultOptions.GetTypeInfo(typeof(StoreState)));
+
+        if (maybeState is not StoreState state)
+        {
+            throw new JsonException("Invalid state format for WorkflowMessageStore.");
+        }
+
+        this._chatMessages.Clear();
+        this._chatMessages.AddRange(state.Messages);
+
+        this._bookmark = state.Bookmark;
+
+        return default;
+    }
+
+    public ValueTask<JsonElement?> SerializeStateAsync(JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
+    {
+        StoreState state = new()
+        {
+            Bookmark = this._bookmark,
+            Messages = this._chatMessages,
+        };
+
+        return new ValueTask<JsonElement?>
+            (JsonSerializer.SerializeToElement(state,
+            WorkflowsJsonUtilities.DefaultOptions.GetTypeInfo(typeof(StoreState))));
+    }
+}

dotnet/src/Microsoft.Agents.Workflows/WorkflowThread.cs

@@ -0,0 +1,60 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Extensions.AI;
+using Microsoft.Extensions.AI.Agents;
+using Microsoft.Shared.Diagnostics;
+
+namespace Microsoft.Agents.Workflows;
+
+internal class WorkflowThread : AgentThread
+{
+    private readonly string _workflowId;
+    private readonly string? _workflowName;
+    private readonly WorkflowMessageStore _messageStore;
+
+    public WorkflowThread(string workflowId, string? workflowName, string runId)
+    {
+        base.MessageStore = this._messageStore = new();
+        this.RunId = Throw.IfNullOrEmpty(runId, nameof(runId));
+
+        this._workflowId = Throw.IfNullOrEmpty(workflowId);
+        this._workflowName = workflowName;
+    }
+
+    public string RunId { get; }
+    public int Halts { get; } = 0;
+
+    public string ResponseId => $"{this.RunId}@{this.Halts}";
+
+    public override Task<JsonElement> SerializeAsync(JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
+    {
+        throw new NotImplementedException("Pending Checkpointing work.");
+    }
+
+    protected override Task DeserializeAsync(JsonElement serializedThread, JsonSerializerOptions? jsonSerializerOptions = null, CancellationToken cancellationToken = default)
+    {
+        throw new NotImplementedException("Pending Checkpointing work.");
+    }
+
+    public AgentRunResponseUpdate CreateUpdate(params AIContent[] parts)
+    {
+        Throw.IfNullOrEmpty(parts);
+
+        AgentRunResponseUpdate update = new(ChatRole.Assistant, parts)
+        {
+            CreatedAt = DateTimeOffset.Now,
+            MessageId = Guid.NewGuid().ToString("N"),
+        };
+
+        this.MessageStore.AddMessages(update.ToChatMessage());
+
+        return update;
+    }
+
+    /// <inheritdoc/>
+    public new WorkflowMessageStore MessageStore => this._messageStore;
+}

dotnet/src/Microsoft.Agents.Workflows/WorkflowsJsonUtilities.cs

@@ -0,0 +1,61 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Diagnostics.CodeAnalysis;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+using Microsoft.Extensions.AI;
+using static Microsoft.Agents.Workflows.WorkflowMessageStore;
+
+namespace Microsoft.Agents.Workflows;
+
+/// <summary>Provides a collection of utility methods for working with JSON data in the context of workflows.</summary>
+internal static partial class WorkflowsJsonUtilities
+{
+    /// <summary>
+    /// Gets the <see cref="JsonSerializerOptions"/> singleton used as the default in JSON serialization operations.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// For Native AOT or applications disabling <see cref="JsonSerializer.IsReflectionEnabledByDefault"/>, this instance
+    /// includes source generated contracts for all common exchange types contained in this library.
+    /// </para>
+    /// <para>
+    /// It additionally turns on the following settings:
+    /// <list type="number">
+    /// <item>Enables <see cref="JsonSerializerDefaults.Web"/> defaults.</item>
+    /// <item>Enables <see cref="JsonIgnoreCondition.WhenWritingNull"/> as the default ignore condition for properties.</item>
+    /// <item>Enables <see cref="JsonNumberHandling.AllowReadingFromString"/> as the default number handling for number types.</item>
+    /// </list>
+    /// </para>
+    /// </remarks>
+    public static JsonSerializerOptions DefaultOptions { get; } = CreateDefaultOptions();
+
+    /// <summary>
+    /// Creates default options to use for agents-related serialization.
+    /// </summary>
+    /// <returns>The configured options.</returns>
+    [UnconditionalSuppressMessage("ReflectionAnalysis", "IL3050:RequiresDynamicCode", Justification = "Converter is guarded by IsReflectionEnabledByDefault check.")]
+    [UnconditionalSuppressMessage("Trimming", "IL2026:Members annotated with 'RequiresUnreferencedCodeAttribute' require dynamic access", Justification = "Converter is guarded by IsReflectionEnabledByDefault check.")]
+    private static JsonSerializerOptions CreateDefaultOptions()
+    {
+        // Copy the configuration from the source generated context.
+        JsonSerializerOptions options = new(JsonContext.Default.Options);
+
+        // Chain with all supported types from Microsoft.Extensions.AI.Abstractions. and Microsoft.Extensions.AI.Agents.Abstractions.
+        options.TypeInfoResolverChain.Add(AIJsonUtilities.DefaultOptions.TypeInfoResolver!);
+
+        options.MakeReadOnly();
+        return options;
+    }
+
+    // Keep in sync with CreateDefaultOptions above.
+    [JsonSourceGenerationOptions(JsonSerializerDefaults.Web,
+        DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
+        NumberHandling = JsonNumberHandling.AllowReadingFromString)]
+
+    // Agent abstraction types
+    [JsonSerializable(typeof(StoreState))]
+
+    [ExcludeFromCodeCoverage]
+    internal sealed partial class JsonContext : JsonSerializerContext;
+}

dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/06_GroupChat_Workflow.cs

@@ -5,6 +5,7 @@ using System.Collections.Generic;
 using System.Diagnostics;
 using System.IO;
 using System.Linq;
+using System.Runtime.CompilerServices;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
@@ -15,17 +16,20 @@ namespace Microsoft.Agents.Workflows.UnitTests.Sample;
 
 internal static class Step6EntryPoint
 {
-    internal static int MaxSteps { get; set; }
+    public static Workflow<List<ChatMessage>> CreateWorkflow(int maxTurns)
+    {
+        GroupChatBuilder builder =
+            GroupChatBuilder.Create<RoundRobinGroupChatManager, RoundRobinGroupChatManagerOptions>
+                                (options => options.MaxTurns = maxTurns)
+                            .AddParticipant(new HelloAgent(), shouldEmitEvents: true)
+                            .AddParticipant(new EchoAgent(), shouldEmitEvents: true);
+
+        return builder.ReduceToWorkflow();
+    }
 
     public static async ValueTask RunAsync(TextWriter writer, int maxSteps = 2)
     {
-        Step6EntryPoint.MaxSteps = maxSteps;
-
-        GroupChatBuilder builder = GroupChatBuilder.Create<RoundRobinGroupChatManager>()
-            .AddParticipant(new HelloAgent(), shouldEmitEvents: true)
-            .AddParticipant(new EchoAgent(), shouldEmitEvents: true);
-
-        Workflow<List<ChatMessage>> workflow = builder.ReduceToWorkflow();
+        Workflow<List<ChatMessage>> workflow = CreateWorkflow(maxSteps);
 
         StreamingRun run = await InProcessExecution.StreamAsync(workflow, [])
                                                    .ConfigureAwait(false);
@@ -37,20 +41,34 @@ internal static class Step6EntryPoint
             {
                 Debug.WriteLine($"{executorComplete.ExecutorId}: {executorComplete.Data}");
             }
-            else if (evt is AgentRunEvent agentRun && agentRun.Data is AgentRunResponse response)
+            else if (evt is AgentRunUpdateEvent update)
             {
+                AgentRunResponse response = update.AsResponse();
+
                 foreach (ChatMessage message in response.Messages)
                 {
-                    writer.WriteLine($"{agentRun.ExecutorId}: {message.Text}");
+                    writer.WriteLine($"{update.ExecutorId}: {message.Text}");
                 }
             }
         }
     }
 
-    private sealed class RoundRobinGroupChatManager : GroupChatManager
+    private sealed class RoundRobinGroupChatManagerOptions : GroupChatManagerOptions
+    {
+        public int? MaxTurns { get; set; } = null;
+    }
+
+    private sealed class RoundRobinGroupChatManager() : GroupChatManager<RoundRobinGroupChatManagerOptions>
     {
         public int TurnCount { get; private set; } = 0;
-        public int MaxTurns { get; init; } = Step6EntryPoint.MaxSteps;
+        public int? MaxTurns { get; private set; } = null;
+
+        protected internal override void Configure(RoundRobinGroupChatManagerOptions options)
+        {
+            base.Configure(options);
+
+            this.MaxTurns = options.MaxTurns;
+        }
 
         public override int? GetNextTurnExecutor(GroupChatHistory history)
         {
@@ -75,17 +93,27 @@ internal sealed class HelloAgent(string id = nameof(HelloAgent)) : AIAgent
     public const string DefaultId = nameof(HelloAgent);
 
     public override string Id => id;
+    public override string? Name => id;
 
-    public override Task<AgentRunResponse> RunAsync(IReadOnlyCollection<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
+    public override async Task<AgentRunResponse> RunAsync(IReadOnlyCollection<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
     {
-        AgentRunResponse response = new(new ChatMessage(ChatRole.Assistant, "Hello World!"));
+        IEnumerable<AgentRunResponseUpdate> update = [
+            await this.RunStreamingAsync(messages, thread, options, cancellationToken)
+                      .SingleAsync(cancellationToken)
+                      .ConfigureAwait(false)];
 
-        return Task.FromResult(response);
+        return update.ToAgentRunResponse();
     }
 
-    public override IAsyncEnumerable<AgentRunResponseUpdate> RunStreamingAsync(IReadOnlyCollection<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
+    public override async IAsyncEnumerable<AgentRunResponseUpdate> RunStreamingAsync(IReadOnlyCollection<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
-        throw new NotImplementedException();
+        AgentRunResponseUpdate response = new(ChatRole.Assistant, "Hello World!")
+        {
+            AgentId = this.Id,
+            AuthorName = this.Name,
+        };
+
+        yield return response;
     }
 }
 
@@ -95,8 +123,19 @@ internal sealed class EchoAgent(string id = nameof(EchoAgent)) : AIAgent
     public const string DefaultId = nameof(EchoAgent);
 
     public override string Id => id;
+    public override string? Name => id;
 
-    public override Task<AgentRunResponse> RunAsync(IReadOnlyCollection<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
+    public override async Task<AgentRunResponse> RunAsync(IReadOnlyCollection<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
+    {
+        IEnumerable<AgentRunResponseUpdate> update = [
+            await this.RunStreamingAsync(messages, thread, options, cancellationToken)
+                      .SingleAsync(cancellationToken)
+                      .ConfigureAwait(false)];
+
+        return update.ToAgentRunResponse();
+    }
+
+    public override async IAsyncEnumerable<AgentRunResponseUpdate> RunStreamingAsync(IReadOnlyCollection<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         if (messages.Count == 0)
         {
@@ -110,13 +149,13 @@ internal sealed class EchoAgent(string id = nameof(EchoAgent)) : AIAgent
             collectedText.AppendLine(messageText);
         }
 
-        AgentRunResponse result = new(new ChatMessage(ChatRole.Assistant, collectedText.ToString()));
-        return Task.FromResult(result);
-    }
+        AgentRunResponseUpdate result = new(ChatRole.Assistant, collectedText.ToString())
+        {
+            AgentId = this.Id,
+            AuthorName = this.Name,
+        };
 
-    public override IAsyncEnumerable<AgentRunResponseUpdate> RunStreamingAsync(IReadOnlyCollection<ChatMessage> messages, AgentThread? thread = null, AgentRunOptions? options = null, CancellationToken cancellationToken = default)
-    {
-        throw new NotImplementedException();
+        yield return result;
     }
 }
 
@@ -144,6 +183,10 @@ internal sealed class GroupChatHistory
     public IEnumerable<ChatMessage> NewMessagesThisTurn => this._messages.Skip(this._bookmark);
 }
 
+internal class GroupChatManagerOptions
+{
+}
+
 internal abstract class GroupChatManager
 {
     public string[] ParticipantIds { get; internal init; } = [];
@@ -151,6 +194,11 @@ internal abstract class GroupChatManager
     public abstract int? GetNextTurnExecutor(GroupChatHistory history);
 }
 
+internal abstract class GroupChatManager<TOptions> : GroupChatManager where TOptions : GroupChatManagerOptions, new()
+{
+    protected internal virtual void Configure(TOptions options) { }
+}
+
 internal sealed class GroupChatBuilder
 {
     private readonly List<ExecutorIsh> _participants = new();
@@ -167,6 +215,21 @@ internal sealed class GroupChatBuilder
         return new GroupChatBuilder(participantIds => new TManager() { ParticipantIds = participantIds });
     }
 
+    public static GroupChatBuilder Create<TManager, TOptions>(Action<TOptions> configure)
+        where TManager : GroupChatManager<TOptions>, new()
+        where TOptions : GroupChatManagerOptions, new()
+    {
+        TOptions options = new();
+        configure(options);
+
+        return new GroupChatBuilder(participantIds =>
+        {
+            TManager manager = new() { ParticipantIds = participantIds };
+            manager.Configure(options);
+            return manager;
+        });
+    }
+
     public GroupChatBuilder AddParticipant(ExecutorIsh executor, bool shouldEmitEvents = false)
     {
         this._participants.Add(executor);
@@ -265,7 +328,7 @@ internal sealed class GroupChatBuilder
             if (this.TryEnterConversation())
             {
                 // Capture the initial turn token's EmitEvents setting
-                this._shouldHostEmitEvents = token.EmitEvents;
+                this._shouldHostEmitEvents = token.EmitEvents.HasValue ? token.EmitEvents.Value : false;
             }
 
             int? nextSpeakerIndex = this._manager.GetNextTurnExecutor(this._history);

dotnet/tests/Microsoft.Agents.Workflows.UnitTests/Sample/07_GroupChat_Workflow_HostAsAgent.cs

@@ -0,0 +1,31 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Threading.Tasks;
+using Microsoft.Extensions.AI;
+using Microsoft.Extensions.AI.Agents;
+
+namespace Microsoft.Agents.Workflows.UnitTests.Sample;
+
+internal static class Step7EntryPoint
+{
+    public static async ValueTask RunAsync(TextWriter writer, int maxSteps = 2)
+    {
+        Workflow<List<ChatMessage>> workflow = Step6EntryPoint.CreateWorkflow(maxSteps);
+        AIAgent agent = workflow.AsAgent("group-chat-agent", "Group Chat Agent");
+
+        AgentThread thread = agent.GetNewThread();
+
+        await foreach (AgentRunResponseUpdate update in agent.RunStreamingAsync(thread).ConfigureAwait(false))
+        {
+            string updateText = $"{update.AuthorName
+                                   ?? update.AgentId
+                                   ?? update.Role.ToString()
+                                   ?? ChatRole.Assistant.ToString()}: {update.Text}";
+            Console.WriteLine(updateText);
+            writer.WriteLine(updateText);
+        }
+    }
+}

dotnet/tests/Microsoft.Agents.Workflows.UnitTests/SampleSmokeTest.cs

@@ -100,6 +100,22 @@ public class SampleSmokeTest
             line => Assert.Contains($"{EchoAgent.DefaultId}: {EchoAgent.Prefix}{HelloAgent.Greeting}", line)
         );
     }
+
+    [Fact]
+    public async Task Test_RunSample_Step7Async()
+    {
+        using StringWriter writer = new();
+
+        await Step7EntryPoint.RunAsync(writer);
+
+        string result = writer.ToString();
+        string[] lines = result.Split([Environment.NewLine], StringSplitOptions.RemoveEmptyEntries);
+
+        Assert.Collection(lines,
+            line => Assert.Contains($"{HelloAgent.DefaultId}: {HelloAgent.Greeting}", line),
+            line => Assert.Contains($"{EchoAgent.DefaultId}: {EchoAgent.Prefix}{HelloAgent.Greeting}", line)
+        );
+    }
 }
 
 internal sealed class VerifyingPlaybackResponder<TInput, TResponse>