Enable more analyzers and various code tweaks/cleanup (#738)

This commit is contained in:
Stephen Toub
2025-09-18 01:48:25 +00:00
committed by GitHub
parent f3264966ff
commit 5e5761b288
326 changed files with 2402 additions and 3642 deletions
@@ -4,7 +4,7 @@ namespace System.Diagnostics.CodeAnalysis;
/// <summary>
/// Indicates that certain members on a specified <see cref="Type"/> are accessed dynamically,
/// for example through <see cref="System.Reflection"/>.
/// for example through <see cref="Reflection"/>.
/// </summary>
/// <remarks>
/// This allows tools to understand which members are being accessed during the execution
@@ -4,7 +4,7 @@ namespace System.Diagnostics.CodeAnalysis;
/// <summary>
/// Indicates that the specified method requires dynamic access to code that is not referenced
/// statically, for example through <see cref="System.Reflection"/>.
/// statically, for example through <see cref="Reflection"/>.
/// </summary>
/// <remarks>
/// This allows tools to understand which methods are unsafe to call when removing unreferenced
@@ -83,12 +83,12 @@ public partial class ConcurrentOrchestration : OrchestratingAgent
tasks.Add(Task.Run(async () =>
{
AIAgent agent = this.Agents[localI];
this.LogOrchestrationSubagentRunning(context, agent);
LogOrchestrationSubagentRunning(context, agent);
completed[localI] = await RunAsync(agent, context, input, options: null, cancellationToken).ConfigureAwait(false);
this.LogOrchestrationSubagentCompleted(context, agent);
await this.CheckpointAsync(input, completed, context, cancellationToken).ConfigureAwait(false);
LogOrchestrationSubagentCompleted(context, agent);
await CheckpointAsync(input, completed, context, cancellationToken).ConfigureAwait(false);
}, cancellationToken));
}
}
@@ -103,8 +103,8 @@ public partial class ConcurrentOrchestration : OrchestratingAgent
return await this.AggregationFunc(completed!, cancellationToken).ConfigureAwait(false);
}
private Task CheckpointAsync(IReadOnlyCollection<ChatMessage> messages, AgentRunResponse?[] completed, OrchestratingAgentContext context, CancellationToken cancellationToken) =>
context.Runtime is not null ? base.WriteCheckpointAsync(JsonSerializer.SerializeToElement(new(messages, completed), OrchestrationJsonContext.Default.ConcurrentState), context, cancellationToken) :
private static Task CheckpointAsync(IReadOnlyCollection<ChatMessage> messages, AgentRunResponse?[] completed, OrchestratingAgentContext context, CancellationToken cancellationToken) =>
context.Runtime is not null ? WriteCheckpointAsync(JsonSerializer.SerializeToElement(new(messages, completed), OrchestrationJsonContext.Default.ConcurrentState), context, cancellationToken) :
Task.CompletedTask;
internal sealed record ConcurrentState(IReadOnlyCollection<ChatMessage> Messages, AgentRunResponse?[] Completed);
@@ -59,7 +59,7 @@ public abstract class GroupChatManager
/// <param name="history">The chat history to filter.</param>
/// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
/// <returns>A <see cref="GroupChatManagerResult{TValue}"/> containing the filtered result as a string.</returns>
protected internal abstract ValueTask<GroupChatManagerResult<string>> FilterResults(IReadOnlyCollection<ChatMessage> history, CancellationToken cancellationToken = default);
protected internal abstract ValueTask<GroupChatManagerResult<string>> FilterResultsAsync(IReadOnlyCollection<ChatMessage> history, CancellationToken cancellationToken = default);
/// <summary>
/// Selects the next agent to participate in the group chat based on the provided chat history and team.
@@ -68,7 +68,7 @@ public abstract class GroupChatManager
/// <param name="team">The group of agents participating in the chat.</param>
/// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
/// <returns>A <see cref="GroupChatManagerResult{TValue}"/> containing the identifier of the next agent as a string.</returns>
protected internal abstract ValueTask<GroupChatManagerResult<string>> SelectNextAgent(IReadOnlyCollection<ChatMessage> history, GroupChatTeam team, CancellationToken cancellationToken = default);
protected internal abstract ValueTask<GroupChatManagerResult<string>> SelectNextAgentAsync(IReadOnlyCollection<ChatMessage> history, GroupChatTeam team, CancellationToken cancellationToken = default);
/// <summary>
/// Determines whether user input should be requested based on the provided chat history.
@@ -76,7 +76,7 @@ public abstract class GroupChatManager
/// <param name="history">The chat history to consider.</param>
/// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
/// <returns>A <see cref="GroupChatManagerResult{TValue}"/> indicating whether user input should be requested.</returns>
protected internal abstract ValueTask<GroupChatManagerResult<bool>> ShouldRequestUserInput(IReadOnlyCollection<ChatMessage> history, CancellationToken cancellationToken = default);
protected internal abstract ValueTask<GroupChatManagerResult<bool>> ShouldRequestUserInputAsync(IReadOnlyCollection<ChatMessage> history, CancellationToken cancellationToken = default);
/// <summary>
/// Determines whether the group chat should be terminated based on the provided chat history and invocation count.
@@ -84,7 +84,7 @@ public abstract class GroupChatManager
/// <param name="history">The chat history to consider.</param>
/// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
/// <returns>A <see cref="GroupChatManagerResult{TValue}"/> indicating whether the chat should be terminated.</returns>
protected internal virtual ValueTask<GroupChatManagerResult<bool>> ShouldTerminate(IReadOnlyCollection<ChatMessage> history, CancellationToken cancellationToken = default)
protected internal virtual ValueTask<GroupChatManagerResult<bool>> ShouldTerminateAsync(IReadOnlyCollection<ChatMessage> history, CancellationToken cancellationToken = default)
{
bool resultValue = false;
string reason = "Maximum number of invocations has not been reached.";
@@ -76,54 +76,51 @@ public sealed partial class GroupChatOrchestration : OrchestratingAgent
// First, check if we should request user input.
if (interactiveCallback is not null)
{
var userInputResult = await this._manager.ShouldRequestUserInput(allMessages, cancellationToken).ConfigureAwait(false);
if (userInputResult.Value)
var userInputResult = await this._manager.ShouldRequestUserInputAsync(allMessages, cancellationToken).ConfigureAwait(false);
if (userInputResult.Value && interactiveCallback is not null)
{
if (interactiveCallback is not null)
ChatMessage userMessage = await interactiveCallback().ConfigureAwait(false);
allMessages.Add(userMessage);
// Broadcast the user input
if (this.ResponseCallback is not null)
{
ChatMessage userMessage = await interactiveCallback().ConfigureAwait(false);
allMessages.Add(userMessage);
// Broadcast the user input
if (this.ResponseCallback is not null)
{
await this.ResponseCallback([userMessage]).ConfigureAwait(false);
}
await this.CheckpointAsync(allMessages, originalMessageCount, context, cancellationToken).ConfigureAwait(false);
continue;
await this.ResponseCallback([userMessage]).ConfigureAwait(false);
}
await CheckpointAsync(allMessages, originalMessageCount, context, cancellationToken).ConfigureAwait(false);
continue;
}
}
// Check if we should terminate the conversation
var terminateResult = await this._manager.ShouldTerminate(allMessages, cancellationToken).ConfigureAwait(false);
var terminateResult = await this._manager.ShouldTerminateAsync(allMessages, cancellationToken).ConfigureAwait(false);
if (terminateResult.Value)
{
// Filter and return final results
var filterResult = await this._manager.FilterResults(allMessages, cancellationToken).ConfigureAwait(false);
var filterResult = await this._manager.FilterResultsAsync(allMessages, cancellationToken).ConfigureAwait(false);
return new AgentRunResponse([new ChatMessage(ChatRole.Assistant, filterResult.Value) { AuthorName = this.DisplayName }]);
}
// Select the next agent to speak
var nextAgentResult = await this._manager.SelectNextAgent(allMessages, team, cancellationToken).ConfigureAwait(false);
var nextAgentResult = await this._manager.SelectNextAgentAsync(allMessages, team, cancellationToken).ConfigureAwait(false);
AIAgent nextAgent = this.FindAgentByName(nextAgentResult.Value) ??
throw new InvalidOperationException($"AIAgent '{nextAgentResult.Value}' not found in the orchestration.");
// Run the selected agent with all messages.
this.LogOrchestrationSubagentRunning(context, nextAgent);
LogOrchestrationSubagentRunning(context, nextAgent);
AgentRunResponse response = await RunAsync(nextAgent, context, allMessages, options: null, cancellationToken).ConfigureAwait(false);
allMessages.AddRange(response.Messages); // Add the agent's response to the conversation.
this.LogOrchestrationSubagentCompleted(context, nextAgent);
LogOrchestrationSubagentCompleted(context, nextAgent);
await this.CheckpointAsync(allMessages, originalMessageCount, context, cancellationToken).ConfigureAwait(false);
await CheckpointAsync(allMessages, originalMessageCount, context, cancellationToken).ConfigureAwait(false);
}
}
private AIAgent? FindAgentByName(string name) => this.Agents.FirstOrDefault(a => a.DisplayName == name);
private Task CheckpointAsync(List<ChatMessage> allMessages, int originalMessageCount, OrchestratingAgentContext context, CancellationToken cancellationToken) =>
context.Runtime is not null ? base.WriteCheckpointAsync(JsonSerializer.SerializeToElement(new(allMessages, originalMessageCount), OrchestrationJsonContext.Default.GroupChatState), context, cancellationToken) :
private static Task CheckpointAsync(List<ChatMessage> allMessages, int originalMessageCount, OrchestratingAgentContext context, CancellationToken cancellationToken) =>
context.Runtime is not null ? WriteCheckpointAsync(JsonSerializer.SerializeToElement(new(allMessages, originalMessageCount), OrchestrationJsonContext.Default.GroupChatState), context, cancellationToken) :
Task.CompletedTask;
internal sealed record GroupChatState(List<ChatMessage> AllMessages, int OriginalMessageCount);
@@ -19,7 +19,7 @@ public class RoundRobinGroupChatManager : GroupChatManager
private int _currentAgentIndex;
/// <inheritdoc/>
protected internal override ValueTask<GroupChatManagerResult<string>> FilterResults(
protected internal override ValueTask<GroupChatManagerResult<string>> FilterResultsAsync(
IReadOnlyCollection<ChatMessage> history, CancellationToken cancellationToken = default)
{
GroupChatManagerResult<string> result = new(history.LastOrDefault()?.Text ?? string.Empty) { Reason = "Default result filter provides the final chat message." };
@@ -27,7 +27,7 @@ public class RoundRobinGroupChatManager : GroupChatManager
}
/// <inheritdoc/>
protected internal override ValueTask<GroupChatManagerResult<string>> SelectNextAgent(
protected internal override ValueTask<GroupChatManagerResult<string>> SelectNextAgentAsync(
IReadOnlyCollection<ChatMessage> history, GroupChatTeam team, CancellationToken cancellationToken = default)
{
string nextAgent = team.Skip(this._currentAgentIndex).First().Key;
@@ -37,7 +37,7 @@ public class RoundRobinGroupChatManager : GroupChatManager
}
/// <inheritdoc/>
protected internal override ValueTask<GroupChatManagerResult<bool>> ShouldRequestUserInput(
protected internal override ValueTask<GroupChatManagerResult<bool>> ShouldRequestUserInputAsync(
IReadOnlyCollection<ChatMessage> history, CancellationToken cancellationToken = default)
{
GroupChatManagerResult<bool> result = new(false) { Reason = "The default round-robin group chat manager does not request user input." };
@@ -85,13 +85,13 @@ public sealed partial class HandoffOrchestration : OrchestratingAgent
while (agent is not null)
{
this.LogOrchestrationSubagentRunning(context, agent);
LogOrchestrationSubagentRunning(context, agent);
if (!this._handoffs.Targets.TryGetValue(agent, out var handoffs) || handoffs.Count == 0)
{
// If no handoff is available, we can run the agent directly and return its response.
response = await RunAsync(agent, context, allMessages, context.Options, cancellationToken).ConfigureAwait(false);
this.LogOrchestrationSubagentCompleted(context, agent);
LogOrchestrationSubagentCompleted(context, agent);
allMessages.AddRange(response.Messages);
agent = null;
await CheckpointAsync().ConfigureAwait(false);
@@ -100,7 +100,7 @@ public sealed partial class HandoffOrchestration : OrchestratingAgent
// Create the options for the next agent request, including handoff functions.
HandoffContext handoffCtx = new(handoffs);
ChatClientAgentRunOptions? options = null;
ChatClientAgentRunOptions? options;
List<AITool> handoffTools = handoffCtx.CreateHandoffFunctions(this.InteractiveCallback is not null);
if (context.Options is ChatClientAgentRunOptions contextOptions)
{
@@ -115,7 +115,7 @@ public sealed partial class HandoffOrchestration : OrchestratingAgent
// Invoke the next agent with all of the messages collected so far.
response = await RunAsync(agent, context, allMessages, options, cancellationToken).ConfigureAwait(false);
this.LogOrchestrationSubagentCompleted(context, agent);
LogOrchestrationSubagentCompleted(context, agent);
allMessages.AddRange(response.Messages);
agent = handoffCtx.TargetedAgent;
RemoveHandoffFunctionCalls(response, handoffTools);
@@ -139,7 +139,7 @@ public sealed partial class HandoffOrchestration : OrchestratingAgent
return response;
Task CheckpointAsync() => context.Runtime is not null ?
base.WriteCheckpointAsync(JsonSerializer.SerializeToElement(new(agent?.Id, allMessages, originalMessageCount), OrchestrationJsonContext.Default.HandoffState), context, cancellationToken) :
WriteCheckpointAsync(JsonSerializer.SerializeToElement(new(agent?.Id, allMessages, originalMessageCount), OrchestrationJsonContext.Default.HandoffState), context, cancellationToken) :
Task.CompletedTask;
}
@@ -17,7 +17,7 @@ public sealed class Handoffs :
IReadOnlyDictionary<AIAgent, IEnumerable<Handoffs.HandoffTarget>>
{
/// <summary>
/// Initializes a new instance of the <see cref="Orchestration.Handoffs"/> class with no handoff relationships.
/// Initializes a new instance of the <see cref="Handoffs"/> class with no handoff relationships.
/// </summary>
/// <param name="initialAgent">The first agent to be invoked (prior to any handoff).</param>
private Handoffs(AIAgent initialAgent)
@@ -41,7 +41,7 @@ public sealed class Handoffs :
/// Creates a new collection of handoffs that start with the specified agent.
/// </summary>
/// <param name="initialAgent">The initial agent.</param>
/// <returns>The new <see cref="Orchestration.Handoffs"/> instance.</returns>
/// <returns>The new <see cref="Handoffs"/> instance.</returns>
public static Handoffs StartWith(AIAgent initialAgent) => new(initialAgent);
/// <summary>Creates a new <see cref="HandoffOrchestration"/> from the described handoffs.</summary>
@@ -54,7 +54,7 @@ public sealed class Handoffs :
/// </summary>
/// <param name="source">The source agent.</param>
/// <param name="targets">The target agents to add as handoff targets for the source agent.</param>
/// <returns>The updated <see cref="Orchestration.Handoffs"/> instance.</returns>
/// <returns>The updated <see cref="Handoffs"/> instance.</returns>
/// <remarks>The handoff reason for each target is derived from its description or name.</remarks>
public Handoffs Add(AIAgent source, AIAgent[] targets)
{
@@ -79,7 +79,7 @@ public sealed class Handoffs :
/// <param name="source">The source agent.</param>
/// <param name="target">The target agent.</param>
/// <param name="handoffReason">The reason the <paramref name="source"/> should hand off to the <paramref name="target"/>.</param>
/// <returns>The updated <see cref="Orchestration.Handoffs"/> instance.</returns>
/// <returns>The updated <see cref="Handoffs"/> instance.</returns>
public Handoffs Add(AIAgent source, AIAgent target, string? handoffReason = null)
{
Throw.IfNull(source);
@@ -127,7 +127,7 @@ public sealed class Handoffs :
/// <inheritdoc />
IEnumerator IEnumerable.GetEnumerator() =>
((IReadOnlyDictionary<AIAgent, IEnumerable<Handoffs.HandoffTarget>>)this).GetEnumerator();
((IReadOnlyDictionary<AIAgent, IEnumerable<HandoffTarget>>)this).GetEnumerator();
/// <inheritdoc />
bool IReadOnlyDictionary<AIAgent, IEnumerable<HandoffTarget>>.TryGetValue(AIAgent key, out IEnumerable<HandoffTarget> value)
@@ -129,7 +129,7 @@ public abstract partial class OrchestratingAgent : AIAgent
CancellationTokenSource cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
cancellationToken = cts.Token;
JsonElement? checkpoint = await this.ReadCheckpointAsync(context, cancellationToken).ConfigureAwait(false);
JsonElement? checkpoint = await ReadCheckpointAsync(context, cancellationToken).ConfigureAwait(false);
Task<AgentRunResponse> completion = checkpoint is null ?
this.RunCoreAsync(readonlyCollectionMessages, context, cancellationToken) :
this.ResumeCoreAsync(checkpoint.Value, readonlyCollectionMessages, context, cancellationToken);
@@ -207,7 +207,7 @@ public abstract partial class OrchestratingAgent : AIAgent
/// <param name="context">The context for the orchestrating operation.</param>
/// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
/// <returns>A Task that completes when the asynchronous operation quiesces.</returns>
protected async Task WriteCheckpointAsync(JsonElement state, OrchestratingAgentContext context, CancellationToken cancellationToken)
protected static async Task WriteCheckpointAsync(JsonElement state, OrchestratingAgentContext context, CancellationToken cancellationToken)
{
_ = Throw.IfNull(context);
@@ -236,7 +236,7 @@ public abstract partial class OrchestratingAgent : AIAgent
/// <param name="context">The context for the orchestrating operation.</param>
/// <param name="cancellationToken">A cancellation token that can be used to cancel the operation.</param>
/// <returns>The loaded state, or null if it doesn't exist.</returns>
protected async ValueTask<JsonElement?> ReadCheckpointAsync(OrchestratingAgentContext context, CancellationToken cancellationToken)
protected static async ValueTask<JsonElement?> ReadCheckpointAsync(OrchestratingAgentContext context, CancellationToken cancellationToken)
{
_ = Throw.IfNull(context);
@@ -276,10 +276,10 @@ public abstract partial class OrchestratingAgent : AIAgent
[LoggerMessage(Level = LogLevel.Trace, Message = "{Orchestration} completed agent '{Agent}' ('{Id}')")]
private static partial void LogOrchestrationSubagentCompleted(ILogger logger, string orchestration, string id, string agent);
private protected void LogOrchestrationSubagentRunning(OrchestratingAgentContext context, AIAgent agent) =>
private protected static void LogOrchestrationSubagentRunning(OrchestratingAgentContext context, AIAgent agent) =>
LogOrchestrationSubagentRunning(context.Logger, context.ToString(), context.Id, agent.DisplayName);
private protected void LogOrchestrationSubagentCompleted(OrchestratingAgentContext context, AIAgent agent) =>
private protected static void LogOrchestrationSubagentCompleted(OrchestratingAgentContext context, AIAgent agent) =>
LogOrchestrationSubagentCompleted(context.Logger, context.ToString(), context.Id, agent.DisplayName);
private static async Task LogCompletionAsync(ILogger logger, OrchestratingAgentContext context, Task<AgentRunResponse> completion)
@@ -48,20 +48,20 @@ public sealed partial class SequentialOrchestration : OrchestratingAgent
AgentRunResponse? response = null;
for (; i < this.Agents.Count; i++)
{
this.LogOrchestrationSubagentRunning(context, this.Agents[i]);
LogOrchestrationSubagentRunning(context, this.Agents[i]);
response = await RunAsync(this.Agents[i], context, input, options: null, cancellationToken).ConfigureAwait(false);
input = response.Messages as IReadOnlyCollection<ChatMessage> ?? [.. response.Messages];
await this.CheckpointAsync(i + 1, input, context, cancellationToken).ConfigureAwait(false);
await CheckpointAsync(i + 1, input, context, cancellationToken).ConfigureAwait(false);
}
Debug.Assert(response is not null, "Response should not be null after processing a positive number of agents.");
return response!;
}
private Task CheckpointAsync(int index, IReadOnlyCollection<ChatMessage> messages, OrchestratingAgentContext context, CancellationToken cancellationToken) =>
context.Runtime is not null ? base.WriteCheckpointAsync(JsonSerializer.SerializeToElement(new(index, messages), OrchestrationJsonContext.Default.SequentialState), context, cancellationToken) :
private static Task CheckpointAsync(int index, IReadOnlyCollection<ChatMessage> messages, OrchestratingAgentContext context, CancellationToken cancellationToken) =>
context.Runtime is not null ? WriteCheckpointAsync(JsonSerializer.SerializeToElement(new(index, messages), OrchestrationJsonContext.Default.SequentialState), context, cancellationToken) :
Task.CompletedTask;
internal sealed record SequentialState(int Index, IReadOnlyCollection<ChatMessage> Messages);
@@ -90,12 +90,8 @@ public sealed class AzureAgentProvider(string projectEndpoint, TokenCredential p
}
/// <inheritdoc/>
public override async Task<AIAgent> GetAgentAsync(string agentId, CancellationToken cancellationToken = default)
{
AIAgent agent = await this.GetAgentsClient().GetAIAgentAsync(agentId, chatOptions: null, cancellationToken).ConfigureAwait(false);
return agent;
}
public override async Task<AIAgent> GetAgentAsync(string agentId, CancellationToken cancellationToken = default) =>
await this.GetAgentsClient().GetAIAgentAsync(agentId, chatOptions: null, cancellationToken).ConfigureAwait(false);
/// <inheritdoc/>
public override async Task<ChatMessage> GetMessageAsync(string conversationId, string messageId, CancellationToken cancellationToken = default)
@@ -31,7 +31,7 @@ public static class DeclarativeWorkflowBuilder
where TInput : notnull
{
using StreamReader yamlReader = File.OpenText(workflowFile);
return Build<TInput>(yamlReader, options, inputTransform);
return Build(yamlReader, options, inputTransform);
}
/// <summary>
@@ -8,14 +8,12 @@ internal static class BotElementExtensions
{
public static string? GetParentId(this BotElement element) => element.Parent?.GetId();
public static string GetId(this BotElement element)
{
return element switch
public static string GetId(this BotElement element) =>
element switch
{
DialogAction action => action.Id.Value,
ConditionItem conditionItem => conditionItem.Id ?? throw new DeclarativeModelException($"Undefined identifier for {nameof(ConditionItem)} that is member of {conditionItem.GetParentId() ?? "(root)"}."),
OnActivity activity => activity.Id.Value,
_ => throw new DeclarativeModelException($"Unknown identify for element type: {element.GetType().Name}"),
};
}
}
@@ -13,7 +13,7 @@ namespace Microsoft.Agents.Workflows.Declarative.Extensions;
internal static class ChatMessageExtensions
{
public static RecordValue ToRecord(this ChatMessage message) =>
RecordValue.NewRecordFromFields(message.GetMessageFields());
FormulaValue.NewRecordFromFields(message.GetMessageFields());
public static TableValue ToTable(this IEnumerable<ChatMessage> messages) =>
FormulaValue.NewTable(s_messageRecordType, messages.Select(message => message.ToRecord()));
@@ -14,7 +14,7 @@ internal static class DataValueExtensions
value switch
{
null => FormulaValue.NewBlank(),
BlankDataValue => BlankValue.NewBlank(),
BlankDataValue => FormulaValue.NewBlank(),
BooleanDataValue boolValue => FormulaValue.New(boolValue.Value),
NumberDataValue numberValue => FormulaValue.New(numberValue.Value),
FloatDataValue floatValue => FormulaValue.New(floatValue.Value),
@@ -71,8 +71,8 @@ internal static class FormulaValueExtensions
DateTimeValue datetimeValue => DateTimeDataValue.Create(datetimeValue.GetConvertedValue(TimeZoneInfo.Utc)),
TimeValue timeValue => TimeDataValue.Create(timeValue.Value),
StringValue stringValue => StringDataValue.Create(stringValue.Value),
BlankValue blankValue => DataValue.Blank(),
VoidValue voidValue => DataValue.Blank(),
BlankValue => DataValue.Blank(),
VoidValue => DataValue.Blank(),
RecordValue recordValue => recordValue.ToRecord(),
TableValue tableValue => tableValue.ToTable(),
_ => throw new DeclarativeModelException($"Unsupported variable type: {value.GetType().Name}"),
@@ -141,10 +141,10 @@ internal static class FormulaValueExtensions
};
public static TableDataValue ToTable(this TableValue value) =>
TableDataValue.TableFromRecords(value.Rows.Select(row => row.Value.ToRecord()).ToImmutableArray());
DataValue.TableFromRecords(value.Rows.Select(row => row.Value.ToRecord()).ToImmutableArray());
public static RecordDataValue ToRecord(this RecordValue value) =>
RecordDataValue.RecordFromFields(value.OriginalFields.Select(field => field.GetKeyValuePair()));
DataValue.RecordFromFields(value.OriginalFields.Select(field => field.GetKeyValuePair()));
private static RecordValue ToRecord(this IDictionary value)
{
@@ -229,7 +229,7 @@ internal static class FormulaValueExtensions
GuidValue guidValue => JsonValue.Create(guidValue.Value),
RecordValue recordValue => recordValue.ToJson(),
TableValue tableValue => tableValue.ToJson(),
BlankValue blankValue => JsonValue.Create(string.Empty),
BlankValue => JsonValue.Create(string.Empty),
_ => $"[{value.GetType().Name}]",
};
@@ -23,12 +23,12 @@ internal static class RecordDataTypeExtensions
FormulaValue? parsedValue =
property.Value.Type switch
{
StringDataType => StringValue.New(propertyElement.GetString()),
NumberDataType => NumberValue.New(propertyElement.GetDecimal()),
BooleanDataType => BooleanValue.New(propertyElement.GetBoolean()),
DateTimeDataType => DateTimeValue.New(propertyElement.GetDateTime()),
DateDataType => DateValue.New(propertyElement.GetDateTime()),
TimeDataType => TimeValue.New(propertyElement.GetDateTimeOffset().TimeOfDay),
StringDataType => FormulaValue.New(propertyElement.GetString()),
NumberDataType => FormulaValue.New(propertyElement.GetDecimal()),
BooleanDataType => FormulaValue.New(propertyElement.GetBoolean()),
DateTimeDataType => FormulaValue.New(propertyElement.GetDateTime()),
DateDataType => FormulaValue.New(propertyElement.GetDateTime()),
TimeDataType => FormulaValue.New(propertyElement.GetDateTimeOffset().TimeOfDay),
RecordDataType recordType => recordType.ParseRecord(propertyElement),
TableDataType tableType => ParseTable(tableType, propertyElement),
_ => throw new InvalidOperationException($"Unsupported data type '{property.Value.Type}' for property '{property.Key}'"),
@@ -5,19 +5,24 @@ using Microsoft.PowerFx.Types;
namespace Microsoft.Agents.Workflows.Declarative.Extensions;
internal static class StringExtensions
internal static partial class StringExtensions
{
private static readonly Regex s_regex = new(@"^```(?:\w*)\s*([\s\S]*?)\s*```$", RegexOptions.Compiled | RegexOptions.Multiline);
#if NET
[GeneratedRegex(@"^```(?:\w*)\s*([\s\S]*?)\s*```$", RegexOptions.Multiline)]
private static partial Regex TrimJsonDelimiterRegex();
#else
private static Regex TrimJsonDelimiterRegex() => s_trimJsonDelimiterRegex;
private static readonly Regex s_trimJsonDelimiterRegex = new(@"^```(?:\w*)\s*([\s\S]*?)\s*```$", RegexOptions.Compiled | RegexOptions.Multiline);
#endif
public static string TrimJsonDelimiter(this string value)
{
Match match = s_regex.Match(value.Trim());
if (match.Success)
{
return match.Groups[1].Value.Trim();
}
value = value.Trim();
return value.Trim();
Match match = TrimJsonDelimiterRegex().Match(value);
return match.Success ?
match.Groups[1].Value.Trim() :
value;
}
public static FormulaValue ToFormula(this string? value) =>
@@ -4,21 +4,18 @@ using System.Collections.Generic;
using System.Linq;
using Microsoft.Bot.ObjectModel;
using Microsoft.PowerFx;
using Microsoft.PowerFx.Types;
namespace Microsoft.Agents.Workflows.Declarative.Extensions;
internal static class TemplateExtensions
{
public static string Format(this RecalcEngine engine, IEnumerable<TemplateLine> template)
{
return string.Concat(template.Select(line => engine.Format(line)));
}
public static string Format(this RecalcEngine engine, IEnumerable<TemplateLine> template) =>
string.Concat(template.Select(engine.Format));
public static string Format(this RecalcEngine engine, TemplateLine? line)
{
return string.Concat(line?.Segments.Select(segment => engine.Format(segment)) ?? [string.Empty]);
}
public static string Format(this RecalcEngine engine, TemplateLine? line) =>
line is not null ?
string.Concat(line.Segments.Select(engine.Format)) :
string.Empty;
public static string Format(this RecalcEngine engine, TemplateSegment segment)
{
@@ -27,20 +24,16 @@ internal static class TemplateExtensions
return textSegment.Value ?? string.Empty;
}
if (segment is ExpressionSegment expressionSegment)
if (segment is ExpressionSegment { Expression: not null } expressionSegment)
{
if (expressionSegment.Expression is not null)
if (expressionSegment.Expression.ExpressionText is not null)
{
if (expressionSegment.Expression.ExpressionText is not null)
{
FormulaValue expressionValue = engine.Eval(expressionSegment.Expression.ExpressionText);
return expressionValue.Format();
}
if (expressionSegment.Expression.VariableReference is not null)
{
FormulaValue expressionValue = engine.Eval(expressionSegment.Expression.VariableReference.ToString());
return expressionValue.Format();
}
return engine.Eval(expressionSegment.Expression.ExpressionText).Format();
}
if (expressionSegment.Expression.VariableReference is not null)
{
return engine.Eval(expressionSegment.Expression.VariableReference.ToString()).Format();
}
}
@@ -23,7 +23,7 @@ internal sealed class DeclarativeWorkflowModel
public int GetDepth(string? nodeId)
{
if (nodeId == null)
if (nodeId is null)
{
return 0;
}
@@ -112,15 +112,7 @@ internal sealed class DeclarativeWorkflowModel
workflowBuilder.AddEdge(GetExecutorIsh(link.Source), GetExecutorIsh(targetNode), link.Condition);
}
ExecutorIsh GetExecutorIsh(ModelNode node)
{
if (node.Port is not null)
{
return node.Port;
}
return node.Executor;
}
static ExecutorIsh GetExecutorIsh(ModelNode node) => node.Port ?? (ExecutorIsh)node.Executor;
}
private ModelNode DefineNode(Executor executor, ModelNode? parentNode = null, Action? completionHandler = null)
@@ -148,7 +140,7 @@ internal sealed class DeclarativeWorkflowModel
return null;
}
while (itemId != null)
while (itemId is not null)
{
if (!this.Nodes.TryGetValue(itemId, out ModelNode? itemNode))
{
@@ -180,7 +172,7 @@ internal sealed class DeclarativeWorkflowModel
public List<ModelNode> Children { get; } = [];
public int Depth => this.Parent?.Depth + 1 ?? 0;
public int Depth => (this.Parent?.Depth + 1) ?? 0;
public Action? CompletionHandler => completionHandler;
}
@@ -137,7 +137,7 @@ internal sealed class WorkflowActionVisitor : DialogActionVisitor
// Create conditional link for else action
string stepId = ConditionGroupExecutor.Steps.Else(item);
this._workflowModel.AddLink(action.Id, stepId, (result) => action.IsElse(result));
this._workflowModel.AddLink(action.Id, stepId, action.IsElse);
}
}
@@ -355,170 +355,74 @@ internal sealed class WorkflowActionVisitor : DialogActionVisitor
this.NotSupported(item);
}
protected override void Visit(GetActivityMembers item)
{
this.NotSupported(item);
}
protected override void Visit(GetActivityMembers item) => this.NotSupported(item);
protected override void Visit(UpdateActivity item)
{
this.NotSupported(item);
}
protected override void Visit(UpdateActivity item) => this.NotSupported(item);
protected override void Visit(ActivateExternalTrigger item)
{
this.NotSupported(item);
}
protected override void Visit(ActivateExternalTrigger item) => this.NotSupported(item);
protected override void Visit(DisableTrigger item)
{
this.NotSupported(item);
}
protected override void Visit(DisableTrigger item) => this.NotSupported(item);
protected override void Visit(WaitForConnectorTrigger item)
{
this.NotSupported(item);
}
protected override void Visit(WaitForConnectorTrigger item) => this.NotSupported(item);
protected override void Visit(InvokeConnectorAction item)
{
this.NotSupported(item);
}
protected override void Visit(InvokeConnectorAction item) => this.NotSupported(item);
protected override void Visit(InvokeCustomModelAction item)
{
this.NotSupported(item);
}
protected override void Visit(InvokeCustomModelAction item) => this.NotSupported(item);
protected override void Visit(InvokeFlowAction item)
{
this.NotSupported(item);
}
protected override void Visit(InvokeFlowAction item) => this.NotSupported(item);
protected override void Visit(InvokeAIBuilderModelAction item)
{
this.NotSupported(item);
}
protected override void Visit(InvokeAIBuilderModelAction item) => this.NotSupported(item);
protected override void Visit(InvokeSkillAction item)
{
this.NotSupported(item);
}
protected override void Visit(InvokeSkillAction item) => this.NotSupported(item);
protected override void Visit(AdaptiveCardPrompt item)
{
this.NotSupported(item);
}
protected override void Visit(AdaptiveCardPrompt item) => this.NotSupported(item);
protected override void Visit(CSATQuestion item)
{
this.NotSupported(item);
}
protected override void Visit(OAuthInput item)
{
this.NotSupported(item);
}
protected override void Visit(OAuthInput item) => this.NotSupported(item);
protected override void Visit(BeginDialog item)
{
this.NotSupported(item);
}
protected override void Visit(BeginDialog item) => this.NotSupported(item);
protected override void Visit(UnknownDialogAction item)
{
this.NotSupported(item);
}
protected override void Visit(UnknownDialogAction item) => this.NotSupported(item);
protected override void Visit(EndDialog item)
{
this.NotSupported(item);
}
protected override void Visit(EndDialog item) => this.NotSupported(item);
protected override void Visit(RepeatDialog item)
{
this.NotSupported(item);
}
protected override void Visit(RepeatDialog item) => this.NotSupported(item);
protected override void Visit(ReplaceDialog item)
{
this.NotSupported(item);
}
protected override void Visit(ReplaceDialog item) => this.NotSupported(item);
protected override void Visit(CancelAllDialogs item)
{
this.NotSupported(item);
}
protected override void Visit(CancelAllDialogs item) => this.NotSupported(item);
protected override void Visit(CancelDialog item)
{
this.NotSupported(item);
}
protected override void Visit(CancelDialog item) => this.NotSupported(item);
protected override void Visit(EmitEvent item)
{
this.NotSupported(item);
}
protected override void Visit(EmitEvent item) => this.NotSupported(item);
protected override void Visit(GetConversationMembers item)
{
this.NotSupported(item);
}
protected override void Visit(GetConversationMembers item) => this.NotSupported(item);
protected override void Visit(HttpRequestAction item)
{
this.NotSupported(item);
}
protected override void Visit(HttpRequestAction item) => this.NotSupported(item);
protected override void Visit(RecognizeIntent item)
{
this.NotSupported(item);
}
protected override void Visit(RecognizeIntent item) => this.NotSupported(item);
protected override void Visit(TransferConversation item)
{
this.NotSupported(item);
}
protected override void Visit(TransferConversation item) => this.NotSupported(item);
protected override void Visit(TransferConversationV2 item)
{
this.NotSupported(item);
}
protected override void Visit(TransferConversationV2 item) => this.NotSupported(item);
protected override void Visit(SignOutUser item)
{
this.NotSupported(item);
}
protected override void Visit(SignOutUser item) => this.NotSupported(item);
protected override void Visit(LogCustomTelemetryEvent item)
{
this.NotSupported(item);
}
protected override void Visit(LogCustomTelemetryEvent item) => this.NotSupported(item);
protected override void Visit(DisconnectedNodeContainer item)
{
this.NotSupported(item);
}
protected override void Visit(DisconnectedNodeContainer item) => this.NotSupported(item);
protected override void Visit(CreateSearchQuery item)
{
this.NotSupported(item);
}
protected override void Visit(CreateSearchQuery item) => this.NotSupported(item);
protected override void Visit(SearchKnowledgeSources item)
{
this.NotSupported(item);
}
protected override void Visit(SearchKnowledgeSources item) => this.NotSupported(item);
protected override void Visit(SearchAndSummarizeWithCustomModel item)
{
this.NotSupported(item);
}
protected override void Visit(SearchAndSummarizeWithCustomModel item) => this.NotSupported(item);
protected override void Visit(SearchAndSummarizeContent item)
{
this.NotSupported(item);
}
protected override void Visit(SearchAndSummarizeContent item) => this.NotSupported(item);
#endregion
@@ -563,10 +467,8 @@ internal sealed class WorkflowActionVisitor : DialogActionVisitor
this.HasUnsupportedActions = true;
}
private void Trace(BotElement item)
{
private void Trace(BotElement item) =>
Debug.WriteLine($"> VISIT: {new string('\t', this._workflowModel.GetDepth(item.GetParentId()))}{FormatItem(item)} => {FormatParent(item)}");
}
private void Trace(DialogAction item)
{
@@ -575,6 +477,7 @@ internal sealed class WorkflowActionVisitor : DialogActionVisitor
{
parentId = Steps.Root(parentId);
}
Debug.WriteLine($"> VISIT: {new string('\t', this._workflowModel.GetDepth(parentId))}{FormatItem(item)} => {FormatParent(item)}");
}
@@ -15,7 +15,7 @@ internal sealed class ClearAllVariablesExecutor(ClearAllVariables model, Workflo
{
protected override ValueTask<object?> ExecuteAsync(IWorkflowContext context, CancellationToken cancellationToken)
{
EvaluationResult<VariablesToClearWrapper> variablesResult = this.State.Evaluator.GetValue<VariablesToClearWrapper>(this.Model.Variables);
EvaluationResult<VariablesToClearWrapper> variablesResult = this.State.Evaluator.GetValue(this.Model.Variables);
variablesResult.Value.Handle(new ScopeHandler(this.Id, this.State));
@@ -24,20 +24,16 @@ internal sealed class ClearAllVariablesExecutor(ClearAllVariables model, Workflo
private sealed class ScopeHandler(string executorId, WorkflowFormulaState state) : IEnumVariablesToClearHandler
{
public void HandleAllGlobalVariables()
{
public void HandleAllGlobalVariables() =>
this.ClearAll(VariableScopeNames.Global);
}
public void HandleConversationHistory()
{
// Not supported....
}
public void HandleConversationScopedVariables()
{
public void HandleConversationScopedVariables() =>
this.ClearAll(WorkflowFormulaState.DefaultScopeName);
}
public void HandleUnknownValue()
{
@@ -47,9 +47,7 @@ internal sealed class ConditionGroupExecutor : DeclarativeActionExecutor<Conditi
return string.Equals(Steps.Else(this.Model), executorMessage.Result as string, StringComparison.Ordinal);
}
#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
protected override async ValueTask<object?> ExecuteAsync(IWorkflowContext context, CancellationToken cancellationToken)
#pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously
{
for (int index = 0; index < this.Model.Conditions.Length; ++index)
{
@@ -69,8 +67,6 @@ internal sealed class ConditionGroupExecutor : DeclarativeActionExecutor<Conditi
return Steps.Else(this.Model);
}
public async ValueTask DoneAsync(IWorkflowContext context, ExecutorResultMessage _, CancellationToken cancellationToken)
{
public async ValueTask DoneAsync(IWorkflowContext context, ExecutorResultMessage _, CancellationToken cancellationToken) =>
await context.RaiseCompletionEventAsync(this.Model).ConfigureAwait(false);
}
}
@@ -61,7 +61,7 @@ internal sealed class InvokeAzureAgentExecutor(InvokeAzureAgent model, WorkflowA
await foreach (AgentRunResponseUpdate update in agentUpdates.ConfigureAwait(false))
{
await AssignConversationId(((ChatResponseUpdate?)update.RawRepresentation)?.ConversationId).ConfigureAwait(false);
await AssignConversationIdAsync(((ChatResponseUpdate?)update.RawRepresentation)?.ConversationId).ConfigureAwait(false);
if (autoSend)
{
@@ -72,7 +72,7 @@ internal sealed class InvokeAzureAgentExecutor(InvokeAzureAgent model, WorkflowA
}
}
async ValueTask AssignConversationId(string? assignValue)
async ValueTask AssignConversationIdAsync(string? assignValue)
{
if (assignValue is not null && conversationId is null)
{
@@ -42,9 +42,9 @@ internal sealed class ParseValueExecutor(ParseValue model, WorkflowFormulaState
parsedResult =
this.Model.ValueType switch
{
StringDataType => StringValue.New(stringValue.Value),
NumberDataType => NumberValue.New(stringValue.Value),
BooleanDataType => BooleanValue.New(stringValue.Value),
StringDataType => FormulaValue.New(stringValue.Value),
NumberDataType => FormulaValue.New(stringValue.Value),
BooleanDataType => FormulaValue.New(stringValue.Value),
RecordDataType recordType => ParseRecord(recordType, stringValue.Value),
_ => null
};
@@ -63,11 +63,10 @@ internal sealed class ParseValueExecutor(ParseValue model, WorkflowFormulaState
RecordValue ParseRecord(RecordDataType recordType, string rawText)
{
string jsonText = rawText.TrimJsonDelimiter();
JsonDocument json = JsonDocument.Parse(jsonText);
JsonElement currentElement = json.RootElement;
using JsonDocument json = JsonDocument.Parse(jsonText);
try
{
return recordType.ParseRecord(currentElement);
return recordType.ParseRecord(json.RootElement);
}
catch (Exception exception)
{
@@ -15,7 +15,7 @@ internal sealed class ResetVariableExecutor(ResetVariable model, WorkflowFormula
{
protected override ValueTask<object?> ExecuteAsync(IWorkflowContext context, CancellationToken cancellationToken)
{
PropertyPath variablePath = Throw.IfNull(this.Model.Variable, $"{nameof(this.Model)}.{nameof(model.Variable)}");
Throw.IfNull(this.Model.Variable, $"{nameof(this.Model)}.{nameof(model.Variable)}");
this.State.Reset(this.Model.Variable);
Debug.WriteLine(
@@ -2,7 +2,6 @@
using System;
using System.Collections.Frozen;
using System.Collections.Generic;
using System.Globalization;
using Microsoft.Agents.Workflows.Declarative.Extensions;
using Microsoft.Bot.ObjectModel;
@@ -31,22 +30,20 @@ internal static class SystemScope
public const string UserLanguage = nameof(UserLanguage);
}
public static FrozenSet<string> AllNames { get; } = GetNames().ToFrozenSet();
public static IEnumerable<string> GetNames()
{
yield return Names.Activity;
yield return Names.Bot;
yield return Names.Conversation;
yield return Names.ConversationId;
yield return Names.InternalId;
yield return Names.LastMessage;
yield return Names.LastMessageId;
yield return Names.LastMessageText;
yield return Names.Recognizer;
yield return Names.User;
yield return Names.UserLanguage;
}
public static FrozenSet<string> AllNames { get; } =
[
Names.Activity,
Names.Bot,
Names.Conversation,
Names.ConversationId,
Names.InternalId,
Names.LastMessage,
Names.LastMessageId,
Names.LastMessageText,
Names.Recognizer,
Names.User,
Names.UserLanguage,
];
public static void InitializeSystem(this WorkflowFormulaState scopes)
{
@@ -59,7 +56,7 @@ internal static class SystemScope
scopes.Set(
Names.Conversation,
RecordValue.NewRecordFromFields(
FormulaValue.NewRecordFromFields(
new NamedValue("Id", FormulaType.String.NewBlank()),
new NamedValue("LocalTimeZone", FormulaValue.New(TimeZoneInfo.Local.StandardName)),
new NamedValue("LocalTimeZoneOffset", FormulaValue.New(TimeZoneInfo.Local.GetUtcOffset(DateTime.UtcNow))),
@@ -70,17 +67,17 @@ internal static class SystemScope
scopes.Set(
Names.Recognizer,
RecordValue.NewRecordFromFields(
FormulaValue.NewRecordFromFields(
new NamedValue("Id", FormulaType.String.NewBlank()),
new NamedValue("Text", FormulaType.String.NewBlank())),
VariableScopeNames.System);
scopes.Set(
Names.User,
RecordValue.NewRecordFromFields(
new NamedValue("Language", StringValue.New(CultureInfo.CurrentCulture.TwoLetterISOLanguageName))),
FormulaValue.NewRecordFromFields(
new NamedValue("Language", FormulaValue.New(CultureInfo.CurrentCulture.TwoLetterISOLanguageName))),
VariableScopeNames.System);
scopes.Set(Names.UserLanguage, StringValue.New(CultureInfo.CurrentCulture.TwoLetterISOLanguageName), VariableScopeNames.System);
scopes.Set(Names.UserLanguage, FormulaValue.New(CultureInfo.CurrentCulture.TwoLetterISOLanguageName), VariableScopeNames.System);
void Set(string key, string? value = null)
{
@@ -53,19 +53,17 @@ internal static class WorkflowDiagnostics
{
foreach (VariableInformationDiagnostic variableDiagnostic in semanticModel.GetVariables(schemaName).Where(x => !x.IsSystemVariable).Select(v => v.ToDiagnostic()))
{
if (variableDiagnostic is null || variableDiagnostic?.Path?.VariableName is null)
if (variableDiagnostic?.Path?.VariableName is null)
{
continue;
}
FormulaValue defaultValue = variableDiagnostic.ConstantValue?.ToFormula() ?? variableDiagnostic.Type.NewBlank();
if (variableDiagnostic.Path.VariableScopeName?.Equals(VariableScopeNames.System, StringComparison.OrdinalIgnoreCase) ?? false)
if (variableDiagnostic.Path.VariableScopeName?.Equals(VariableScopeNames.System, StringComparison.OrdinalIgnoreCase) is true &&
!SystemScope.AllNames.Contains(variableDiagnostic.Path.VariableName))
{
if (!SystemScope.AllNames.Contains(variableDiagnostic.Path.VariableName))
{
throw new DeclarativeModelException($"Variable '{variableDiagnostic.Path.VariableName}' is not a supported system variable.");
}
throw new DeclarativeModelException($"Variable '{variableDiagnostic.Path.VariableName}' is not a supported system variable.");
}
scopes.Set(variableDiagnostic.Path.VariableName, defaultValue, variableDiagnostic.Path.VariableScopeName ?? WorkflowFormulaState.DefaultScopeName);
@@ -13,7 +13,7 @@ using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows.Declarative.PowerFx;
internal class WorkflowExpressionEngine
internal sealed class WorkflowExpressionEngine
{
private readonly RecalcEngine _engine;
@@ -39,7 +39,7 @@ internal class WorkflowExpressionEngine
public ImmutableArray<T> GetValue<T>(ArrayExpressionOnly<T> expression) => this.Evaluate(expression).Value;
public EvaluationResult<TValue> GetValue<TValue>(EnumExpression<TValue> expression) where TValue : EnumWrapper =>
this.Evaluate<TValue>(expression);
this.Evaluate(expression);
private EvaluationResult<bool> Evaluate(BoolExpression expression)
{
@@ -186,7 +186,7 @@ internal class WorkflowExpressionEngine
{
Throw.IfNull(expression, nameof(expression));
if (expression.LiteralValue != null)
if (expression.LiteralValue is not null)
{
return new EvaluationResult<TValue?>(expression.LiteralValue, SensitivityLevel.None);
}
@@ -240,7 +240,7 @@ internal class WorkflowExpressionEngine
{
if (value is BlankValue)
{
return ImmutableArray<TValue>.Empty;
return [];
}
if (value is not TableValue tableValue)
@@ -254,8 +254,7 @@ internal class WorkflowExpressionEngine
List<TValue> list = [];
foreach (RecordDataValue row in tableDataValue.Values)
{
TValue? s = TableItemParser<TValue>.Parse(row);
if (s != null)
if (TableItemParser<TValue>.Parse(row) is TValue s)
{
list.Add(s);
}
@@ -105,10 +105,11 @@ internal sealed class WorkflowFormulaState
foreach (string key in keys)
{
object? value = await context.ReadStateAsync<object>(key, scopeName).ConfigureAwait(false);
if (value is null || value is UnassignedValue)
if (value is null or UnassignedValue)
{
value = FormulaValue.NewBlank();
}
this.Set(key, value.ToFormula(), scopeName);
}
@@ -142,7 +143,7 @@ internal sealed class WorkflowFormulaState
private WorkflowScope GetScope(string? scopeName)
{
scopeName ??= WorkflowFormulaState.DefaultScopeName;
scopeName ??= DefaultScopeName;
if (!VariableScopeNames.IsValidName(scopeName))
{
@@ -1,6 +1,5 @@
// Copyright (c) Microsoft. All rights reserved.
using System.Collections.Generic;
using System.Diagnostics;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.AI.Agents;
@@ -9,34 +8,29 @@ namespace Microsoft.Agents.Workflows;
internal static class AIAgentsAbstractionsExtensions
{
public static ChatMessage ToChatMessage(this AgentRunResponseUpdate update)
{
return new ChatMessage
public static ChatMessage ToChatMessage(this AgentRunResponseUpdate update) =>
new()
{
AuthorName = update.AuthorName,
Contents = update.Contents,
Role = update.Role ?? ChatRole.User,
CreatedAt = update.CreatedAt,
MessageId = update.MessageId,
RawRepresentation = update.RawRepresentation,
RawRepresentation = update.RawRepresentation ?? update,
};
}
public static ChatMessage UpdateWith(this ChatMessage baseMessage, AgentRunResponseUpdate update)
{
Debug.Assert(update.MessageId == null || baseMessage.MessageId == update.MessageId);
Debug.Assert(update.MessageId is null || baseMessage.MessageId == update.MessageId);
List<AIContent> mergedContent = new(baseMessage.Contents);
mergedContent.AddRange(update.Contents);
return new ChatMessage
return new()
{
AuthorName = update.AuthorName ?? baseMessage.AuthorName,
Contents = mergedContent,
Contents = [.. baseMessage.Contents, .. update.Contents],
Role = update.Role ?? baseMessage.Role,
CreatedAt = update.CreatedAt ?? baseMessage.CreatedAt,
MessageId = baseMessage.MessageId,
RawRepresentation = update.RawRepresentation ?? baseMessage.RawRepresentation,
RawRepresentation = update.RawRepresentation ?? baseMessage.RawRepresentation ?? update,
};
}
}
@@ -1,11 +1,12 @@
// Copyright (c) Microsoft. All rights reserved.
using Microsoft.Extensions.AI.Agents;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows;
/// <summary>
/// Event triggered when an agent run produces an update.
/// Represents an event triggered when an agent run produces an update.
/// </summary>
public class AgentRunResponseEvent : ExecutorEvent
{
@@ -13,14 +14,14 @@ public class AgentRunResponseEvent : ExecutorEvent
/// 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="response"></param>
/// <param name="response">The agent run response.</param>
public AgentRunResponseEvent(string executorId, AgentRunResponse response) : base(executorId, data: response)
{
this.Response = response;
this.Response = Throw.IfNull(response);
}
/// <summary>
/// Gets the content of the agent response.
/// Gets the agent run response.
/// </summary>
public AgentRunResponse Response { get; }
}
@@ -2,11 +2,12 @@
using System.Collections.Generic;
using Microsoft.Extensions.AI.Agents;
using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows;
/// <summary>
/// Event triggered when an agent run produces an update.
/// Represents an event triggered when an agent run produces an update.
/// </summary>
public class AgentRunUpdateEvent : ExecutorEvent
{
@@ -14,14 +15,14 @@ public class AgentRunUpdateEvent : ExecutorEvent
/// 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>
/// <param name="update">The agent run response update.</param>
public AgentRunUpdateEvent(string executorId, AgentRunResponseUpdate update) : base(executorId, data: update)
{
this.Update = update;
this.Update = Throw.IfNull(update);
}
/// <summary>
/// Gets the content of the agent response.
/// Gets the agent run response update.
/// </summary>
public AgentRunResponseUpdate Update { get; }
@@ -9,7 +9,7 @@ namespace Microsoft.Agents.Workflows;
/// <summary>
/// Represents a checkpoint with a unique identifier and a timestamp indicating when it was created.
/// </summary>
public class CheckpointInfo : IEquatable<CheckpointInfo>
public sealed class CheckpointInfo : IEquatable<CheckpointInfo>
{
/// <summary>
/// Gets the unique identifier for the current run.
@@ -37,27 +37,16 @@ public class CheckpointInfo : IEquatable<CheckpointInfo>
}
/// <inheritdoc/>
public bool Equals(CheckpointInfo? other)
{
if (other == null)
{
return false;
}
return this.RunId == other.RunId && this.CheckpointId == other.CheckpointId;
}
public bool Equals(CheckpointInfo? other) =>
other is not null &&
this.RunId == other.RunId &&
this.CheckpointId == other.CheckpointId;
/// <inheritdoc/>
public override bool Equals(object? obj)
{
return this.Equals(obj as CheckpointInfo);
}
public override bool Equals(object? obj) => this.Equals(obj as CheckpointInfo);
/// <inheritdoc/>
public override int GetHashCode()
{
return HashCode.Combine(this.RunId, this.CheckpointId);
}
public override int GetHashCode() => HashCode.Combine(this.RunId, this.CheckpointId);
/// <inheritdoc/>
public override string ToString() => $"CheckpointInfo(RunId: {this.RunId}, CheckpointId: {this.CheckpointId})";
@@ -11,21 +11,21 @@ namespace Microsoft.Agents.Workflows;
/// <summary>
/// Represents a workflow run that supports checkpointing.
/// </summary>
/// <typeparam name="TRun">The type of the underlying workflow run handle</typeparam>
/// <typeparam name="TRun">The type of the underlying workflow run handle.</typeparam>
/// <seealso cref="Run"/>
/// <seealso cref="Run{TResult}"/>
/// <seealso cref="StreamingRun"/>
/// <seealso cref="StreamingRun{TResult}"/>
public class Checkpointed<TRun>
{
private readonly ICheckpointingRunner _runner;
internal Checkpointed(TRun run, ICheckpointingRunner runner)
{
this.Run = Throw.IfNull(run);
this._runner = Throw.IfNull(runner);
}
private readonly ICheckpointingRunner _runner;
/// <summary>
/// Gets the workflow run associated with this <see cref="Checkpointed{TRun}"/> instance.
/// </summary>
@@ -41,7 +41,14 @@ public class Checkpointed<TRun>
/// <summary>
/// Gets the most recent checkpoint information.
/// </summary>
public CheckpointInfo? LastCheckpoint => this.Checkpoints.Count > 0 ? this.Checkpoints[this.Checkpoints.Count - 1] : null;
public CheckpointInfo? LastCheckpoint
{
get
{
var checkpoints = this.Checkpoints;
return checkpoints.Count > 0 ? checkpoints[checkpoints.Count - 1] : null;
}
}
/// <inheritdoc cref="ICheckpointingRunner.RestoreCheckpointAsync"/>
public ValueTask RestoreCheckpointAsync(CheckpointInfo checkpointInfo, CancellationToken cancellation = default)
@@ -7,7 +7,7 @@ using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows.Checkpointing;
internal class Checkpoint
internal sealed class Checkpoint
{
[JsonConstructor]
internal Checkpoint(
@@ -33,8 +33,8 @@ internal class Checkpoint
public WorkflowInfo Workflow { get; }
public RunnerStateData RunnerData { get; }
public Dictionary<ScopeKey, PortableValue> StateData { get; } = new();
public Dictionary<EdgeId, PortableValue> EdgeStateData { get; } = new();
public Dictionary<ScopeKey, PortableValue> StateData { get; } = [];
public Dictionary<EdgeId, PortableValue> EdgeStateData { get; } = [];
public CheckpointInfo? Parent { get; }
}
@@ -10,7 +10,7 @@ namespace Microsoft.Agents.Workflows.Checkpointing;
/// </summary>
public sealed class DirectEdgeInfo : EdgeInfo
{
internal DirectEdgeInfo(DirectEdgeData data) : this(data.Condition != null, data.Connection) { }
internal DirectEdgeInfo(DirectEdgeData data) : this(data.Condition is not null, data.Connection) { }
[JsonConstructor]
internal DirectEdgeInfo(bool hasCondition, EdgeConnection connection) : base(EdgeKind.Direct, connection)
@@ -26,6 +26,6 @@ public sealed class DirectEdgeInfo : EdgeInfo
internal override bool IsMatchInternal(EdgeData edgeData)
{
return edgeData is DirectEdgeData directEdge
&& this.HasCondition == (directEdge.Condition != null);
&& this.HasCondition == (directEdge.Condition is not null);
}
}
@@ -2,23 +2,17 @@
namespace Microsoft.Agents.Workflows.Checkpointing;
internal record class ExecutorInfo(TypeId ExecutorType, string ExecutorId)
internal sealed record class ExecutorInfo(TypeId ExecutorType, string ExecutorId)
{
public bool IsMatch<T>() where T : Executor
{
return this.ExecutorType.IsMatch<T>()
public bool IsMatch<T>() where T : Executor =>
this.ExecutorType.IsMatch<T>()
&& this.ExecutorId == typeof(T).Name;
}
public bool IsMatch(Executor executor)
{
return this.ExecutorType.IsMatch(executor.GetType())
public bool IsMatch(Executor executor) =>
this.ExecutorType.IsMatch(executor.GetType())
&& this.ExecutorId == executor.Id;
}
public bool IsMatch(ExecutorRegistration registration)
{
return this.ExecutorType.IsMatch(registration.ExecutorType)
public bool IsMatch(ExecutorRegistration registration) =>
this.ExecutorType.IsMatch(registration.ExecutorType)
&& this.ExecutorId == registration.Id;
}
}
@@ -10,7 +10,7 @@ namespace Microsoft.Agents.Workflows.Checkpointing;
/// </summary>
public sealed class FanOutEdgeInfo : EdgeInfo
{
internal FanOutEdgeInfo(FanOutEdgeData data) : this(data.EdgeAssigner != null, data.Connection) { }
internal FanOutEdgeInfo(FanOutEdgeData data) : this(data.EdgeAssigner is not null, data.Connection) { }
[JsonConstructor]
internal FanOutEdgeInfo(bool hasAssigner, EdgeConnection connection) : base(EdgeKind.FanOut, connection)
@@ -26,6 +26,6 @@ public sealed class FanOutEdgeInfo : EdgeInfo
internal override bool IsMatchInternal(EdgeData edgeData)
{
return edgeData is FanOutEdgeData fanOutEdge
&& this.HasAssigner == (fanOutEdge.EdgeAssigner != null);
&& this.HasAssigner == (fanOutEdge.EdgeAssigner is not null);
}
}
@@ -55,12 +55,11 @@ public sealed class FileSystemJsonCheckpointStore : JsonCheckpointStore, IDispos
try
{
// read the lines of indexfile and parse them as CheckpointInfos
this.CheckpointIndex = new HashSet<CheckpointInfo>();
this.CheckpointIndex = [];
using StreamReader reader = new(this._indexFile, encoding: Encoding.UTF8, detectEncodingFromByteOrderMarks: false, bufferSize: -1, leaveOpen: true);
while (reader.ReadLine() is string line)
{
CheckpointInfo? info = JsonSerializer.Deserialize<CheckpointInfo>(line, this.KeyTypeInfo);
if (info != null)
if (JsonSerializer.Deserialize(line, this.KeyTypeInfo) is { } info)
{
this.CheckpointIndex.Add(info);
}
@@ -83,7 +82,7 @@ public sealed class FileSystemJsonCheckpointStore : JsonCheckpointStore, IDispos
Justification = "Throw helper does not exist in NetFx 4.7.2")]
private void CheckDisposed()
{
if (this._indexFile == null)
if (this._indexFile is null)
{
throw new ObjectDisposedException($"{nameof(FileSystemJsonCheckpointStore)}({this.Directory.FullName})");
}
@@ -10,7 +10,7 @@ namespace Microsoft.Agents.Workflows.Checkpointing;
/// </summary>
internal sealed class InMemoryCheckpointManager : ICheckpointManager
{
private readonly Dictionary<string, RunCheckpointCache<Checkpoint>> _store = new();
private readonly Dictionary<string, RunCheckpointCache<Checkpoint>> _store = [];
private RunCheckpointCache<Checkpoint> GetRunStore(string runId)
{
@@ -18,18 +18,11 @@ internal abstract class JsonConverterBase<T> : JsonConverter<T>
public override T? Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
{
SequencePosition position = reader.Position;
T? maybeValue = JsonSerializer.Deserialize<T>(ref reader, this.TypeInfo);
if (maybeValue is null)
{
return
JsonSerializer.Deserialize(ref reader, this.TypeInfo) ??
throw new JsonException($"Could not deserialize a {typeof(T).Name} from JSON at position {position}");
}
return maybeValue;
}
public override void Write(Utf8JsonWriter writer, T value, JsonSerializerOptions options)
{
public override void Write(Utf8JsonWriter writer, T value, JsonSerializerOptions options) =>
JsonSerializer.Serialize(writer, value, this.TypeInfo);
}
}
@@ -23,7 +23,7 @@ internal abstract class JsonConverterDictionarySupportBase<T> : JsonConverterBas
SequencePosition position = reader.Position;
string? propertyName = reader.GetString();
if (propertyName == null)
if (propertyName is null)
{
throw new JsonException($"Got null trying to read property name at position {position}");
}
@@ -6,7 +6,7 @@ using System.Text.Json.Serialization.Metadata;
namespace Microsoft.Agents.Workflows.Checkpointing;
internal class JsonMarshaller : IWireMarshaller<JsonElement>
internal sealed class JsonMarshaller : IWireMarshaller<JsonElement>
{
private readonly JsonSerializerOptions _internalOptions;
private readonly JsonSerializerOptions? _externalOptions;
@@ -26,7 +26,7 @@ internal class JsonMarshaller : IWireMarshaller<JsonElement>
{
if (!this._internalOptions.TryGetTypeInfo(type, out JsonTypeInfo? typeInfo))
{
if (this._externalOptions == null ||
if (this._externalOptions is null ||
!this._externalOptions.TryGetTypeInfo(type, out typeInfo))
{
throw new InvalidOperationException($"No JSON type info is available for type '{type}'.");
@@ -44,13 +44,8 @@ internal class JsonMarshaller : IWireMarshaller<JsonElement>
public TValue Marshal<TValue>(JsonElement data)
{
Type type = typeof(TValue);
object? value = JsonSerializer.Deserialize(data, this.LookupTypeInfo(type));
if (value is null)
{
object value = data.Deserialize(this.LookupTypeInfo(typeof(TValue))) ??
throw new InvalidOperationException($"Could not deserialize the value as the expected type {typeof(TValue)}.");
}
if (value is TValue typedValue)
{
@@ -62,12 +57,8 @@ internal class JsonMarshaller : IWireMarshaller<JsonElement>
public object Marshal(Type targetType, JsonElement data)
{
object? value = JsonSerializer.Deserialize(data, this.LookupTypeInfo(targetType));
if (value is null)
{
object value = data.Deserialize(this.LookupTypeInfo(targetType)) ??
throw new InvalidOperationException($"Could not deserialize the value as the expected type {targetType}.");
}
if (targetType.IsInstanceOfType(value))
{
@@ -23,7 +23,7 @@ internal sealed class JsonWireSerializedValue(JsonMarshaller serializer, JsonEle
public override bool Equals(object? obj)
{
if (obj == null)
if (obj is null)
{
return false;
}
@@ -24,7 +24,7 @@ internal sealed class PortableValueConverter(JsonMarshaller marshaller) : JsonCo
SequencePosition initial = reader.Position;
JsonTypeInfo<PortableValue> baseTypeInfo = WorkflowsJsonUtilities.JsonContext.Default.PortableValue;
PortableValue? maybeValue = JsonSerializer.Deserialize<PortableValue>(ref reader, baseTypeInfo);
PortableValue? maybeValue = JsonSerializer.Deserialize(ref reader, baseTypeInfo);
if (maybeValue is null)
{
@@ -7,8 +7,8 @@ namespace Microsoft.Agents.Workflows.Checkpointing;
internal sealed class RunCheckpointCache<TStoreObject>
{
private readonly HashSet<CheckpointInfo> _checkpointIndex = new();
private readonly Dictionary<CheckpointInfo, TStoreObject> _cache = new();
private readonly HashSet<CheckpointInfo> _checkpointIndex = [];
private readonly Dictionary<CheckpointInfo, TStoreObject> _cache = [];
public IEnumerable<CheckpointInfo> Index => this._checkpointIndex;
@@ -39,12 +39,12 @@ internal sealed class ScopeKeyConverter : JsonConverterDictionarySupportBase<Sco
[return: NotNull]
private static string Escape(string? value, bool allowNullAndPad = false, [CallerArgumentExpression("value")] string componentName = "ScopeKey")
{
if (!allowNullAndPad && value == null)
if (!allowNullAndPad && value is null)
{
throw new JsonException($"Invalid {componentName} '{value}'. Expecting non-null string.");
}
if (value == null)
if (value is null)
{
return string.Empty;
}
@@ -9,7 +9,7 @@ namespace Microsoft.Agents.Workflows.Checkpointing;
/// <summary>
/// A representation of a type's identity, including its assembly and type names.
/// </summary>
public class TypeId
public sealed class TypeId : IEquatable<TypeId>
{
/// <inheritdoc cref="System.Reflection.Assembly.FullName"/>
public string AssemblyName { get; }
@@ -43,15 +43,29 @@ public class TypeId
/// <inheritdoc />
public override bool Equals(object? obj)
=> obj is TypeId other
&& this.AssemblyName == other.AssemblyName
&& this.TypeName == other.TypeName;
=> this.Equals(obj as TypeId);
/// <inheritdoc />
public bool Equals(TypeId? other)
{
if (other is null)
{
return false;
}
if (ReferenceEquals(this, other))
{
return true;
}
return this.AssemblyName == other.AssemblyName && this.TypeName == other.TypeName;
}
/// <inheritdoc />
public override int GetHashCode() => HashCode.Combine(this.AssemblyName, this.TypeName);
/// <inheritdoc />
public static bool operator ==(TypeId? left, TypeId? right) => object.ReferenceEquals(left, right) || (!object.ReferenceEquals(left, null) && left.Equals(right));
public static bool operator ==(TypeId? left, TypeId? right) => left is null ? right is null : left.Equals(right);
/// <inheritdoc />
public static bool operator !=(TypeId? left, TypeId? right) => !(left == right);
@@ -84,7 +98,7 @@ public class TypeId
{
Type? candidateType = type;
while (candidateType != null)
while (candidateType is not null)
{
if (this.IsMatch(candidateType))
{
@@ -8,7 +8,7 @@ using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows.Checkpointing;
internal class WorkflowInfo
internal sealed class WorkflowInfo
{
[JsonConstructor]
internal WorkflowInfo(
@@ -27,12 +27,12 @@ internal class WorkflowInfo
this.InputType = Throw.IfNull(inputType);
this.StartExecutorId = Throw.IfNullOrEmpty(startExecutorId);
if (outputType != null && outputCollectorId != null)
if (outputType is not null && outputCollectorId is not null)
{
this.OutputType = outputType;
this.OutputCollectorId = outputCollectorId;
}
else if (outputCollectorId != null)
else if (outputCollectorId is not null)
{
throw new InvalidOperationException(
$"Either both or none of OutputType and OutputCollectorId must be set. ({nameof(outputType)}: {outputType} vs. {nameof(outputCollectorId)}: {outputCollectorId})"
@@ -108,6 +108,6 @@ internal class WorkflowInfo
public bool IsMatch<TInput, TResult>(Workflow<TInput, TResult> workflow)
=> this.IsMatch(workflow as Workflow)
&& this.OutputType != null && this.OutputType.IsMatch(typeof(TResult))
&& this.OutputCollectorId != null && this.OutputCollectorId == workflow.OutputCollectorId;
&& this.OutputType?.IsMatch(typeof(TResult)) is true
&& this.OutputCollectorId is not null && this.OutputCollectorId == workflow.OutputCollectorId;
}
@@ -3,13 +3,13 @@
namespace Microsoft.Agents.Workflows;
/// <summary>
/// Configuration for an object with a string identifier. For example, <see cref="IIdentified"/> object.
/// Represents a configuration for an object with a string identifier. For example, <see cref="IIdentified"/> object.
/// </summary>
/// <param name="id">A unique identifier for the configurable object.</param>
public class Config(string? id = null)
{
/// <summary>
/// A unique identifier for the configurable object.
/// Gets a unique identifier for the configurable object.
/// </summary>
/// <remarks>
/// If not provided, the configured object will generate its own identifier.
@@ -18,7 +18,7 @@ public class Config(string? id = null)
}
/// <summary>
/// Configuration for an object with a string identifier and options of type <typeparamref name="TOptions"/>.
/// Represents a configuration for an object with a string identifier and options of type <typeparamref name="TOptions"/>.
/// </summary>
/// <typeparam name="TOptions">The type of options for the configurable object.</typeparam>
/// <param name="options">The options for the configurable object.</param>
@@ -26,7 +26,7 @@ public class Config(string? id = null)
public class Config<TOptions>(TOptions? options = default, string? id = null) : Config(id)
{
/// <summary>
/// Options for the configured object.
/// Gets the options for the configured object.
/// </summary>
public TOptions? Options => options;
}
@@ -3,7 +3,7 @@
namespace Microsoft.Agents.Workflows;
/// <summary>
/// Extensions methods for creating Configured objects
/// Provides extensions methods for creating <see cref="Configured{TSubject}"/> objects
/// </summary>
public static class ConfigurationExtensions
{
@@ -6,7 +6,7 @@ using System.Threading.Tasks;
namespace Microsoft.Agents.Workflows;
/// <summary>
/// Helper methods for creating <see cref="Configured{TSubject}"/> instances.
/// Provides methods for creating <see cref="Configured{TSubject}"/> instances.
/// </summary>
public static class Configured
{
@@ -29,20 +29,20 @@ public static class Configured
{
if (subject is IIdentified identified)
{
if (id != null && identified.Id != id)
if (id is not null && identified.Id != id)
{
throw new ArgumentException($"Provided ID '{id}' does not match subject's ID '{identified.Id}'.", nameof(id));
}
return new Configured<TSubject>((_) => new(subject), id: identified.Id, raw: raw ?? subject);
return new Configured<TSubject>(_ => new(subject), id: identified.Id, raw: raw ?? subject);
}
if (id == null)
if (id is null)
{
throw new ArgumentNullException(nameof(id), "ID must be provided when the subject does not implement IIdentified.");
}
return new Configured<TSubject>((_) => new(subject), id, raw: raw ?? subject);
return new Configured<TSubject>(_ => new(subject), id, raw: raw ?? subject);
}
}
@@ -56,7 +56,7 @@ public static class Configured
public class Configured<TSubject>(Func<Config, ValueTask<TSubject>> factoryAsync, string id, object? raw = null)
{
/// <summary>
/// The raw representation of the configured object, if any.
/// Gets the raw representation of the configured object, if any.
/// </summary>
public object? Raw => raw;
@@ -137,7 +137,7 @@ public class Configured<TSubject, TOptions>(Func<Config<TOptions>, ValueTask<TSu
TSubject subject = await this.FactoryAsync(this.Configuration).ConfigureAwait(false);
if (this.Id != null && subject is IIdentified identified && identified.Id != this.Id)
if (this.Id is not null && subject is IIdentified identified && identified.Id != this.Id)
{
throw new InvalidOperationException($"Created instance ID '{identified.Id}' does not match configured ID '{this.Id}'.");
}
@@ -21,7 +21,7 @@ public readonly struct EdgeId : IEquatable<EdgeId>
/// <inheritdoc />
public override bool Equals(object? obj)
{
if (obj == null)
if (obj is null)
{
return false;
}
@@ -19,18 +19,18 @@ internal sealed class CallResult
/// If the call was successful, this property contains the result of the call. For calls to
/// void handlers, this will be <c>null</c>.
/// </summary>
public object? Result { get; init; } = null;
public object? Result { get; init; }
/// <summary>
/// If the call failed, this property contains the exception that was raised during the call.
/// </summary>
public Exception? Exception { get; init; } = null;
public Exception? Exception { get; init; }
/// <summary>
/// Indicates whether the call was successful. A call is considered successful if it returned
/// without throwing an exception.
/// </summary>
public bool IsSuccess => this.Exception == null;
public bool IsSuccess => this.Exception is null;
private CallResult(bool isVoid = false)
{
@@ -43,19 +43,13 @@ internal sealed class CallResult
/// </summary>
/// <param name="result">The result to return.</param>
/// <returns>A <see cref="CallResult"/> indicating the result of the call.</returns>
public static CallResult ReturnResult(object? result = null)
{
return new() { Result = result };
}
public static CallResult ReturnResult(object? result = null) => new() { Result = result };
/// <summary>
/// Create a <see cref="CallResult"/> indicating a successful call that returned no result (void).
/// </summary>
/// <returns>A <see cref="CallResult"/> indicating the result of the call.</returns>
public static CallResult ReturnVoid()
{
return new(isVoid: true);
}
public static CallResult ReturnVoid() => new(isVoid: true);
/// <summary>
/// Create a <see cref="CallResult"/> indicating that an exception was raised during the call.
@@ -5,26 +5,23 @@ using System.Threading.Tasks;
namespace Microsoft.Agents.Workflows.Execution;
internal class DirectEdgeRunner(IRunnerContext runContext, DirectEdgeData edgeData) :
internal sealed class DirectEdgeRunner(IRunnerContext runContext, DirectEdgeData edgeData) :
EdgeRunner<DirectEdgeData>(runContext, edgeData)
{
public IWorkflowContext WorkflowContext { get; } = runContext.Bind(edgeData.SinkId);
private async ValueTask<Executor> FindRouterAsync(IStepTracer? tracer)
{
return await this.RunContext.EnsureExecutorAsync(this.EdgeData.SinkId, tracer)
private async ValueTask<Executor> FindRouterAsync(IStepTracer? tracer) => await this.RunContext.EnsureExecutorAsync(this.EdgeData.SinkId, tracer)
.ConfigureAwait(false);
}
public async ValueTask<IEnumerable<object?>> ChaseAsync(MessageEnvelope envelope, IStepTracer? tracer)
{
if (envelope.TargetId != null && this.EdgeData.SinkId != envelope.TargetId)
if (envelope.TargetId is not null && this.EdgeData.SinkId != envelope.TargetId)
{
return [];
}
object message = envelope.Message;
if (this.EdgeData.Condition != null && !this.EdgeData.Condition(message))
if (this.EdgeData.Condition is not null && !this.EdgeData.Condition(message))
{
return [];
}
@@ -15,7 +15,7 @@ namespace Microsoft.Agents.Workflows.Execution;
/// Ordering is relevant because in at least one case, the order of sinks is significant for the execution of
/// the edge: <see cref="FanOutEdgeData"/>.
/// </remarks>
public class EdgeConnection : IEquatable<EdgeConnection>
public sealed class EdgeConnection : IEquatable<EdgeConnection>
{
/// <summary>
/// Create an <see cref="EdgeConnection"/> instance with the specified source and sink IDs.
@@ -65,7 +65,7 @@ public class EdgeConnection : IEquatable<EdgeConnection>
return false;
}
if (object.ReferenceEquals(this, other))
if (ReferenceEquals(this, other))
{
return true;
}
@@ -8,10 +8,10 @@ using Microsoft.Agents.Workflows.Checkpointing;
namespace Microsoft.Agents.Workflows.Execution;
internal class EdgeMap
internal sealed class EdgeMap
{
private readonly Dictionary<EdgeId, object> _edgeRunners = new();
private readonly Dictionary<EdgeId, FanInEdgeState> _fanInState = new();
private readonly Dictionary<EdgeId, object> _edgeRunners = [];
private readonly Dictionary<EdgeId, FanInEdgeState> _fanInState = [];
private readonly Dictionary<string, InputEdgeRunner> _portEdgeRunners;
private readonly InputEdgeRunner _inputRunner;
private readonly IStepTracer? _stepTracer;
@@ -68,14 +68,14 @@ internal class EdgeMap
// between the Runners, we can normalize it behind an IFace.
case EdgeKind.Direct:
{
DirectEdgeRunner runner = (DirectEdgeRunner)this._edgeRunners[id];
DirectEdgeRunner runner = (DirectEdgeRunner)edgeRunner;
edgeResults = await runner.ChaseAsync(message, this._stepTracer).ConfigureAwait(false);
break;
}
case EdgeKind.FanOut:
{
FanOutEdgeRunner runner = (FanOutEdgeRunner)this._edgeRunners[id];
FanOutEdgeRunner runner = (FanOutEdgeRunner)edgeRunner;
edgeResults = await runner.ChaseAsync(message, this._stepTracer).ConfigureAwait(false);
break;
}
@@ -83,7 +83,7 @@ internal class EdgeMap
case EdgeKind.FanIn:
{
FanInEdgeState state = this._fanInState[id];
FanInEdgeRunner runner = (FanInEdgeRunner)this._edgeRunners[id];
FanInEdgeRunner runner = (FanInEdgeRunner)edgeRunner;
edgeResults = [await runner.ChaseAsync(sourceId, message, state, this._stepTracer).ConfigureAwait(false)];
break;
}
@@ -114,7 +114,7 @@ internal class EdgeMap
internal ValueTask<Dictionary<EdgeId, PortableValue>> ExportStateAsync()
{
Dictionary<EdgeId, PortableValue> exportedStates = new();
Dictionary<EdgeId, PortableValue> exportedStates = [];
// Right now there is only fan-in state
foreach (EdgeId id in this._fanInState.Keys)
@@ -7,25 +7,23 @@ namespace Microsoft.Agents.Workflows.Execution;
internal readonly struct ExecutorIdentity : IEquatable<ExecutorIdentity>
{
public static ExecutorIdentity None { get; } = new ExecutorIdentity();
public static ExecutorIdentity None { get; }
public string? Id { get; init; }
public bool Equals(ExecutorIdentity other)
{
return this.Id == null
? other.Id == null
: other.Id != null && StringComparer.OrdinalIgnoreCase.Equals(this.Id, other.Id);
}
public bool Equals(ExecutorIdentity other) =>
this.Id is null
? other.Id is null
: other.Id is not null && StringComparer.OrdinalIgnoreCase.Equals(this.Id, other.Id);
public override bool Equals([NotNullWhen(true)] object? obj)
{
if (this.Id == null)
if (this.Id is null)
{
return obj == null;
return obj is null;
}
if (obj == null)
if (obj is null)
{
return false;
}
@@ -43,18 +41,9 @@ internal readonly struct ExecutorIdentity : IEquatable<ExecutorIdentity>
return false;
}
public override int GetHashCode()
{
return this.Id == null ? 0 : StringComparer.OrdinalIgnoreCase.GetHashCode(this.Id);
}
public override int GetHashCode() => this.Id is null ? 0 : StringComparer.OrdinalIgnoreCase.GetHashCode(this.Id);
public static implicit operator ExecutorIdentity(string? id)
{
return new ExecutorIdentity { Id = id };
}
public static implicit operator ExecutorIdentity(string? id) => new() { Id = id };
public static implicit operator string?(ExecutorIdentity identity)
{
return identity.Id;
}
public static implicit operator string?(ExecutorIdentity identity) => identity.Id;
}
@@ -6,7 +6,7 @@ using System.Threading.Tasks;
namespace Microsoft.Agents.Workflows.Execution;
internal class FanInEdgeRunner(IRunnerContext runContext, FanInEdgeData edgeData) :
internal sealed class FanInEdgeRunner(IRunnerContext runContext, FanInEdgeData edgeData) :
EdgeRunner<FanInEdgeData>(runContext, edgeData)
{
private IWorkflowContext BoundContext { get; } = runContext.Bind(edgeData.SinkId);
@@ -15,13 +15,12 @@ internal class FanInEdgeRunner(IRunnerContext runContext, FanInEdgeData edgeData
public ValueTask<IEnumerable<object?>> ChaseAsync(string sourceId, MessageEnvelope envelope, FanInEdgeState state, IStepTracer? tracer)
{
if (envelope.TargetId != null && this.EdgeData.SinkId != envelope.TargetId)
if (envelope.TargetId is not null && this.EdgeData.SinkId != envelope.TargetId)
{
// This message is not for us.
return new([]);
}
object message = envelope.Message;
IEnumerable<MessageEnvelope>? releasedMessages = state.ProcessMessage(sourceId, envelope);
if (releasedMessages is null)
{
@@ -8,7 +8,7 @@ using Microsoft.Agents.Workflows.Checkpointing;
namespace Microsoft.Agents.Workflows.Execution;
internal class FanInEdgeState
internal sealed class FanInEdgeState
{
private List<PortableMessageEnvelope> _pendingMessages;
public FanInEdgeState(FanInEdgeData fanInEdge)
@@ -6,26 +6,27 @@ using System.Threading.Tasks;
namespace Microsoft.Agents.Workflows.Execution;
internal class FanOutEdgeRunner(IRunnerContext runContext, FanOutEdgeData edgeData) :
internal sealed class FanOutEdgeRunner(IRunnerContext runContext, FanOutEdgeData edgeData) :
EdgeRunner<FanOutEdgeData>(runContext, edgeData)
{
private Dictionary<string, IWorkflowContext> BoundContexts { get; }
= edgeData.SinkIds.ToDictionary(
sinkId => sinkId,
sinkId => runContext.Bind(sinkId));
runContext.Bind);
public async ValueTask<IEnumerable<object?>> ChaseAsync(MessageEnvelope envelope, IStepTracer? tracer)
{
object message = envelope.Message;
List<string> targets =
this.EdgeData.EdgeAssigner == null
this.EdgeData.EdgeAssigner is null
? this.EdgeData.SinkIds
: this.EdgeData.EdgeAssigner(message, this.BoundContexts.Count)
.Select(i => this.EdgeData.SinkIds[i]).ToList();
IEnumerable<string> filteredTargets = envelope.TargetId != null
? targets.Where(IsValidTarget)
: targets;
IEnumerable<string> filteredTargets =
envelope.TargetId is not null
? targets.Where(IsValidTarget)
: targets;
object?[] result = await Task.WhenAll(filteredTargets.Select(ProcessTargetAsync)).ConfigureAwait(false);
return result.Where(r => r is not null);
@@ -47,7 +48,7 @@ internal class FanOutEdgeRunner(IRunnerContext runContext, FanOutEdgeData edgeDa
bool IsValidTarget(string targetId)
{
return envelope.TargetId == null || targetId == envelope.TargetId;
return envelope.TargetId is null || targetId == envelope.TargetId;
}
}
}
@@ -6,7 +6,7 @@ using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows.Execution;
internal class InputEdgeRunner(IRunnerContext runContext, string sinkId)
internal sealed class InputEdgeRunner(IRunnerContext runContext, string sinkId)
: EdgeRunner<string>(runContext, sinkId)
{
public IWorkflowContext WorkflowContext { get; } = runContext.Bind(sinkId);
@@ -19,10 +19,7 @@ internal class InputEdgeRunner(IRunnerContext runContext, string sinkId)
return new InputEdgeRunner(runContext, port.Id);
}
private async ValueTask<Executor> FindExecutorAsync(IStepTracer? tracer)
{
return await this.RunContext.EnsureExecutorAsync(this.EdgeData, tracer).ConfigureAwait(false);
}
private async ValueTask<Executor> FindExecutorAsync(IStepTracer? tracer) => await this.RunContext.EnsureExecutorAsync(this.EdgeData, tracer).ConfigureAwait(false);
public async ValueTask<object?> ChaseAsync(MessageEnvelope envelope, IStepTracer? tracer)
{
@@ -16,7 +16,7 @@ using MessageHandlerF =
namespace Microsoft.Agents.Workflows.Execution;
internal class MessageRouter
internal sealed class MessageRouter
{
private readonly Dictionary<Type, MessageHandlerF> _typedHandlers;
private readonly Dictionary<TypeId, Type> _runtimeTypeMap;
@@ -5,7 +5,7 @@ using Microsoft.Agents.Workflows.Checkpointing;
namespace Microsoft.Agents.Workflows.Execution;
internal class RunnerStateData(HashSet<string> instantiatedExecutors, Dictionary<ExecutorIdentity, List<PortableMessageEnvelope>> queuedMessages, List<ExternalRequest> outstandingRequests)
internal sealed class RunnerStateData(HashSet<string> instantiatedExecutors, Dictionary<ExecutorIdentity, List<PortableMessageEnvelope>> queuedMessages, List<ExternalRequest> outstandingRequests)
{
public HashSet<string> InstantiatedExecutors { get; } = instantiatedExecutors;
public Dictionary<ExecutorIdentity, List<PortableMessageEnvelope>> QueuedMessages { get; } = queuedMessages;
@@ -9,10 +9,10 @@ using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows.Execution;
internal class StateManager
internal sealed class StateManager
{
private readonly Dictionary<ScopeId, StateScope> _scopes = new();
private readonly Dictionary<UpdateKey, StateUpdate> _queuedUpdates = new();
private readonly Dictionary<ScopeId, StateScope> _scopes = [];
private readonly Dictionary<UpdateKey, StateUpdate> _queuedUpdates = [];
private StateScope GetOrCreateScope(ScopeId scopeId)
{
@@ -125,15 +125,14 @@ internal class StateManager
}
public ValueTask WriteStateAsync<T>(string executorId, string? scopeName, string key, T value)
=> this.WriteStateAsync<T>(new ScopeId(Throw.IfNullOrEmpty(executorId), scopeName), key, value);
=> this.WriteStateAsync(new ScopeId(Throw.IfNullOrEmpty(executorId), scopeName), key, value);
public ValueTask WriteStateAsync<T>(ScopeId scopeId, string key, T value)
{
Throw.IfNullOrEmpty(key);
UpdateKey stateKey = new(scopeId, key);
StateUpdate update = StateUpdate.Update(key, value);
this._queuedUpdates[stateKey] = update;
this._queuedUpdates[stateKey] = StateUpdate.Update(key, value);
return default;
}
@@ -169,7 +168,7 @@ internal class StateManager
stateUpdates.Add(this._queuedUpdates[key]);
}
if (tracer != null && (updatesByScope.Count > 0))
if (tracer is not null && (updatesByScope.Count > 0))
{
tracer.TraceStatePublished();
}
@@ -8,9 +8,9 @@ using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows.Execution;
internal class StateScope
internal sealed class StateScope
{
private readonly Dictionary<string, PortableValue> _stateData = new();
private readonly Dictionary<string, PortableValue> _stateData = [];
public ScopeId ScopeId { get; }
public StateScope(ScopeId scopeId)
@@ -63,7 +63,7 @@ internal class StateScope
foreach (string key in updates.Keys)
{
if (updates == null || updates[key].Count == 0)
if (updates is null || updates[key].Count == 0)
{
continue;
}
@@ -17,10 +17,7 @@ internal sealed class StateUpdate
this.IsDelete = isDelete;
}
public static StateUpdate Update<T>(string key, T? value)
{
return new StateUpdate(key, value, value is null);
}
public static StateUpdate Update<T>(string key, T? value) => new(key, value, value is null);
public static StateUpdate Delete(string key)
{
@@ -6,9 +6,9 @@ using Microsoft.Agents.Workflows.Checkpointing;
namespace Microsoft.Agents.Workflows.Execution;
internal class StepContext
internal sealed class StepContext
{
public Dictionary<ExecutorIdentity, List<MessageEnvelope>> QueuedMessages { get; } = new();
public Dictionary<ExecutorIdentity, List<MessageEnvelope>> QueuedMessages { get; } = [];
public bool HasMessages => this.QueuedMessages.Values.Any(messageList => messageList.Count > 0);
@@ -16,7 +16,7 @@ internal class StepContext
{
if (!this.QueuedMessages.TryGetValue(executorId, out var messages))
{
this.QueuedMessages[executorId] = messages = new();
this.QueuedMessages[executorId] = messages = [];
}
return messages;
@@ -29,8 +29,7 @@ internal class StepContext
return this.QueuedMessages.Keys.ToDictionary(
keySelector: identity => identity,
elementSelector: identity => this.QueuedMessages[identity]
.Select(v => new PortableMessageEnvelope(v))
.ToList()
.ConvertAll(v => new PortableMessageEnvelope(v))
);
}
@@ -38,9 +37,9 @@ internal class StepContext
{
foreach (ExecutorIdentity identity in messages.Keys)
{
this.QueuedMessages[identity] = messages[identity].Select(UnwrapExportedState).ToList();
this.QueuedMessages[identity] = messages[identity].ConvertAll(UnwrapExportedState);
}
MessageEnvelope UnwrapExportedState(PortableMessageEnvelope es) => es.ToMessageEnvelope();
static MessageEnvelope UnwrapExportedState(PortableMessageEnvelope es) => es.ToMessageEnvelope();
}
}
@@ -15,7 +15,7 @@ namespace Microsoft.Agents.Workflows.Execution;
/// appropriate) and published during a step transition.</remarks>
/// <param name="scopeId"></param>
/// <param name="key"></param>
internal class UpdateKey(ScopeId scopeId, string key)
internal sealed class UpdateKey(ScopeId scopeId, string key)
{
public ScopeId ScopeId { get; } = Throw.IfNull(scopeId);
public string Key { get; } = Throw.IfNullOrEmpty(key);
@@ -24,15 +24,9 @@ internal class UpdateKey(ScopeId scopeId, string key)
: this(new ScopeId(Throw.IfNullOrEmpty(executorId), scopeName), key)
{ }
public override string ToString()
{
return $"{this.ScopeId}/{this.Key}";
}
public override string ToString() => $"{this.ScopeId}/{this.Key}";
public bool IsMatchingScope(ScopeId scopeId, bool strict = false)
{
return this.ScopeId == scopeId && (!strict || this.ScopeId.ExecutorId == scopeId.ExecutorId);
}
public bool IsMatchingScope(ScopeId scopeId, bool strict = false) => this.ScopeId == scopeId && (!strict || this.ScopeId.ExecutorId == scopeId.ExecutorId);
public override bool Equals(object? obj)
{
@@ -46,8 +40,5 @@ internal class UpdateKey(ScopeId scopeId, string key)
return false;
}
public override int GetHashCode()
{
return HashCode.Combine(this.ScopeId.ExecutorId, this.ScopeId.ScopeName, this.Key);
}
public override int GetHashCode() => HashCode.Combine(this.ScopeId.ExecutorId, this.ScopeId.ScopeName, this.Key);
}
@@ -41,12 +41,12 @@ public abstract class Executor : IIdentified
/// </summary>
protected abstract RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder);
private MessageRouter? _router = null;
private MessageRouter? _router;
internal MessageRouter Router
{
get
{
if (this._router == null)
if (this._router is null)
{
RouteBuilder routeBuilder = this.ConfigureRoutes(new RouteBuilder());
this._router = routeBuilder.Build();
@@ -74,7 +74,7 @@ public abstract class Executor : IIdentified
.ConfigureAwait(false);
ExecutorEvent executionResult;
if (result == null || result.IsSuccess)
if (result?.IsSuccess is not false)
{
executionResult = new ExecutorCompletedEvent(this.Id, result?.Result);
}
@@ -85,7 +85,7 @@ public abstract class Executor : IIdentified
await context.AddEventAsync(executionResult).ConfigureAwait(false);
if (result == null)
if (result is null)
{
throw new NotSupportedException(
$"No handler found for message type {message.GetType().Name} in executor {this.GetType().Name}.");
@@ -102,7 +102,7 @@ public abstract class Executor : IIdentified
}
// If we had a real return type, raise it as a SendMessage; TODO: Should we have a way to disable this behaviour?
if (result.Result != null && this._options.AutoSendMessageHandlerResultObject)
if (result.Result is not null && this._options.AutoSendMessageHandlerResultObject)
{
await context.SendMessageAsync(result.Result).ConfigureAwait(false);
}
@@ -156,10 +156,8 @@ public abstract class Executor<TInput>(string? id = null, ExecutorOptions? optio
: Executor(id, options), IMessageHandler<TInput>
{
/// <inheritdoc/>
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder)
{
return routeBuilder.AddHandler<TInput>(this.HandleAsync);
}
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
routeBuilder.AddHandler<TInput>(this.HandleAsync);
/// <inheritdoc/>
public abstract ValueTask HandleAsync(TInput message, IWorkflowContext context);
@@ -177,10 +175,8 @@ public abstract class Executor<TInput, TOutput>(string? id = null, ExecutorOptio
IMessageHandler<TInput, TOutput>
{
/// <inheritdoc/>
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder)
{
return routeBuilder.AddHandler<TInput, TOutput>(this.HandleAsync);
}
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
routeBuilder.AddHandler<TInput, TOutput>(this.HandleAsync);
/// <inheritdoc/>
public abstract ValueTask<TOutput> HandleAsync(TInput message, IWorkflowContext context);
@@ -18,13 +18,8 @@ public class ExecutorEvent(string executorId, object? data) : WorkflowEvent(data
public string ExecutorId => executorId;
/// <inheritdoc/>
public override string ToString()
{
if (this.Data != null)
{
return $"{this.GetType().Name}(Executor = {this.ExecutorId}, Data: {this.Data.GetType()} = {this.Data})";
}
return $"{this.GetType().Name}(Executor = {this.ExecutorId})";
}
public override string ToString() =>
this.Data is not null ?
$"{this.GetType().Name}(Executor = {this.ExecutorId}, Data: {this.Data.GetType()} = {this.Data})" :
$"{this.GetType().Name}(Executor = {this.ExecutorId})";
}
@@ -39,21 +39,15 @@ public static class ExecutorIshConfigurationExtensions
return new ExecutorIsh(configured.Super<TExecutor, Executor, TOptions>(), typeof(TExecutor), ExecutorIsh.Type.Executor);
}
private static ExecutorIsh ToExecutorIsh<TInput>(this FunctionExecutor<TInput> executor, Delegate raw)
{
return new ExecutorIsh(Configured.FromInstance(executor, raw: raw)
private static ExecutorIsh ToExecutorIsh<TInput>(this FunctionExecutor<TInput> executor, Delegate raw) => new(Configured.FromInstance(executor, raw: raw)
.Super<FunctionExecutor<TInput>, Executor>(),
typeof(FunctionExecutor<TInput>),
ExecutorIsh.Type.Function);
}
private static ExecutorIsh ToExecutorIsh<TInput, TOutput>(this FunctionExecutor<TInput, TOutput> executor, Delegate raw)
{
return new ExecutorIsh(Configured.FromInstance(executor, raw: raw)
private static ExecutorIsh ToExecutorIsh<TInput, TOutput>(this FunctionExecutor<TInput, TOutput> executor, Delegate raw) => new(Configured.FromInstance(executor, raw: raw)
.Super<FunctionExecutor<TInput, TOutput>, Executor>(),
typeof(FunctionExecutor<TInput, TOutput>),
ExecutorIsh.Type.Function);
}
/// <summary>
/// Configures a function-based asynchronous message handler as an executor with the specified identifier and
@@ -141,7 +135,7 @@ public sealed class ExecutorIsh :
this._idValue = Throw.IfNull(id);
}
internal ExecutorIsh(Configured<Executor> configured, System.Type configuredExecutorType, ExecutorIsh.Type type)
internal ExecutorIsh(Configured<Executor> configured, System.Type configuredExecutorType, Type type)
{
this.ExecutorType = type;
this._configuredExecutor = configured;
@@ -256,73 +250,42 @@ public sealed class ExecutorIsh :
/// Defines an implicit conversion from a string to an <see cref="ExecutorIsh"/> instance.
/// </summary>
/// <param name="id">The string ID to convert to an <see cref="ExecutorIsh"/>.</param>
public static implicit operator ExecutorIsh(string id)
{
return new ExecutorIsh(id);
}
public static implicit operator ExecutorIsh(string id) => new(id);
/// <inheritdoc/>
public bool Equals(ExecutorIsh? other)
{
return other is not null &&
other.Id == this.Id;
}
public bool Equals(ExecutorIsh? other) =>
other is not null && other.Id == this.Id;
/// <inheritdoc/>
public bool Equals(IIdentified? other)
{
return other is not null &&
other.Id == this.Id;
}
public bool Equals(IIdentified? other) =>
other is not null && other.Id == this.Id;
/// <inheritdoc/>
public bool Equals(string? other)
{
return other is not null &&
other == this.Id;
}
public bool Equals(string? other) =>
other is not null && other == this.Id;
/// <inheritdoc/>
public override bool Equals(object? obj)
{
if (obj is null)
public override bool Equals(object? obj) =>
obj switch
{
return false;
}
if (obj is ExecutorIsh ish)
{
return this.Equals(ish);
}
else if (obj is IIdentified identified)
{
return this.Equals(identified);
}
else if (obj is string str)
{
return this.Equals(str);
}
return false;
}
/// <inheritdoc/>
public override int GetHashCode()
{
return this.Id.GetHashCode();
}
/// <inheritdoc/>
public override string ToString()
{
return this.ExecutorType switch
{
Type.Unbound => $"'{this.Id}':<unbound>",
Type.Executor => $"'{this.Id}':{this._configuredExecutorType!.Name}",
Type.InputPort => $"'{this.Id}':Input({this._inputPortValue!.Request.Name}->{this._inputPortValue!.Response.Name})",
Type.Agent => $"{this.Id}':AIAgent(@{this._aiAgentValue!.GetType().Name})",
Type.Function => $"'{this.Id}':{this._configuredExecutorType!.Name}",
_ => $"'{this.Id}':<unknown[{this.ExecutorType}]>"
null => false,
ExecutorIsh ish => this.Equals(ish),
IIdentified identified => this.Equals(identified),
string str => this.Equals(str),
_ => false
};
}
/// <inheritdoc/>
public override int GetHashCode() => this.Id.GetHashCode();
/// <inheritdoc/>
public override string ToString() => this.ExecutorType switch
{
Type.Unbound => $"'{this.Id}':<unbound>",
Type.Executor => $"'{this.Id}':{this._configuredExecutorType!.Name}",
Type.InputPort => $"'{this.Id}':Input({this._inputPortValue!.Request.Name}->{this._inputPortValue!.Response.Name})",
Type.Agent => $"{this.Id}':AIAgent(@{this._aiAgentValue!.GetType().Name})",
Type.Function => $"'{this.Id}':{this._configuredExecutorType!.Name}",
_ => $"'{this.Id}':<unknown[{this.ExecutorType}]>"
};
}
@@ -8,7 +8,7 @@ using ExecutorFactoryF = System.Func<System.Threading.Tasks.ValueTask<Microsoft.
namespace Microsoft.Agents.Workflows;
internal class ExecutorRegistration(string id, Type executorType, ExecutorFactoryF provider, object? rawData)
internal sealed class ExecutorRegistration(string id, Type executorType, ExecutorFactoryF provider, object? rawData)
{
public string Id { get; } = Throw.IfNullOrEmpty(id);
public Type ExecutorType { get; } = Throw.IfNull(executorType);
@@ -36,7 +36,8 @@ public class FunctionExecutor<TInput>(Func<TInput, IWorkflowContext, Cancellatio
/// </summary>
/// <param name="handlerSync">A synchronous function to execute for each input message and workflow context.</param>
public FunctionExecutor(Action<TInput, IWorkflowContext, CancellationToken> handlerSync) : this(WrapAction(handlerSync))
{ }
{
}
}
/// <summary>
@@ -61,12 +62,15 @@ public class FunctionExecutor<TInput, TOutput>(Func<TInput, IWorkflowContext, Ca
return new ValueTask<TOutput>(result);
}
}
/// <inheritdoc/>
public override ValueTask<TOutput> HandleAsync(TInput message, IWorkflowContext context) => handlerAsync(message, context, default);
/// <summary>
/// Creates a new instance of the <see cref="FunctionExecutor{TInput,TOutput}"/> class.
/// </summary>
/// <param name="handlerSync">A synchronous function to execute for each input message and workflow context.</param>
public FunctionExecutor(Func<TInput, IWorkflowContext, CancellationToken, TOutput> handlerSync) : this(WrapFunc(handlerSync))
{ }
{
}
}
@@ -10,11 +10,11 @@ namespace Microsoft.Agents.Workflows.InProc;
internal sealed class InProcStepTracer : IStepTracer
{
private int _nextStepNumber = 0;
private int _nextStepNumber;
public int StepNumber => this._nextStepNumber - 1;
public bool StateUpdated { get; private set; } = false;
public CheckpointInfo? Checkpoint { get; private set; } = null;
public bool StateUpdated { get; private set; }
public CheckpointInfo? Checkpoint { get; private set; }
public HashSet<string> Instantiated { get; } = [];
public HashSet<string> Activated { get; } = [];
@@ -60,32 +60,33 @@ internal sealed class InProcStepTracer : IStepTracer
});
}
public SuperStepCompletedEvent Complete(bool nextStepHasActions, bool hasPendingRequests)
public SuperStepCompletedEvent Complete(bool nextStepHasActions, bool hasPendingRequests) => new(this.StepNumber, new SuperStepCompletionInfo(this.Activated, this.Instantiated)
{
return new SuperStepCompletedEvent(this.StepNumber, new SuperStepCompletionInfo(this.Activated, this.Instantiated)
{
HasPendingMessages = nextStepHasActions,
HasPendingRequests = hasPendingRequests,
StateUpdated = this.StateUpdated,
Checkpoint = this.Checkpoint,
});
}
HasPendingMessages = nextStepHasActions,
HasPendingRequests = hasPendingRequests,
StateUpdated = this.StateUpdated,
Checkpoint = this.Checkpoint,
});
public override string ToString()
{
StringBuilder sb = new();
if (this.Instantiated.Count != 0)
{
sb.Append("Instantiated: ");
sb.Append(string.Join(", ", this.Instantiated.OrderBy(id => id, StringComparer.Ordinal)));
sb.AppendLine();
sb.Append("Instantiated: ").Append(string.Join(", ", this.Instantiated.OrderBy(id => id, StringComparer.Ordinal)));
}
if (this.Activated.Count != 0)
{
sb.Append("Activated: ");
sb.Append(string.Join(", ", this.Activated.OrderBy(id => id, StringComparer.Ordinal)));
sb.AppendLine();
if (sb.Length != 0)
{
sb.AppendLine();
}
sb.Append("Activated: ").Append(string.Join(", ", this.Activated.OrderBy(id => id, StringComparer.Ordinal)));
}
return sb.ToString();
}
}
@@ -20,7 +20,7 @@ namespace Microsoft.Agents.Workflows.InProc;
/// within the current process, without distributed coordination. It is primarily intended for testing, debugging, or
/// scenarios where workflow execution does not require executor distribution. </para></remarks>
/// <typeparam name="TInput">The type of input accepted by the workflow. Must be non-nullable.</typeparam>
internal class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointingRunner where TInput : notnull
internal sealed class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointingRunner where TInput : notnull
{
public InProcessRunner(Workflow<TInput> workflow, ICheckpointManager? checkpointManager, string? runId = null)
{
@@ -61,7 +61,7 @@ internal class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointingRunner
return false;
}
await this.RunContext.AddExternalMessageAsync<T>(message).ConfigureAwait(false);
await this.RunContext.AddExternalMessageAsync(message).ConfigureAwait(false);
return true;
}
@@ -71,7 +71,6 @@ internal class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointingRunner
}
private InProcStepTracer StepTracer { get; } = new();
private Dictionary<string, string> PendingCalls { get; } = new();
private Workflow<TInput> Workflow { get; init; }
private InProcessRunnerContext<TInput> RunContext { get; init; }
private ICheckpointManager? CheckpointManager { get; }
@@ -90,14 +89,9 @@ internal class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointingRunner
this.WorkflowEvent?.Invoke(this, workflowEvent);
}
private bool IsResponse(object message)
{
return message is ExternalResponse;
}
private ValueTask<IEnumerable<object?>> RouteExternalMessageAsync(MessageEnvelope envelope)
{
Debug.Assert(envelope.TargetId == null, "External Messages cannot be targeted to a specific executor.");
Debug.Assert(envelope.TargetId is null, "External Messages cannot be targeted to a specific executor.");
object message = envelope.Message;
return message is ExternalResponse response
@@ -155,7 +149,6 @@ internal class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointingRunner
bool ISuperStepRunner.HasUnprocessedMessages => this.RunContext.NextStepHasActions;
public IReadOnlyList<CheckpointInfo> Checkpoints => this._checkpoints;
private CheckpointInfo? LastCheckpoint => this.Checkpoints[this.Checkpoints.Count - 1];
async ValueTask<bool> ISuperStepRunner.RunSuperStepAsync(CancellationToken cancellation)
{
@@ -190,7 +183,7 @@ internal class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointingRunner
this.RaiseWorkflowEvent(this.StepTracer.Advance(currentStep));
// Deliver the messages and queue the next step
List<Task<IEnumerable<object?>>> edgeTasks = new();
List<Task<IEnumerable<object?>>> edgeTasks = [];
foreach (ExecutorIdentity sender in currentStep.QueuedMessages.Keys)
{
IEnumerable<MessageEnvelope> senderMessages = currentStep.QueuedMessages[sender];
@@ -220,11 +213,11 @@ internal class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointingRunner
this.RaiseWorkflowEvent(this.StepTracer.Complete(this.RunContext.NextStepHasActions, this.RunContext.HasUnservicedRequests));
}
private WorkflowInfo? _workflowInfoCache = null;
private WorkflowInfo? _workflowInfoCache;
private readonly List<CheckpointInfo> _checkpoints = [];
internal async ValueTask CheckpointAsync(CancellationToken cancellation = default)
{
if (this.CheckpointManager == null)
if (this.CheckpointManager is null)
{
// Always publish the state updates, even in the absence of a CheckpointManager.
await this.RunContext.StateManager.PublishUpdatesAsync(this.StepTracer).ConfigureAwait(false);
@@ -235,10 +228,7 @@ internal class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointingRunner
Task prepareTask = this.RunContext.PrepareForCheckpointAsync(cancellation);
// Create a representation of the current workflow if it does not already exist.
if (this._workflowInfoCache == null)
{
this._workflowInfoCache = this.Workflow.ToWorkflowInfo();
}
this._workflowInfoCache ??= this.Workflow.ToWorkflowInfo();
Dictionary<EdgeId, PortableValue> edgeData = await this.EdgeMap.ExportStateAsync().ConfigureAwait(false);
@@ -284,13 +274,11 @@ internal class InProcessRunner<TInput> : ISuperStepRunner, ICheckpointingRunner
this.StepTracer.Reload(this.StepTracer.StepNumber);
}
protected virtual bool CheckWorkflowMatch(Checkpoint checkpoint)
{
return checkpoint.Workflow.IsMatch<TInput>(this.Workflow);
}
private bool CheckWorkflowMatch(Checkpoint checkpoint) =>
checkpoint.Workflow.IsMatch(this.Workflow);
}
internal class InProcessRunner<TInput, TResult> : IRunnerWithOutput<TResult>, ICheckpointingRunner where TInput : notnull
internal sealed class InProcessRunner<TInput, TResult> : IRunnerWithOutput<TResult>, ICheckpointingRunner where TInput : notnull
{
private readonly Workflow<TInput, TResult> _workflow;
private readonly InProcessRunner<TInput> _innerRunner;
@@ -299,8 +287,7 @@ internal class InProcessRunner<TInput, TResult> : IRunnerWithOutput<TResult>, IC
{
this._workflow = Throw.IfNull(workflow);
InProcessRunner<TInput> runner = new(workflow, checkpointManager, runId);
this._innerRunner = runner;
this._innerRunner = new(workflow, checkpointManager, runId);
}
internal async ValueTask<StreamingRun<TResult>> ResumeStreamAsync(CheckpointInfo checkpoint, CancellationToken cancellation = default)
@@ -14,7 +14,7 @@ using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows.InProc;
internal class InProcessRunnerContext<TExternalInput> : IRunnerContext
internal sealed class InProcessRunnerContext<TExternalInput> : IRunnerContext
{
private StepContext _nextStep = new();
private readonly Dictionary<string, ExecutorRegistration> _executorRegistrations;
@@ -66,10 +66,7 @@ internal class InProcessRunnerContext<TExternalInput> : IRunnerContext
public bool NextStepHasActions => this._nextStep.HasMessages;
public bool HasUnservicedRequests => this._externalRequests.Count > 0;
public StepContext Advance()
{
return Interlocked.Exchange(ref this._nextStep, new StepContext());
}
public StepContext Advance() => Interlocked.Exchange(ref this._nextStep, new StepContext());
public ValueTask AddEventAsync(WorkflowEvent workflowEvent)
{
@@ -83,10 +80,7 @@ internal class InProcessRunnerContext<TExternalInput> : IRunnerContext
return default;
}
public IWorkflowContext Bind(string executorId)
{
return new BoundContext(this, executorId);
}
public IWorkflowContext Bind(string executorId) => new BoundContext(this, executorId);
public ValueTask PostAsync(ExternalRequest request)
{
@@ -100,7 +94,7 @@ internal class InProcessRunnerContext<TExternalInput> : IRunnerContext
internal StateManager StateManager { get; } = new();
private class BoundContext(InProcessRunnerContext<TExternalInput> RunnerContext, string ExecutorId) : IWorkflowContext
private sealed class BoundContext(InProcessRunnerContext<TExternalInput> RunnerContext, string ExecutorId) : IWorkflowContext
{
public ValueTask AddEventAsync(WorkflowEvent workflowEvent) => RunnerContext.AddEventAsync(workflowEvent);
public ValueTask SendMessageAsync(object message, string? targetId = null) => RunnerContext.SendMessageAsync(ExecutorId, message, targetId);
@@ -118,15 +112,9 @@ internal class InProcessRunnerContext<TExternalInput> : IRunnerContext
=> RunnerContext.StateManager.ClearStateAsync(ExecutorId, scopeName);
}
internal Task PrepareForCheckpointAsync(CancellationToken cancellation = default)
{
return Task.WhenAll(this._executors.Values.Select(executor => executor.OnCheckpointingAsync(this.Bind(executor.Id), cancellation).AsTask()));
}
internal Task PrepareForCheckpointAsync(CancellationToken cancellation = default) => Task.WhenAll(this._executors.Values.Select(executor => executor.OnCheckpointingAsync(this.Bind(executor.Id), cancellation).AsTask()));
internal Task NotifyCheckpointLoadedAsync(CancellationToken cancellationToken = default)
{
return Task.WhenAll(this._executors.Values.Select(executor => executor.OnCheckpointRestoredAsync(this.Bind(executor.Id), cancellationToken).AsTask()));
}
internal Task NotifyCheckpointLoadedAsync(CancellationToken cancellationToken = default) => Task.WhenAll(this._executors.Values.Select(executor => executor.OnCheckpointRestoredAsync(this.Bind(executor.Id), cancellationToken).AsTask()));
internal ValueTask<RunnerStateData> ExportStateAsync()
{
@@ -10,7 +10,7 @@ namespace Microsoft.Agents.Workflows;
/// <param name="Id"></param>
/// <param name="Request"></param>
/// <param name="Response"></param>
public record InputPort(string Id, Type Request, Type Response)
public sealed record InputPort(string Id, Type Request, Type Response)
{
/// <summary>
/// Creates a new <see cref="InputPort"/> instance configured for the specified request and response types.
@@ -9,14 +9,14 @@ using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows;
internal class MessageMerger
internal sealed class MessageMerger
{
private class ResponseMergeState(string? responseId)
private sealed class ResponseMergeState(string? responseId)
{
public string? ResponseId { get; } = responseId;
public Dictionary<string, List<AgentRunResponseUpdate>> UpdatesByMessageId { get; } = new();
public List<AgentRunResponseUpdate> DanglingUpdates { get; } = new();
public Dictionary<string, List<AgentRunResponseUpdate>> UpdatesByMessageId { get; } = [];
public List<AgentRunResponseUpdate> DanglingUpdates { get; } = [];
public void AddUpdate(AgentRunResponseUpdate update)
{
@@ -28,15 +28,13 @@ internal class MessageMerger
{
if (!this.UpdatesByMessageId.TryGetValue(update.MessageId, out List<AgentRunResponseUpdate>? updates))
{
this.UpdatesByMessageId[update.MessageId] = updates = new List<AgentRunResponseUpdate>();
this.UpdatesByMessageId[update.MessageId] = updates = [];
}
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))
@@ -80,7 +78,7 @@ internal class MessageMerger
return updates.Aggregate(null,
(ChatMessage? previous, AgentRunResponseUpdate current) =>
{
return previous == null
return previous is null
? current.ToChatMessage()
: previous.UpdateWith(current);
})!;
@@ -88,7 +86,7 @@ internal class MessageMerger
}
}
private readonly Dictionary<string, ResponseMergeState> _mergeStates = new();
private readonly Dictionary<string, ResponseMergeState> _mergeStates = [];
private readonly ResponseMergeState _danglingState = new(null);
public void AddUpdate(AgentRunResponseUpdate update)
@@ -133,8 +131,8 @@ internal class MessageMerger
public AgentRunResponse ComputeMerged(string primaryResponseId, string? primaryAgentId = null, string? primaryAgentName = null)
{
List<ChatMessage> messages = [];
Dictionary<string, AgentRunResponse> responses = new();
HashSet<string> agentIds = new();
Dictionary<string, AgentRunResponse> responses = [];
HashSet<string> agentIds = [];
foreach (string responseId in this._mergeStates.Keys)
{
@@ -153,11 +151,11 @@ internal class MessageMerger
UsageDetails? usage = null;
AdditionalPropertiesDictionary? additionalProperties = null;
HashSet<DateTimeOffset> createdTimes = new();
HashSet<DateTimeOffset> createdTimes = [];
foreach (AgentRunResponse response in responses.Values)
{
if (response.AgentId != null)
if (response.AgentId is not null)
{
agentIds.Add(response.AgentId);
}
@@ -183,7 +181,7 @@ internal class MessageMerger
AdditionalProperties = additionalProperties
};
AgentRunResponse MergeResponses(AgentRunResponse? current, AgentRunResponse incoming)
static AgentRunResponse MergeResponses(AgentRunResponse? current, AgentRunResponse incoming)
{
if (current is null)
{
@@ -237,12 +235,12 @@ internal class MessageMerger
static AdditionalPropertiesDictionary? MergeProperties(AdditionalPropertiesDictionary? current, AdditionalPropertiesDictionary? incoming)
{
if (current == null)
if (current is null)
{
return incoming;
}
if (incoming == null)
if (incoming is null)
{
return current;
}
@@ -258,22 +256,22 @@ internal class MessageMerger
static UsageDetails? MergeUsage(UsageDetails? current, UsageDetails? incoming)
{
if (current == null)
if (current is null)
{
return incoming;
}
AdditionalPropertiesDictionary<long>? additionalCounts = current.AdditionalCounts;
if (incoming == null)
if (incoming is null)
{
return current;
}
if (additionalCounts == null)
if (additionalCounts is null)
{
additionalCounts = incoming.AdditionalCounts;
}
else if (incoming.AdditionalCounts != null)
else if (incoming.AdditionalCounts is not null)
{
foreach (string key in incoming.AdditionalCounts.Keys)
{
@@ -30,7 +30,7 @@ public sealed class PortableValue
/// <inheritdoc />
public override bool Equals(object? obj)
{
if (obj == null)
if (obj is null)
{
return false;
}
@@ -38,12 +38,12 @@ public sealed class PortableValue
if (obj is not PortableValue other)
{
Type targetType = obj.GetType();
return this.AsType(targetType)?.Equals(obj) ?? false;
return this.AsType(targetType)?.Equals(obj) is true;
}
return this.TypeId == other.TypeId
&& ((this.Value == null && other.Value == null)
|| this.Value != null && this.Value.Equals(other.Value));
&& ((this.Value is null && other.Value is null)
|| this.Value?.Equals(other.Value) is true);
}
/// <inheritdoc />
@@ -75,10 +75,10 @@ public sealed class PortableValue
internal bool IsDelayedDeserialization => this.Value is IDelayedDeserialization;
[JsonIgnore]
internal bool IsDeserialized => this._deserializedValueCache != null;
internal bool IsDeserialized => this._deserializedValueCache is not null;
private readonly object _value;
private object? _deserializedValueCache = null;
private object? _deserializedValueCache;
/// <summary>
/// Gets the raw underlying value represented by this instance.
@@ -103,10 +103,7 @@ public sealed class PortableValue
{
if (this.Value is IDelayedDeserialization delayedDeserialization)
{
if (this._deserializedValueCache == null)
{
this._deserializedValueCache = delayedDeserialization.Deserialize<TValue>();
}
this._deserializedValueCache ??= delayedDeserialization.Deserialize<TValue>();
}
if (this.Value is TValue typedValue)
@@ -172,5 +169,5 @@ public sealed class PortableValue
/// </summary>
/// <param name="targetType">The type to compare with the current instance. Cannot be null.</param>
/// <returns>true if the current instance can be assigned to targetType; otherwise, false.</returns>
public bool IsType(Type targetType) => this.AsType(targetType) != null;
public bool IsType(Type targetType) => this.AsType(targetType) is not null;
}
@@ -10,13 +10,13 @@ using Microsoft.Agents.Workflows.Execution;
namespace Microsoft.Agents.Workflows.Reflection;
internal struct MessageHandlerInfo
internal readonly struct MessageHandlerInfo
{
public Type InType { get; init; }
public Type? OutType { get; init; } = null;
public Type? OutType { get; init; }
public MethodInfo HandlerInfo { get; init; }
public Func<object, ValueTask<object?>>? Unwrapper { get; init; } = null;
public Func<object, ValueTask<object?>>? Unwrapper { get; init; }
public MessageHandlerInfo(MethodInfo handlerInfo)
{
@@ -67,7 +67,7 @@ internal struct MessageHandlerInfo
async ValueTask<CallResult> InvokeHandlerAsync(object message, IWorkflowContext workflowContext)
{
bool expectingVoid = resultType == null || resultType == typeof(void);
bool expectingVoid = resultType is null || resultType == typeof(void);
try
{
@@ -86,14 +86,14 @@ internal struct MessageHandlerInfo
$"{maybeValueTask?.GetType().Name ?? "null"}.");
}
Debug.Assert(resultType != null, "Expected resultType to be non-null when not expecting void.");
if (unwrapper == null)
Debug.Assert(resultType is not null, "Expected resultType to be non-null when not expecting void.");
if (unwrapper is null)
{
throw new InvalidOperationException(
$"Handler method is expected to return ValueTask<{resultType!.Name}>, but no unwrapper is available.");
}
if (maybeValueTask == null)
if (maybeValueTask is null)
{
throw new InvalidOperationException(
$"Handler method returned null, but a ValueTask<{resultType!.Name}> was expected.");
@@ -101,7 +101,7 @@ internal struct MessageHandlerInfo
object? result = await unwrapper(maybeValueTask).ConfigureAwait(false);
if (checkType && result != null && !resultType.IsInstanceOfType(result))
if (checkType && result is not null && !resultType.IsInstanceOfType(result))
{
throw new InvalidOperationException(
$"Handler method returned an incompatible type: expected {resultType.Name}, got {result.GetType().Name}.");
@@ -126,11 +126,11 @@ internal struct MessageHandlerInfo
where TExecutor : ReflectingExecutor<TExecutor>
{
MethodInfo handlerMethod = this.HandlerInfo;
return MessageHandlerInfo.Bind(InvokeHandler, checkType, this.OutType, this.Unwrapper);
return Bind(InvokeHandler, checkType, this.OutType, this.Unwrapper);
object? InvokeHandler(object message, IWorkflowContext workflowContext)
{
return handlerMethod.Invoke(executor, new object[] { message, workflowContext });
return handlerMethod.Invoke(executor, [message, workflowContext]);
}
}
}
@@ -16,13 +16,12 @@ public class ReflectingExecutor<
] TExecutor
> : Executor where TExecutor : ReflectingExecutor<TExecutor>
{
/// <inheritdoc cref="Executor.Executor(string?, ExecutorOptions?)"/>
/// <inheritdoc cref="Executor(string?, ExecutorOptions?)"/>
protected ReflectingExecutor(string? id = null, ExecutorOptions? options = null) : base(id, options)
{ }
{
}
/// <inheritdoc />
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder)
{
return routeBuilder.ReflectHandlers<TExecutor>(this);
}
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
routeBuilder.ReflectHandlers(this);
}
@@ -15,7 +15,7 @@ internal static class IMessageHandlerReflection
internal static readonly MethodInfo HandleAsync_1 = typeof(IMessageHandler<>).GetMethod(Nameof_HandleAsync, BindingFlags.Public | BindingFlags.Instance)!;
internal static readonly MethodInfo HandleAsync_2 = typeof(IMessageHandler<,>).GetMethod(Nameof_HandleAsync, BindingFlags.Public | BindingFlags.Instance)!;
internal static MethodInfo ReflectHandleAsync(this Type specializedType, int genericArgumentCount)
internal static MethodInfo ReflectHandle(this Type specializedType, int genericArgumentCount)
{
Debug.Assert(specializedType.IsGenericType &&
(specializedType.GetGenericTypeDefinition() == typeof(IMessageHandler<>) ||
@@ -60,16 +60,16 @@ internal static class RouteBuilderExtensions
// Get the generic arguments of the interface.
Type[] genericArguments = interfaceType.GetGenericArguments();
if (genericArguments.Length < 1 || genericArguments.Length > 2)
if (genericArguments.Length is < 1 or > 2)
{
continue; // Invalid handler signature.
}
Type inType = genericArguments[0];
Type? outType = genericArguments.Length == 2 ? genericArguments[1] : null;
MethodInfo? method = interfaceType.ReflectHandleAsync(genericArguments.Length);
MethodInfo? method = interfaceType.ReflectHandle(genericArguments.Length);
if (method != null)
if (method is not null)
{
yield return new MessageHandlerInfo(method) { InType = inType, OutType = outType };
}
@@ -68,9 +68,7 @@ internal static class ValueTaskTypeErasure
Task task = (Task)asTaskMethod.ReflectionInvoke(maybeGenericVT)!;
await task.ConfigureAwait(false); // TODO: Could we need to capture the context here?
object? result = getResultMethod.ReflectionInvoke(task);
return result;
return getResultMethod.ReflectionInvoke(task);
}
}
}
@@ -23,7 +23,7 @@ namespace Microsoft.Agents.Workflows;
/// </remarks>
public class RouteBuilder
{
private readonly Dictionary<Type, MessageHandlerF> _typedHandlers = new();
private readonly Dictionary<Type, MessageHandlerF> _typedHandlers = [];
internal RouteBuilder AddHandler(Type messageType, MessageHandlerF handler, bool overwrite = false)
{
@@ -126,8 +126,5 @@ public class RouteBuilder
}
}
internal MessageRouter Build()
{
return new MessageRouter(this._typedHandlers);
}
internal MessageRouter Build() => new(this._typedHandlers);
}
+2 -2
View File
@@ -46,7 +46,7 @@ public class Run
return result;
}
private readonly List<WorkflowEvent> _eventSink = new();
private readonly List<WorkflowEvent> _eventSink = [];
private readonly StreamingRun _streamingRun;
internal Run(StreamingRun streamingRun)
{
@@ -91,7 +91,7 @@ public class Run
/// </summary>
public IEnumerable<WorkflowEvent> OutgoingEvents => this._eventSink;
private int _lastBookmark = 0;
private int _lastBookmark;
/// <summary>
/// Gets all events emitted by the workflow since the last access to <see cref="NewEvents" />.
@@ -12,7 +12,7 @@ namespace Microsoft.Agents.Workflows;
/// <param name="executorId">The unique identifier for the executor associated with this ScopeId.</param>
/// <param name="scopeName">The name of the scope, if any. If <see langword="null"/>, this ScopeId
/// corresponds to the Executor's private scope.</param>
public class ScopeId(string executorId, string? scopeName = null)
public sealed class ScopeId(string executorId, string? scopeName = null)
{
/// <summary>
/// Gets the unique identifier of the executor.
@@ -25,10 +25,7 @@ public class ScopeId(string executorId, string? scopeName = null)
public string? ScopeName { get; } = scopeName;
/// <inheritdoc/>
public override string ToString()
{
return $"{this.ExecutorId}/{this.ScopeName ?? "default"}";
}
public override string ToString() => $"{this.ExecutorId}/{this.ScopeName ?? "default"}";
/// <inheritdoc/>
public override bool Equals(object? obj)
@@ -39,14 +36,13 @@ public class ScopeId(string executorId, string? scopeName = null)
{
return this.ExecutorId == other.ExecutorId;
}
else if (other.ScopeName is not null && this.ScopeName is not null)
if (other.ScopeName is not null && this.ScopeName is not null)
{
return this.ScopeName == other.ScopeName;
}
else
{
return false; // One has a scope name, the other does not.
}
// One has a scope name, the other does not.
}
return false;
@@ -55,7 +51,7 @@ public class ScopeId(string executorId, string? scopeName = null)
/// <inheritdoc/>
public static bool operator ==(ScopeId? left, ScopeId? right)
{
if (left is null && right == null)
if (left is null && right is null)
{
return true;
}
@@ -9,7 +9,7 @@ namespace Microsoft.Agents.Workflows;
/// <summary>
/// Represents a unique key within a specific scope, combining a scope identifier and a key string.
/// </summary>
public class ScopeKey
public sealed class ScopeKey
{
/// <summary>
/// The identifier for the scope associated with this key.
@@ -9,12 +9,12 @@ using Microsoft.Extensions.AI.Agents;
namespace Microsoft.Agents.Workflows.Specialized;
internal class AIAgentHostExecutor : Executor
internal sealed class AIAgentHostExecutor : Executor
{
private readonly bool _emitEvents;
private readonly AIAgent _agent;
private readonly List<ChatMessage> _pendingMessages = new();
private AgentThread? _thread = null;
private readonly List<ChatMessage> _pendingMessages = [];
private AgentThread? _thread;
public AIAgentHostExecutor(AIAgent agent, bool emitEvents = false) : base(id: agent.Id)
{
@@ -22,22 +22,13 @@ internal class AIAgentHostExecutor : Executor
this._emitEvents = emitEvents;
}
private AgentThread EnsureThread(IWorkflowContext context)
{
if (this._thread != null)
{
return this._thread;
}
private AgentThread EnsureThread(IWorkflowContext context) =>
this._thread ??= this._agent.GetNewThread();
return this._thread = this._agent.GetNewThread();
}
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder)
{
return routeBuilder.AddHandler<ChatMessage>(this.QueueMessageAsync)
.AddHandler<List<ChatMessage>>(this.QueueMessagesAsync)
.AddHandler<TurnToken>(this.TakeTurnAsync);
}
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
routeBuilder.AddHandler<ChatMessage>(this.QueueMessageAsync)
.AddHandler<List<ChatMessage>>(this.QueueMessagesAsync)
.AddHandler<TurnToken>(this.TakeTurnAsync);
public ValueTask QueueMessagesAsync(List<ChatMessage> messages, IWorkflowContext context)
{
@@ -51,12 +42,12 @@ internal class AIAgentHostExecutor : Executor
return default;
}
private const string ThreadStateKey = nameof(AIAgentHostExecutor._thread);
private const string PendingMessagesStateKey = nameof(AIAgentHostExecutor._pendingMessages);
private const string ThreadStateKey = nameof(_thread);
private const string PendingMessagesStateKey = nameof(_pendingMessages);
protected internal override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellation = default)
{
Task threadTask = Task.CompletedTask;
if (this._thread != null)
if (this._thread is not null)
{
JsonElement threadValue = await this._thread.SerializeAsync(cancellationToken: cancellation).ConfigureAwait(false);
threadTask = context.QueueStateUpdateAsync(ThreadStateKey, threadValue).AsTask();
@@ -90,10 +81,10 @@ internal class AIAgentHostExecutor : Executor
public async ValueTask TakeTurnAsync(TurnToken token, IWorkflowContext context)
{
bool emitEvents = token.EmitEvents.HasValue ? token.EmitEvents.Value : this._emitEvents;
bool emitEvents = token.EmitEvents ?? this._emitEvents;
IAsyncEnumerable<AgentRunResponseUpdate> agentStream = this._agent.RunStreamingAsync(this._pendingMessages, this.EnsureThread(context));
List<AIContent> updates = new();
List<AIContent> updates = [];
ChatMessage? currentStreamingMessage = null;
await foreach (AgentRunResponseUpdate update in agentStream.ConfigureAwait(false))
@@ -114,7 +105,7 @@ internal class AIAgentHostExecutor : Executor
// providing some mechanisms to help the user complete the request, or route it out of the
// workflow.
if (currentStreamingMessage == null || currentStreamingMessage.MessageId != update.MessageId)
if (currentStreamingMessage is null || currentStreamingMessage.MessageId != update.MessageId)
{
await PublishCurrentMessageAsync().ConfigureAwait(false);
currentStreamingMessage = new(update.Role ?? ChatRole.Assistant, update.Contents)
@@ -135,7 +126,7 @@ internal class AIAgentHostExecutor : Executor
async ValueTask PublishCurrentMessageAsync()
{
if (currentStreamingMessage != null)
if (currentStreamingMessage is not null)
{
currentStreamingMessage.Contents = updates;
updates = [];
@@ -6,7 +6,7 @@ using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows.Specialized;
internal class OutputCollectorExecutor<TInput, TResult> : Executor, IOutputSink<TResult>
internal sealed class OutputCollectorExecutor<TInput, TResult> : Executor, IOutputSink<TResult>
{
private readonly StreamingAggregator<TInput, TResult> _aggregator;
private readonly Func<TInput, TResult?, bool>? _completionCondition;
@@ -19,10 +19,8 @@ internal class OutputCollectorExecutor<TInput, TResult> : Executor, IOutputSink<
this._completionCondition = completionCondition;
}
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder)
{
return routeBuilder.AddHandler<TInput>(this.HandleAsync);
}
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) =>
routeBuilder.AddHandler<TInput>(this.HandleAsync);
public ValueTask HandleAsync(TInput message, IWorkflowContext context)
{
@@ -7,7 +7,7 @@ using Microsoft.Shared.Diagnostics;
namespace Microsoft.Agents.Workflows.Specialized;
internal class RequestInfoExecutor : Executor
internal sealed class RequestInfoExecutor : Executor
{
private InputPort Port { get; }
private IExternalRequestSink? RequestSink { get; set; }
@@ -20,7 +20,7 @@ internal class RequestInfoExecutor : Executor
};
private readonly bool _allowWrapped;
public RequestInfoExecutor(InputPort port, bool allowWrapped = true) : base(port.Id, RequestInfoExecutor.DefaultOptions)
public RequestInfoExecutor(InputPort port, bool allowWrapped = true) : base(port.Id, DefaultOptions)
{
this.Port = port;
@@ -30,25 +30,22 @@ internal class RequestInfoExecutor : Executor
protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder)
{
routeBuilder = routeBuilder
// Handle incoming requests (as raw request payloads)
.AddHandler(this.Port.Request, this.HandleAsync)
.AddHandler(typeof(object), this.HandleAsync);
// Handle incoming requests (as raw request payloads)
.AddHandler(this.Port.Request, this.HandleAsync)
.AddHandler(typeof(object), this.HandleAsync);
if (this._allowWrapped)
{
routeBuilder = routeBuilder
.AddHandler<ExternalRequest, ExternalRequest>((request, context) => this.HandleAsync(request.Data, context));
.AddHandler<ExternalRequest, ExternalRequest>((request, context) => this.HandleAsync(request.Data, context));
}
return routeBuilder
// Handle incoming responses (as wrapped Response object)
.AddHandler<ExternalResponse, ExternalResponse>(this.HandleAsync);
// Handle incoming responses (as wrapped Response object)
.AddHandler<ExternalResponse, ExternalResponse>(this.HandleAsync);
}
internal void AttachRequestSink(IExternalRequestSink requestSink)
{
this.RequestSink = Throw.IfNull(requestSink);
}
internal void AttachRequestSink(IExternalRequestSink requestSink) => this.RequestSink = Throw.IfNull(requestSink);
public async ValueTask<ExternalRequest> HandleAsync(object message, IWorkflowContext context)
{
@@ -65,13 +62,9 @@ internal class RequestInfoExecutor : Executor
Throw.IfNull(message);
Throw.IfNull(message.Data);
object? data = message.DataAs(this.Port.Response);
if (data == null)
{
object data = message.DataAs(this.Port.Response) ??
throw new InvalidOperationException(
$"Message type {message.Data.TypeId} is not assignable to the response type {this.Port.Response.Name} of input port {this.Port.Id}.");
}
await context.SendMessageAsync(message).ConfigureAwait(false);
await context.SendMessageAsync(data).ConfigureAwait(false);
@@ -16,16 +16,12 @@ internal static partial class WorkflowJsonUtilities
[JsonSerializable(typeof(List<ChatMessage>))]
internal sealed partial class WorkflowJsonContext : JsonSerializerContext;
public static JsonElement SerializeToJson(this List<ChatMessage> messages)
{
return JsonSerializer.SerializeToElement(messages, Default.ListChatMessage);
}
public static JsonElement SerializeToJson(this List<ChatMessage> messages) =>
JsonSerializer.SerializeToElement(messages, Default.ListChatMessage);
public static JsonElement SerializeToJson(this IEnumerable<ChatMessage> messages)
=> messages.ToList().SerializeToJson();
public static List<ChatMessage> DeserializeMessageList(this JsonElement element)
{
return element.Deserialize<List<ChatMessage>>(Default.ListChatMessage) ?? [];
}
public static List<ChatMessage> DeserializeMessageList(this JsonElement element) =>
element.Deserialize(Default.ListChatMessage) ?? [];
}
@@ -104,7 +104,7 @@ public static class StreamingAggregators
IEnumerable<TResult> Aggregate(TInput input, IEnumerable<TResult>? runningResult)
{
return runningResult != null ? runningResult.Append(conversion(input)) : [conversion(input)];
return runningResult is not null ? runningResult.Append(conversion(input)) : [conversion(input)];
}
}
@@ -120,9 +120,9 @@ public static class StreamingAggregators
{
return Aggregate;
IEnumerable<TInput> Aggregate(TInput input, IEnumerable<TInput>? runningResult)
static IEnumerable<TInput> Aggregate(TInput input, IEnumerable<TInput>? runningResult)
{
return runningResult != null ? runningResult.Append(input) : new[] { input };
return runningResult is not null ? runningResult.Append(input) : [input];
}
}
}
@@ -17,7 +17,7 @@ namespace Microsoft.Agents.Workflows;
/// </summary>
public class StreamingRun
{
private TaskCompletionSource<object>? _waitForResponseSource = null;
private TaskCompletionSource<object>? _waitForResponseSource;
private readonly ISuperStepRunner _stepRunner;
/// <summary>
@@ -86,7 +86,7 @@ public class StreamingRun
bool blockOnPendingRequest,
[EnumeratorCancellation] CancellationToken cancellation = default)
{
List<WorkflowEvent> eventSink = new();
List<WorkflowEvent> eventSink = [];
this._stepRunner.WorkflowEvent += OnWorkflowEvent;
@@ -102,8 +102,7 @@ public class StreamingRun
}
bool hadCompletionEvent = false;
List<WorkflowEvent> outputEvents = Interlocked.Exchange(ref eventSink, new());
foreach (WorkflowEvent raisedEvent in outputEvents)
foreach (WorkflowEvent raisedEvent in Interlocked.Exchange(ref eventSink, []))
{
yield return raisedEvent;
@@ -132,15 +131,9 @@ public class StreamingRun
!this._stepRunner.HasUnprocessedMessages &&
this._stepRunner.HasUnservicedRequests)
{
if (this._waitForResponseSource == null)
{
this._waitForResponseSource = new();
}
this._waitForResponseSource ??= new();
using CancellationTokenRegistration registration = cancellation.Register(() =>
{
this._waitForResponseSource?.SetResult(new());
});
using CancellationTokenRegistration registration = cancellation.Register(() => this._waitForResponseSource?.SetResult(new()));
await this._waitForResponseSource.Task.ConfigureAwait(false);
this._waitForResponseSource = null;
@@ -203,7 +196,7 @@ public static class StreamingRunExtensions
await foreach (WorkflowEvent @event in handle.WatchStreamAsync(cancellation).ConfigureAwait(false))
{
ExternalResponse? maybeResponse = eventCallback?.Invoke(@event);
if (maybeResponse != null)
if (maybeResponse is not null)
{
await handle.SendResponseAsync(maybeResponse).ConfigureAwait(false);
}
@@ -40,5 +40,5 @@ public sealed class SuperStepCompletionInfo(HashSet<string> activatedExecutors,
/// Gets the <see cref="CheckpointInfo"/> corresponding to the checkpoint created at the end of this SuperStep.
/// <see langword="null"/> if checkpointing was not enabled when the run was started.
/// </summary>
public CheckpointInfo? Checkpoint { get; init; } = null;
public CheckpointInfo? Checkpoint { get; init; }
}
@@ -17,13 +17,8 @@ public class SuperStepEvent(int stepNumber, object? data = null) : WorkflowEvent
public int StepNumber => stepNumber;
/// <inheritdoc/>
public override string ToString()
{
if (this.Data != null)
{
return $"{this.GetType().Name}(Step = {this.StepNumber}, Data: {this.Data.GetType()} = {this.Data})";
}
return $"{this.GetType().Name}(Step = {this.StepNumber})";
}
public override string ToString() =>
this.Data is not null ?
$"{this.GetType().Name}(Step = {this.StepNumber}, Data: {this.Data.GetType()} = {this.Data})" :
$"{this.GetType().Name}(Step = {this.StepNumber})";
}
@@ -6,7 +6,7 @@ namespace Microsoft.Agents.Workflows;
/// <summary>
/// Sent to an <see cref="AIAgent"/>-based executor to request
/// a response to accumulated <see cref="Microsoft.Extensions.AI.ChatMessage"/>.
/// a response to accumulated <see cref="Extensions.AI.ChatMessage"/>.
/// </summary>
/// <param name="emitEvents">Whether to raise AgentRunEvents for this executor.</param>
public class TurnToken(bool? emitEvents = null)
@@ -17,9 +17,9 @@ public class Workflow
/// <summary>
/// A dictionary of executor providers, keyed by executor ID.
/// </summary>
internal Dictionary<string, ExecutorRegistration> Registrations { get; init; } = new();
internal Dictionary<string, ExecutorRegistration> Registrations { get; init; } = [];
internal Dictionary<string, HashSet<Edge>> Edges { get; init; } = new();
internal Dictionary<string, HashSet<Edge>> Edges { get; init; } = [];
/// <summary>
/// Gets the collection of edges grouped by their source node identifier.
@@ -32,7 +32,7 @@ public class Workflow
);
}
internal Dictionary<string, InputPort> Ports { get; init; } = new();
internal Dictionary<string, InputPort> Ports { get; init; } = [];
/// <summary>
/// Gets the collection of external request ports, keyed by their ID.
@@ -25,12 +25,12 @@ public class WorkflowBuilder
public override string ToString() => $"{this.SourceId} -> {this.TargetId}";
}
private int _edgeCount = 0;
private readonly Dictionary<string, ExecutorRegistration> _executors = new();
private readonly Dictionary<string, HashSet<Edge>> _edges = new();
private readonly HashSet<string> _unboundExecutors = new();
private readonly HashSet<EdgeConnection> _conditionlessConnections = new();
private readonly Dictionary<string, InputPort> _inputPorts = new();
private int _edgeCount;
private readonly Dictionary<string, ExecutorRegistration> _executors = [];
private readonly Dictionary<string, HashSet<Edge>> _edges = [];
private readonly HashSet<string> _unboundExecutors = [];
private readonly HashSet<EdgeConnection> _conditionlessConnections = [];
private readonly Dictionary<string, InputPort> _inputPorts = [];
private readonly string _startExecutorId;
@@ -65,8 +65,8 @@ public class WorkflowBuilder
$"Cannot bind executor with ID '{executorish.Id}' because an executor with the same ID but a different type ({existing.ExecutorType.Name} vs {incoming.ExecutorType.Name}) is already bound.");
}
if (existing.RawExecutorishData != null &&
!object.ReferenceEquals(existing.RawExecutorishData, incoming.RawExecutorishData))
if (existing.RawExecutorishData is not null &&
!ReferenceEquals(existing.RawExecutorishData, incoming.RawExecutorishData))
{
throw new InvalidOperationException(
$"Cannot bind executor with ID '{executorish.Id}' because an executor with the same ID but different instance is already bound.");
@@ -116,7 +116,7 @@ public class WorkflowBuilder
// If it does not exist, create a new one.
if (!this._edges.TryGetValue(sourceId, out HashSet<Edge>? edges))
{
this._edges[sourceId] = edges = new HashSet<Edge>();
this._edges[sourceId] = edges = [];
}
return edges;
@@ -136,7 +136,7 @@ public class WorkflowBuilder
internal static Func<object?, bool>? CreateConditionFunc<T>(Func<T?, bool>? condition)
{
if (condition == null)
if (condition is null)
{
return null;
}
@@ -152,7 +152,7 @@ public class WorkflowBuilder
internal static Func<object?, bool>? CreateConditionFunc<T>(Func<object?, bool>? condition)
{
if (condition == null)
if (condition is null)
{
return null;
}
@@ -188,7 +188,7 @@ public class WorkflowBuilder
Throw.IfNull(target);
EdgeConnection connection = new(source.Id, target.Id);
if (condition == null && this._conditionlessConnections.Contains(connection))
if (condition is null && this._conditionlessConnections.Contains(connection))
{
throw new InvalidOperationException(
$"An edge from '{source.Id}' to '{target.Id}' already exists without a condition. " +
@@ -216,7 +216,7 @@ public class WorkflowBuilder
internal static Func<object?, int, IEnumerable<int>>? CreateEdgeAssignerFunc<T>(Func<T?, int, IEnumerable<int>>? partitioner)
{
if (partitioner == null)
if (partitioner is null)
{
return null;
}
@@ -252,7 +252,7 @@ public class WorkflowBuilder
this.Track(source).Id,
targets.Select(target => this.Track(target).Id).ToList(),
this.TakeEdgeId(),
CreateEdgeAssignerFunc<T>(partitioner));
CreateEdgeAssignerFunc(partitioner));
this.EnsureEdgesFor(source.Id).Add(new(fanOutEdge));
@@ -302,9 +302,9 @@ public class WorkflowBuilder
var culture = System.Globalization.CultureInfo.CurrentCulture;
var uiCulture = System.Globalization.CultureInfo.CurrentUICulture;
return factory.StartNew(PropagateCultureAndInvoke).Unwrap().GetAwaiter().GetResult();
return factory.StartNew(PropagateCultureAndInvokeAsync).Unwrap().GetAwaiter().GetResult();
Task<TResult> PropagateCultureAndInvoke()
Task<TResult> PropagateCultureAndInvokeAsync()
{
// Set the culture and UI culture to the captured values
System.Globalization.CultureInfo.CurrentCulture = culture;

Some files were not shown because too many files have changed in this diff Show More