// 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; /// /// Routes messages from a source executor to a single target executor with optional condition evaluation. /// /// /// /// Created by during construction — one instance per (source, sink) edge. /// When an edge has a condition (e.g., order => order.Total > 1000), 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 false, the /// message is not forwarded and the target executor will not run for this edge. /// /// /// For sources with multiple successors, individual instances /// are wrapped in a so a single RouteMessage call /// fans the same message out to all targets, each evaluating its own condition independently. /// /// internal sealed class DurableDirectEdgeRouter : IDurableEdgeRouter { private readonly string _sourceId; private readonly string _sinkId; private readonly Func? _condition; private readonly Type? _sourceOutputType; /// /// Initializes a new instance of . /// /// The source executor ID. /// The target executor ID. /// Optional condition function to evaluate before routing. /// The output type of the source executor for deserialization. internal DurableDirectEdgeRouter( string sourceId, string sinkId, Func? condition, Type? sourceOutputType) { this._sourceId = sourceId; this._sinkId = sinkId; this._condition = condition; this._sourceOutputType = sourceOutputType; } /// public void RouteMessage( DurableMessageEnvelope envelope, Dictionary> 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); } /// /// Deserializes a JSON message to an object for condition evaluation. /// /// /// 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. /// /// The JSON string representation of the message. /// /// The expected type of the message. When provided, enables strongly-typed deserialization /// so the condition function receives the correct type to evaluate against. /// /// /// The deserialized object, or null if the JSON is empty. /// /// Thrown when the JSON is invalid or cannot be deserialized to the target type. [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(json, DurableSerialization.Options) : JsonSerializer.Deserialize(json, targetType, DurableSerialization.Options); } private static void EnqueueMessage( Dictionary> queues, string executorId, DurableMessageEnvelope envelope) { if (!queues.TryGetValue(executorId, out Queue? queue)) { queue = new Queue(); queues[executorId] = queue; } queue.Enqueue(envelope); } }