feat: WorkflowHostAgent forwards AgentResponseEvent unconditionally under Futures-on

Aligns the .NET Workflow-as-Agent surface with Python `as_agent`. Under
`Futures.EnableAgentResponseOutputTaggingAndFiltering = true`,
`WorkflowSession.InvokeStageAsync` now forwards `AgentResponseEvent`
unconditionally — joining `AgentResponseUpdateEvent` in ignoring the host's
`includeWorkflowOutputsInResponse` switch. That switch keeps governing the
generic `WorkflowOutputEvent` path for non-AIAgent payloads, where it is
further short-circuited by an `IsIntermediate()` check (tagged intermediate
outputs always surface).

Under Futures-off the legacy asymmetry is preserved: `AgentResponseUpdateEvent`
always forwarded, `AgentResponseEvent` gated by `includeWorkflowOutputsInResponse`.

Back-compat: with `Futures.EnableAgentResponseOutputTaggingAndFiltering` left at
its default `false`, observable behavior is identical to before.

`Futures` documentation gains a remark explaining the `Workflow.AsAIAgent()`
interaction in both flag states.

Runner fix
----------
`InProcessRunnerContext.YieldOutputAsync` now skips `Executor.CanOutput` for
AgentResponse-shaped payloads under both Futures branches. `AIAgentHostExecutor`
doesn't declare AgentResponse(Update) in its `Yields` set, so the historical
legacy bypass had silently skipped the check; Phase 3's Futures-on path was
running it and would reject AIAgent payloads. AIAgent-shaped payloads are now
always a valid output shape, matching the legacy bypass semantics.

Phase 4 follow-on
-----------------
Switched the three orchestration-builder designation-replay loops to iterate
`Dictionary.Keys` with a value lookup instead of constructing/destructuring
`KeyValuePair<,>`. Cleaner shape and avoids the netstandard2.0 / net472
`KeyValuePair<,>.Deconstruct` unavailability that surfaced when this branch
multi-TFM-built.

Tests
-----
`WorkflowHostSmokeTests.IntermediateForwarding` (new nested class, 6 tests):
- intermediate AgentResponse forwarded past the include-outputs gate (Futures on)
- terminal AgentResponse forwarded unconditionally (Futures on)
- terminal AgentResponse gated by include flag (Futures off, legacy)
- undesignated AIAgent executor emits no AgentResponseEvent under Futures-on
- legacy bypass still emits AgentResponseEvent under Futures-off
- intermediate tag is observable via `update.RawRepresentation`

The class joins the `FuturesSerial` xUnit collection so the process-global flag
is serialized against other Futures-toggling tests.

599/599 unit tests pass on net10.0 (593 baseline + 6 new).

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

dotnet/src/Microsoft.Agents.AI.Workflows/Futures.cs

@@ -9,8 +9,6 @@ namespace Microsoft.Agents.AI.Workflows;
 /// </summary>
 public static class Futures
 {
-    private static bool s_enableAgentResponseOutputTaggingAndFiltering;
-
     /// <summary>
     /// When <see langword="true"/>, <see cref="AgentResponse"/> and
     /// <see cref="AgentResponseUpdate"/> payloads yielded by an executor participate
@@ -21,14 +19,22 @@ public static class Futures
     /// <see cref="WorkflowOutputEvent.Tags"/> reflecting that designation.
     /// </summary>
     /// <remarks>
+    /// <para>
     /// When <see langword="false"/> (the current default), the runner emits
     /// <see cref="AgentResponseEvent"/> and <see cref="AgentResponseUpdateEvent"/> unconditionally,
     /// bypassing the output filter (historical behavior). Lifecycle: opt-in today, marked
     /// <c>[Obsolete]</c> in v2.0.0 when the new behavior becomes default, and removed in v3.0.0.
+    /// </para>
+    /// <para>
+    /// <b>Interaction with <see cref="WorkflowHostingExtensions.AsAIAgent"/>.</b> When this flag
+    /// is <see langword="true"/>, <see cref="AgentResponseEvent"/> joins
+    /// <see cref="AgentResponseUpdateEvent"/> in being forwarded out of the agent surface
+    /// unconditionally — neither honors the host's <c>includeWorkflowOutputsInResponse</c>
+    /// switch. That switch only governs the generic <see cref="WorkflowOutputEvent"/> path for
+    /// non-AIAgent payloads. When this flag is <see langword="false"/>, the legacy asymmetry
+    /// is preserved: <see cref="AgentResponseUpdateEvent"/> is always forwarded but
+    /// <see cref="AgentResponseEvent"/> stays gated by <c>includeWorkflowOutputsInResponse</c>.
+    /// </para>
     /// </remarks>
-    public static bool EnableAgentResponseOutputTaggingAndFiltering
-    {
-        get => s_enableAgentResponseOutputTaggingAndFiltering;
-        set => s_enableAgentResponseOutputTaggingAndFiltering = value;
-    }
+    public static bool EnableAgentResponseOutputTaggingAndFiltering { get; set; }
 }

