diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Futures.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Futures.cs index f19ba88832..8403d5fbb0 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Futures.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Futures.cs @@ -9,8 +9,6 @@ namespace Microsoft.Agents.AI.Workflows; /// public static class Futures { - private static bool s_enableAgentResponseOutputTaggingAndFiltering; - /// /// When , and /// payloads yielded by an executor participate @@ -21,14 +19,22 @@ public static class Futures /// reflecting that designation. /// /// + /// /// When (the current default), the runner emits /// and unconditionally, /// bypassing the output filter (historical behavior). Lifecycle: opt-in today, marked /// [Obsolete] in v2.0.0 when the new behavior becomes default, and removed in v3.0.0. + /// + /// + /// Interaction with . When this flag + /// is , joins + /// in being forwarded out of the agent surface + /// unconditionally — neither honors the host's includeWorkflowOutputsInResponse + /// switch. That switch only governs the generic path for + /// non-AIAgent payloads. When this flag is , the legacy asymmetry + /// is preserved: is always forwarded but + /// stays gated by includeWorkflowOutputsInResponse. + /// /// - public static bool EnableAgentResponseOutputTaggingAndFiltering - { - get => s_enableAgentResponseOutputTaggingAndFiltering; - set => s_enableAgentResponseOutputTaggingAndFiltering = value; - } + public static bool EnableAgentResponseOutputTaggingAndFiltering { get; set; } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/GroupChatWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/GroupChatWorkflowBuilder.cs index 253377137f..a19915f5d5 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/GroupChatWorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/GroupChatWorkflowBuilder.cs @@ -176,7 +176,7 @@ public sealed class GroupChatWorkflowBuilder return; } - foreach ((AIAgent agent, HashSet 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 tags = this._outputDesignations[agent]; if (tags.Count == 0) { builder.WithOutputFrom(binding); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/HandoffWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/HandoffWorkflowBuilder.cs index fe09a6beba..26e83c84f5 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/HandoffWorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/HandoffWorkflowBuilder.cs @@ -720,8 +720,8 @@ public class HandoffWorkflowBuilderCore where TBuilder : HandoffWorkfl return; } - // User took control — replay only their designations, in dictionary order. - foreach ((AIAgent agent, HashSet 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 where TBuilder : HandoffWorkfl $"Output designation references agent '{agent.Name ?? agent.Id}', which is not a participant in this handoff workflow."); } + HashSet tags = this._outputDesignations[agent]; if (tags.Count == 0) { builder.WithOutputFrom(binding); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs index 8c8eac3562..dec2b31bb8 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs +++ b/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)}]."); } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/MagenticWorkflowBuilder.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/MagenticWorkflowBuilder.cs index 24926ceebb..0c8782ceb2 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/MagenticWorkflowBuilder.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/MagenticWorkflowBuilder.cs @@ -205,7 +205,7 @@ public class MagenticWorkflowBuilder(AIAgent managerAgent) return; } - foreach ((AIAgent agent, HashSet 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 tags = this._outputDesignations[agent]; if (tags.Count == 0) { builder.WithOutputFrom(binding); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs index 719b72e112..2caba8f3ec 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs +++ b/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; } diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs index 7b9b428871..4402142fae 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs +++ b/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> 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 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 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 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 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 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 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 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 }); + } + } }