// Copyright (c) Microsoft. All rights reserved. using System; using System.Collections.Generic; using System.Diagnostics; using System.Reflection; using System.Threading; using System.Threading.Tasks; using Microsoft.Agents.AI.Workflows.Checkpointing; using Microsoft.Agents.AI.Workflows.Execution; using Microsoft.Agents.AI.Workflows.Observability; using Microsoft.Agents.AI.Workflows.Reflection; namespace Microsoft.Agents.AI.Workflows; /// /// A component that processes messages in a . /// [DebuggerDisplay("{GetType().Name}{Id}")] public abstract class Executor : IIdentified { /// /// A unique identifier for the executor. /// public string Id { get; } private static readonly string s_namespace = typeof(Executor).Namespace!; private static readonly ActivitySource s_activitySource = new(s_namespace); // TODO: Add overloads for binding with a configuration/options object once the Configured hierarchy goes away. /// /// Initialize the executor with a unique identifier /// /// A unique identifier for the executor. /// Configuration options for the executor. If null, default options will be used. /// Declare that this executor may be used simultaneously by multiple runs safely. protected Executor(string id, ExecutorOptions? options = null, bool declareCrossRunShareable = false) { this.Id = id; this.Options = options ?? ExecutorOptions.Default; //if (declareCrossRunShareable && this is IResettableExecutor) //{ // // We need a way to be able to let the user override this at the workflow level too, because knowing the fine // // details of when to use which of these paths seems like it could be tricky, and we should not force users // // to do this; instead container agents should set this when they intiate the run (via WorkflowHostAgent). // throw new ArgumentException("An executor that is declared as cross-run shareable cannot also be resettable."); //} this.IsCrossRunShareable = declareCrossRunShareable; } internal bool IsCrossRunShareable { get; } /// /// Gets the configuration options for the executor. /// protected ExecutorOptions Options { get; } /// /// Override this method to register handlers for the executor. /// protected abstract RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder); /// /// Perform any asynchronous initialization required by the executor. This method is called once per executor instance, /// /// The workflow context in which the executor executes. /// The to monitor for cancellation requests. /// The default is . /// A representing the asynchronous operation. protected internal virtual ValueTask InitializeAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => default; /// /// Override this method to declare the types of messages this executor can send. /// /// protected virtual ISet ConfigureSentTypes() => new HashSet([typeof(object)]); /// /// Override this method to declare the types of messages this executor can yield as workflow outputs. /// /// protected virtual ISet ConfigureYieldTypes() { if (this.Options.AutoYieldOutputHandlerResultObject) { return this.Router.DefaultOutputTypes; } return new HashSet(); } private MessageRouter? _router; internal MessageRouter Router { get { if (this._router is null) { RouteBuilder routeBuilder = this.ConfigureRoutes(new RouteBuilder()); this._router = routeBuilder.Build(); } return this._router; } } /// /// Process an incoming message using the registered handlers. /// /// The message to be processed by the executor. /// The "declared" type of the message (captured when it was being sent). This is /// used to enable routing messages as their base types, in absence of true polymorphic type routing. /// The workflow context in which the executor executes. /// The to monitor for cancellation requests. /// The default is . /// A ValueTask representing the asynchronous operation, wrapping the output from the executor. /// No handler found for the message type. /// An exception is generated while handling the message. public async ValueTask ExecuteAsync(object message, TypeId messageType, IWorkflowContext context, CancellationToken cancellationToken = default) { using var activity = s_activitySource.StartActivity(ActivityNames.ExecutorProcess, ActivityKind.Internal); activity?.SetTag(Tags.ExecutorId, this.Id) .SetTag(Tags.ExecutorType, this.GetType().FullName) .SetTag(Tags.MessageType, messageType.TypeName) .CreateSourceLinks(context.TraceContext); await context.AddEventAsync(new ExecutorInvokedEvent(this.Id, message), cancellationToken).ConfigureAwait(false); CallResult? result = await this.Router.RouteMessageAsync(message, context, requireRoute: true, cancellationToken) .ConfigureAwait(false); ExecutorEvent executionResult; if (result?.IsSuccess is not false) { executionResult = new ExecutorCompletedEvent(this.Id, result?.Result); } else { executionResult = new ExecutorFailedEvent(this.Id, result.Exception); } await context.AddEventAsync(executionResult, cancellationToken).ConfigureAwait(false); if (result is null) { throw new NotSupportedException( $"No handler found for message type {message.GetType().Name} in executor {this.GetType().Name}."); } if (!result.IsSuccess) { throw new TargetInvocationException($"Error invoking handler for {message.GetType()}", result.Exception); } if (result.IsVoid) { return null; // Void result. } // 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 is not null && this.Options.AutoSendMessageHandlerResultObject) { await context.SendMessageAsync(result.Result, cancellationToken: cancellationToken).ConfigureAwait(false); } if (result.Result is not null && this.Options.AutoYieldOutputHandlerResultObject) { await context.YieldOutputAsync(result.Result, cancellationToken).ConfigureAwait(false); } return result.Result; } /// /// Invoked before a checkpoint is saved, allowing custom pre-save logic in derived classes. /// /// The workflow context. /// A ValueTask representing the asynchronous operation. /// The to monitor for cancellation requests. /// The default is . protected internal virtual ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => default; /// /// Invoked after a checkpoint is loaded, allowing custom post-load logic in derived classes. /// /// The workflow context. /// A ValueTask representing the asynchronous operation. /// The to monitor for cancellation requests. /// The default is . protected internal virtual ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default) => default; /// /// A set of s, representing the messages this executor can handle. /// public ISet InputTypes => this.Router.IncomingTypes; /// /// A set of s, representing the messages this executor can produce as output. /// public ISet OutputTypes { get; } = new HashSet([typeof(object)]); /// /// Describes the protocol for communication with this . /// /// public ProtocolDescriptor DescribeProtocol() { // TODO: Once burden of annotating yield/output messages becomes easier for the non-Auto case, // we should (1) start checking for validity on output/send side, and (2) add the Yield/Send // types to the ProtocolDescriptor. return new(this.InputTypes); } /// /// Checks if the executor can handle a specific message type. /// /// /// public bool CanHandle(Type messageType) => this.Router.CanHandle(messageType); internal bool CanHandle(TypeId messageType) => this.Router.CanHandle(messageType); internal bool CanOutput(Type messageType) { foreach (Type type in this.OutputTypes) { if (type.IsAssignableFrom(messageType)) { return true; } } return false; } } /// /// Provides a simple executor implementation that uses a single message handler function to process incoming messages. /// /// The type of input message. /// A unique identifier for the executor. /// Configuration options for the executor. If null, default options will be used. /// Declare that this executor may be used simultaneously by multiple runs safely. public abstract class Executor(string id, ExecutorOptions? options = null, bool declareCrossRunShareable = false) : Executor(id, options, declareCrossRunShareable), IMessageHandler { /// protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) => routeBuilder.AddHandler(this.HandleAsync); /// public abstract ValueTask HandleAsync(TInput message, IWorkflowContext context, CancellationToken cancellationToken = default); } /// /// Provides a simple executor implementation that uses a single message handler function to process incoming messages. /// /// The type of input message. /// The type of output message. /// A unique identifier for the executor. /// Configuration options for the executor. If null, default options will be used. /// Declare that this executor may be used simultaneously by multiple runs safely. public abstract class Executor(string id, ExecutorOptions? options = null, bool declareCrossRunShareable = false) : Executor(id, options ?? ExecutorOptions.Default, declareCrossRunShareable), IMessageHandler { /// protected override RouteBuilder ConfigureRoutes(RouteBuilder routeBuilder) => routeBuilder.AddHandler(this.HandleAsync); /// public abstract ValueTask HandleAsync(TInput message, IWorkflowContext context, CancellationToken cancellationToken = default); }