dotnet/src/Microsoft.Agents.AI.Workflows/GroupChatWorkflowBuilder.cs

@@ -176,7 +176,7 @@ public sealed class GroupChatWorkflowBuilder
             return;
         }
 
-        foreach ((AIAgent agent, HashSet<OutputTag> tags) in this._outputDesignations)
+        foreach (AIAgent agent in this._outputDesignations.Keys)
         {
             if (!agentMap.TryGetValue(agent, out ExecutorBinding? binding))
             {
@@ -184,6 +184,7 @@ public sealed class GroupChatWorkflowBuilder
                     $"Output designation references agent '{agent.Name ?? agent.Id}', which is not a participant in this group chat workflow.");
             }
 
+            HashSet<OutputTag> tags = this._outputDesignations[agent];
             if (tags.Count == 0)
             {
                 builder.WithOutputFrom(binding);

dotnet/src/Microsoft.Agents.AI.Workflows/HandoffWorkflowBuilder.cs

@@ -720,8 +720,8 @@ public class HandoffWorkflowBuilderCore<TBuilder> where TBuilder : HandoffWorkfl
             return;
         }
 
-        // User took control — replay only their designations, in dictionary order.
-        foreach ((AIAgent agent, HashSet<OutputTag> tags) in this._outputDesignations)
+        // User took control — replay only their designations.
+        foreach (AIAgent agent in this._outputDesignations.Keys)
         {
             if (!executors.TryGetValue(agent.Id, out ExecutorBinding? binding))
             {
@@ -729,6 +729,7 @@ public class HandoffWorkflowBuilderCore<TBuilder> where TBuilder : HandoffWorkfl
                     $"Output designation references agent '{agent.Name ?? agent.Id}', which is not a participant in this handoff workflow.");
             }
 
+            HashSet<OutputTag> tags = this._outputDesignations[agent];
             if (tags.Count == 0)
             {
                 builder.WithOutputFrom(binding);

dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs

@@ -260,8 +260,12 @@ internal sealed class InProcessRunnerContext : IRunnerContext
         }
 
         Executor sourceExecutor = await this.EnsureExecutorAsync(sourceId, tracer: null, cancellationToken).ConfigureAwait(false);
-        if (!sourceExecutor.CanOutput(output.GetType()))
+        if (!isAgentResponseShaped && !sourceExecutor.CanOutput(output.GetType()))
         {
+            // AIAgent-shaped payloads bypass the per-executor declared-yield check (matching the
+            // legacy bypass branch above). The AIAgent host executor relays the agent's output
+            // without declaring AgentResponse(Update) in its Yields set, so a CanOutput probe
+            // here would always reject — but those payloads are always a valid output shape.
             throw new InvalidOperationException($"Cannot output object of type {output.GetType().Name}. Expecting one of [{string.Join(", ", sourceExecutor.OutputTypes)}].");
         }
 

dotnet/src/Microsoft.Agents.AI.Workflows/MagenticWorkflowBuilder.cs

@@ -205,7 +205,7 @@ public class MagenticWorkflowBuilder(AIAgent managerAgent)
             return;
         }
 
-        foreach ((AIAgent agent, HashSet<OutputTag> tags) in this._outputDesignations)
+        foreach (AIAgent agent in this._outputDesignations.Keys)
         {
             if (!teamMap.TryGetValue(agent, out ExecutorBinding? binding))
             {
@@ -213,6 +213,7 @@ public class MagenticWorkflowBuilder(AIAgent managerAgent)
                     $"Output designation references agent '{agent.Name ?? agent.Id}', which is not a participant in this Magentic workflow.");
             }
 
