.NET: Align WorkflowOutputEvent streaming deserialization with SourceId to ExecutorId rename

dotnet/src/Microsoft.Agents.AI.DurableTask/CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## [Unreleased]
 
+- Fixed `WorkflowOutputEvent` streaming deserialization to read `executorId` instead of the renamed `sourceId` property, with fallback for backward compatibility ([#6896](https://github.com/microsoft/agent-framework/pull/6896))
 - Fix issue with resuming checkpoint after package version upgrade ([#6670](https://github.com/microsoft/agent-framework/pull/6670))
 - Bind MCP threadId to the current agent and guard cross-agent session dispatch ([#6531](https://github.com/microsoft/agent-framework/pull/6531))
 - Added support for durable workflows ([#4436](https://github.com/microsoft/agent-framework/pull/4436))

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

@@ -425,9 +425,13 @@ private static bool TryParseWorkflowResult(string? serializedOutput, [NotNullWhe
             }
 
             // WorkflowOutputEvent
-            string sourceId = root.GetProperty("sourceId").GetString() ?? string.Empty;
+            string outputExecutorId = root.TryGetProperty("executorId", out JsonElement execIdElem)
+                ? execIdElem.GetString() ?? string.Empty
+                : root.TryGetProperty("sourceId", out JsonElement srcIdElem)
+                    ? srcIdElem.GetString() ?? string.Empty
+                    : throw new JsonException("WorkflowOutputEvent is missing required 'executorId' (or legacy 'sourceId') property.");
             object? outputData = GetDataProperty(root);
-            return new WorkflowOutputEvent(outputData!, sourceId);
+            return new WorkflowOutputEvent(outputData!, outputExecutorId);
         }
 
         return JsonSerializer.Deserialize(json, eventType, DurableSerialization.Options) as WorkflowEvent;

dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableStreamingWorkflowRunTests.cs

@@ -233,6 +233,66 @@ public async Task WatchStreamAsync_CompletedWithEventsInOutput_YieldsEventsAndCo
         Assert.Equal("result", completedResult.Result);
     }
 
+    [Fact]
+    public async Task WatchStreamAsync_WorkflowOutputEvent_RoundTripsCorrectlyAsync()
+    {
+        // Arrange
+        WorkflowOutputEvent outputEvent = new("test-data", "executor-1");
+        string serializedEvent = SerializeEvent(outputEvent);
+        string serializedOutput = SerializeWorkflowResult("done", [serializedEvent]);
+
+        Mock<DurableTaskClient> mockClient = new("test");
+        mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
+            .ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput));
+
+        DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+        // Act
+        List<WorkflowEvent> events = [];
+        await foreach (WorkflowEvent evt in run.WatchStreamAsync())
+        {
+            events.Add(evt);
+        }
+
+        // Assert
+        Assert.Equal(2, events.Count);
+        WorkflowOutputEvent result = Assert.IsType<WorkflowOutputEvent>(events[0]);
+        Assert.Equal("executor-1", result.ExecutorId);
+        Assert.Equal("test-data", result.Data?.ToString());
+        Assert.Empty(result.Tags);
+    }
+
+    [Fact]
+    public async Task WatchStreamAsync_WorkflowOutputEvent_LegacySourceId_RoundTripsCorrectlyAsync()
+    {
+        // Arrange — simulate a legacy payload that uses "sourceId" instead of "executorId"
+        const string LegacyEventJson = """{"sourceId":"legacy-executor","data":"legacy-data"}""";
+        string typeName = typeof(WorkflowOutputEvent).AssemblyQualifiedName!;
+        string serializedEvent = JsonSerializer.Serialize(
+            new { typeName, data = LegacyEventJson },
+            DurableSerialization.Options);
+        string serializedOutput = SerializeWorkflowResult("done", [serializedEvent]);
+
+        Mock<DurableTaskClient> mockClient = new("test");
+        mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny<CancellationToken>()))
+            .ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput));
+
+        DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+        // Act
+        List<WorkflowEvent> events = [];
+        await foreach (WorkflowEvent evt in run.WatchStreamAsync())
+        {
+            events.Add(evt);
+        }
+
+        // Assert
+        Assert.Equal(2, events.Count);
+        WorkflowOutputEvent result = Assert.IsType<WorkflowOutputEvent>(events[0]);
+        Assert.Equal("legacy-executor", result.ExecutorId);
+        Assert.Equal("legacy-data", result.Data?.ToString());
+    }
+
     [Fact]
     public async Task WatchStreamAsync_CompletedWithoutWrapper_YieldsFailedEventAsync()
     {