feat: route AgentResponse(Update) through the output filter under a Futures flag

`InProcessRunnerContext.YieldOutputAsync` historically special-cased AgentResponse and AgentResponseUpdate payloads: it built the typed event subclass and emitted it directly, bypassing the output filter. Rewrites the method so that: - When `Futures.EnableAgentResponseOutputTaggingAndFiltering` is `false` (the current default), AgentResponse(Update) keep the legacy bypass — emitted as AgentResponseEvent / AgentResponseUpdateEvent with no tags. Existing callers see no behavior change. - When the flag is `true`, AIAgent payloads flow through the output filter just like every other payload type: undesignated sources are dropped, and the emitted event carries the source's tag set (empty for terminal `WithOutputFrom`, `{Intermediate}` for `WithIntermediateOutputFrom`, the set union when both designations apply). Non-AIAgent (POCO) outputs also now carry the source's tag set on the emitted WorkflowOutputEvent unconditionally — additive, since no existing assertion inspected Tags. Subclass events (`AgentResponseEvent` / `AgentResponseUpdateEvent`) continue to be emitted under both modes so `switch (evt) { case AgentResponseEvent: ... }` consumer code keeps matching. Adds `OutputFilter.TryGetTags` as the tag-aware lookup used by the runner. `OutputFilter.CanOutput` is kept (still used by the existing sync tests in `OutputFilterTests.cs`). Tests ----- - `Futures/Futures.AgentResponseOutputFilteringAndTaggingTests.cs` (new): the F1–F13 matrix from the plan, covering every combination of `(flag on/off) × (designation) × (payload shape)`. Uses a `FuturesScope` IDisposable + a `FuturesSerial` xUnit collection (DisableParallelization = true) to keep the process-global flag from leaking across parallel tests. - `OutputFilterTests.cs`: four new `Test_OutputFilter_…` cases for the `TryGetTags` surface (empty-tag-set for terminal designation, `{Intermediate}` for intermediate designation, union for accumulated designation, `false` for unregistered). 582/582 unit tests pass on net10.0 (565 baseline + 17 new). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
2026-06-16 21:04:09 +08:00 · 2026-05-22 13:11:04 -04:00
parent 59b6e1f6e0
commit e03a93c60f
6 changed files with 422 additions and 12 deletions
@@ -1,5 +1,8 @@
 // Copyright (c) Microsoft. All rights reserved.

+using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
+
 namespace Microsoft.Agents.AI.Workflows.Execution;

 internal sealed class OutputFilter(Workflow workflow)
@@ -8,4 +11,7 @@ internal sealed class OutputFilter(Workflow workflow)
    {
        return workflow.OutputExecutors.ContainsKey(sourceExecutorId);
    }
+
+    public bool TryGetTags(string sourceExecutorId, [NotNullWhen(true)] out HashSet<OutputTag>? tags)
+        => workflow.OutputExecutors.TryGetValue(sourceExecutorId, out tags);
 }
@@ -241,17 +241,21 @@ internal sealed class InProcessRunnerContext : IRunnerContext
        this.CheckEnded();
        Throw.IfNull(output);

-        // Special-case AgentResponse and AgentResponseUpdate to create their specific event types
-        // and bypass the output filter (for backwards compatibility - these events were previously
-        // emitted directly via AddEventAsync without filtering)
-        if (output is AgentResponseUpdate update)
+        bool isAgentResponseShaped = output is AgentResponse or AgentResponseUpdate;
+
+        if (isAgentResponseShaped && !Futures.EnableAgentResponseOutputTaggingAndFiltering)
        {
-            await this.AddEventAsync(new AgentResponseUpdateEvent(sourceId, update), cancellationToken).ConfigureAwait(false);
-            return;
-        }
-        else if (output is AgentResponse response)
-        {
-            await this.AddEventAsync(new AgentResponseEvent(sourceId, response), cancellationToken).ConfigureAwait(false);
+            // Legacy bypass: AgentResponse/AgentResponseUpdate skip the output filter and are
+            // emitted as their typed event subclasses with no tags. Preserved verbatim for
+            // back-compat; once Futures.EnableAgentResponseOutputTaggingAndFiltering becomes the
+            // default in v2.0.0, this branch goes away.
+            WorkflowEvent typedEvent = output switch
+            {
+                AgentResponseUpdate u => new AgentResponseUpdateEvent(sourceId, u),
+                AgentResponse r => new AgentResponseEvent(sourceId, r),
+                _ => throw new InvalidOperationException("Unexpected AIAgent-shaped payload type."),
+            };
+            await this.AddEventAsync(typedEvent, cancellationToken).ConfigureAwait(false);
            return;
        }

