PR feedback fixes

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<T>(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<T> 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<T>(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}";
 
+    /// <summary>
+    /// Checks whether the given key exists in local updates or initial state,
+    /// respecting cleared scopes.
+    /// </summary>
+    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);
+    }
+
     /// <summary>
     /// Returns the key prefix for the given scope. Scopes partition shared state
     /// into logical namespaces, allowing different workflow executors to manage

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;
 

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;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="DurableStreamingWorkflowRun"/> 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;
     }
 
     /// <inheritdoc/>
@@ -43,7 +42,7 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun
     /// <summary>
     /// Gets the name of the workflow being executed.
     /// </summary>
-    public string WorkflowName => this._workflow.Name ?? string.Empty;
+    public string WorkflowName { get; }
 
     /// <summary>
     /// 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<WorkflowEvent> 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
     /// <summary>
     /// Extracts a typed result from the orchestration output, unwrapping the
     /// <see cref="DurableWorkflowResult"/> wrapper if present.
+    /// Falls back to deserializing the raw output when the wrapper is absent
+    /// (e.g., runs started before the wrapper was introduced).
     /// </summary>
     [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow result.")]
     [UnconditionalSuppressMessage("Trimming", "IL2026", Justification = "Deserializing workflow result.")]
     internal static TResult? ExtractResult<TResult>(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<TResult>(resultJson);
         }
 
-        return JsonSerializer.Deserialize<TResult>(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<string>(serializedOutput);
+            if (typeof(TResult) == typeof(string) && innerString is not null)
+            {
+                return (TResult)(object)innerString;
+            }
+
+            if (innerString is not null)
+            {
+                return JsonSerializer.Deserialize<TResult>(innerString);
+            }
+        }
+        catch (JsonException)
+        {
+            // Not a JSON-encoded string; try direct deserialization below.
+        }
+
+        if (typeof(TResult) == typeof(string))
+        {
+            return (TResult)(object)serializedOutput;
+        }
+
+        return JsonSerializer.Deserialize<TResult>(serializedOutput);
     }
 
     [UnconditionalSuppressMessage("AOT", "IL3050", Justification = "Deserializing workflow event types.")]

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);
     }
+
+    /// <inheritdoc/>
+    public ValueTask<IStreamingWorkflowRun> StreamAsync(
+        Workflow workflow,
+        string input,
+        string? runId = null,
+        CancellationToken cancellationToken = default)
+        => this.StreamAsync<string>(workflow, input, runId, cancellationToken);
 }

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

@@ -394,6 +394,10 @@ internal sealed class DurableWorkflowRunner
     /// <summary>
     /// Merges state updates from an executor into the shared state.
     /// </summary>
+    /// <remarks>
+    /// When concurrent executors in the same superstep modify keys in the same scope,
+    /// last-write-wins semantics apply.
+    /// </remarks>
     private static void MergeStateUpdates(
         SuperstepState state,
         Dictionary<string, string?> stateUpdates,

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;
+
+    /// <summary>
+    /// Starts a workflow with string input and returns a streaming handle to watch events in real-time.
+    /// </summary>
+    /// <param name="workflow">The workflow to execute.</param>
+    /// <param name="input">The string input to pass to the workflow.</param>
+    /// <param name="runId">Optional identifier for the run. If not provided, a new ID will be generated.</param>
+    /// <param name="cancellationToken">A cancellation token to observe.</param>
+    /// <returns>An <see cref="IStreamingWorkflowRun"/> that can be used to stream workflow events.</returns>
+    ValueTask<IStreamingWorkflowRun> StreamAsync(
+        Workflow workflow,
+        string input,
+        string? runId = null,
+        CancellationToken cancellationToken = default);
 }