+            HashSet<OutputTag> tags = this._outputDesignations[agent];
             if (tags.Count == 0)
             {
                 builder.WithOutputFrom(binding);

dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs

@@ -520,7 +520,12 @@ internal sealed class WorkflowSession : AgentSession
                     goto default;
 
                 case AgentResponseEvent agentResponse:
-                    if (!this._includeWorkflowOutputsInResponse)
+                    // Under Futures.EnableAgentResponseOutputTaggingAndFiltering=true, mirror
+                    // AgentResponseUpdateEvent's behavior: always forward, regardless of the
+                    // _includeWorkflowOutputsInResponse host flag. Under the legacy default,
+                    // keep today's behavior — gated by the include flag.
+                    if (!Futures.EnableAgentResponseOutputTaggingAndFiltering &&
+                        !this._includeWorkflowOutputsInResponse)
                     {
                         goto default;
                     }
@@ -539,7 +544,11 @@ internal sealed class WorkflowSession : AgentSession
                         _ => null
                     };
 
-                    if (!this._includeWorkflowOutputsInResponse || updateMessages == null)
+                    // Same gating asymmetry as AgentResponseEvent: intermediate outputs are
+                    // forwarded unconditionally; terminal/untagged outputs require the host
+                    // to opt in via _includeWorkflowOutputsInResponse.
+                    if (updateMessages == null ||
+                        (!output.IsIntermediate() && !this._includeWorkflowOutputsInResponse))
                     {
                         goto default;
                     }

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

@@ -824,4 +824,130 @@ public class WorkflowHostSmokeTests : AIAgentHostingExecutorTestsBase
         Workflow handoffWorkflow = new HandoffWorkflowBuilder(agent).Build();
         return this.Run_AsAgent_OutgoingMessagesInHistoryAsync(handoffWorkflow, runAsync);
     }
