Merge branch 'main' into dev/dotnet_workflow/remove_timeout

.github/workflows/python-check-coverage.py

@@ -34,15 +34,14 @@ from dataclasses import dataclass
 #   (e.g., "packages/core/agent_framework/observability.py")
 # =============================================================================
 ENFORCED_TARGETS: set[str] = {
-    # Packages (sorted alphabetically)
-    "packages.anthropic.agent_framework_anthropic",
-    "packages.azure-ai-search.agent_framework_azure_ai_search",
+    # Packages
     "packages.azure-ai.agent_framework_azure_ai",
     "packages.core.agent_framework",
     "packages.core.agent_framework._workflows",
-    "packages.foundry.agent_framework_foundry",
-    "packages.openai.agent_framework_openai",
     "packages.purview.agent_framework_purview",
+    "packages.anthropic.agent_framework_anthropic",
+    "packages.azure-ai-search.agent_framework_azure_ai_search",
+    "packages.openai.agent_framework_openai",
     # Individual files (if you want to enforce specific files instead of whole packages)
     "packages/core/agent_framework/observability.py",
     # Add more targets here as coverage improves

dotnet/src/Microsoft.Agents.AI.Workflows/Execution/StreamingRunEventStream.cs

@@ -23,7 +23,8 @@ internal sealed class StreamingRunEventStream : IRunEventStream
     private readonly CancellationTokenSource _runLoopCancellation;
     private readonly bool _disableRunLoop;
     private Task? _runLoopTask;
-    private RunStatus _runStatus = RunStatus.NotStarted;
+    private volatile RunStatus _runStatus = RunStatus.NotStarted;
+
     private int _completionEpoch; // Tracks which completion signal belongs to which consumer iteration
 
     public StreamingRunEventStream(ISuperStepRunner stepRunner, bool disableRunLoop = false)
@@ -127,7 +128,7 @@ internal sealed class StreamingRunEventStream : IRunEventStream
 
                 // Wait for next input from the consumer
                 // Works for both Idle (no work) and PendingRequests (waiting for responses)
-                await this._inputWaiter.WaitForInputAsync(TimeSpan.FromSeconds(1), linkedSource.Token).ConfigureAwait(false);
+                await this._inputWaiter.WaitForInputAsync(linkedSource.Token).ConfigureAwait(false);
 
                 // When signaled, resume running
                 this._runStatus = RunStatus.Running;
@@ -209,7 +210,10 @@ internal sealed class StreamingRunEventStream : IRunEventStream
         [EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         // Get the current epoch - we'll only respond to completion signals from this epoch or later
-        int myEpoch = Volatile.Read(ref this._completionEpoch) + 1;
+        int currentEpoch = Volatile.Read(ref this._completionEpoch);
+
+        bool expectingFreshWork = this._stepRunner.HasUnprocessedMessages || this._runStatus == RunStatus.Running;
+        int myEpoch = expectingFreshWork ? currentEpoch + 1 : currentEpoch;
 
         // Use custom async enumerable to avoid exceptions on cancellation.
         NonThrowingChannelReaderAsyncEnumerable<WorkflowEvent> eventStream = new(this._eventChannel.Reader);

dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/InProcessExecutionTests.cs

@@ -132,6 +132,53 @@ public class InProcessExecutionTests
             "both versions should produce the same number of agent events");
     }
 
+    /// <summary>
+    /// This test checks that the logic around waiting for input and halting appropriately works right when the
+    /// workflow runs to halting before the EventStream is watched by the user.
+    /// </summary>
+    [Fact]
+    public async Task RunStreamingAsyncWaitToTakeStreamAsync()
+    {
+        // Arrange: Create a simple agent that responds to messages
+        var agent = new SimpleTestAgent("test-agent");
+        var workflow = AgentWorkflowBuilder.BuildSequential(agent);
+        var inputMessage = new ChatMessage(ChatRole.User, "Hello");
+
+        // Act: Execute using streaming version with TurnToken
+        await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, new List<ChatMessage> { inputMessage });
+
+        // Send TurnToken to actually trigger execution (this is the key step)
+        bool messageSent = await run.TrySendMessageAsync(new TurnToken(emitEvents: true));
+        messageSent.Should().BeTrue("TurnToken should be accepted");
+
+        while (await run.GetStatusAsync() != RunStatus.Idle)
+        {
+            await Task.Delay(200);
+        }
+
+        // Collect events
+        List<WorkflowEvent> events = [];
+
+        await foreach (WorkflowEvent evt in run.WatchStreamAsync())
+        {
+            events.Add(evt);
+        }
+
+        // Assert: The workflow should have executed and produced events
+        RunStatus status = await run.GetStatusAsync();
+        status.Should().Be(RunStatus.Idle, "workflow should complete execution");
+
+        events.Should().NotBeEmpty("workflow should produce events during execution");
+
+        // Check that we have agent execution events
+        var agentEvents = events.OfType<AgentResponseUpdateEvent>().ToList();
+        agentEvents.Should().NotBeEmpty("agent should have executed and produced update events");
+
+        // Check that we have output events
+        var outputEvents = events.OfType<WorkflowOutputEvent>().ToList();
+        outputEvents.Should().NotBeEmpty("workflow should produce output events");
+    }
+
     /// <summary>
     /// Simple test agent that echoes back the input message.
     /// </summary>