diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/OutputFilter.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/OutputFilter.cs
index c5bcf62466..929e5cb6cc 100644
--- a/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/OutputFilter.cs
+++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Execution/OutputFilter.cs
@@ -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);
 }
diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs
index d6c7d301e3..8c8eac3562 100644
--- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs
+++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs
@@ -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)
diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Futures/Futures.AgentResponseOutputFilteringAndTaggingTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Futures/Futures.AgentResponseOutputFilteringAndTaggingTests.cs
new file mode 100644
index 0000000000..ba364f7e64
--- /dev/null
+++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Futures/Futures.AgentResponseOutputFilteringAndTaggingTests.cs
@@ -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));
+        }
+    }
+}
diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Futures/FuturesScope.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Futures/FuturesScope.cs
new file mode 100644
index 0000000000..9eba02cf06
--- /dev/null
+++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Futures/FuturesScope.cs
@@ -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;
+    }
+}
diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Futures/FuturesSerialCollection.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Futures/FuturesSerialCollection.cs
new file mode 100644
index 0000000000..2725fe6a50
--- /dev/null
+++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Futures/FuturesSerialCollection.cs
@@ -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";
+}
diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/OutputFilterTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/OutputFilterTests.cs
index bb4a0e59bd..55bfacf9a1 100644
--- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/OutputFilterTests.cs
+++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/OutputFilterTests.cs
@@ -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)