@@ -261,10 +265,19 @@ internal sealed class InProcessRunnerContext : IRunnerContext
            throw new InvalidOperationException($"Cannot output object of type {output.GetType().Name}. Expecting one of [{string.Join(", ", sourceExecutor.OutputTypes)}].");
        }

-        if (this._outputFilter.CanOutput(sourceId, output))
+        if (!this._outputFilter.TryGetTags(sourceId, out HashSet<OutputTag>? tags))
        {
-            await this.AddEventAsync(new WorkflowOutputEvent(output, sourceId), cancellationToken).ConfigureAwait(false);
+            // Not designated as an output source — drop silently (matches Python semantics).
+            return;
        }
+
+        WorkflowOutputEvent evt = output switch
+        {
+            AgentResponseUpdate u => new AgentResponseUpdateEvent(sourceId, u, tags),
+            AgentResponse r => new AgentResponseEvent(sourceId, r, tags),
+            _ => new WorkflowOutputEvent(output, sourceId, tags),
+        };
+        await this.AddEventAsync(evt, cancellationToken).ConfigureAwait(false);
    }

    public IExternalRequestContext BindExternalRequestContext(string executorId)
@@ -0,0 +1,290 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using FluentAssertions;
+using Microsoft.Extensions.AI;
+using Xunit;
+
+namespace Microsoft.Agents.AI.Workflows.UnitTests.Futures;
+
+/// <summary>
+/// Runner-level coverage for <see cref="Workflows.Futures.EnableAgentResponseOutputTaggingAndFiltering"/>.
+/// Exercises every combination of (flag on/off) × (designation kind) × (payload shape) to pin the
+/// runner's behavior in both the legacy bypass path and the unified filter-and-tag path.
+/// </summary>
+public static partial class FuturesTests
+{
+    [Collection(FuturesSerialCollection.Name)]
+    public class AgentResponseOutputFilteringAndTaggingTests
+    {
+        private const string SourceId = "yielder";
+
+        private static AgentResponse SampleResponse(string text = "hi")
+            => new(new ChatMessage(ChatRole.Assistant, text));
+
+        private static AgentResponseUpdate SampleUpdate(string text = "tick")
+            => new(ChatRole.Assistant, text);
+
+        private static async Task<List<WorkflowEvent>> RunAsync<T>(Workflow workflow, T input) where T : notnull
+        {
+            List<WorkflowEvent> events = [];
+            await using StreamingRun run = await InProcessExecution.RunStreamingAsync(workflow, input).ConfigureAwait(false);
+            await foreach (WorkflowEvent evt in run.WatchStreamAsync().ConfigureAwait(false))
+            {
+                events.Add(evt);
+            }
+            return events;
+        }
+
+        private static Workflow BuildAgentResponseWorkflow(Action<WorkflowBuilder, YieldAgentResponseExecutor>? designate = null)
+        {
+            YieldAgentResponseExecutor exec = new(SourceId);
+            WorkflowBuilder builder = new(exec);
+            designate?.Invoke(builder, exec);
+            return builder.Build();
+        }
+
+        private static Workflow BuildAgentResponseUpdateWorkflow(Action<WorkflowBuilder, YieldAgentResponseUpdateExecutor>? designate = null)
+        {
+            YieldAgentResponseUpdateExecutor exec = new(SourceId);
+            WorkflowBuilder builder = new(exec);
+            designate?.Invoke(builder, exec);
+            return builder.Build();
+        }
+
+        private static Workflow BuildPocoWorkflow(Action<WorkflowBuilder, YieldPocoExecutor>? designate = null)
+        {
+            YieldPocoExecutor exec = new(SourceId);
+            WorkflowBuilder builder = new(exec);
+            designate?.Invoke(builder, exec);
+            return builder.Build();
+        }
+
+        // F1
+        [Fact]
+        public async Task Test_Runner_LegacyAgentResponseBypass_RaisesUntaggedEventAsync()
+        {
+            using FuturesScope _ = new(enabled: false);
+            Workflow workflow = BuildAgentResponseWorkflow(designate: null);
+
+            List<WorkflowEvent> events = await RunAsync(workflow, "go");
+
+            AgentResponseEvent emitted = events.OfType<AgentResponseEvent>().Should().ContainSingle().Subject;
+            emitted.ExecutorId.Should().Be(SourceId);
+            emitted.Tags.Should().BeEmpty("legacy bypass attaches no tags");
+            emitted.IsIntermediate().Should().BeFalse();
+        }
+
+        // F2
+        [Fact]
+        public async Task Test_Runner_LegacyAgentResponseUpdateBypass_RaisesUntaggedEventAsync()
+        {
+            using FuturesScope _ = new(enabled: false);
+            Workflow workflow = BuildAgentResponseUpdateWorkflow(designate: null);
+
+            List<WorkflowEvent> events = await RunAsync(workflow, "go");
+
+            AgentResponseUpdateEvent emitted = events.OfType<AgentResponseUpdateEvent>().Should().ContainSingle().Subject;
+            emitted.Tags.Should().BeEmpty();
+        }
+
+        // F3
+        [Fact]
+        public async Task Test_Runner_LegacyBypassIgnoresDesignationAsync()
+        {
+            using FuturesScope _ = new(enabled: false);
+            Workflow workflow = BuildAgentResponseWorkflow(static (b, e) => b.WithIntermediateOutputFrom([e]));
+
+            List<WorkflowEvent> events = await RunAsync(workflow, "go");
+
+            AgentResponseEvent emitted = events.OfType<AgentResponseEvent>().Should().ContainSingle().Subject;
+            emitted.Tags.Should().BeEmpty("legacy bypass ignores the designation entirely");
+            emitted.IsIntermediate().Should().BeFalse("legacy bypass does not propagate tags");
+        }
+
+        // F4
+        [Fact]
+        public async Task Test_Runner_LegacyPocoIsFilteredAsync()
+        {
+            using FuturesScope _ = new(enabled: false);
+            Workflow workflow = BuildPocoWorkflow(designate: null);
+
+            List<WorkflowEvent> events = await RunAsync(workflow, "go");
+
+            events.OfType<WorkflowOutputEvent>().Should().BeEmpty("POCO outputs always go through the filter; undesignated source is dropped");
+        }
+
+        // F5
+        [Fact]
+        public async Task Test_Runner_UndesignatedAgentResponseIsFilteredWhenFuturesOnAsync()
+        {
+            using FuturesScope _ = new(enabled: true);
+            Workflow workflow = BuildAgentResponseWorkflow(designate: null);
+
+            List<WorkflowEvent> events = await RunAsync(workflow, "go");
+
+            events.OfType<WorkflowOutputEvent>().Should().BeEmpty(
+                "with the future on, AgentResponse must be designated to surface");
+        }
+
+        // F6
+        [Fact]
+        public async Task Test_Runner_DesignatedTerminalAgentResponseHasEmptyTagsAsync()
+        {
+            using FuturesScope _ = new(enabled: true);
+            Workflow workflow = BuildAgentResponseWorkflow(static (b, e) => b.WithOutputFrom(e));
+
+            List<WorkflowEvent> events = await RunAsync(workflow, "go");
+
+            AgentResponseEvent emitted = events.OfType<AgentResponseEvent>().Should().ContainSingle().Subject;
+            emitted.Tags.Should().BeEmpty("terminal designation carries no tag");
+            emitted.IsIntermediate().Should().BeFalse();
+        }
+
+        // F7
+        [Fact]
+        public async Task Test_Runner_DesignatedIntermediateAgentResponseHasIntermediateTagAsync()
+        {
+            using FuturesScope _ = new(enabled: true);
+            Workflow workflow = BuildAgentResponseWorkflow(static (b, e) => b.WithIntermediateOutputFrom([e]));
+
+            List<WorkflowEvent> events = await RunAsync(workflow, "go");
+
+            AgentResponseEvent emitted = events.OfType<AgentResponseEvent>().Should().ContainSingle().Subject;
+            emitted.Tags.Should().BeEquivalentTo(new[] { OutputTag.Intermediate });
+            emitted.IsIntermediate().Should().BeTrue();
+        }
+
+        // F8
+        [Fact]
+        public async Task Test_Runner_DesignatedIntermediateAgentResponseUpdateHasIntermediateTagAsync()
+        {
+            using FuturesScope _ = new(enabled: true);
+            Workflow workflow = BuildAgentResponseUpdateWorkflow(static (b, e) => b.WithIntermediateOutputFrom([e]));
+
+            List<WorkflowEvent> events = await RunAsync(workflow, "go");
+
+            AgentResponseUpdateEvent emitted = events.OfType<AgentResponseUpdateEvent>().Should().ContainSingle().Subject;
+            emitted.Tags.Should().BeEquivalentTo(new[] { OutputTag.Intermediate });
+            emitted.IsIntermediate().Should().BeTrue();
+        }
+
+        // F9
+        [Fact]
+        public async Task Test_Runner_TagsAccumulateOutputThenIntermediateAsync()
+        {
+            using FuturesScope _ = new(enabled: true);
+            Workflow workflow = BuildAgentResponseWorkflow(static (b, e) =>
+            {
+                b.WithOutputFrom(e);
+                b.WithIntermediateOutputFrom([e]);
+            });
+
+            List<WorkflowEvent> events = await RunAsync(workflow, "go");
+
+            AgentResponseEvent emitted = events.OfType<AgentResponseEvent>().Should().ContainSingle().Subject;
+            emitted.Tags.Should().BeEquivalentTo(new[] { OutputTag.Intermediate },
+                "terminal+intermediate union is {{ Intermediate }} (terminal contributes the entry but no tag)");
+            emitted.IsIntermediate().Should().BeTrue();
+        }
+
+        // F10
+        [Fact]
+        public async Task Test_Runner_TagsAccumulateIntermediateThenOutputAsync()
+        {
+            using FuturesScope _ = new(enabled: true);
+            Workflow workflow = BuildAgentResponseWorkflow(static (b, e) =>
+            {
+                b.WithIntermediateOutputFrom([e]);
+                b.WithOutputFrom(e);
+            });
+
+            List<WorkflowEvent> events = await RunAsync(workflow, "go");
+
+            AgentResponseEvent emitted = events.OfType<AgentResponseEvent>().Should().ContainSingle().Subject;
+            emitted.Tags.Should().BeEquivalentTo(new[] { OutputTag.Intermediate }, "designation order is irrelevant");
+            emitted.IsIntermediate().Should().BeTrue();
+        }
+
+        // F11
+        [Fact]
+        public async Task Test_Runner_DesignatedIntermediatePocoHasIntermediateTagAsync()
+        {
+            using FuturesScope _ = new(enabled: true);
+            Workflow workflow = BuildPocoWorkflow(static (b, e) => b.WithIntermediateOutputFrom([e]));
+
+            List<WorkflowEvent> events = await RunAsync(workflow, "go");
+
+            WorkflowOutputEvent emitted = events.OfType<WorkflowOutputEvent>().Should().ContainSingle().Subject;
+            emitted.Should().NotBeOfType<AgentResponseEvent>();
+            emitted.Tags.Should().BeEquivalentTo(new[] { OutputTag.Intermediate });
+            emitted.IsIntermediate().Should().BeTrue();
+        }
+
+        // F12
+        [Fact]
+        public async Task Test_Runner_DesignatedTerminalPocoHasEmptyTagsAsync()
+        {
+            using FuturesScope _ = new(enabled: true);
+            Workflow workflow = BuildPocoWorkflow(static (b, e) => b.WithOutputFrom(e));
+
+            List<WorkflowEvent> events = await RunAsync(workflow, "go");
+
+            WorkflowOutputEvent emitted = events.OfType<WorkflowOutputEvent>().Should().ContainSingle().Subject;
+            emitted.Tags.Should().BeEmpty();
+            emitted.IsIntermediate().Should().BeFalse();
+        }
+
+        // F13
+        [Fact]
+        public async Task Test_Runner_RepeatedTerminalDesignationDedupesAsync()
+        {
+            using FuturesScope _ = new(enabled: true);
+            Workflow workflow = BuildAgentResponseWorkflow(static (b, e) =>
+            {
+                b.WithOutputFrom(e);
+                b.WithOutputFrom(e);
+            });
+
+            List<WorkflowEvent> events = await RunAsync(workflow, "go");
+
+            AgentResponseEvent emitted = events.OfType<AgentResponseEvent>().Should().ContainSingle().Subject;
+            emitted.Tags.Should().BeEmpty("repeated terminal designation contributes no tag");
+        }
+
+        // ---- Executors -----------------------------------------------------------
+
+        internal sealed class YieldAgentResponseExecutor(string id) : Executor(id)
+        {
+            protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
+                => protocolBuilder.ConfigureRoutes(rb => rb.AddHandler<string, AgentResponse>(this.HandleAsync));
+
+            private ValueTask<AgentResponse> HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken)
+                => new(SampleResponse(input));
+        }
+
+        internal sealed class YieldAgentResponseUpdateExecutor(string id) : Executor(id)
+        {
+            protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
+                => protocolBuilder.ConfigureRoutes(rb => rb.AddHandler<string, AgentResponseUpdate>(this.HandleAsync));
+
+            private ValueTask<AgentResponseUpdate> HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken)
+                => new(SampleUpdate(input));
+        }
+
+        public sealed record Poco(string Value);
+
+        internal sealed class YieldPocoExecutor(string id) : Executor(id)
+        {
+            protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)
+                => protocolBuilder.ConfigureRoutes(rb => rb.AddHandler<string, Poco>(this.HandleAsync));
+
+            private ValueTask<Poco> HandleAsync(string input, IWorkflowContext context, CancellationToken cancellationToken)
+                => new(new Poco(input));
+        }
+    }
+}
@@ -0,0 +1,27 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System;
+
+namespace Microsoft.Agents.AI.Workflows.UnitTests.Futures;
+
+/// <summary>
+/// Sets <see cref="Workflows.Futures.EnableAgentResponseOutputTaggingAndFiltering"/> for
+/// the lifetime of the scope, restoring the prior value on dispose. Pair every use with
+/// <c>using</c> and run inside the <c>FuturesSerial</c> xUnit collection to avoid leaking
+/// state across parallel tests.
+/// </summary>
+internal sealed class FuturesScope : IDisposable
+{
+    private readonly bool _previous;
+
+    public FuturesScope(bool enabled)
+    {
+        this._previous = Workflows.Futures.EnableAgentResponseOutputTaggingAndFiltering;
+        Workflows.Futures.EnableAgentResponseOutputTaggingAndFiltering = enabled;
+    }
+
+    public void Dispose()
+    {
+        Workflows.Futures.EnableAgentResponseOutputTaggingAndFiltering = this._previous;
+    }
+}
@@ -0,0 +1,19 @@
+// Copyright (c) Microsoft. All rights reserved.
+
+using System.Diagnostics.CodeAnalysis;
+
+namespace Microsoft.Agents.AI.Workflows.UnitTests.Futures;
+
+/// <summary>
+/// xUnit collection marker for tests that mutate the process-global
+/// <see cref="Workflows.Futures"/> switches. Membership in this collection serializes
+/// the tests against each other so that <see cref="FuturesScope"/> cannot leak state
+/// into a concurrently running test.
+/// </summary>
+[CollectionDefinition(Name, DisableParallelization = true)]
+[SuppressMessage("Naming", "CA1711:Identifiers should not have incorrect suffix",
+    Justification = "xUnit's [CollectionDefinition] pattern names the marker type after the collection's purpose; the 'Collection' suffix is idiomatic.")]
+public sealed class FuturesSerialCollection
+{
+    public const string Name = "FuturesSerial";
+}
@@ -1,5 +1,6 @@
 // Copyright (c) Microsoft. All rights reserved.

