mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
.NET: Add durable workflow support (#4436)
* .NET: [Feature Branch] Add basic durable workflow support (#3648) * Add basic durable workflow support. * PR feedback fixes * Add conditional edge sample. * PR feedback fixes. * Minor cleanup. * Minor cleanup * Minor formatting improvements. * Improve comments/documentation on the execution flow. * .NET: [Feature Branch] Add Azure Functions hosting support for durable workflows (#3935) * Adding azure functions workflow support. * - PR feedback fixes. - Add example to demonstrate complex Object as payload. * rename instanceId to runId. * Use custom ITaskOrchestrator to run orchestrator function. * .NET: [Feature Branch] Adding support for events & shared state in durable workflows (#4020) * Adding support for events & shared state in durable workflows. * PR feedback fixes * PR feedback fixes. * Add YieldOutputAsync calls to 05_WorkflowEvents sample executors The integration test asserts that WorkflowOutputEvent is found in the stream, but the sample executors only used AddEventAsync for custom events and never called YieldOutputAsync. Since WorkflowOutputEvent is only emitted via explicit YieldOutputAsync calls, the assertion would fail. Added YieldOutputAsync to each executor to match the test expectation and demonstrate the API in the sample. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Fix deserialization to use shared serializer options. * PR feedback updates. * Sample cleanup * PR feedback fixes * Addressing PR review feedback for DurableStreamingWorkflowRun - Use -1 instead of 0 for taskId in TaskFailedException when task ID is not relevant. - Add [NotNullWhen(true)] to TryParseWorkflowResult out parameter following .NET TryXXX conventions. --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * .NET: [Feature Branch] Add nested sub-workflow support for durable workflows (#4190) * .NET: [Feature Branch] Add nested sub-workflow support for durable workflows * fix readme path * Switch Orchestration output from string to DurableWorkflowResult. * PR feedback fixes * Minor cleanup based on PR feedback. * .NET: [Feature Branch] Add Human In the Loop support for durable workflows (#4358) * Add Azure Functions HITL workflow sample Add 06_WorkflowHITL Azure Functions sample demonstrating Human-in-the-Loop workflow support with HTTP endpoints for status checking and approval responses. The sample includes: - ExpenseReimbursement workflow with RequestPort for manager approval - Custom HTTP endpoint to check workflow status and pending approvals - Custom HTTP endpoint to send approval responses via RaiseEventAsync - demo.http file with step-by-step interaction examples * PR feedback fixes * Minor comment cleanup * Minor comment clReverted the `!context.IsReplaying` guards on `PendingEvents.Add`/`RemoveAll` and `SetCustomStatus` in `ExecuteRequestPortAsync`. The guards broke fan-out scenarios where parallel RequestPorts need to be discoverable after replay. `SetCustomStatus` is idempotent metadata that doesn't affect replay determinism.eanup * fix for PR feedback * PR feedback updates * Improvements to samples * Improvements to README * Update samples to use parallel request ports. * Unit tests * Introduce local variables to improve readability of Workflows.Workflows access patter * Use GitHub-style callouts and add PowerShell command variants in HITL sample README * Add changelog entries for durable workflow support (#4436) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Bump Microsoft.DurableTask.Worker to 1.19.1 to fix version downgrade Microsoft.Azure.Functions.Worker.Extensions.DurableTask 1.13.1 requires Microsoft.DurableTask.Worker >= 1.19.1 via its transitive dependency on Microsoft.DurableTask.Worker.Grpc 1.19.1. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Fix broken markdown links in durable workflow sample READMEs - Create Workflow/README.md with environment setup docs - Fix ../README.md -> ../../README.md in ConsoleApps 01, 02, 03, 08 - Fix SubWorkflows relative path (3 levels -> 4 levels up) - Fix dead Durable Task Scheduler URL Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Fix build errors from main merge: Throw conflict, ExecuteAsync rename, GetNewSessionAsync rename - Remove InjectSharedThrow from DurableTask csproj (uses Workflows' internal Throw via InternalsVisibleTo) - Update ExecuteAsync -> ExecuteCoreAsync with WorkflowTelemetryContext.Disabled - Update GetNewSessionAsync -> CreateSessionAsync Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Move durable workflow samples to 04-hosting/DurableWorkflows Aligns with main branch sample reorganization where durable samples live under 04-hosting/ (alongside DurableAgents/). - Move samples/Durable/Workflow/ -> samples/04-hosting/DurableWorkflows/ - Add Directory.Build.props matching DurableAgents pattern - Update slnx project paths - Update integration test sample paths - Update README cd paths and cross-references Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Fix build errors: remove duplicate base class members, update renamed APIs - Remove duplicate OutputLog, WriteInputAsync, CreateTestTimeoutCts, etc. from ConsoleAppSamplesValidation (already in SamplesValidationBase) - Update AddFanInEdge -> AddFanInBarrierEdge in workflow samples - Update GetNewSessionAsync -> CreateSessionAsync in workflow samples - Update SourceId -> ExecutorId (obsolete) in workflow samples Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Fix dotnet format issues: add UTF-8 BOM and remove unused using - Add UTF-8 BOM to 20 .cs files across DurableTask, AzureFunctions, unit tests, and workflow samples - Remove unnecessary using directive in 07_SubWorkflows/Executors.cs Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Fix typo PaymentProcesser -> PaymentProcessor and garbled arrows in README Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Fix GetExecutorName to handle agent names with underscores Split on last underscore instead of first, and validate that the suffix is a 32-char hex string (sanitized GUID) before stripping it. This prevents truncation of agent names like 'my_agent' when the executor ID is 'my_agent_<guid>'. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Align DurableTask.Client.AzureManaged to 1.19.1 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Bump DurableTask and Azure Functions extension package versions - DurableTask.* packages: 1.19.1 -> 1.22.0 - Functions.Worker.Extensions.DurableTask: 1.13.1 -> 1.16.0 - Functions.Worker.Extensions.DurableTask.AzureManaged: 1.0.1 -> 1.5.0 (telemetry bug fix) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Bump DurableTask SDK packages to 1.22.0 - DurableTask.Client: 1.19.1 -> 1.22.0 - DurableTask.Client.AzureManaged: 1.19.1 -> 1.22.0 - DurableTask.Worker: 1.19.1 -> 1.22.0 - DurableTask.Worker.AzureManaged: 1.19.1 -> 1.22.0 - Azure Functions extensions kept at original versions (1.13.1/1.0.1) due to host-side DurableTask.Core 3.7.0 incompatibility with newer extensions Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * Update Microsoft.Azure.Functions.Worker.Extensions.DurableTask to "1.16.0" * Add the local.settings.json files to the sample which were previously ignored. This aligns with our other samples. * Increase timeout for tests as CI has them failing transiently. * increaset timeout value for azure functions integration tests. * Add YieldsOutput(string) to workflow shared state sample executors ValidateOrder and EnrichOrder call YieldOutputAsync with string messages, but only their TOutput (OrderDetails) was in the allowed yield types. This caused TargetInvocationException in the WorkflowSharedState sample validation integration test. * Downgrade the durable packages to 1.18.0 * Downgrading Worker.Extensions.DurableTask to 1.12.1 --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
Shyju Krishnankutty
co-authored by
Copilot
Copilot
parent
0fdcfd0f4c
commit
cbcdb2d29e
+156
@@ -0,0 +1,156 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
// Routing decision flow for a single edge.
|
||||
// Example: the B→D edge from a workflow like below:
|
||||
//
|
||||
// [A] ──► [B] ──► [C] ──► [E] (B→D has condition: x => x.NeedsReview)
|
||||
// │ ▲
|
||||
// └──► [D] ──────┘
|
||||
//
|
||||
// (condition: x => x.NeedsReview, _sourceOutputType: typeof(Order))
|
||||
//
|
||||
// RouteMessage(envelope) envelope.Message = "{\"NeedsReview\":true, ...}"
|
||||
// │
|
||||
// ▼
|
||||
// Has condition? ──── No ────► Enqueue to sink's queue
|
||||
// │
|
||||
// Yes (B→D has one)
|
||||
// │
|
||||
// ▼
|
||||
// Deserialize message JSON string → Order object using _sourceOutputType
|
||||
// │
|
||||
// ▼
|
||||
// Evaluate _condition(order) order => order.NeedsReview
|
||||
// │
|
||||
// ┌──┴──┐
|
||||
// true false
|
||||
// │ │
|
||||
// ▼ └──► Skip (log and return, D will not run)
|
||||
// Enqueue to
|
||||
// D's queue
|
||||
|
||||
using System.Diagnostics.CodeAnalysis;
|
||||
using System.Text.Json;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Microsoft.Agents.AI.DurableTask.Workflows.EdgeRouters;
|
||||
|
||||
/// <summary>
|
||||
/// Routes messages from a source executor to a single target executor with optional condition evaluation.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Created by <see cref="DurableEdgeMap"/> during construction — one instance per (source, sink) edge.
|
||||
/// When an edge has a condition (e.g., <c>order => order.Total > 1000</c>), the router deserialises
|
||||
/// the serialised JSON message back to the source executor's output type so the condition delegate
|
||||
/// can evaluate it against strongly-typed properties. If the condition returns <c>false</c>, the
|
||||
/// message is not forwarded and the target executor will not run for this edge.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// For sources with multiple successors, individual <see cref="DurableDirectEdgeRouter"/> instances
|
||||
/// are wrapped in a <see cref="DurableFanOutEdgeRouter"/> so a single <c>RouteMessage</c> call
|
||||
/// fans the same message out to all targets, each evaluating its own condition independently.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
internal sealed class DurableDirectEdgeRouter : IDurableEdgeRouter
|
||||
{
|
||||
private readonly string _sourceId;
|
||||
private readonly string _sinkId;
|
||||
private readonly Func<object?, bool>? _condition;
|
||||
private readonly Type? _sourceOutputType;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of <see cref="DurableDirectEdgeRouter"/>.
|
||||
/// </summary>
|
||||
/// <param name="sourceId">The source executor ID.</param>
|
||||
/// <param name="sinkId">The target executor ID.</param>
|
||||
/// <param name="condition">Optional condition function to evaluate before routing.</param>
|
||||
/// <param name="sourceOutputType">The output type of the source executor for deserialization.</param>
|
||||
internal DurableDirectEdgeRouter(
|
||||
string sourceId,
|
||||
string sinkId,
|
||||
Func<object?, bool>? condition,
|
||||
Type? sourceOutputType)
|
||||
{
|
||||
this._sourceId = sourceId;
|
||||
this._sinkId = sinkId;
|
||||
this._condition = condition;
|
||||
this._sourceOutputType = sourceOutputType;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void RouteMessage(
|
||||
DurableMessageEnvelope envelope,
|
||||
Dictionary<string, Queue<DurableMessageEnvelope>> messageQueues,
|
||||
ILogger logger)
|
||||
{
|
||||
if (this._condition is not null)
|
||||
{
|
||||
try
|
||||
{
|
||||
object? messageObj = DeserializeForCondition(envelope.Message, this._sourceOutputType);
|
||||
if (!this._condition(messageObj))
|
||||
{
|
||||
logger.LogEdgeConditionFalse(this._sourceId, this._sinkId);
|
||||
return;
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogEdgeConditionEvaluationFailed(ex, this._sourceId, this._sinkId);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
logger.LogEdgeRoutingMessage(this._sourceId, this._sinkId);
|
||||
EnqueueMessage(messageQueues, this._sinkId, envelope);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Deserializes a JSON message to an object for condition evaluation.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Messages travel through the durable workflow as serialized JSON strings, but condition
|
||||
/// delegates need typed objects to evaluate (e.g., order => order.Status == "Approved").
|
||||
/// This method converts the JSON back to an object the condition delegate can evaluate.
|
||||
/// </remarks>
|
||||
/// <param name="json">The JSON string representation of the message.</param>
|
||||
/// <param name="targetType">
|
||||
/// The expected type of the message. When provided, enables strongly-typed deserialization
|
||||
/// so the condition function receives the correct type to evaluate against.
|
||||
/// </param>
|
||||
/// <returns>
|
||||
/// The deserialized object, or null if the JSON is empty.
|
||||
/// </returns>
|
||||
/// <exception cref="JsonException">Thrown when the JSON is invalid or cannot be deserialized to the target type.</exception>
|
||||
[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)
|
||||
{
|
||||
if (string.IsNullOrEmpty(json))
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
// If we know the source executor's output type, deserialize to that specific type
|
||||
// so the condition function can access strongly-typed properties.
|
||||
// Otherwise, deserialize as a generic object for basic inspection.
|
||||
return targetType is null
|
||||
? JsonSerializer.Deserialize<object>(json, DurableSerialization.Options)
|
||||
: JsonSerializer.Deserialize(json, targetType, DurableSerialization.Options);
|
||||
}
|
||||
|
||||
private static void EnqueueMessage(
|
||||
Dictionary<string, Queue<DurableMessageEnvelope>> queues,
|
||||
string executorId,
|
||||
DurableMessageEnvelope envelope)
|
||||
{
|
||||
if (!queues.TryGetValue(executorId, out Queue<DurableMessageEnvelope>? queue))
|
||||
{
|
||||
queue = new Queue<DurableMessageEnvelope>();
|
||||
queues[executorId] = queue;
|
||||
}
|
||||
|
||||
queue.Enqueue(envelope);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,205 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
// How WorkflowGraphInfo maps to DurableEdgeMap at runtime.
|
||||
// For a workflow like below:
|
||||
//
|
||||
// [A] ──► [B] ──► [C] ──► [E]
|
||||
// │ ▲
|
||||
// └──► [D] ──────┘
|
||||
// (condition: x => x.NeedsReview)
|
||||
//
|
||||
// WorkflowGraphInfo DurableEdgeMap
|
||||
// ┌──────────────────────────┐ ┌──────────────────────────────────────┐
|
||||
// │ Successors: │ │ _routersBySource: │
|
||||
// │ A → [B] │──constructs──►│ A → [DirectRouter(A→B)] │
|
||||
// │ B → [C, D] │ │ B → [FanOutRouter([C, D])] │
|
||||
// │ C → [E] │ │ C → [DirectRouter(C→E)] │
|
||||
// │ D → [E] │ │ D → [DirectRouter(D→E)] │
|
||||
// └──────────────────────────┘ │ │
|
||||
// ┌──────────────────────────┐ │ _predecessorCounts: │
|
||||
// │ Predecessors: │ │ A → 0 │
|
||||
// │ E → [C, D] (fan-in!) │──constructs──►│ B → 1, C → 1, D → 1 │
|
||||
// └──────────────────────────┘ │ E → 2 ◄── IsFanInExecutor = true │
|
||||
// └──────────────────────────────────────┘
|
||||
//
|
||||
// Usage during superstep execution (continuing the example):
|
||||
//
|
||||
// 1. EnqueueInitialInput(msg) ──► MessageQueues["A"].Enqueue(envelope)
|
||||
//
|
||||
// 2. After B completes, RouteMessage("B", resultB) ──► _routersBySource["B"]
|
||||
// │
|
||||
// ▼
|
||||
// FanOutRouter (B has 2 successors)
|
||||
// ├─► DirectRouter(B→C) ──► no condition ──► enqueue to C
|
||||
// └─► DirectRouter(B→D) ──► evaluate x => x.NeedsReview ──► enqueue to D (or skip)
|
||||
//
|
||||
// 3. Before superstep 4, IsFanInExecutor("E") returns true (count=2)
|
||||
// → CollectExecutorInputs aggregates C and D results into ["resultC","resultD"]
|
||||
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Microsoft.Agents.AI.DurableTask.Workflows.EdgeRouters;
|
||||
|
||||
/// <summary>
|
||||
/// Manages message routing through workflow edges for durable orchestrations.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// This is the durable equivalent of <c>EdgeMap</c> in the in-process runner.
|
||||
/// It is constructed from <see cref="WorkflowGraphInfo"/> (produced by <see cref="WorkflowAnalyzer.BuildGraphInfo"/>)
|
||||
/// and converts the static graph structure into an active routing layer used during superstep execution.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>What it stores:</b>
|
||||
/// </para>
|
||||
/// <list type="bullet">
|
||||
/// <item><description><c>_routersBySource</c> — For each source executor, a list of <see cref="IDurableEdgeRouter"/> instances
|
||||
/// that know how to deliver messages to successor executors. When a source has multiple successors, a single
|
||||
/// <see cref="DurableFanOutEdgeRouter"/> wraps the individual <see cref="DurableDirectEdgeRouter"/> instances.</description></item>
|
||||
/// <item><description><c>_predecessorCounts</c> — The number of predecessors for each executor, used to detect
|
||||
/// fan-in points where multiple incoming messages should be aggregated before execution.</description></item>
|
||||
/// <item><description><c>_startExecutorId</c> — The entry-point executor that receives the initial workflow input.</description></item>
|
||||
/// </list>
|
||||
/// <para>
|
||||
/// <b>How it is used during execution:</b>
|
||||
/// </para>
|
||||
/// <list type="number">
|
||||
/// <item><description><see cref="EnqueueInitialInput"/> seeds the start executor's queue before the first superstep.</description></item>
|
||||
/// <item><description>After each superstep, <c>DurableWorkflowRunner.RouteOutputToSuccessors</c> calls
|
||||
/// <see cref="RouteMessage"/> which looks up the routers for the completed executor and forwards the
|
||||
/// result to successor queues. Each router may evaluate an edge condition before enqueueing.</description></item>
|
||||
/// <item><description><see cref="IsFanInExecutor"/> is checked during input collection to decide whether
|
||||
/// to aggregate multiple queued messages into a single JSON array before dispatching.</description></item>
|
||||
/// </list>
|
||||
/// </remarks>
|
||||
internal sealed class DurableEdgeMap
|
||||
{
|
||||
private readonly Dictionary<string, List<IDurableEdgeRouter>> _routersBySource = [];
|
||||
private readonly Dictionary<string, int> _predecessorCounts = [];
|
||||
private readonly string _startExecutorId;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of <see cref="DurableEdgeMap"/> from workflow graph info.
|
||||
/// </summary>
|
||||
/// <param name="graphInfo">The workflow graph information containing routing structure.</param>
|
||||
internal DurableEdgeMap(WorkflowGraphInfo graphInfo)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(graphInfo);
|
||||
|
||||
this._startExecutorId = graphInfo.StartExecutorId;
|
||||
|
||||
// Build edge routers for each source executor
|
||||
foreach (KeyValuePair<string, List<string>> entry in graphInfo.Successors)
|
||||
{
|
||||
string sourceId = entry.Key;
|
||||
List<string> successorIds = entry.Value;
|
||||
|
||||
if (successorIds.Count == 0)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
graphInfo.ExecutorOutputTypes.TryGetValue(sourceId, out Type? sourceOutputType);
|
||||
|
||||
List<IDurableEdgeRouter> routers = [];
|
||||
foreach (string sinkId in successorIds)
|
||||
{
|
||||
graphInfo.EdgeConditions.TryGetValue((sourceId, sinkId), out Func<object?, bool>? condition);
|
||||
|
||||
routers.Add(new DurableDirectEdgeRouter(sourceId, sinkId, condition, sourceOutputType));
|
||||
}
|
||||
|
||||
// If multiple successors, wrap in a fan-out router
|
||||
if (routers.Count > 1)
|
||||
{
|
||||
this._routersBySource[sourceId] = [new DurableFanOutEdgeRouter(sourceId, routers)];
|
||||
}
|
||||
else
|
||||
{
|
||||
this._routersBySource[sourceId] = routers;
|
||||
}
|
||||
}
|
||||
|
||||
// Store predecessor counts for fan-in detection
|
||||
foreach (KeyValuePair<string, List<string>> entry in graphInfo.Predecessors)
|
||||
{
|
||||
this._predecessorCounts[entry.Key] = entry.Value.Count;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Routes a message from a source executor to its successors.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Called by <c>DurableWorkflowRunner.RouteOutputToSuccessors</c> after each superstep.
|
||||
/// Wraps the message in a <see cref="DurableMessageEnvelope"/> and delegates to the
|
||||
/// appropriate <see cref="IDurableEdgeRouter"/>(s) for the source executor. Each router
|
||||
/// may evaluate an edge condition and, if satisfied, enqueue the envelope into the
|
||||
/// target executor's message queue for the next superstep.
|
||||
/// </remarks>
|
||||
/// <param name="sourceId">The source executor ID.</param>
|
||||
/// <param name="message">The serialized message to route.</param>
|
||||
/// <param name="inputTypeName">The type name of the message.</param>
|
||||
/// <param name="messageQueues">The message queues to enqueue messages into.</param>
|
||||
/// <param name="logger">The logger for tracing.</param>
|
||||
internal void RouteMessage(
|
||||
string sourceId,
|
||||
string message,
|
||||
string? inputTypeName,
|
||||
Dictionary<string, Queue<DurableMessageEnvelope>> messageQueues,
|
||||
ILogger logger)
|
||||
{
|
||||
if (!this._routersBySource.TryGetValue(sourceId, out List<IDurableEdgeRouter>? routers))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
DurableMessageEnvelope envelope = DurableMessageEnvelope.Create(message, inputTypeName, sourceId);
|
||||
|
||||
foreach (IDurableEdgeRouter router in routers)
|
||||
{
|
||||
router.RouteMessage(envelope, messageQueues, logger);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Enqueues the initial workflow input to the start executor.
|
||||
/// </summary>
|
||||
/// <param name="message">The serialized initial input message.</param>
|
||||
/// <param name="messageQueues">The message queues to enqueue into.</param>
|
||||
/// <remarks>
|
||||
/// This method is used only at workflow startup to provide input to the first executor.
|
||||
/// No input type hint is required because the start executor determines its expected input type from its own <c>InputTypes</c> configuration.
|
||||
/// </remarks>
|
||||
internal void EnqueueInitialInput(
|
||||
string message,
|
||||
Dictionary<string, Queue<DurableMessageEnvelope>> messageQueues)
|
||||
{
|
||||
DurableMessageEnvelope envelope = DurableMessageEnvelope.Create(message, inputTypeName: null);
|
||||
EnqueueMessage(messageQueues, this._startExecutorId, envelope);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Determines if an executor is a fan-in point (has multiple predecessors).
|
||||
/// </summary>
|
||||
/// <param name="executorId">The executor ID to check.</param>
|
||||
/// <returns><c>true</c> if the executor has multiple predecessors; otherwise, <c>false</c>.</returns>
|
||||
internal bool IsFanInExecutor(string executorId)
|
||||
{
|
||||
return this._predecessorCounts.TryGetValue(executorId, out int count) && count > 1;
|
||||
}
|
||||
|
||||
private static void EnqueueMessage(
|
||||
Dictionary<string, Queue<DurableMessageEnvelope>> queues,
|
||||
string executorId,
|
||||
DurableMessageEnvelope envelope)
|
||||
{
|
||||
if (!queues.TryGetValue(executorId, out Queue<DurableMessageEnvelope>? queue))
|
||||
{
|
||||
queue = new Queue<DurableMessageEnvelope>();
|
||||
queues[executorId] = queue;
|
||||
}
|
||||
|
||||
queue.Enqueue(envelope);
|
||||
}
|
||||
}
|
||||
+67
@@ -0,0 +1,67 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
// Fan-out routing: one source message is forwarded to multiple targets.
|
||||
// Example from a workflow like below:
|
||||
//
|
||||
// [A] ──► [B] ──► [C] ──► [E] (B→D has condition: x => x.NeedsReview)
|
||||
// │ ▲
|
||||
// └──► [D] ──────┘
|
||||
//
|
||||
// B has two successors (C and D), so DurableEdgeMap wraps them:
|
||||
//
|
||||
// Executor B completes with resultB (type: Order)
|
||||
// │
|
||||
// ▼
|
||||
// FanOutRouter(B)
|
||||
// ├──► DirectRouter(B→C) ──► no condition ──► enqueue to C
|
||||
// └──► DirectRouter(B→D) ──► x => x.NeedsReview ──► enqueue to D (or skip)
|
||||
//
|
||||
// Each DirectRouter independently evaluates its condition,
|
||||
// so resultB always reaches C, but only reaches D if NeedsReview is true.
|
||||
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Microsoft.Agents.AI.DurableTask.Workflows.EdgeRouters;
|
||||
|
||||
/// <summary>
|
||||
/// Routes messages from a source executor to multiple target executors (fan-out pattern).
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Created by <see cref="DurableEdgeMap"/> when a source executor has more than one successor.
|
||||
/// Wraps the individual <see cref="DurableDirectEdgeRouter"/> instances and delegates
|
||||
/// <see cref="RouteMessage"/> to each of them, so the same message is evaluated and
|
||||
/// potentially enqueued for every target independently.
|
||||
/// </remarks>
|
||||
internal sealed class DurableFanOutEdgeRouter : IDurableEdgeRouter
|
||||
{
|
||||
private readonly string _sourceId;
|
||||
private readonly List<IDurableEdgeRouter> _targetRouters;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of <see cref="DurableFanOutEdgeRouter"/>.
|
||||
/// </summary>
|
||||
/// <param name="sourceId">The source executor ID.</param>
|
||||
/// <param name="targetRouters">The routers for each target executor.</param>
|
||||
internal DurableFanOutEdgeRouter(string sourceId, List<IDurableEdgeRouter> targetRouters)
|
||||
{
|
||||
this._sourceId = sourceId;
|
||||
this._targetRouters = targetRouters;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void RouteMessage(
|
||||
DurableMessageEnvelope envelope,
|
||||
Dictionary<string, Queue<DurableMessageEnvelope>> messageQueues,
|
||||
ILogger logger)
|
||||
{
|
||||
if (logger.IsEnabled(LogLevel.Debug))
|
||||
{
|
||||
logger.LogDebug("Fan-Out from {Source}: routing to {Count} targets", this._sourceId, this._targetRouters.Count);
|
||||
}
|
||||
|
||||
foreach (IDurableEdgeRouter targetRouter in this._targetRouters)
|
||||
{
|
||||
targetRouter.RouteMessage(envelope, messageQueues, logger);
|
||||
}
|
||||
}
|
||||
}
|
||||
+26
@@ -0,0 +1,26 @@
|
||||
// Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Microsoft.Agents.AI.DurableTask.Workflows.EdgeRouters;
|
||||
|
||||
/// <summary>
|
||||
/// Defines the contract for routing messages through workflow edges in durable orchestrations.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// Implementations include <see cref="DurableDirectEdgeRouter"/> for single-target routing
|
||||
/// and <see cref="DurableFanOutEdgeRouter"/> for multi-target fan-out patterns.
|
||||
/// </remarks>
|
||||
internal interface IDurableEdgeRouter
|
||||
{
|
||||
/// <summary>
|
||||
/// Routes a message from the source executor to its target(s).
|
||||
/// </summary>
|
||||
/// <param name="envelope">The message envelope containing the message and metadata.</param>
|
||||
/// <param name="messageQueues">The message queues to enqueue messages into.</param>
|
||||
/// <param name="logger">The logger for tracing.</param>
|
||||
void RouteMessage(
|
||||
DurableMessageEnvelope envelope,
|
||||
Dictionary<string, Queue<DurableMessageEnvelope>> messageQueues,
|
||||
ILogger logger);
|
||||
}
|
||||
Reference in New Issue
Block a user