diff --git a/dotnet/samples/Durable/Workflow/ConsoleApps/05_WorkflowEvents/05_WorkflowEvents.csproj b/dotnet/samples/Durable/Workflow/ConsoleApps/05_WorkflowEvents/05_WorkflowEvents.csproj
index fdb25d48b1..09e20ef622 100644
--- a/dotnet/samples/Durable/Workflow/ConsoleApps/05_WorkflowEvents/05_WorkflowEvents.csproj
+++ b/dotnet/samples/Durable/Workflow/ConsoleApps/05_WorkflowEvents/05_WorkflowEvents.csproj
@@ -24,6 +24,5 @@
-->
-
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableActivityContext.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableActivityContext.cs
index 6bd66fbd40..9eb8d9d6c1 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableActivityContext.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableActivityContext.cs
@@ -133,6 +133,8 @@ internal sealed class DurableActivityContext : IWorkflowContext
string? scopeName = null,
CancellationToken cancellationToken = default)
{
+ ArgumentException.ThrowIfNullOrEmpty(key);
+
string scopeKey = GetScopeKey(scopeName, key);
string normalizedScope = scopeName ?? DefaultScopeName;
bool scopeCleared = this.ClearedScopes.Contains(normalizedScope);
@@ -165,11 +167,19 @@ internal sealed class DurableActivityContext : IWorkflowContext
string? scopeName = null,
CancellationToken cancellationToken = default)
{
- T? value = await this.ReadStateAsync(key, scopeName, cancellationToken).ConfigureAwait(false);
+ ArgumentException.ThrowIfNullOrEmpty(key);
+ ArgumentNullException.ThrowIfNull(initialStateFactory);
- if (value is not null)
+ // Cannot rely on `value is not null` because T? on an unconstrained generic
+ // parameter does not become Nullable for value types — the null check is
+ // always true for types like int. Instead, check key existence directly.
+ if (this.HasStateKey(key, scopeName))
{
- return value;
+ T? value = await this.ReadStateAsync(key, scopeName, cancellationToken).ConfigureAwait(false);
+ if (value is not null)
+ {
+ return value;
+ }
}
T initialValue = initialStateFactory();
@@ -231,6 +241,8 @@ internal sealed class DurableActivityContext : IWorkflowContext
string? scopeName = null,
CancellationToken cancellationToken = default)
{
+ ArgumentException.ThrowIfNullOrEmpty(key);
+
string scopeKey = GetScopeKey(scopeName, key);
this.StateUpdates[scopeKey] = value is null ? null : SerializeState(value);
return default;
@@ -265,6 +277,28 @@ internal sealed class DurableActivityContext : IWorkflowContext
private static string GetScopeKey(string? scopeName, string key)
=> $"{GetScopePrefix(scopeName)}{key}";
+ ///
+ /// Checks whether the given key exists in local updates or initial state,
+ /// respecting cleared scopes.
+ ///
+ private bool HasStateKey(string key, string? scopeName)
+ {
+ string scopeKey = GetScopeKey(scopeName, key);
+
+ if (this.StateUpdates.TryGetValue(scopeKey, out string? updated))
+ {
+ return updated is not null;
+ }
+
+ string normalizedScope = scopeName ?? DefaultScopeName;
+ if (this.ClearedScopes.Contains(normalizedScope))
+ {
+ return false;
+ }
+
+ return this._initialState.ContainsKey(scopeKey);
+ }
+
///
/// Returns the key prefix for the given scope. Scopes partition shared state
/// into logical namespaces, allowing different workflow executors to manage
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableSerialization.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableSerialization.cs
index e88e5a2779..245ec36fb8 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableSerialization.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableSerialization.cs
@@ -1,4 +1,4 @@
-// Copyright (c) Microsoft. All rights reserved.
+// Copyright (c) Microsoft. All rights reserved.
using System.Text.Json;
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableStreamingWorkflowRun.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableStreamingWorkflowRun.cs
index ac5ea12f1b..25bff4148b 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableStreamingWorkflowRun.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableStreamingWorkflowRun.cs
@@ -22,7 +22,6 @@ namespace Microsoft.Agents.AI.DurableTask.Workflows;
internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
{
private readonly DurableTaskClient _client;
- private readonly Workflow _workflow;
///
/// Initializes a new instance of the class.
@@ -34,7 +33,7 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
{
this._client = client;
this.RunId = instanceId;
- this._workflow = workflow;
+ this.WorkflowName = workflow.Name ?? string.Empty;
}
///
@@ -43,7 +42,7 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
///
/// Gets the name of the workflow being executed.
///
- public string WorkflowName => this._workflow.Name ?? string.Empty;
+ public string WorkflowName { get; }
///
/// Gets the current execution status of the workflow run.
@@ -88,13 +87,17 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
TimeSpan? pollingInterval,
[EnumeratorCancellation] CancellationToken cancellationToken = default)
{
- TimeSpan interval = pollingInterval ?? TimeSpan.FromMilliseconds(100);
+ TimeSpan minInterval = pollingInterval ?? TimeSpan.FromMilliseconds(100);
+ TimeSpan maxInterval = TimeSpan.FromSeconds(2);
+ TimeSpan currentInterval = minInterval;
// Track how many events we've already read from custom status
int lastReadEventIndex = 0;
while (!cancellationToken.IsCancellationRequested)
{
+ // Poll with getInputsAndOutputs: true because SerializedCustomStatus
+ // (used for event streaming) is only populated when this flag is set.
OrchestrationMetadata? metadata = await this._client.GetInstanceAsync(
this.RunId,
getInputsAndOutputs: true,
@@ -105,6 +108,8 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
yield break;
}
+ bool hasNewEvents = false;
+
// Always drain any unread events from custom status before checking terminal states.
// The orchestration may complete before the next poll, so events would be lost if we
// check terminal status first.
@@ -116,11 +121,13 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
(List events, lastReadEventIndex) = DrainNewEvents(customStatus.Events, lastReadEventIndex);
foreach (WorkflowEvent evt in events)
{
+ hasNewEvents = true;
yield return evt;
}
}
}
+ // On terminal status, re-fetch with outputs to get the final result.
// Check terminal states after draining events from custom status
if (metadata.RuntimeStatus == OrchestrationRuntimeStatus.Completed)
{
@@ -158,7 +165,19 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
yield break;
}
- await Task.Delay(interval, cancellationToken).ConfigureAwait(false);
+ // Adaptive backoff: reset to minimum when events were found, increase otherwise
+ currentInterval = hasNewEvents
+ ? minInterval
+ : TimeSpan.FromMilliseconds(Math.Min(currentInterval.TotalMilliseconds * 2, maxInterval.TotalMilliseconds));
+
+ try
+ {
+ await Task.Delay(currentInterval, cancellationToken).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+ {
+ yield break;
+ }
}
}
@@ -263,25 +282,58 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
///
/// Extracts a typed result from the orchestration output, unwrapping the
/// wrapper if present.
+ /// Falls back to deserializing the raw output when the wrapper is absent
+ /// (e.g., runs started before the wrapper was introduced).
///
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow result.")]
[UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow result.")]
internal static TResult? ExtractResult(string? serializedOutput)
{
- DurableWorkflowResult? workflowResult = TryParseWorkflowResult(serializedOutput);
- string? resultJson = workflowResult?.Result;
-
- if (resultJson is null)
+ if (serializedOutput is null)
{
return default;
}
- if (typeof(TResult) == typeof(string))
+ DurableWorkflowResult? workflowResult = TryParseWorkflowResult(serializedOutput);
+ string? resultJson = workflowResult?.Result;
+
+ if (resultJson is not null)
{
- return (TResult)(object)resultJson;
+ if (typeof(TResult) == typeof(string))
+ {
+ return (TResult)(object)resultJson;
+ }
+
+ return JsonSerializer.Deserialize(resultJson);
}
- return JsonSerializer.Deserialize(resultJson);
+ // Fallback: the output is not wrapped in DurableWorkflowResult.
+ // The DurableDataConverter wraps string results in JSON quotes, so
+ // we unwrap the outer JSON string first.
+ try
+ {
+ string? innerString = JsonSerializer.Deserialize(serializedOutput);
+ if (typeof(TResult) == typeof(string) && innerString is not null)
+ {
+ return (TResult)(object)innerString;
+ }
+
+ if (innerString is not null)
+ {
+ return JsonSerializer.Deserialize(innerString);
+ }
+ }
+ catch (JsonException)
+ {
+ // Not a JSON-encoded string; try direct deserialization below.
+ }
+
+ if (typeof(TResult) == typeof(string))
+ {
+ return (TResult)(object)serializedOutput;
+ }
+
+ return JsonSerializer.Deserialize(serializedOutput);
}
[UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow event types.")]
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowClient.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowClient.cs
index 588dc79e52..5944d578ef 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowClient.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowClient.cs
@@ -84,4 +84,12 @@ internal sealed class DurableWorkflowClient : IWorkflowClient
return new DurableStreamingWorkflowRun(this._client, instanceId, workflow);
}
+
+ ///
+ public ValueTask StreamAsync(
+ Workflow workflow,
+ string input,
+ string? runId = null,
+ CancellationToken cancellationToken = default)
+ => this.StreamAsync(workflow, input, runId, cancellationToken);
}
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs
index 8003cac508..bdd045213d 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs
@@ -394,6 +394,10 @@ internal sealed class DurableWorkflowRunner
///
/// Merges state updates from an executor into the shared state.
///
+ ///
+ /// When concurrent executors in the same superstep modify keys in the same scope,
+ /// last-write-wins semantics apply.
+ ///
private static void MergeStateUpdates(
SuperstepState state,
Dictionary stateUpdates,
diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/IWorkflowClient.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/IWorkflowClient.cs
index c3a471f8e1..e84f3fe4cd 100644
--- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/IWorkflowClient.cs
+++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/IWorkflowClient.cs
@@ -54,4 +54,18 @@ public interface IWorkflowClient
string? runId = null,
CancellationToken cancellationToken = default)
where TInput : notnull;
+
+ ///
+ /// Starts a workflow with string input and returns a streaming handle to watch events in real-time.
+ ///
+ /// The workflow to execute.
+ /// The string input to pass to the workflow.
+ /// Optional identifier for the run. If not provided, a new ID will be generated.
+ /// A cancellation token to observe.
+ /// An that can be used to stream workflow events.
+ ValueTask StreamAsync(
+ Workflow workflow,
+ string input,
+ string? runId = null,
+ CancellationToken cancellationToken = default);
}
diff --git a/dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableActivityContextTests.cs b/dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableActivityContextTests.cs
new file mode 100644
index 0000000000..50c5602516
--- /dev/null
+++ b/dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableActivityContextTests.cs
@@ -0,0 +1,504 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using Microsoft.Agents.AI.DurableTask.Workflows;
+using Microsoft.Agents.AI.Workflows;
+
+namespace Microsoft.Agents.AI.DurableTask.UnitTests.Workflows;
+
+public sealed class DurableActivityContextTests
+{
+ private static FunctionExecutor CreateTestExecutor(string id = "test-executor")
+ => new(id, (_, _, _) => default);
+
+ #region ReadStateAsync
+
+ [Fact]
+ public async Task ReadStateAsync_KeyExistsInInitialState_ReturnsValue()
+ {
+ // Arrange
+ Dictionary state = new() { ["__default__:counter"] = "42" };
+ DurableActivityContext context = new(state, CreateTestExecutor());
+
+ // Act
+ int? result = await context.ReadStateAsync("counter");
+
+ // Assert
+ Assert.Equal(42, result);
+ }
+
+ [Fact]
+ public async Task ReadStateAsync_KeyDoesNotExist_ReturnsNull()
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Act
+ string? result = await context.ReadStateAsync("missing");
+
+ // Assert
+ Assert.Null(result);
+ }
+
+ [Fact]
+ public async Task ReadStateAsync_LocalUpdateTakesPriorityOverInitialState()
+ {
+ // Arrange
+ Dictionary state = new() { ["__default__:key"] = "\"old\"" };
+ DurableActivityContext context = new(state, CreateTestExecutor());
+ await context.QueueStateUpdateAsync("key", "new");
+
+ // Act
+ string? result = await context.ReadStateAsync("key");
+
+ // Assert
+ Assert.Equal("new", result);
+ }
+
+ [Fact]
+ public async Task ReadStateAsync_ScopeCleared_IgnoresInitialState()
+ {
+ // Arrange
+ Dictionary state = new() { ["__default__:key"] = "\"value\"" };
+ DurableActivityContext context = new(state, CreateTestExecutor());
+ await context.QueueClearScopeAsync();
+
+ // Act
+ string? result = await context.ReadStateAsync("key");
+
+ // Assert
+ Assert.Null(result);
+ }
+
+ [Fact]
+ public async Task ReadStateAsync_WithNamedScope_ReadsFromCorrectScope()
+ {
+ // Arrange
+ Dictionary state = new()
+ {
+ ["scopeA:key"] = "\"fromA\"",
+ ["scopeB:key"] = "\"fromB\""
+ };
+ DurableActivityContext context = new(state, CreateTestExecutor());
+
+ // Act
+ string? resultA = await context.ReadStateAsync("key", "scopeA");
+ string? resultB = await context.ReadStateAsync("key", "scopeB");
+
+ // Assert
+ Assert.Equal("fromA", resultA);
+ Assert.Equal("fromB", resultB);
+ }
+
+ [Theory]
+ [InlineData(null)]
+ [InlineData("")]
+ public async Task ReadStateAsync_NullOrEmptyKey_ThrowsArgumentException(string? key)
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Act & Assert
+ await Assert.ThrowsAnyAsync(() => context.ReadStateAsync(key!).AsTask());
+ }
+
+ #endregion
+
+ #region ReadOrInitStateAsync
+
+ [Fact]
+ public async Task ReadOrInitStateAsync_KeyDoesNotExist_CallsFactoryAndQueuesUpdate()
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Act
+ string result = await context.ReadOrInitStateAsync("key", () => "initialized");
+
+ // Assert
+ Assert.Equal("initialized", result);
+ Assert.True(context.StateUpdates.ContainsKey("__default__:key"));
+ }
+
+ [Fact]
+ public async Task ReadOrInitStateAsync_KeyExists_ReturnsExistingValue()
+ {
+ // Arrange
+ Dictionary state = new() { ["__default__:key"] = "\"existing\"" };
+ DurableActivityContext context = new(state, CreateTestExecutor());
+ bool factoryCalled = false;
+
+ // Act
+ string result = await context.ReadOrInitStateAsync("key", () =>
+ {
+ factoryCalled = true;
+ return "should-not-be-used";
+ });
+
+ // Assert
+ Assert.Equal("existing", result);
+ Assert.False(factoryCalled);
+ }
+
+ [Theory]
+ [InlineData(null)]
+ [InlineData("")]
+ public async Task ReadOrInitStateAsync_NullOrEmptyKey_ThrowsArgumentException(string? key)
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Act & Assert
+ await Assert.ThrowsAnyAsync(
+ () => context.ReadOrInitStateAsync(key!, () => "value").AsTask());
+ }
+
+ [Fact]
+ public async Task ReadOrInitStateAsync_ValueType_MissingKey_CallsFactory()
+ {
+ // Arrange
+ // Validates that ReadStateAsync returns null (not 0) for missing keys,
+ // because the return type is int? (Nullable). This ensures the factory
+ // is correctly invoked for value types when the key does not exist.
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Act
+ int result = await context.ReadOrInitStateAsync("counter", () => 42);
+
+ // Assert
+ Assert.Equal(42, result);
+ Assert.True(context.StateUpdates.ContainsKey("__default__:counter"));
+ }
+
+ [Fact]
+ public async Task ReadOrInitStateAsync_NullFactory_ThrowsArgumentNullException()
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Act & Assert
+ await Assert.ThrowsAsync(
+ () => context.ReadOrInitStateAsync("key", null!).AsTask());
+ }
+
+ #endregion
+
+ #region QueueStateUpdateAsync
+
+ [Fact]
+ public async Task QueueStateUpdateAsync_SetsValue_VisibleToSubsequentRead()
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Act
+ await context.QueueStateUpdateAsync("key", "hello");
+ string? result = await context.ReadStateAsync("key");
+
+ // Assert
+ Assert.Equal("hello", result);
+ }
+
+ [Fact]
+ public async Task QueueStateUpdateAsync_NullValue_RecordsDeletion()
+ {
+ // Arrange
+ Dictionary state = new() { ["__default__:key"] = "\"value\"" };
+ DurableActivityContext context = new(state, CreateTestExecutor());
+
+ // Act
+ await context.QueueStateUpdateAsync("key", null);
+
+ // Assert
+ Assert.True(context.StateUpdates.ContainsKey("__default__:key"));
+ Assert.Null(context.StateUpdates["__default__:key"]);
+ }
+
+ [Theory]
+ [InlineData(null)]
+ [InlineData("")]
+ public async Task QueueStateUpdateAsync_NullOrEmptyKey_ThrowsArgumentException(string? key)
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Act & Assert
+ await Assert.ThrowsAnyAsync(
+ () => context.QueueStateUpdateAsync(key!, "value").AsTask());
+ }
+
+ #endregion
+
+ #region QueueClearScopeAsync
+
+ [Fact]
+ public async Task QueueClearScopeAsync_DefaultScope_ClearsStateAndPendingUpdates()
+ {
+ // Arrange
+ Dictionary state = new() { ["__default__:key"] = "\"value\"" };
+ DurableActivityContext context = new(state, CreateTestExecutor());
+ await context.QueueStateUpdateAsync("pending", "data");
+
+ // Act
+ await context.QueueClearScopeAsync();
+
+ // Assert
+ Assert.Contains("__default__", context.ClearedScopes);
+ Assert.Empty(context.StateUpdates);
+ }
+
+ [Fact]
+ public async Task QueueClearScopeAsync_NamedScope_OnlyClearsThatScope()
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+ await context.QueueStateUpdateAsync("keyA", "valueA", scopeName: "scopeA");
+ await context.QueueStateUpdateAsync("keyB", "valueB", scopeName: "scopeB");
+
+ // Act
+ await context.QueueClearScopeAsync("scopeA");
+
+ // Assert
+ Assert.DoesNotContain("scopeA:keyA", context.StateUpdates.Keys);
+ Assert.Contains("scopeB:keyB", context.StateUpdates.Keys);
+ }
+
+ #endregion
+
+ #region ReadStateKeysAsync
+
+ [Fact]
+ public async Task ReadStateKeysAsync_ReturnsKeysFromInitialState()
+ {
+ // Arrange
+ Dictionary state = new()
+ {
+ ["__default__:alpha"] = "\"a\"",
+ ["__default__:beta"] = "\"b\""
+ };
+ DurableActivityContext context = new(state, CreateTestExecutor());
+
+ // Act
+ HashSet keys = await context.ReadStateKeysAsync();
+
+ // Assert
+ Assert.Equal(2, keys.Count);
+ Assert.Contains("alpha", keys);
+ Assert.Contains("beta", keys);
+ }
+
+ [Fact]
+ public async Task ReadStateKeysAsync_MergesLocalUpdatesAndDeletions()
+ {
+ // Arrange
+ Dictionary state = new()
+ {
+ ["__default__:existing"] = "\"val\"",
+ ["__default__:toDelete"] = "\"val\""
+ };
+ DurableActivityContext context = new(state, CreateTestExecutor());
+ await context.QueueStateUpdateAsync("newKey", "value");
+ await context.QueueStateUpdateAsync("toDelete", null);
+
+ // Act
+ HashSet keys = await context.ReadStateKeysAsync();
+
+ // Assert
+ Assert.Contains("existing", keys);
+ Assert.Contains("newKey", keys);
+ Assert.DoesNotContain("toDelete", keys);
+ }
+
+ [Fact]
+ public async Task ReadStateKeysAsync_AfterClearScope_ExcludesInitialState()
+ {
+ // Arrange
+ Dictionary state = new() { ["__default__:old"] = "\"val\"" };
+ DurableActivityContext context = new(state, CreateTestExecutor());
+ await context.QueueClearScopeAsync();
+ await context.QueueStateUpdateAsync("new", "value");
+
+ // Act
+ HashSet keys = await context.ReadStateKeysAsync();
+
+ // Assert
+ Assert.DoesNotContain("old", keys);
+ Assert.Contains("new", keys);
+ }
+
+ [Fact]
+ public async Task ReadStateKeysAsync_WithNamedScope_OnlyReturnsKeysFromThatScope()
+ {
+ // Arrange
+ Dictionary state = new()
+ {
+ ["scopeA:key1"] = "\"val\"",
+ ["scopeB:key2"] = "\"val\""
+ };
+ DurableActivityContext context = new(state, CreateTestExecutor());
+
+ // Act
+ HashSet keysA = await context.ReadStateKeysAsync("scopeA");
+
+ // Assert
+ Assert.Single(keysA);
+ Assert.Contains("key1", keysA);
+ }
+
+ #endregion
+
+ #region AddEventAsync
+
+ [Fact]
+ public async Task AddEventAsync_AddsEventToCollection()
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+ WorkflowEvent evt = new ExecutorInvokedEvent("test", "test-data");
+
+ // Act
+ await context.AddEventAsync(evt);
+
+ // Assert
+ Assert.Single(context.Events);
+ Assert.Same(evt, context.Events[0]);
+ }
+
+ [Fact]
+ public async Task AddEventAsync_NullEvent_DoesNotAdd()
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Act
+#pragma warning disable CS8625 // Cannot convert null literal to non-nullable reference type.
+ await context.AddEventAsync(null);
+#pragma warning restore CS8625
+
+ // Assert
+ Assert.Empty(context.Events);
+ }
+
+ #endregion
+
+ #region SendMessageAsync
+
+ [Fact]
+ public async Task SendMessageAsync_SerializesMessageWithTypeName()
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Act
+ await context.SendMessageAsync("hello");
+
+ // Assert
+ Assert.Single(context.SentMessages);
+ Assert.Equal(typeof(string).AssemblyQualifiedName, context.SentMessages[0].TypeName);
+ Assert.NotNull(context.SentMessages[0].Data);
+ }
+
+ [Fact]
+ public async Task SendMessageAsync_NullMessage_DoesNotAdd()
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Act
+#pragma warning disable CS8625 // Cannot convert null literal to non-nullable reference type.
+ await context.SendMessageAsync(null);
+#pragma warning restore CS8625
+
+ // Assert
+ Assert.Empty(context.SentMessages);
+ }
+
+ #endregion
+
+ #region YieldOutputAsync
+
+ [Fact]
+ public async Task YieldOutputAsync_AddsWorkflowOutputEvent()
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Act
+ await context.YieldOutputAsync("result");
+
+ // Assert
+ Assert.Single(context.Events);
+ WorkflowOutputEvent outputEvent = Assert.IsType(context.Events[0]);
+ Assert.Equal("result", outputEvent.Data);
+ }
+
+ [Fact]
+ public async Task YieldOutputAsync_NullOutput_DoesNotAdd()
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Act
+#pragma warning disable CS8625 // Cannot convert null literal to non-nullable reference type.
+ await context.YieldOutputAsync(null);
+#pragma warning restore CS8625
+
+ // Assert
+ Assert.Empty(context.Events);
+ }
+
+ #endregion
+
+ #region RequestHaltAsync
+
+ [Fact]
+ public async Task RequestHaltAsync_SetsHaltRequestedAndAddsEvent()
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Act
+ await context.RequestHaltAsync();
+
+ // Assert
+ Assert.True(context.HaltRequested);
+ Assert.Single(context.Events);
+ Assert.IsType(context.Events[0]);
+ }
+
+ #endregion
+
+ #region Properties
+
+ [Fact]
+ public void TraceContext_ReturnsNull()
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Assert
+ Assert.Null(context.TraceContext);
+ }
+
+ [Fact]
+ public void ConcurrentRunsEnabled_ReturnsFalse()
+ {
+ // Arrange
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Assert
+ Assert.False(context.ConcurrentRunsEnabled);
+ }
+
+ [Fact]
+ public async Task Constructor_NullInitialState_CreatesEmptyState()
+ {
+ // Arrange & Act
+ DurableActivityContext context = new(null, CreateTestExecutor());
+
+ // Assert
+ string? result = await context.ReadStateAsync("anything");
+ Assert.Null(result);
+ }
+
+ #endregion
+}
diff --git a/dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableStreamingWorkflowRunTests.cs b/dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableStreamingWorkflowRunTests.cs
new file mode 100644
index 0000000000..8a291ddf93
--- /dev/null
+++ b/dotnet/tests/Microsoft.Agents.AI.DurableTask.UnitTests/Workflows/DurableStreamingWorkflowRunTests.cs
@@ -0,0 +1,598 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Text.Json;
+using Microsoft.Agents.AI.DurableTask.Workflows;
+using Microsoft.Agents.AI.Workflows;
+using Microsoft.DurableTask;
+using Microsoft.DurableTask.Client;
+using Moq;
+
+namespace Microsoft.Agents.AI.DurableTask.UnitTests.Workflows;
+
+public sealed class DurableStreamingWorkflowRunTests
+{
+ private const string InstanceId = "test-instance-123";
+ private const string WorkflowTestName = "TestWorkflow";
+
+ private static Workflow CreateTestWorkflow() =>
+ new WorkflowBuilder(new FunctionExecutor("start", (_, _, _) => default))
+ .WithName(WorkflowTestName)
+ .Build();
+
+ private static OrchestrationMetadata CreateMetadata(
+ OrchestrationRuntimeStatus status,
+ string? serializedCustomStatus = null,
+ string? serializedOutput = null,
+ TaskFailureDetails? failureDetails = null)
+ {
+ return new OrchestrationMetadata(WorkflowTestName, InstanceId)
+ {
+ RuntimeStatus = status,
+ SerializedCustomStatus = serializedCustomStatus,
+ SerializedOutput = serializedOutput,
+ FailureDetails = failureDetails,
+ };
+ }
+
+ private static string SerializeCustomStatus(List events)
+ {
+ DurableWorkflowCustomStatus status = new() { Events = events };
+ return JsonSerializer.Serialize(status, DurableWorkflowJsonContext.Default.DurableWorkflowCustomStatus);
+ }
+
+ private static string SerializeWorkflowResult(string? result, List events)
+ {
+ DurableWorkflowResult workflowResult = new() { Result = result, Events = events };
+ string inner = JsonSerializer.Serialize(workflowResult, DurableWorkflowJsonContext.Default.DurableWorkflowResult);
+ return JsonSerializer.Serialize(inner);
+ }
+
+ private static string SerializeEvent(WorkflowEvent evt)
+ {
+ Type eventType = evt.GetType();
+ TypedPayload wrapper = new()
+ {
+ TypeName = eventType.AssemblyQualifiedName,
+ Data = JsonSerializer.Serialize(evt, eventType, DurableSerialization.Options)
+ };
+
+ return JsonSerializer.Serialize(wrapper, DurableWorkflowJsonContext.Default.TypedPayload);
+ }
+
+ #region Constructor and Properties
+
+ [Fact]
+ public void Constructor_SetsRunIdAndWorkflowName()
+ {
+ // Arrange
+ Mock mockClient = new("test");
+
+ // Act
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Assert
+ Assert.Equal(InstanceId, run.RunId);
+ Assert.Equal(WorkflowTestName, run.WorkflowName);
+ }
+
+ [Fact]
+ public void Constructor_NoWorkflowName_SetsEmptyString()
+ {
+ // Arrange
+ Mock mockClient = new("test");
+ Workflow workflow = new WorkflowBuilder(new FunctionExecutor("start", (_, _, _) => default)).Build();
+
+ // Act
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, workflow);
+
+ // Assert
+ Assert.Equal(string.Empty, run.WorkflowName);
+ }
+
+ #endregion
+
+ #region GetStatusAsync
+
+ [Theory]
+ [InlineData(OrchestrationRuntimeStatus.Pending, DurableRunStatus.Pending)]
+ [InlineData(OrchestrationRuntimeStatus.Running, DurableRunStatus.Running)]
+ [InlineData(OrchestrationRuntimeStatus.Completed, DurableRunStatus.Completed)]
+ [InlineData(OrchestrationRuntimeStatus.Failed, DurableRunStatus.Failed)]
+ [InlineData(OrchestrationRuntimeStatus.Terminated, DurableRunStatus.Terminated)]
+ [InlineData(OrchestrationRuntimeStatus.Suspended, DurableRunStatus.Suspended)]
+
+ public async Task GetStatusAsync_MapsRuntimeStatusCorrectlyAsync(
+ OrchestrationRuntimeStatus runtimeStatus,
+ DurableRunStatus expectedStatus)
+ {
+ // Arrange
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.GetInstanceAsync(InstanceId, false, It.IsAny()))
+ .ReturnsAsync(CreateMetadata(runtimeStatus));
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act
+ DurableRunStatus status = await run.GetStatusAsync();
+
+ // Assert
+ Assert.Equal(expectedStatus, status);
+ }
+
+ [Fact]
+ public async Task GetStatusAsync_InstanceNotFound_ReturnsNotFoundAsync()
+ {
+ // Arrange
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.GetInstanceAsync(InstanceId, false, It.IsAny()))
+ .ReturnsAsync((OrchestrationMetadata?)null);
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act
+ DurableRunStatus status = await run.GetStatusAsync();
+
+ // Assert
+ Assert.Equal(DurableRunStatus.NotFound, status);
+ }
+
+ #endregion
+
+ #region WatchStreamAsync
+
+ [Fact]
+ public async Task WatchStreamAsync_InstanceNotFound_YieldsNoEventsAsync()
+ {
+ // Arrange
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny()))
+ .ReturnsAsync((OrchestrationMetadata?)null);
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act
+ List events = [];
+ await foreach (WorkflowEvent evt in run.WatchStreamAsync())
+ {
+ events.Add(evt);
+ }
+
+ // Assert
+ Assert.Empty(events);
+ }
+
+ [Fact]
+ public async Task WatchStreamAsync_CompletedWithResult_YieldsCompletedEventAsync()
+ {
+ // Arrange
+ string serializedOutput = SerializeWorkflowResult("done", []);
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny()))
+ .ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput));
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act
+ List events = [];
+ await foreach (WorkflowEvent evt in run.WatchStreamAsync())
+ {
+ events.Add(evt);
+ }
+
+ // Assert
+ Assert.Single(events);
+ DurableWorkflowCompletedEvent completedEvent = Assert.IsType(events[0]);
+ Assert.Equal("done", completedEvent.Data);
+ }
+
+ [Fact]
+ public async Task WatchStreamAsync_CompletedWithEventsInOutput_YieldsEventsAndCompletionAsync()
+ {
+ // Arrange
+ DurableHaltRequestedEvent haltEvent = new("executor-1");
+ string serializedEvent = SerializeEvent(haltEvent);
+ string serializedOutput = SerializeWorkflowResult("result", [serializedEvent]);
+
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny()))
+ .ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput));
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act
+ List events = [];
+ await foreach (WorkflowEvent evt in run.WatchStreamAsync())
+ {
+ events.Add(evt);
+ }
+
+ // Assert
+ Assert.Equal(2, events.Count);
+ Assert.IsType(events[0]);
+ Assert.IsType(events[1]);
+ }
+
+ [Fact]
+ public async Task WatchStreamAsync_CompletedWithoutWrapper_YieldsRawOutputAsync()
+ {
+ // Arrange
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny()))
+ .ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: "\"raw output\""));
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act
+ List events = [];
+ await foreach (WorkflowEvent evt in run.WatchStreamAsync())
+ {
+ events.Add(evt);
+ }
+
+ // Assert
+ Assert.Single(events);
+ DurableWorkflowCompletedEvent completedEvent = Assert.IsType(events[0]);
+ Assert.Equal("\"raw output\"", completedEvent.Data);
+ }
+
+ [Fact]
+ public async Task WatchStreamAsync_Failed_YieldsFailedEventAsync()
+ {
+ // Arrange
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny()))
+ .ReturnsAsync(CreateMetadata(
+ OrchestrationRuntimeStatus.Failed,
+ failureDetails: new TaskFailureDetails("ErrorType", "Something went wrong", null, null, null)));
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act
+ List events = [];
+ await foreach (WorkflowEvent evt in run.WatchStreamAsync())
+ {
+ events.Add(evt);
+ }
+
+ // Assert
+ Assert.Single(events);
+ DurableWorkflowFailedEvent failedEvent = Assert.IsType(events[0]);
+ Assert.Equal("Something went wrong", failedEvent.Data);
+ }
+
+ [Fact]
+ public async Task WatchStreamAsync_FailedWithNoDetails_YieldsDefaultMessageAsync()
+ {
+ // Arrange
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny()))
+ .ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Failed));
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act
+ List events = [];
+ await foreach (WorkflowEvent evt in run.WatchStreamAsync())
+ {
+ events.Add(evt);
+ }
+
+ // Assert
+ Assert.Single(events);
+ DurableWorkflowFailedEvent failedEvent = Assert.IsType(events[0]);
+ Assert.Equal("Workflow execution failed.", failedEvent.Data);
+ }
+
+ [Fact]
+ public async Task WatchStreamAsync_Terminated_YieldsFailedEventAsync()
+ {
+ // Arrange
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny()))
+ .ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Terminated));
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act
+ List events = [];
+ await foreach (WorkflowEvent evt in run.WatchStreamAsync())
+ {
+ events.Add(evt);
+ }
+
+ // Assert
+ Assert.Single(events);
+ DurableWorkflowFailedEvent failedEvent = Assert.IsType(events[0]);
+ Assert.Equal("Workflow was terminated.", failedEvent.Data);
+ }
+
+ [Fact]
+ public async Task WatchStreamAsync_EventsInCustomStatus_YieldsEventsBeforeCompletionAsync()
+ {
+ // Arrange
+ DurableHaltRequestedEvent haltEvent = new("exec-1");
+ string serializedEvent = SerializeEvent(haltEvent);
+ string customStatus = SerializeCustomStatus([serializedEvent]);
+ string serializedOutput = SerializeWorkflowResult("final", []);
+
+ int callCount = 0;
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny()))
+ .ReturnsAsync(() =>
+ {
+ callCount++;
+ if (callCount == 1)
+ {
+ return CreateMetadata(OrchestrationRuntimeStatus.Running, serializedCustomStatus: customStatus);
+ }
+
+ return CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput);
+ });
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act
+ List events = [];
+ await foreach (WorkflowEvent evt in run.WatchStreamAsync())
+ {
+ events.Add(evt);
+ }
+
+ // Assert
+ Assert.Equal(2, events.Count);
+ Assert.IsType(events[0]);
+ Assert.IsType(events[1]);
+ }
+
+ [Fact]
+ public async Task WatchStreamAsync_IncrementalEvents_YieldsOnlyNewEventsPerPollAsync()
+ {
+ // Arrange — simulate 3 poll cycles where events accumulate in custom status,
+ // then a final completion poll. This validates:
+ // 1. Events arriving across multiple poll cycles are yielded incrementally
+ // 2. Already-seen events are not re-yielded (lastReadEventIndex dedup)
+ // 3. Completion event follows all streamed events
+ DurableHaltRequestedEvent event1 = new("executor-1");
+ DurableHaltRequestedEvent event2 = new("executor-2");
+ DurableHaltRequestedEvent event3 = new("executor-3");
+
+ string serializedEvent1 = SerializeEvent(event1);
+ string serializedEvent2 = SerializeEvent(event2);
+ string serializedEvent3 = SerializeEvent(event3);
+
+ // Poll 1: 1 event in custom status
+ string customStatus1 = SerializeCustomStatus([serializedEvent1]);
+ // Poll 2: same event + 1 new event (accumulating list)
+ string customStatus2 = SerializeCustomStatus([serializedEvent1, serializedEvent2]);
+ // Poll 3: all 3 events accumulated
+ string customStatus3 = SerializeCustomStatus([serializedEvent1, serializedEvent2, serializedEvent3]);
+ // Poll 4: completed, all events also in output
+ string serializedOutput = SerializeWorkflowResult("done", [serializedEvent1, serializedEvent2, serializedEvent3]);
+
+ int callCount = 0;
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny()))
+ .ReturnsAsync(() =>
+ {
+ callCount++;
+ return callCount switch
+ {
+ 1 => CreateMetadata(OrchestrationRuntimeStatus.Running, serializedCustomStatus: customStatus1),
+ 2 => CreateMetadata(OrchestrationRuntimeStatus.Running, serializedCustomStatus: customStatus2),
+ 3 => CreateMetadata(OrchestrationRuntimeStatus.Running, serializedCustomStatus: customStatus3),
+ _ => CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput),
+ };
+ });
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act
+ List events = [];
+ await foreach (WorkflowEvent evt in run.WatchStreamAsync())
+ {
+ events.Add(evt);
+ }
+
+ // Assert — exactly 4 events: 3 incremental halt events + 1 completion
+ Assert.Equal(4, events.Count);
+ DurableHaltRequestedEvent halt1 = Assert.IsType(events[0]);
+ DurableHaltRequestedEvent halt2 = Assert.IsType(events[1]);
+ DurableHaltRequestedEvent halt3 = Assert.IsType(events[2]);
+ Assert.Equal("executor-1", halt1.ExecutorId);
+ Assert.Equal("executor-2", halt2.ExecutorId);
+ Assert.Equal("executor-3", halt3.ExecutorId);
+ DurableWorkflowCompletedEvent completed = Assert.IsType(events[3]);
+ Assert.Equal("done", completed.Data);
+ }
+
+ [Fact]
+ public async Task WatchStreamAsync_NoNewEventsOnRepoll_DoesNotDuplicateAsync()
+ {
+ // Arrange — simulate polling where custom status doesn't change between polls,
+ // validating that events are not duplicated when the list is unchanged.
+ DurableHaltRequestedEvent event1 = new("executor-1");
+ string serializedEvent1 = SerializeEvent(event1);
+ string customStatus = SerializeCustomStatus([serializedEvent1]);
+ string serializedOutput = SerializeWorkflowResult("result", [serializedEvent1]);
+
+ int callCount = 0;
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny()))
+ .ReturnsAsync(() =>
+ {
+ callCount++;
+ return callCount switch
+ {
+ // First 3 polls return the same custom status (no new events after first)
+ <= 3 => CreateMetadata(OrchestrationRuntimeStatus.Running, serializedCustomStatus: customStatus),
+ _ => CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput),
+ };
+ });
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act
+ List events = [];
+ await foreach (WorkflowEvent evt in run.WatchStreamAsync())
+ {
+ events.Add(evt);
+ }
+
+ // Assert — event1 appears exactly once despite 3 polls with the same status
+ Assert.Equal(2, events.Count);
+ Assert.IsType(events[0]);
+ Assert.IsType(events[1]);
+ }
+
+ [Fact]
+ public async Task WatchStreamAsync_Cancellation_EndsGracefullyAsync()
+ {
+ // Arrange
+ using CancellationTokenSource cts = new();
+ int pollCount = 0;
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.GetInstanceAsync(InstanceId, true, It.IsAny()))
+ .ReturnsAsync(() =>
+ {
+ if (++pollCount >= 2)
+ {
+ cts.Cancel();
+ }
+
+ return CreateMetadata(OrchestrationRuntimeStatus.Running);
+ });
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act
+ List events = [];
+ await foreach (WorkflowEvent evt in run.WatchStreamAsync(cts.Token))
+ {
+ events.Add(evt);
+ }
+
+ // Assert — no exception thrown, stream ends cleanly
+ Assert.Empty(events);
+ }
+
+ #endregion
+
+ #region WaitForCompletionAsync
+
+ [Fact]
+ public async Task WaitForCompletionAsync_Completed_ReturnsResultAsync()
+ {
+ // Arrange
+ string serializedOutput = SerializeWorkflowResult("hello world", []);
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.WaitForInstanceCompletionAsync(InstanceId, true, It.IsAny()))
+ .ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Completed, serializedOutput: serializedOutput));
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act
+ string? result = await run.WaitForCompletionAsync();
+
+ // Assert
+ Assert.Equal("hello world", result);
+ }
+
+ [Fact]
+ public async Task WaitForCompletionAsync_Failed_ThrowsWithErrorMessageAsync()
+ {
+ // Arrange
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.WaitForInstanceCompletionAsync(InstanceId, true, It.IsAny()))
+ .ReturnsAsync(CreateMetadata(
+ OrchestrationRuntimeStatus.Failed,
+ failureDetails: new TaskFailureDetails("Error", "kaboom", null, null, null)));
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act & Assert
+ InvalidOperationException ex = await Assert.ThrowsAsync(
+ () => run.WaitForCompletionAsync().AsTask());
+ Assert.Equal("kaboom", ex.Message);
+ }
+
+ [Fact]
+ public async Task WaitForCompletionAsync_UnexpectedStatus_ThrowsAsync()
+ {
+ // Arrange
+ Mock mockClient = new("test");
+ mockClient.Setup(c => c.WaitForInstanceCompletionAsync(InstanceId, true, It.IsAny()))
+ .ReturnsAsync(CreateMetadata(OrchestrationRuntimeStatus.Terminated));
+
+ DurableStreamingWorkflowRun run = new(mockClient.Object, InstanceId, CreateTestWorkflow());
+
+ // Act & Assert
+ await Assert.ThrowsAsync(
+ () => run.WaitForCompletionAsync().AsTask());
+ }
+
+ #endregion
+
+ #region ExtractResult
+
+ [Fact]
+ public void ExtractResult_NullOutput_ReturnsDefault()
+ {
+ // Act
+ string? result = DurableStreamingWorkflowRun.ExtractResult(null);
+
+ // Assert
+ Assert.Null(result);
+ }
+
+ [Fact]
+ public void ExtractResult_WrappedStringResult_ReturnsUnwrappedString()
+ {
+ // Arrange
+ string serializedOutput = SerializeWorkflowResult("hello", []);
+
+ // Act
+ string? result = DurableStreamingWorkflowRun.ExtractResult(serializedOutput);
+
+ // Assert
+ Assert.Equal("hello", result);
+ }
+
+ [Fact]
+ public void ExtractResult_UnwrappedStringOutput_FallsBackToDirectDeserialization()
+ {
+ // Arrange — raw DurableDataConverter-style output (JSON-encoded string)
+ string serializedOutput = JsonSerializer.Serialize("raw value");
+
+ // Act
+ string? result = DurableStreamingWorkflowRun.ExtractResult(serializedOutput);
+
+ // Assert
+ Assert.Equal("raw value", result);
+ }
+
+ [Fact]
+ public void ExtractResult_WrappedObjectResult_DeserializesCorrectly()
+ {
+ // Arrange
+ TestPayload original = new() { Name = "test", Value = 42 };
+ string resultJson = JsonSerializer.Serialize(original);
+ string serializedOutput = SerializeWorkflowResult(resultJson, []);
+
+ // Act
+ TestPayload? result = DurableStreamingWorkflowRun.ExtractResult(serializedOutput);
+
+ // Assert
+ Assert.NotNull(result);
+ Assert.Equal("test", result.Name);
+ Assert.Equal(42, result.Value);
+ }
+
+ #endregion
+
+ private sealed class TestPayload
+ {
+ public string? Name { get; set; }
+
+ public int Value { get; set; }
+ }
+}