+using System.Collections.Generic;
 using FluentAssertions;
 using Microsoft.Agents.AI.Workflows.Execution;

@@ -44,6 +45,60 @@ public class OutputFilterTests
        filter.CanOutput("nonexistent", "some output").Should().BeFalse("an executor not in the workflow should not be an output executor");
    }

+    [Fact]
+    public void Test_OutputFilter_ReturnsEmptyTagSetWhenRegisteredViaWithOutputFrom()
+    {
+        OutputFilter filter = CreateFilterWithOutputFrom("end");
+
+        filter.TryGetTags("end", out HashSet<OutputTag>? tags).Should().BeTrue();
+        tags.Should().NotBeNull().And.BeEmpty("terminal designation carries no tag");
+    }
+
+    [Fact]
+    public void Test_OutputFilter_ReturnsIntermediateTagWhenRegisteredViaWithIntermediateOutputFrom()
+    {
+        NoOpExecutor start = new("start");
+        NoOpExecutor end = new("end");
+
+        Workflow workflow = new WorkflowBuilder("start")
+            .AddEdge(start, end)
+            .WithIntermediateOutputFrom([end])
+            .Build();
+
+        OutputFilter filter = new(workflow);
+
+        filter.TryGetTags("end", out HashSet<OutputTag>? tags).Should().BeTrue();
+        tags.Should().BeEquivalentTo(new[] { OutputTag.Intermediate });
+    }
+
+    [Fact]
+    public void Test_OutputFilter_ReturnsIntermediateTagForAccumulatedDesignation()
+    {
+        NoOpExecutor start = new("start");
+        NoOpExecutor end = new("end");
+
+        Workflow workflow = new WorkflowBuilder("start")
+            .AddEdge(start, end)
+            .WithOutputFrom(end)
+            .WithIntermediateOutputFrom([end])
+            .Build();
+
+        OutputFilter filter = new(workflow);
+
+        filter.TryGetTags("end", out HashSet<OutputTag>? tags).Should().BeTrue();
+        tags.Should().BeEquivalentTo(new[] { OutputTag.Intermediate },
+            "terminal designation contributes no tag; the union is the intermediate set");
+    }
+
+    [Fact]
+    public void Test_OutputFilter_TryGetTagsReturnsFalseForUnregisteredExecutor()
+    {
+        OutputFilter filter = CreateFilterWithOutputFrom("end");
+
+        filter.TryGetTags("start", out HashSet<OutputTag>? tags).Should().BeFalse();
+        tags.Should().BeNull();
+    }
+
    private sealed class NoOpExecutor(string id) : Executor(id)
    {
        protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder)