diff --git a/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/OrderCancelExecutor.cs b/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/OrderCancelExecutor.cs index 153830c84b..83dd4d2366 100644 --- a/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/OrderCancelExecutor.cs +++ b/dotnet/samples/DurableAgents/ConsoleApps/08_SingleWorkflow/OrderCancelExecutor.cs @@ -45,7 +45,7 @@ internal sealed class OrderLookup() : Executor("OrderLookup") Order order = new() { Id = message, - OrderDate = DateTime.UtcNow.AddDays(-3), + OrderDate = DateTime.UtcNow.AddDays(-1), IsCancelled = false, Customer = new Customer { Name = "Jerry", Email = "jerry@example.com" } }; @@ -78,11 +78,11 @@ internal sealed class OrderCancel() : Executor("OrderCancel") // Simulate a slow cancellation process (e.g., calling external payment system) // This is where you can kill the process to test durability - for (int i = 1; i <= 5; i++) + for (int i = 1; i <= 10; i++) { await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); Console.ForegroundColor = ConsoleColor.DarkYellow; - Console.WriteLine($"│ [Activity] OrderCancel: Processing... {i}/5 seconds"); + Console.WriteLine($"│ [Activity] OrderCancel: Processing... {i}/10 seconds"); Console.ResetColor(); } diff --git a/dotnet/samples/DurableAgents/ConsoleApps/12_WorkflowLoop/FeedbackExecutor.cs b/dotnet/samples/DurableAgents/ConsoleApps/12_WorkflowLoop/FeedbackExecutor.cs index 84f59f4ac6..40efc88d60 100644 --- a/dotnet/samples/DurableAgents/ConsoleApps/12_WorkflowLoop/FeedbackExecutor.cs +++ b/dotnet/samples/DurableAgents/ConsoleApps/12_WorkflowLoop/FeedbackExecutor.cs @@ -62,6 +62,7 @@ internal sealed class FeedbackExecutor : Executor return; } + Console.WriteLine("Sending back for refining"); await context.SendMessageAsync(feedback, cancellationToken: cancellationToken); this._attempts++; } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableStreamingRun.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableStreamingRun.cs index 71f310210f..65830d1e31 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableStreamingRun.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableStreamingRun.cs @@ -246,8 +246,8 @@ public sealed class DurableStreamingRun : IAsyncDisposable try { // First try to deserialize as SerializedWorkflowEvent (new format with type info) - DurableWorkflowRunner.SerializedWorkflowEvent? wrapper = - JsonSerializer.Deserialize(serializedEvent); + SerializedWorkflowEvent? wrapper = + JsonSerializer.Deserialize(serializedEvent); if (wrapper?.TypeName is not null && wrapper.Data is not null) { diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowJsonContext.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowJsonContext.cs new file mode 100644 index 0000000000..b57492a874 --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowJsonContext.cs @@ -0,0 +1,42 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Text.Json.Serialization; + +namespace Microsoft.Agents.AI.DurableTask; + +/// +/// Source-generated JSON serialization context for durable workflow types. +/// +/// +/// +/// This context provides AOT-compatible and trimmer-safe JSON serialization for the +/// internal data transfer types used by the durable workflow infrastructure: +/// +/// +/// : Activity input wrapper with state +/// : Activity output wrapper with results and events +/// : Messages sent via SendMessageAsync +/// : Workflow event wrapper +/// : Orchestrator-to-activity input wrapper +/// +/// +/// Note: User-defined executor input/output types still use reflection-based serialization +/// since their types are not known at compile time. +/// +/// +[JsonSourceGenerationOptions( + WriteIndented = false, + DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull, + PropertyNamingPolicy = JsonKnownNamingPolicy.CamelCase)] +[JsonSerializable(typeof(DurableActivityInput))] +[JsonSerializable(typeof(DurableActivityOutput))] +[JsonSerializable(typeof(SentMessageInfo))] +[JsonSerializable(typeof(List))] +[JsonSerializable(typeof(SerializedWorkflowEvent))] +[JsonSerializable(typeof(ActivityInputWithState))] +[JsonSerializable(typeof(List))] +[JsonSerializable(typeof(Dictionary))] +[JsonSerializable(typeof(Dictionary))] +internal partial class DurableWorkflowJsonContext : JsonSerializerContext +{ +} diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs index 52a7cee234..3f59aaea86 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowRunner.cs @@ -18,6 +18,17 @@ namespace Microsoft.Agents.AI.DurableTask; /// The serialized input data that was passed to the RequestPort. /// The full type name of the request type. /// The full type name of the expected response type. +/// +/// +/// This status is set when a workflow reaches a human-in-the-loop point (RequestPort executor). +/// External actors can query this status to: +/// +/// +/// Discover which event name to raise +/// See the input/context that prompted the request +/// Understand the expected request and response types for proper serialization +/// +/// public sealed record PendingExternalEventStatus( string EventName, string Input, @@ -27,6 +38,25 @@ public sealed record PendingExternalEventStatus( /// /// Represents the complete custom status for a durable workflow orchestration. /// +/// +/// +/// This status object is serialized and stored as the orchestration's custom status, +/// making it visible in the Durable Task dashboard and queryable via the client API. +/// +/// +/// It serves two primary purposes: +/// +/// +/// +/// Event Accumulation: Collects workflow events (yields, halts, etc.) emitted +/// by executors during the workflow run for final result determination. +/// +/// +/// HITL Status: Indicates when the workflow is waiting for external input, +/// including the event name to raise and context about the request. +/// +/// +/// public sealed class DurableWorkflowCustomStatus { /// @@ -44,6 +74,116 @@ public sealed class DurableWorkflowCustomStatus /// Core workflow runner that executes workflow orchestrations using Durable Tasks. /// This class contains the core workflow execution logic independent of the hosting environment. /// +/// +/// +/// Architecture Overview: +/// +/// +/// The DurableWorkflowRunner implements a message-driven execution model that naturally supports +/// both directed acyclic graphs (DAGs) and cyclic workflows. Each executor in the workflow is +/// treated as a message processor that receives input, executes logic, and routes output to successors. +/// +/// +/// Execution Flow: +/// +/// +/// +/// +/// Initialization: The workflow starts by queueing the initial input to the start executor. +/// +/// +/// +/// +/// Superstep Processing: In each superstep, all executors with pending messages are processed. +/// This continues until no messages remain or a halt is requested. +/// +/// +/// +/// +/// Activity Execution: Each executor runs as a Durable Task activity. The activity receives +/// wrapped input containing the message, type information, and shared workflow state. +/// +/// +/// +/// +/// Result Unwrapping: Activity results are unwrapped to extract the actual result, +/// state updates, workflow events, and any messages sent via SendMessageAsync. +/// +/// +/// +/// +/// Message Routing: Results are routed to successor executors based on workflow edges. +/// Edge conditions are evaluated to determine if a message should be forwarded. +/// +/// +/// +/// +/// Key Data Structures: +/// +/// +/// +/// +/// Message Queues: Each executor has a queue of pending messages (input + type information). +/// +/// +/// +/// +/// Shared State: A dictionary of scope-prefixed key-value pairs shared across executors. +/// +/// +/// +/// +/// Custom Status: Tracks workflow events and pending external event info for observability. +/// +/// +/// +/// +/// Executor Types: +/// +/// +/// +/// +/// Regular Executors: Run as Durable Task activities. Can return values or use SendMessageAsync. +/// +/// +/// +/// +/// Agent Executors: AI agents that run via Durable Entities for stateful conversations. +/// +/// +/// +/// +/// Request Port Executors: Human-in-the-loop points that wait for external events. +/// +/// +/// +/// +/// Input/Output Flow: +/// +/// +/// ┌─────────────────────────────────────────────────────────────────────────────────┐ +/// │ Orchestrator (DurableWorkflowRunner) │ +/// │ │ +/// │ ┌─────────────┐ ActivityInputWithState ┌─────────────────────────────┐ │ +/// │ │ Message │ ──────────────────────────▶ │ Activity │ │ +/// │ │ Queue │ {Input, InputTypeName, │ (ExecuteActivityAsync) │ │ +/// │ │ │ State} │ │ │ +/// │ │ │ │ - Deserialize input │ │ +/// │ │ │ │ - Execute executor logic │ │ +/// │ │ │ │ - Collect state updates │ │ +/// │ │ │ DurableActivityOutput │ - Collect events │ │ +/// │ │ │ ◀────────────────────────── │ - Collect sent messages │ │ +/// │ └─────────────┘ {Result, StateUpdates, └─────────────────────────────┘ │ +/// │ │ ClearedScopes, Events, │ +/// │ │ SentMessages} │ +/// │ ▼ │ +/// │ ┌─────────────┐ │ +/// │ │ Route to │ ── Evaluate edge conditions ──▶ Queue to successor executors │ +/// │ │ Successors │ │ +/// │ └─────────────┘ │ +/// └─────────────────────────────────────────────────────────────────────────────────┘ +/// +/// public class DurableWorkflowRunner { /// @@ -95,7 +235,7 @@ public class DurableWorkflowRunner logger.LogRunningWorkflow(workflow.Name); - return await this.ExecuteWorkflowLevelsAsync(context, workflow, input, logger).ConfigureAwait(true); + return await this.ExecuteWorkflowAsync(context, workflow, input, logger).ConfigureAwait(true); } /// @@ -123,13 +263,11 @@ public class DurableWorkflowRunner } /// - /// Serializes a list of strings to JSON. + /// Serializes a list of strings to JSON using source-generated serialization. /// - [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing known types.")] - [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing known types.")] protected static string SerializeToJson(List values) { - return JsonSerializer.Serialize(values); + return JsonSerializer.Serialize(values, DurableWorkflowJsonContext.Default.ListString); } /// @@ -187,7 +325,21 @@ public class DurableWorkflowRunner ?? throw new InvalidOperationException($"Failed to deserialize input to type '{targetType.Name}'."); } - private async Task ExecuteWorkflowLevelsAsync( + /// + /// Executes a workflow by building an execution plan and delegating to message-driven execution. + /// + /// The Durable Task orchestration context. + /// The workflow definition to execute. + /// The initial input string to pass to the start executor. + /// The replay-safe logger for orchestration logging. + /// The final result of the workflow execution. + /// + /// This method serves as the entry point for workflow execution after the workflow has been + /// resolved from the orchestration name. It builds a that + /// contains the graph structure (successors, predecessors), edge conditions, and executor + /// output types needed for message routing. + /// + private async Task ExecuteWorkflowAsync( TaskOrchestrationContext context, Workflow workflow, string initialInput, @@ -195,18 +347,51 @@ public class DurableWorkflowRunner { WorkflowExecutionPlan plan = WorkflowHelper.GetExecutionPlan(workflow); - // Use message-driven execution for all workflows + // Use superstep-based execution for all workflows // This approach naturally handles both DAGs and cyclic workflows - return await this.ExecuteMessageDrivenAsync(context, workflow, plan, initialInput, logger).ConfigureAwait(true); + return await this.RunSuperstepLoopAsync(context, workflow, plan, initialInput, logger).ConfigureAwait(true); } /// - /// Executes a workflow using message-driven execution. - /// Messages are routed through edges dynamically, naturally supporting both DAGs and cycles. + /// Runs the workflow execution loop using superstep-based processing. /// + /// + /// + /// This method implements a Bulk Synchronous Parallel (BSP) style execution where each "superstep" + /// processes all pending messages for all active executors. The workflow terminates when: + /// + /// + /// No more messages are pending (natural completion) + /// An executor calls RequestHaltAsync (explicit halt) + /// Maximum superstep limit is reached (safety limit) + /// + /// + /// Message Routing Rules: + /// + /// + /// + /// + /// Messages sent via SendMessageAsync take priority and include explicit type information. + /// This is the primary mechanism for void-returning executors. + /// + /// + /// + /// + /// If no messages were sent explicitly, the executor's return value is routed to successors. + /// The type information comes from . + /// + /// + /// + /// + /// Edge conditions are evaluated before routing. If a condition returns false, the message + /// is not forwarded to that particular successor. + /// + /// + /// + /// [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")] [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")] - private async Task ExecuteMessageDrivenAsync( + private async Task RunSuperstepLoopAsync( TaskOrchestrationContext context, Workflow workflow, WorkflowExecutionPlan plan, @@ -270,14 +455,14 @@ public class DurableWorkflowRunner WorkflowExecutorInfo executorInfo = CreateExecutorInfo(executorId, executorBindings); // Execute the activity with type information - string rawResult = await this.ExecuteExecutorAsync( + string rawResult = await this.DispatchExecutorAsync( context, executorInfo, input, inputTypeName, logger, customStatus, sharedState).ConfigureAwait(true); (string result, List sentMessages) = UnwrapActivityResult(rawResult, customStatus, sharedState); lastResults[executorId] = result; // Check for explicit halt request (via RequestHaltAsync) - if (CheckForHalt(customStatus, executorId)) + if (HasHaltBeenRequested(customStatus, executorId)) { haltRequested = true; finalOutput = result; @@ -321,12 +506,20 @@ public class DurableWorkflowRunner } // Return final output or last result from output executors - return finalOutput ?? GetMessageDrivenFinalResult(workflow, lastResults, customStatus); + return finalOutput ?? DetermineFinalResult(workflow, lastResults, customStatus); } /// /// Enqueues a message to an executor's message queue with type information. /// + /// The dictionary of message queues, keyed by executor ID. + /// The target executor ID to queue the message for. + /// The serialized message content. + /// The full type name of the message, used for deserialization hints. + /// + /// Creates a new queue for the executor if one doesn't exist. Messages are processed + /// in FIFO order during each superstep. + /// private static void EnqueueMessage( Dictionary> queues, string executorId, @@ -343,8 +536,16 @@ public class DurableWorkflowRunner } /// - /// Creates a WorkflowExecutorInfo for the given executor ID. + /// Creates a for the given executor ID. /// + /// The executor ID to look up. + /// The dictionary of executor bindings from the workflow. + /// A containing metadata about the executor. + /// Thrown when the executor ID is not found in bindings. + /// + /// This method determines the executor type (agentic vs regular) and extracts request port + /// information for human-in-the-loop executors. + /// private static WorkflowExecutorInfo CreateExecutorInfo( string executorId, Dictionary executorBindings) @@ -362,10 +563,21 @@ public class DurableWorkflowRunner /// /// Checks if the workflow should halt based on halt request events. - /// Note: YieldOutputAsync does NOT halt the workflow - it just yields intermediate output. - /// Only explicit RequestHaltAsync calls should halt the workflow. /// - private static bool CheckForHalt( + /// The custom status containing accumulated workflow events. + /// The executor ID that just completed execution. + /// true if a halt was requested; otherwise, false. + /// + /// + /// Only DurableHaltRequestedEvent triggers a halt. Note that YieldOutputAsync + /// does NOT halt the workflow - it just yields intermediate output that can be used as + /// the final result if no other output is produced. + /// + /// + /// Events are searched in reverse order (most recent first) for efficiency. + /// + /// + private static bool HasHaltBeenRequested( DurableWorkflowCustomStatus customStatus, string executorId) { @@ -434,8 +646,22 @@ public class DurableWorkflowRunner } /// - /// Core implementation for routing messages to successors. + /// Core implementation for routing messages to successor executors. /// + /// The source executor ID that produced the message. + /// The serialized message content to route. + /// The type name to pass to successors for deserialization. + /// The resolved for edge condition evaluation. + /// The workflow execution plan containing the graph structure. + /// The message queues to enqueue messages into. + /// The logger for debug output. + /// + /// For each successor of the source executor, this method: + /// + /// Evaluates the edge condition (if any) + /// Enqueues the message if the condition passes or no condition exists + /// + /// [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")] [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")] private static void RouteMessageToSuccessorsCore( @@ -465,9 +691,22 @@ public class DurableWorkflowRunner } /// - /// Evaluates an edge condition if one exists. + /// Evaluates an edge condition if one exists for the given source-sink pair. /// - /// True if the message should be routed (no condition or condition passed); false otherwise. + /// The source executor ID. + /// The target (sink) executor ID. + /// The serialized message to evaluate. + /// The type to deserialize the message to for condition evaluation. + /// The execution plan containing edge conditions. + /// The logger for debug/warning output. + /// + /// true if the message should be routed (no condition exists or condition passed); + /// false if the condition returned false or evaluation failed. + /// + /// + /// Edge conditions are user-defined predicates that filter which messages flow through an edge. + /// If evaluation throws an exception, the edge is skipped (fail-safe behavior) and a warning is logged. + /// [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")] [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")] private static bool TryEvaluateEdgeCondition( @@ -485,7 +724,7 @@ public class DurableWorkflowRunner try { - object? messageObj = DeserializeForCondition(message, messageType); + object? messageObj = DeserializeMessageForCondition(message, messageType); if (!condition(messageObj)) { logger.LogDebug("Edge {Source} -> {Sink}: condition returned false, skipping", sourceId, sinkId); @@ -502,12 +741,28 @@ public class DurableWorkflowRunner } /// - /// Gets the final result for a message-driven workflow execution. - /// Checks yielded outputs first (from YieldOutputAsync calls), then falls back to executor results. + /// Determines the final result for a workflow execution. /// + /// The workflow definition. + /// Dictionary of last results from each executor. + /// The custom status containing workflow events. + /// The final workflow result string. + /// + /// + /// Result priority (highest to lowest): + /// + /// + /// Most recent YieldOutputAsync call (explicit workflow output) + /// Results from designated output executors (via WithOutputFrom) + /// Last non-empty result from any executor + /// + /// + /// If multiple output executors are defined, their results are joined with \n---\n. + /// + /// [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing event types.")] [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing event types.")] - private static string GetMessageDrivenFinalResult( + private static string DetermineFinalResult( Workflow workflow, Dictionary lastResults, DurableWorkflowCustomStatus customStatus) @@ -549,6 +804,23 @@ public class DurableWorkflowRunner /// /// Extracts the most recent yielded output from the custom status events. /// + /// The custom status containing serialized workflow events. + /// The yielded output string, or null if no yield event was found. + /// + /// + /// Searches for DurableYieldedOutputEvent in reverse order (most recent first). + /// The event structure is: + /// + /// + /// { + /// "TypeName": "...DurableYieldedOutputEvent...", + /// "Data": "{\"Output\": \"actual output value\"}" + /// } + /// + /// + /// The output can be either a string or a serialized object (returned as raw JSON). + /// + /// [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing event types.")] [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing event types.")] private static string? GetLastYieldedOutput(DurableWorkflowCustomStatus customStatus) @@ -591,10 +863,30 @@ public class DurableWorkflowRunner } /// - /// Unwraps an activity result, extracting state updates, events, sent messages, and returning the actual result. + /// Unwraps an activity result, extracting state updates, events, sent messages, and the actual result. /// - [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing known wrapper type.")] - [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing known wrapper type.")] + /// The raw JSON string returned from the activity. + /// The custom status to append events to. + /// The shared state dictionary to apply updates to. + /// + /// A tuple containing the executor's result and any messages sent via SendMessageAsync. + /// + /// + /// + /// Activities return a wrapper containing: + /// + /// + /// Result: The serialized return value of the executor + /// StateUpdates: Key-value pairs to add/update/remove from shared state + /// ClearedScopes: Scope prefixes to bulk-remove from shared state + /// Events: Workflow events (yield, halt, etc.) to propagate + /// SentMessages: Messages sent via SendMessageAsync + /// + /// + /// If the raw result is not a valid , it's returned as-is + /// (backward compatibility for activities that return plain values). + /// + /// private static (string Result, List SentMessages) UnwrapActivityResult( string rawResult, DurableWorkflowCustomStatus customStatus, @@ -607,7 +899,7 @@ public class DurableWorkflowRunner try { - DurableActivityOutput? output = JsonSerializer.Deserialize(rawResult); + DurableActivityOutput? output = JsonSerializer.Deserialize(rawResult, DurableWorkflowJsonContext.Default.DurableActivityOutput); if (output is null || !IsValidActivityOutput(output)) { @@ -631,8 +923,14 @@ public class DurableWorkflowRunner } /// - /// Checks if the deserialized output is a valid DurableActivityOutput (not just default values). + /// Checks if the deserialized output is a valid (not just default values). /// + /// The deserialized output to validate. + /// true if the output has any meaningful content; otherwise, false. + /// + /// This check distinguishes actual activity output from arbitrary JSON that happened to + /// deserialize successfully but with all default/empty values. + /// private static bool IsValidActivityOutput(DurableActivityOutput output) { return output.Result is not null @@ -645,6 +943,17 @@ public class DurableWorkflowRunner /// /// Clears all state entries matching the specified scopes. /// + /// The list of scope names to clear. + /// The shared state dictionary to modify. + /// + /// + /// State keys are prefixed with their scope (e.g., myScope:keyName). + /// The special scope __default__ is used for keys without an explicit scope. + /// + /// + /// Clearing a scope removes all keys that start with {scope}:. + /// + /// private static void ApplyClearedScopes(List clearedScopes, Dictionary sharedState) { foreach (string clearedScope in clearedScopes) @@ -664,6 +973,12 @@ public class DurableWorkflowRunner /// /// Applies state updates to the shared state dictionary. /// + /// The updates to apply (key -> value, where null value means delete). + /// The shared state dictionary to modify. + /// + /// A null value in the updates dictionary signals that the key should be removed + /// from the shared state, enabling executors to delete state entries. + /// private static void ApplyStateUpdates(Dictionary stateUpdates, Dictionary sharedState) { foreach (KeyValuePair update in stateUpdates) @@ -682,6 +997,17 @@ public class DurableWorkflowRunner /// /// Updates the orchestration custom status with current events and pending event info. /// + /// The orchestration context to set the status on. + /// The custom status object containing events and pending event info. + /// + /// The custom status is visible in the Durable Task dashboard and can be queried via the + /// Durable Task client. It includes: + /// + /// Accumulated workflow events from all executors + /// Pending external event info when waiting for human-in-the-loop input + /// + /// Only updates if there's meaningful content to report. + /// private static void UpdateCustomStatus(TaskOrchestrationContext context, DurableWorkflowCustomStatus customStatus) { // Only update if there are events or a pending event @@ -692,48 +1018,20 @@ public class DurableWorkflowRunner } /// - /// Wrapper for serialized workflow events that includes type information for proper deserialization. - /// - public sealed class SerializedWorkflowEvent - { - /// - /// Gets or sets the assembly-qualified type name of the event. - /// - public string? TypeName { get; set; } - - /// - /// Gets or sets the serialized JSON data of the event. - /// - public string? Data { get; set; } - } - - /// - /// Wrapper for activity input that includes shared state and type information. - /// - internal sealed class ActivityInputWithState - { - /// - /// Gets or sets the serialized executor input. - /// - public string? Input { get; set; } - - /// - /// Gets or sets the assembly-qualified type name of the input, used for proper deserialization. - /// - public string? InputTypeName { get; set; } - - /// - /// Gets or sets the shared state dictionary. - /// - public Dictionary State { get; set; } = []; - } - - /// - /// Deserializes a JSON string result into an object for condition evaluation. + /// Deserializes a JSON message string into an object for edge condition evaluation. /// + /// The JSON string to deserialize. + /// The target type to deserialize to, or null to deserialize as . + /// + /// The deserialized object, or the original string if deserialization fails (graceful fallback). + /// + /// + /// This method is used to prepare the message for edge condition predicates. + /// If the JSON is invalid, it returns the raw string to allow conditions to handle string inputs. + /// [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow types registered at startup.")] [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow types registered at startup.")] - private static object? DeserializeForCondition(string json, Type? targetType) + private static object? DeserializeMessageForCondition(string json, Type? targetType) { if (string.IsNullOrEmpty(json)) { @@ -756,9 +1054,35 @@ public class DurableWorkflowRunner } } - [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Serializing known wrapper type.")] - [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Serializing known wrapper type.")] - private async Task ExecuteExecutorAsync( + /// + /// Dispatches execution to the appropriate handler based on executor type. + /// + /// + /// + /// This method routes execution to the appropriate handler based on executor type: + /// + /// + /// + /// + /// Request Port Executors: Handled via . + /// Sets custom status and waits for an external event (human-in-the-loop). + /// + /// + /// + /// + /// Regular Executors: Invoked as Durable Task activities. Input is wrapped + /// with state and type information via . + /// + /// + /// + /// + /// Agent Executors: Handled via . + /// AI agents run through Durable Entities for stateful conversation management. + /// + /// + /// + /// + private async Task DispatchExecutorAsync( TaskOrchestrationContext context, WorkflowExecutorInfo executorInfo, string input, @@ -786,13 +1110,37 @@ public class DurableWorkflowRunner State = new Dictionary(sharedState) // Pass a copy of the state }; - string wrappedInput = JsonSerializer.Serialize(inputWithState); + string wrappedInput = JsonSerializer.Serialize(inputWithState, DurableWorkflowJsonContext.Default.ActivityInputWithState); return await context.CallActivityAsync(triggerName, wrappedInput).ConfigureAwait(true); } return await ExecuteAgentAsync(context, executorInfo, input, logger).ConfigureAwait(true); } + /// + /// Executes a request port executor by waiting for an external event (human-in-the-loop). + /// + /// The orchestration context for waiting on external events. + /// The executor info containing the request port configuration. + /// The input data that prompted this request (visible to the external actor). + /// The logger for tracing. + /// The custom status to update with pending event info. + /// The response string from the external actor. + /// + /// + /// Request ports enable human-in-the-loop workflows. When a request port executor is reached: + /// + /// + /// The custom status is updated with including + /// the event name, input, and expected request/response types. + /// The orchestration waits for an external event with the port's ID as the event name. + /// Once the event is received, the custom status is cleared and the response is returned. + /// + /// + /// External actors can query the custom status to discover what input is needed and raise + /// the appropriate event via the Durable Task client. + /// + /// private static async Task ExecuteRequestPortAsync( TaskOrchestrationContext context, WorkflowExecutorInfo executorInfo, @@ -828,6 +1176,32 @@ public class DurableWorkflowRunner return response; } + /// + /// Executes an AI agent executor through Durable Entities. + /// + /// The orchestration context for entity communication. + /// The executor info containing the agent configuration. + /// The input/prompt to send to the AI agent. + /// The logger for warnings. + /// The agent's response text. + /// + /// + /// AI agents are stateful components that maintain conversation history and context. + /// They are implemented using Durable Entities to persist their state across orchestration replays. + /// + /// + /// Each agent invocation: + /// + /// + /// Retrieves the agent proxy via context.GetAgent() + /// Creates a new thread for this conversation turn + /// Runs the agent with the input and returns the response text + /// + /// + /// If the agent is not found (not registered in ), + /// an error message is returned instead of throwing. + /// + /// private static async Task ExecuteAgentAsync( TaskOrchestrationContext context, WorkflowExecutorInfo executorInfo, @@ -848,3 +1222,67 @@ public class DurableWorkflowRunner return response.Text; } } + +/// +/// Wrapper for serialized workflow events that includes type information for proper deserialization. +/// +/// +/// Workflow events (e.g., DurableYieldedOutputEvent, DurableHaltRequestedEvent) are +/// serialized with their type information so they can be properly deserialized and processed +/// by the orchestrator. This enables type-safe event handling across the activity boundary. +/// +internal sealed class SerializedWorkflowEvent +{ + /// + /// Gets or sets the assembly-qualified type name of the event. + /// + public string? TypeName { get; set; } + + /// + /// Gets or sets the serialized JSON data of the event. + /// + public string? Data { get; set; } +} + +/// +/// Wrapper for activity input that includes shared state and type information. +/// +/// +/// +/// This wrapper is serialized and passed to each activity execution. It contains: +/// +/// +/// +/// Input: The serialized message/data for the executor to process. +/// +/// +/// +/// InputTypeName: The full type name of the input, used to deserialize to the correct type. +/// This is especially important when an executor accepts multiple input types. +/// +/// +/// +/// +/// State: A snapshot of the shared workflow state at the time of execution. +/// Activities can read from and write to this state via . +/// +/// +/// +/// +internal sealed class ActivityInputWithState +{ + /// + /// Gets or sets the serialized executor input. + /// + public string? Input { get; set; } + + /// + /// Gets or sets the assembly-qualified type name of the input, used for proper deserialization. + /// + public string? InputTypeName { get; set; } + + /// + /// Gets or sets the shared state dictionary. + /// + public Dictionary State { get; set; } = []; +} diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowServiceCollectionExtensions.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowServiceCollectionExtensions.cs index c73e12d7a5..daa796f72d 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowServiceCollectionExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/DurableWorkflowServiceCollectionExtensions.cs @@ -122,21 +122,19 @@ public static class DurableWorkflowServiceCollectionExtensions { string workflowName = workflow.Name!; string orchestrationName = WorkflowNamingHelper.ToOrchestrationFunctionName(workflowName); - - // Get all executor IDs from the workflow - HashSet executorIds = GetAllExecutorIds(workflow); Dictionary executorBindings = workflow.ReflectExecutors(); List activities = []; - foreach (string executorId in executorIds) + foreach (KeyValuePair entry in executorBindings) { - if (!executorBindings.TryGetValue(executorId, out ExecutorBinding? binding)) + // Skip agent executors - they're handled differently + if (entry.Value is AIAgentBinding) { continue; } - string executorName = WorkflowNamingHelper.GetExecutorName(executorId); + string executorName = WorkflowNamingHelper.GetExecutorName(entry.Key); string activityName = WorkflowNamingHelper.ToOrchestrationFunctionName(executorName); // Skip if already registered (same executor used in multiple workflows) @@ -145,37 +143,12 @@ public static class DurableWorkflowServiceCollectionExtensions continue; } - // Skip agent executors - they're handled differently - if (binding is AIAgentBinding) - { - continue; - } - - activities.Add(new ActivityRegistrationInfo(activityName, binding)); + activities.Add(new ActivityRegistrationInfo(activityName, entry.Value)); } return new WorkflowRegistrationInfo(orchestrationName, activities); } - private static HashSet GetAllExecutorIds(Workflow workflow) - { - HashSet executorIds = [workflow.StartExecutorId]; - - foreach (KeyValuePair> edgeGroup in workflow.ReflectEdges()) - { - executorIds.Add(edgeGroup.Key); - foreach (EdgeInfo edge in edgeGroup.Value) - { - foreach (string sinkId in edge.Connection.SinkIds) - { - executorIds.Add(sinkId); - } - } - } - - return executorIds; - } - private static async Task RunWorkflowOrchestrationAsync( TaskOrchestrationContext context, string input, @@ -193,38 +166,45 @@ public static class DurableWorkflowServiceCollectionExtensions [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "Executor types are registered at startup.")] private static async Task ExecuteActivityAsync(ExecutorBinding binding, string input) { - // Deserialize the input wrapper that includes state DurableActivityInput? inputWithState = TryDeserializeActivityInput(input); string executorInput = inputWithState?.Input ?? input; Dictionary sharedState = inputWithState?.State ?? []; - // Create executor instance from binding Executor executor = await binding.FactoryAsync!("activity-run").ConfigureAwait(false); - - // Determine the input type - prefer the provided type name, fall back to first supported type Type inputType = ResolveInputType(inputWithState?.InputTypeName, executor.InputTypes); object typedInput = DeserializeInput(executorInput, inputType); - // Create a pipeline context that has access to shared state and executor PipelineActivityContext workflowContext = new(sharedState, executor); - object? result = await executor.ExecuteAsync( typedInput, new TypeId(inputType), workflowContext, CancellationToken.None).ConfigureAwait(false); - // Always return wrapped output with state updates, events, sent messages, and result + return SerializeActivityOutput(result, workflowContext); + } + + /// + /// Serializes the activity output using source-generated serialization. + /// + [UnconditionalSuppressMessage("Trimming", "IL2026:RequiresUnreferencedCode", Justification = "SerializeResult uses reflection for user types.")] + [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "SerializeResult uses reflection for user types.")] + private static string SerializeActivityOutput(object? result, PipelineActivityContext context) + { DurableActivityOutput output = new() { Result = SerializeResult(result), - StateUpdates = workflowContext.StateUpdates, - ClearedScopes = [.. workflowContext.ClearedScopes], - Events = workflowContext.Events.ConvertAll(SerializeEvent), - SentMessages = workflowContext.SentMessages.ConvertAll(m => new SentMessageInfo { Message = m.Message, TypeName = m.TypeName }) + StateUpdates = context.StateUpdates, + ClearedScopes = [.. context.ClearedScopes], + Events = context.Events.ConvertAll(SerializeEvent), + SentMessages = context.SentMessages.ConvertAll(m => new SentMessageInfo + { + Message = m.Message, + TypeName = m.TypeName + }) }; - return JsonSerializer.Serialize(output); + return JsonSerializer.Serialize(output, DurableWorkflowJsonContext.Default.DurableActivityOutput); } /// @@ -235,38 +215,34 @@ public static class DurableWorkflowServiceCollectionExtensions [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "Type resolution for registered executor types.")] private static Type ResolveInputType(string? inputTypeName, ISet supportedTypes) { - if (!string.IsNullOrEmpty(inputTypeName)) + if (string.IsNullOrEmpty(inputTypeName)) { - // Try to find a matching type in the supported types - foreach (Type supportedType in supportedTypes) - { - if (supportedType.AssemblyQualifiedName == inputTypeName || - supportedType.FullName == inputTypeName || - supportedType.Name == inputTypeName) - { - return supportedType; - } - } - - // Try to load the type directly (for types not in supported types) - Type? resolvedType = Type.GetType(inputTypeName); - if (resolvedType is not null) - { - return resolvedType; - } + return supportedTypes.FirstOrDefault() ?? typeof(string); } - // Fall back to first supported type or string - return supportedTypes.FirstOrDefault() ?? typeof(string); + // Try to find a matching type in the supported types + Type? matchedType = supportedTypes.FirstOrDefault(t => + t.AssemblyQualifiedName == inputTypeName || + t.FullName == inputTypeName || + t.Name == inputTypeName); + + if (matchedType is not null) + { + return matchedType; + } + + // Try to load the type directly (for types not in supported types) + return Type.GetType(inputTypeName) ?? supportedTypes.FirstOrDefault() ?? typeof(string); } - [UnconditionalSuppressMessage("Trimming", "IL2026:RequiresUnreferencedCode", Justification = "Deserializing known wrapper type.")] - [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "Deserializing known wrapper type.")] + /// + /// Attempts to deserialize activity input using source-generated serialization. + /// private static DurableActivityInput? TryDeserializeActivityInput(string input) { try { - return JsonSerializer.Deserialize(input); + return JsonSerializer.Deserialize(input, DurableWorkflowJsonContext.Default.DurableActivityInput); } catch (JsonException) { @@ -274,17 +250,24 @@ public static class DurableWorkflowServiceCollectionExtensions } } - [UnconditionalSuppressMessage("Trimming", "IL2026:RequiresUnreferencedCode", Justification = "Serializing workflow event types.")] - [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "Serializing workflow event types.")] + /// + /// Serializes a workflow event with type information. + /// + /// + /// The event data itself uses reflection-based serialization since event types + /// are user-defined, but the wrapper uses source generation. + /// + [UnconditionalSuppressMessage("Trimming", "IL2026:RequiresUnreferencedCode", Justification = "Event data uses reflection for user types.")] + [UnconditionalSuppressMessage("AOT", "IL3050:RequiresDynamicCode", Justification = "Event data uses reflection for user types.")] private static string SerializeEvent(WorkflowEvent evt) { // Serialize with type information so we can deserialize to the correct type later - DurableWorkflowRunner.SerializedWorkflowEvent wrapper = new() + SerializedWorkflowEvent wrapper = new() { TypeName = evt.GetType().AssemblyQualifiedName, Data = JsonSerializer.Serialize(evt, evt.GetType()) }; - return JsonSerializer.Serialize(wrapper); + return JsonSerializer.Serialize(wrapper, DurableWorkflowJsonContext.Default.SerializedWorkflowEvent); } [UnconditionalSuppressMessage("Trimming", "IL2026:RequiresUnreferencedCode", Justification = "Deserializing workflow types registered at startup.")] @@ -346,13 +329,8 @@ public static class DurableWorkflowServiceCollectionExtensions return JsonSerializer.Deserialize(data, DurableAgentStateJsonContext.Default.DurableAgentState); } - JsonTypeInfo? typeInfo = s_options.GetTypeInfo(targetType); - if (typeInfo is JsonTypeInfo typedInfo) - { - return JsonSerializer.Deserialize(data, typedInfo); - } - - return JsonSerializer.Deserialize(data, targetType, s_options); + return TryDeserializeWithTypeInfo(data, targetType) + ?? JsonSerializer.Deserialize(data, targetType, s_options); } [return: NotNullIfNotNull(nameof(value))] @@ -370,13 +348,20 @@ public static class DurableWorkflowServiceCollectionExtensions return JsonSerializer.Serialize(durableAgentState, DurableAgentStateJsonContext.Default.DurableAgentState); } - JsonTypeInfo? typeInfo = s_options.GetTypeInfo(value.GetType()); - if (typeInfo is JsonTypeInfo typedInfo) - { - return JsonSerializer.Serialize(value, typedInfo); - } + return TrySerializeWithTypeInfo(value) + ?? JsonSerializer.Serialize(value, s_options); + } - return JsonSerializer.Serialize(value, s_options); + private static object? TryDeserializeWithTypeInfo(string data, Type targetType) + { + JsonTypeInfo? typeInfo = s_options.GetTypeInfo(targetType); + return typeInfo is not null ? JsonSerializer.Deserialize(data, typeInfo) : null; + } + + private static string? TrySerializeWithTypeInfo(object value) + { + JsonTypeInfo? typeInfo = s_options.GetTypeInfo(value.GetType()); + return typeInfo is not null ? JsonSerializer.Serialize(value, typeInfo) : null; } } } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs index 50c95140ab..a520d59e7f 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Logs.cs @@ -103,31 +103,31 @@ internal static partial class Logs [LoggerMessage( EventId = 12, - Level = LogLevel.Debug, + Level = LogLevel.Information, Message = "Attempting to run workflow: {WorkflowName}")] public static partial void LogAttemptingToRunWorkflow(this ILogger logger, string workflowName); [LoggerMessage( EventId = 13, - Level = LogLevel.Debug, + Level = LogLevel.Warning, Message = "Running workflow: {WorkflowName}")] public static partial void LogRunningWorkflow(this ILogger logger, string? workflowName); [LoggerMessage( EventId = 14, - Level = LogLevel.Debug, + Level = LogLevel.Warning, Message = "Attempting to execute activity in workflow '{WorkflowName}' for executor '{ExecutorName}'")] public static partial void LogAttemptingToExecuteActivity(this ILogger logger, string workflowName, string executorName); [LoggerMessage( EventId = 15, - Level = LogLevel.Debug, + Level = LogLevel.Warning, Message = "Executing activity for executor '{ExecutorId}' of type '{ExecutorType}'")] public static partial void LogExecutingActivity(this ILogger logger, string executorId, string executorType); [LoggerMessage( EventId = 16, - Level = LogLevel.Debug, + Level = LogLevel.Warning, Message = "Activity executed for executor '{ExecutorId}' with result: {Result}")] public static partial void LogActivityExecuted(this ILogger logger, string executorId, string result); diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs index ab209b54c1..f3d4496074 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.AzureFunctions/DurableWorkflowRunner.cs @@ -107,9 +107,7 @@ internal sealed class FunctionsWorkflowRunner : DurableWorkflowRunner private static string SerializeEvent(WorkflowEvent evt) { // Serialize with type information so we can deserialize to the correct type later -#pragma warning disable IDE0001 // Simplify name - cannot simplify cross-assembly reference - Microsoft.Agents.AI.DurableTask.DurableWorkflowRunner.SerializedWorkflowEvent wrapper = new() -#pragma warning restore IDE0001 + Microsoft.Agents.AI.DurableTask.SerializedWorkflowEvent wrapper = new() { TypeName = evt.GetType().AssemblyQualifiedName, Data = JsonSerializer.Serialize(evt, evt.GetType())