Add OpenTelemetry observability instrumentation to DurableTask workflows

- Add centralized DurableTaskInstrumentation class with shared ActivitySource
- Add workflow.run span to DurableWorkflowRunner (emitted after superstep loop to avoid replay semantics)
- Add edge_group.process spans to DurableDirectEdgeRouter and DurableFanOutEdgeRouter
- Add message.send span to DurableWorkflowContext
- Add TraceContext propagation support to DurableWorkflowContext
- Add 09_Observability sample for durable workflows with Azure Monitor + OTLP export
- Update CHANGELOG.md

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableExecutorDispatcher.cs

@@ -43,6 +43,9 @@ internal static class DurableExecutorDispatcher
     {
         logger.LogDispatchingExecutor(executorInfo.ExecutorId, executorInfo.IsAgenticExecutor);
 
+        // Note: executor.process telemetry is emitted by core Executor.ExecuteAsync inside the
+        // activity worker. We don't duplicate it here because orchestration replay semantics
+        // make wrapping spans unreliable (context.IsReplaying is true at method entry).
         if (executorInfo.IsAgenticExecutor)
         {
             return await ExecuteAgentAsync(context, executorInfo, logger, envelope.Message).ConfigureAwait(true);

dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableRunStatus.cs

@@ -1,4 +1,4 @@
-// Copyright (c) Microsoft. All rights reserved.
+// Copyright (c) Microsoft. All rights reserved.
 
 namespace Microsoft.Agents.AI.DurableTask.Workflows;
 

dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowContext.cs

@@ -1,4 +1,4 @@
-// Copyright (c) Microsoft. All rights reserved.
+// Copyright (c) Microsoft. All rights reserved.
 
 using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
@@ -33,10 +33,12 @@ internal sealed class DurableWorkflowContext : IWorkflowContext
     /// </summary>
     /// <param name="initialState">The shared state passed from the orchestration.</param>
     /// <param name="executor">The executor running in this context.</param>
-    internal DurableWorkflowContext(Dictionary<string, string>? initialState, Executor executor)
+    /// <param name="traceContext">Optional trace context for correlation.</param>
+    internal DurableWorkflowContext(Dictionary<string, string>? initialState, Executor executor, IReadOnlyDictionary<string, string>? traceContext = null)
     {
         this._executor = executor;
         this._initialState = initialState ?? [];
+        this.TraceContext = traceContext;
     }
 
     /// <summary>
@@ -87,6 +89,8 @@ internal sealed class DurableWorkflowContext : IWorkflowContext
     {
         if (message is not null)
         {
+            using Activity? activity = DurableWorkflowInstrumentation.ActivitySource.StartActivity("message.send", ActivityKind.Producer);
+
             Type messageType = message.GetType();
             this.SentMessages.Add(new TypedPayload
             {
@@ -269,7 +273,7 @@ internal sealed class DurableWorkflowContext : IWorkflowContext
     }
 
     /// <inheritdoc/>
-    public IReadOnlyDictionary<string, string>? TraceContext => null;
+    public IReadOnlyDictionary<string, string>? TraceContext { get; }
 
     /// <inheritdoc/>
     public bool ConcurrentRunsEnabled => false;

dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowInstrumentation.cs

@@ -0,0 +1,16 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Diagnostics;
+
+namespace Microsoft.Agents.AI.DurableTask.Workflows;
+
+/// <summary>
+/// Provides centralized OpenTelemetry instrumentation for durable workflow execution.
+/// </summary>
+internal static class DurableWorkflowInstrumentation
+{
+    /// <summary>
+    /// The shared <see cref="ActivitySource"/> used by all durable workflow components.
+    /// </summary>
+    internal static readonly ActivitySource ActivitySource = new("Microsoft.Agents.AI.DurableTask.Workflows");
+}

dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs

@@ -44,6 +44,7 @@
 //  Superstep 5 — loop exits (no pending messages)
 //    GetFinalResult returns resultE
 
+using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
 using System.Text.Json;
 using Microsoft.Agents.AI.DurableTask.Workflows.EdgeRouters;
@@ -115,7 +116,23 @@ internal sealed class DurableWorkflowRunner
         // Extract input - the start executor determines the expected input type from its own InputTypes
         object input = workflowInput.Input;
 
-        return await RunSuperstepLoopAsync(context, workflow, edgeMap, input, logger).ConfigureAwait(true);
+        string result = await RunSuperstepLoopAsync(context, workflow, edgeMap, input, logger).ConfigureAwait(true);
+
+        // Emit the workflow.run span after the superstep loop completes.
+        // Durable Task orchestrations replay the method from the top, where context.IsReplaying
+        // is true. It only transitions to false during forward progress (after all cached activity
+        // results are replayed). By emitting here — after RunSuperstepLoopAsync — we are guaranteed
+        // to be in non-replay mode, so the span is always captured exactly once.
+        if (!context.IsReplaying)
+        {
+            using Activity? activity = DurableWorkflowInstrumentation.ActivitySource.StartActivity("workflow.run");
+            activity?.SetTag("workflow.id", workflow.StartExecutorId)
+                .SetTag("workflow.name", workflowName)
+                .SetTag("run.id", instanceId)
+                .AddEvent(new ActivityEvent("workflow.completed"));
+        }
+
+        return result;
     }
 
     private Workflow GetWorkflowOrThrow(string orchestrationName)

dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/EdgeRouters/DurableDirectEdgeRouter.cs

@@ -29,6 +29,7 @@
 //  Enqueue to
 //  D's queue
 
+using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
 using System.Text.Json;
 using Microsoft.Extensions.Logging;
@@ -84,6 +85,12 @@ internal sealed class DurableDirectEdgeRouter : IDurableEdgeRouter
         Dictionary<string, Queue<DurableMessageEnvelope>> messageQueues,
         ILogger logger)
     {
+        using Activity? activity = DurableWorkflowInstrumentation.ActivitySource.StartActivity("edge_group.process");
+        activity?
+            .SetTag("edge_group.type", nameof(DurableDirectEdgeRouter))
+            .SetTag("message.source_id", this._sourceId)
+            .SetTag("message.target_id", this._sinkId);
+
         if (this._condition is not null)
         {
             try
@@ -92,17 +99,23 @@ internal sealed class DurableDirectEdgeRouter : IDurableEdgeRouter
                 if (!this._condition(messageObj))
                 {
                     logger.LogEdgeConditionFalse(this._sourceId, this._sinkId);
+                    activity?.SetTag("edge_group.delivered", false)
+                        .SetTag("edge_group.delivery_status", "dropped condition false");
                     return;
                 }
             }
             catch (Exception ex)
             {
                 logger.LogEdgeConditionEvaluationFailed(ex, this._sourceId, this._sinkId);
+                activity?.SetTag("edge_group.delivered", false)
+                    .SetTag("edge_group.delivery_status", "exception");
                 return;
             }
         }
 
         logger.LogEdgeRoutingMessage(this._sourceId, this._sinkId);
+        activity?.SetTag("edge_group.delivered", true)
+            .SetTag("edge_group.delivery_status", "delivered");
         EnqueueMessage(messageQueues, this._sinkId, envelope);
     }
 

dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/EdgeRouters/DurableFanOutEdgeRouter.cs

@@ -19,6 +19,7 @@
 //  Each DirectRouter independently evaluates its condition,
 //  so resultB always reaches C, but only reaches D if NeedsReview is true.
 
+using System.Diagnostics;
 using Microsoft.Extensions.Logging;
 
 namespace Microsoft.Agents.AI.DurableTask.Workflows.EdgeRouters;
@@ -54,6 +55,11 @@ internal sealed class DurableFanOutEdgeRouter : IDurableEdgeRouter
         Dictionary<string, Queue<DurableMessageEnvelope>> messageQueues,
         ILogger logger)
     {
+        using Activity? activity = DurableWorkflowInstrumentation.ActivitySource.StartActivity("edge_group.process");
+        activity?
+            .SetTag("edge_group.type", nameof(DurableFanOutEdgeRouter))
+            .SetTag("message.source_id", this._sourceId);
+
         if (logger.IsEnabled(LogLevel.Debug))
         {
             logger.LogDebug("Fan-Out from {Source}: routing to {Count} targets", this._sourceId, this._targetRouters.Count);