+
+    // ----- Phase 5: Workflow-as-Agent intermediate forwarding -----------------
+
+    [Collection(Futures.FuturesSerialCollection.Name)]
+    public class IntermediateForwarding
+    {
+        private const string InterText = "progress";
+        private const string FinalText = "final";
+
+        private static async Task<List<AgentResponseUpdate>> RunStreamingAsync(
+            Workflow workflow,
+            bool includeWorkflowOutputsInResponse = false)
+        {
+            return await workflow
+                .AsAIAgent("WorkflowAgent", includeWorkflowOutputsInResponse: includeWorkflowOutputsInResponse)
+                .RunStreamingAsync(new ChatMessage(ChatRole.User, "hi"))
+                .ToListAsync();
+        }
+
+        [Fact]
+        public async Task Test_WorkflowHostAgent_IntermediateAgentResponseForwardedInStreamingAsync()
+        {
+            using Futures.FuturesScope _ = new(enabled: true);
+            TestReplayAgent agent = new(TestReplayAgent.ToChatMessages(InterText));
+            ExecutorBinding binding = agent.BindAsExecutor(new AIAgentHostOptions { EmitAgentResponseEvents = true });
+            Workflow workflow = new WorkflowBuilder(binding)
+                .WithIntermediateOutputFrom([binding])
+                .Build();
+
+            // Under Futures-on, AgentResponseEvent mirrors AgentResponseUpdateEvent: always
+            // forwarded regardless of the include flag. The intermediate tag is observable on
+            // the surfaced event for consumers that care to distinguish.
+            List<AgentResponseUpdate> updates = await RunStreamingAsync(workflow, includeWorkflowOutputsInResponse: false);
+
+            updates.Any(u => u.RawRepresentation is AgentResponseEvent are && are.IsIntermediate() && u.Text == InterText)
+                .Should().BeTrue("AgentResponseEvent is forwarded under Futures-on regardless of the include flag");
+        }
+
+        [Fact]
+        public async Task Test_WorkflowHostAgent_TerminalAgentResponseForwardedUnconditionallyWhenFuturesOnAsync()
+        {
+            using Futures.FuturesScope _ = new(enabled: true);
+            TestReplayAgent agent = new(TestReplayAgent.ToChatMessages(FinalText));
+            ExecutorBinding binding = agent.BindAsExecutor(new AIAgentHostOptions { EmitAgentResponseEvents = true });
+            Workflow workflow = new WorkflowBuilder(binding)
+                .WithOutputFrom(binding)
+                .Build();
+
+            // Even a terminal-only designation surfaces without the include flag — the gating
+            // asymmetry between AgentResponse and AgentResponseUpdate is gone under Futures-on.
+            List<AgentResponseUpdate> updates = await RunStreamingAsync(workflow, includeWorkflowOutputsInResponse: false);
+
+            updates.Any(u => u.RawRepresentation is AgentResponseEvent && u.Text == FinalText)
+                .Should().BeTrue("terminal AgentResponseEvent is forwarded under Futures-on regardless of the include flag");
+        }
+
+        [Fact]
+        public async Task Test_WorkflowHostAgent_TerminalAgentResponseGatedWhenFuturesOffAsync()
+        {
+            using Futures.FuturesScope _ = new(enabled: false);
+
+            static Workflow Build()
+            {
+                TestReplayAgent agent = new(TestReplayAgent.ToChatMessages(FinalText));
+                ExecutorBinding binding = agent.BindAsExecutor(new AIAgentHostOptions { EmitAgentResponseEvents = true });
+                return new WorkflowBuilder(binding).WithOutputFrom(binding).Build();
+            }
+
+            // Legacy semantics: AgentResponseEvent stays behind the include flag when Futures
+            // is off. Two fresh workflows because in-process runs aren't reentrant.
+            List<AgentResponseUpdate> gated = await RunStreamingAsync(Build(), includeWorkflowOutputsInResponse: false);
+            gated.Any(u => u.RawRepresentation is AgentResponseEvent && u.Text == FinalText)
+                .Should().BeFalse("terminal AgentResponseEvent stays gated under Futures-off");
+
+            List<AgentResponseUpdate> included = await RunStreamingAsync(Build(), includeWorkflowOutputsInResponse: true);
+            included.Any(u => u.RawRepresentation is AgentResponseEvent && u.Text == FinalText)
+                .Should().BeTrue("opting in via includeWorkflowOutputsInResponse surfaces it");
+        }
+
+        [Fact]
+        public async Task Test_WorkflowHostAgent_UndesignatedExecutorEmitsNoAgentResponseEventWhenFuturesOnAsync()
+        {
+            using Futures.FuturesScope _ = new(enabled: true);
+            TestReplayAgent agent = new(TestReplayAgent.ToChatMessages(InterText));
+            ExecutorBinding binding = agent.BindAsExecutor(new AIAgentHostOptions { EmitAgentResponseEvents = true });
+            // No designation — under Futures-on, the AgentResponse is dropped by the filter.
+            Workflow workflow = new WorkflowBuilder(binding).Build();
+
+            List<AgentResponseUpdate> updates = await RunStreamingAsync(workflow, includeWorkflowOutputsInResponse: true);
+
+            updates.Any(u => u.RawRepresentation is AgentResponseEvent)
+                .Should().BeFalse("an undesignated AIAgent executor produces no AgentResponseEvent under Futures-on");
+        }
+
+        [Fact]
+        public async Task Test_WorkflowHostAgent_UndesignatedAgentResponseSurfacesWhenFuturesOffAsync()
+        {
+            using Futures.FuturesScope _ = new(enabled: false);
+            TestReplayAgent agent = new(TestReplayAgent.ToChatMessages(InterText));
+            ExecutorBinding binding = agent.BindAsExecutor(new AIAgentHostOptions { EmitAgentResponseEvents = true });
+            Workflow workflow = new WorkflowBuilder(binding).Build();
+
+            List<AgentResponseUpdate> updates = await RunStreamingAsync(workflow, includeWorkflowOutputsInResponse: true);
+
+            updates.Any(u => u.RawRepresentation is AgentResponseEvent && u.Text == InterText)
+                .Should().BeTrue("legacy bypass still emits AgentResponseEvent regardless of designation");
+        }
+
+        [Fact]
+        public async Task Test_WorkflowHostAgent_IntermediateTagAvailableViaRawRepresentationAsync()
+        {
+            using Futures.FuturesScope _ = new(enabled: true);
+            TestReplayAgent agent = new(TestReplayAgent.ToChatMessages(InterText));
+            ExecutorBinding binding = agent.BindAsExecutor(new AIAgentHostOptions { EmitAgentResponseEvents = true });
+            Workflow workflow = new WorkflowBuilder(binding)
+                .WithIntermediateOutputFrom([binding])
+                .Build();
+
+            List<AgentResponseUpdate> updates = await RunStreamingAsync(workflow);
+
+            AgentResponseUpdate progress = updates.First(u => u.RawRepresentation is AgentResponseEvent && u.Text == InterText);
+            AgentResponseEvent raw = (AgentResponseEvent)progress.RawRepresentation!;
+            raw.IsIntermediate().Should().BeTrue();
+            raw.Tags.Should().BeEquivalentTo(new[] { OutputTag.Intermediate });
+        }
+    }
 }