// Copyright (c) Microsoft. All rights reserved. using System; using System.Collections.Generic; using System.Linq; using System.Runtime.CompilerServices; using System.Text.Json; using System.Threading; using System.Threading.Tasks; using Azure.AI.AgentServer.Responses; using Azure.AI.AgentServer.Responses.Models; using Microsoft.Agents.AI.Foundry.Hosting; using Microsoft.Agents.AI.Workflows; using Microsoft.Extensions.AI; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using Moq; using MeaiTextContent = Microsoft.Extensions.AI.TextContent; namespace Microsoft.Agents.AI.Foundry.UnitTests.Hosting; /// /// Integration tests that verify workflow execution through the /// pipeline. /// These use real workflow builders and the InProcessExecution environment /// to produce authentic streaming event patterns. /// public class WorkflowIntegrationTests { // ===== Sequential Workflow Tests ===== [Fact] public async Task SequentialWorkflow_SingleAgent_ProducesTextOutputAsync() { // Arrange: single-agent sequential workflow var echoAgent = new StreamingTextAgent("echo", "Hello from the workflow!"); var workflow = AgentWorkflowBuilder.BuildSequential("test-sequential", echoAgent); var workflowAgent = workflow.AsAIAgent( id: "workflow-agent", name: "Test Workflow", executionEnvironment: InProcessExecution.OffThread, includeExceptionDetails: true); var (handler, request, context) = CreateHandlerWithAgent(workflowAgent, "Hello"); // Act var events = await CollectEventsAsync(handler, request, context); // Assert: should have lifecycle events + at least one text output + terminal Assert.IsType(events[0]); Assert.IsType(events[1]); Assert.True(events.Count >= 4, $"Expected at least 4 events, got {events.Count}"); var lastEvent = events[^1]; Assert.True( lastEvent is ResponseCompletedEvent || lastEvent is ResponseFailedEvent, $"Expected terminal event, got {lastEvent.GetType().Name}"); } [Fact] public async Task SequentialWorkflow_TwoAgents_ProducesOutputFromBothAsync() { // Arrange: two agents in sequence var agent1 = new StreamingTextAgent("agent1", "First agent says hello"); var agent2 = new StreamingTextAgent("agent2", "Second agent says goodbye"); var workflow = AgentWorkflowBuilder.BuildSequential("test-sequential-2", agent1, agent2); var workflowAgent = workflow.AsAIAgent( id: "seq-workflow", name: "Sequential Workflow", executionEnvironment: InProcessExecution.OffThread, includeExceptionDetails: true); var (handler, request, context) = CreateHandlerWithAgent(workflowAgent, "Process this"); // Act var events = await CollectEventsAsync(handler, request, context); // Assert: should have workflow action events for executor lifecycle var lastEvent = events[^1]; Assert.True( lastEvent is ResponseCompletedEvent || lastEvent is ResponseFailedEvent, $"Expected terminal event, got {lastEvent.GetType().Name}"); // Should have output item events (either text messages or workflow actions) Assert.True(events.OfType().Any(), "Expected at least one output item from the workflow"); } // ===== Workflow Error Propagation ===== [Fact] public async Task Workflow_AgentThrowsException_ProducesErrorOutputAsync() { // Arrange: workflow with an agent that throws var throwingAgent = new ThrowingStreamingAgent("thrower", new InvalidOperationException("Agent crashed")); var workflow = AgentWorkflowBuilder.BuildSequential("test-error", throwingAgent); var workflowAgent = workflow.AsAIAgent( id: "error-workflow", name: "Error Workflow", executionEnvironment: InProcessExecution.OffThread, includeExceptionDetails: true); var (handler, request, context) = CreateHandlerWithAgent(workflowAgent, "Trigger error"); // Act var events = await CollectEventsAsync(handler, request, context); // Assert: should have lifecycle events + error/failure indicator Assert.IsType(events[0]); Assert.IsType(events[1]); var lastEvent = events[^1]; // Workflow errors surface as either Failed or Completed (depending on error handling) Assert.True( lastEvent is ResponseCompletedEvent || lastEvent is ResponseFailedEvent, $"Expected terminal event, got {lastEvent.GetType().Name}"); } // ===== Workflow Action Lifecycle Events ===== [Fact] public async Task Workflow_ExecutorEvents_ProduceWorkflowActionItemsAsync() { // Arrange var agent = new StreamingTextAgent("test-agent", "Result"); var workflow = AgentWorkflowBuilder.BuildSequential("test-actions", agent); var workflowAgent = workflow.AsAIAgent( id: "actions-workflow", name: "Actions Workflow", executionEnvironment: InProcessExecution.OffThread); var (handler, request, context) = CreateHandlerWithAgent(workflowAgent, "Hello"); // Act var events = await CollectEventsAsync(handler, request, context); // Assert: workflow should produce OutputItemAdded events for executor lifecycle var addedEvents = events.OfType().ToList(); Assert.True(addedEvents.Count >= 1, $"Expected at least 1 output item added event, got {addedEvents.Count}"); } // ===== Keyed Workflow Registration ===== [Fact] public async Task WorkflowAgent_RegisteredWithKey_ResolvesCorrectlyAsync() { // Arrange: workflow agent registered with a keyed service name var agent = new StreamingTextAgent("inner", "Keyed workflow response"); var workflow = AgentWorkflowBuilder.BuildSequential("keyed-wf", agent); var workflowAgent = workflow.AsAIAgent( id: "keyed-workflow", name: "Keyed Workflow", executionEnvironment: InProcessExecution.OffThread); var services = new ServiceCollection(); services.AddSingleton(new InMemoryAgentSessionStore()); services.AddKeyedSingleton("my-workflow", workflowAgent); var sp = services.BuildServiceProvider(); var handler = new AgentFrameworkResponseHandler(sp, NullLogger.Instance); var request = new CreateResponse { Model = "test", AgentReference = new AgentReference("my-workflow") }; request.Input = CreateUserInput("Test keyed workflow"); var mockContext = CreateMockContext(); // Act var events = await CollectEventsAsync(handler, request, mockContext.Object); // Assert Assert.IsType(events[0]); Assert.True(events.Count >= 3, $"Expected at least 3 events, got {events.Count}"); } // ===== OutputConverter Direct Workflow Pattern Tests ===== // These test the OutputConverter directly with update patterns that mirror real workflows. [Fact] public async Task OutputConverter_SequentialWorkflowPattern_ProducesCorrectEventsAsync() { // Simulate what WorkflowSession produces for a 2-agent sequential workflow var (stream, _) = CreateTestStream(); var updates = new[] { // Superstep 1: Agent 1 new AgentResponseUpdate { RawRepresentation = new SuperStepStartedEvent(1) }, new AgentResponseUpdate { RawRepresentation = new ExecutorInvokedEvent("agent_1", "start") }, new AgentResponseUpdate { MessageId = "msg_a1", Contents = [new MeaiTextContent("Agent 1 output")] }, new AgentResponseUpdate { RawRepresentation = new ExecutorCompletedEvent("agent_1", null) }, new AgentResponseUpdate { RawRepresentation = new SuperStepCompletedEvent(1) }, // Superstep 2: Agent 2 new AgentResponseUpdate { RawRepresentation = new SuperStepStartedEvent(2) }, new AgentResponseUpdate { RawRepresentation = new ExecutorInvokedEvent("agent_2", "start") }, new AgentResponseUpdate { MessageId = "msg_a2", Contents = [new MeaiTextContent("Agent 2 output")] }, new AgentResponseUpdate { RawRepresentation = new ExecutorCompletedEvent("agent_2", null) }, new AgentResponseUpdate { RawRepresentation = new SuperStepCompletedEvent(2) }, }; var events = new List(); await foreach (var evt in OutputConverter.ConvertUpdatesToEventsAsync(ToAsync(updates), stream)) { events.Add(evt); } // 4 workflow action items + 2 text messages = 6 output items Assert.Equal(6, events.OfType().Count()); Assert.Equal(2, events.OfType().Count()); Assert.IsType(events[^1]); } [Fact] public async Task OutputConverter_GroupChatPattern_ProducesCorrectEventsAsync() { // Simulate round-robin group chat: agent1 → agent2 → agent1 → terminate var (stream, _) = CreateTestStream(); var updates = new[] { new AgentResponseUpdate { RawRepresentation = new SuperStepStartedEvent(1) }, new AgentResponseUpdate { RawRepresentation = new ExecutorInvokedEvent("chat_agent_1", "turn") }, new AgentResponseUpdate { MessageId = "msg_gc_1", Contents = [new MeaiTextContent("Agent 1 turn 1")] }, new AgentResponseUpdate { RawRepresentation = new ExecutorCompletedEvent("chat_agent_1", null) }, new AgentResponseUpdate { RawRepresentation = new SuperStepCompletedEvent(1) }, new AgentResponseUpdate { RawRepresentation = new SuperStepStartedEvent(2) }, new AgentResponseUpdate { RawRepresentation = new ExecutorInvokedEvent("chat_agent_2", "turn") }, new AgentResponseUpdate { MessageId = "msg_gc_2", Contents = [new MeaiTextContent("Agent 2 turn 1")] }, new AgentResponseUpdate { RawRepresentation = new ExecutorCompletedEvent("chat_agent_2", null) }, new AgentResponseUpdate { RawRepresentation = new SuperStepCompletedEvent(2) }, new AgentResponseUpdate { RawRepresentation = new SuperStepStartedEvent(3) }, new AgentResponseUpdate { RawRepresentation = new ExecutorInvokedEvent("chat_agent_1", "turn") }, new AgentResponseUpdate { MessageId = "msg_gc_3", Contents = [new MeaiTextContent("Agent 1 turn 2")] }, new AgentResponseUpdate { RawRepresentation = new ExecutorCompletedEvent("chat_agent_1", null) }, new AgentResponseUpdate { RawRepresentation = new SuperStepCompletedEvent(3) }, }; var events = new List(); await foreach (var evt in OutputConverter.ConvertUpdatesToEventsAsync(ToAsync(updates), stream)) { events.Add(evt); } // 6 workflow actions + 3 text messages = 9 output items Assert.Equal(9, events.OfType().Count()); Assert.Equal(3, events.OfType().Count()); Assert.IsType(events[^1]); } [Fact] public async Task OutputConverter_CodeExecutorPattern_ProducesCorrectEventsAsync() { // Simulate a code-based FunctionExecutor: invoked → completed, no text content // (code executors don't produce AgentResponseUpdateEvent, just executor lifecycle) var (stream, _) = CreateTestStream(); var updates = new[] { new AgentResponseUpdate { RawRepresentation = new SuperStepStartedEvent(1) }, new AgentResponseUpdate { RawRepresentation = new ExecutorInvokedEvent("uppercase_fn", "hello") }, new AgentResponseUpdate { RawRepresentation = new ExecutorCompletedEvent("uppercase_fn", "HELLO") }, new AgentResponseUpdate { RawRepresentation = new SuperStepCompletedEvent(1) }, // Second executor uses the output new AgentResponseUpdate { RawRepresentation = new SuperStepStartedEvent(2) }, new AgentResponseUpdate { RawRepresentation = new ExecutorInvokedEvent("format_agent", "start") }, new AgentResponseUpdate { MessageId = "msg_fmt", Contents = [new MeaiTextContent("Formatted: HELLO")] }, new AgentResponseUpdate { RawRepresentation = new ExecutorCompletedEvent("format_agent", null) }, new AgentResponseUpdate { RawRepresentation = new SuperStepCompletedEvent(2) }, }; var events = new List(); await foreach (var evt in OutputConverter.ConvertUpdatesToEventsAsync(ToAsync(updates), stream)) { events.Add(evt); } // 4 workflow actions + 1 text message = 5 output items Assert.Equal(5, events.OfType().Count()); Assert.Single(events.OfType()); Assert.IsType(events[^1]); } [Fact] public async Task OutputConverter_SubworkflowPattern_ProducesCorrectEventsAsync() { // Simulate a parent workflow that invokes a sub-workflow executor var (stream, _) = CreateTestStream(); var updates = new[] { new AgentResponseUpdate { RawRepresentation = new WorkflowStartedEvent("parent") }, new AgentResponseUpdate { RawRepresentation = new SuperStepStartedEvent(1) }, // Sub-workflow executor invoked new AgentResponseUpdate { RawRepresentation = new ExecutorInvokedEvent("sub_workflow_host", "start") }, // Inner agent within sub-workflow produces text (unwrapped by WorkflowSession) new AgentResponseUpdate { MessageId = "msg_sub_1", Contents = [new MeaiTextContent("Sub-workflow agent output")] }, // Sub-workflow executor completed new AgentResponseUpdate { RawRepresentation = new ExecutorCompletedEvent("sub_workflow_host", null) }, new AgentResponseUpdate { RawRepresentation = new SuperStepCompletedEvent(1) }, }; var events = new List(); await foreach (var evt in OutputConverter.ConvertUpdatesToEventsAsync(ToAsync(updates), stream)) { events.Add(evt); } // 2 workflow actions + 1 text message = 3 output items Assert.Equal(3, events.OfType().Count()); Assert.Single(events.OfType()); Assert.IsType(events[^1]); } [Fact] public async Task OutputConverter_WorkflowWithMultipleContentTypes_HandlesAllCorrectlyAsync() { // Simulate a workflow producing reasoning, text, function calls, and usage var (stream, _) = CreateTestStream(); var updates = new[] { new AgentResponseUpdate { RawRepresentation = new ExecutorInvokedEvent("planner", "start") }, // Reasoning new AgentResponseUpdate { Contents = [new TextReasoningContent("Let me think about this...")] }, // Function call (tool use) new AgentResponseUpdate { Contents = [new FunctionCallContent("call_search", "web_search", new Dictionary { ["query"] = "latest news" })] }, new AgentResponseUpdate { RawRepresentation = new ExecutorCompletedEvent("planner", null) }, // Next executor uses tool result new AgentResponseUpdate { RawRepresentation = new ExecutorInvokedEvent("writer", "start") }, new AgentResponseUpdate { MessageId = "msg_w1", Contents = [new MeaiTextContent("Based on my research, ")] }, new AgentResponseUpdate { MessageId = "msg_w1", Contents = [new MeaiTextContent("here are the findings.")] }, new AgentResponseUpdate { Contents = [new UsageContent(new UsageDetails { InputTokenCount = 500, OutputTokenCount = 200, TotalTokenCount = 700 })] }, new AgentResponseUpdate { RawRepresentation = new ExecutorCompletedEvent("writer", null) }, }; var events = new List(); await foreach (var evt in OutputConverter.ConvertUpdatesToEventsAsync(ToAsync(updates), stream)) { events.Add(evt); } // Workflow actions: 4 (2 invoked + 2 completed) // Content: 1 reasoning + 1 function call + 1 text message = 3 // Total: 7 output items Assert.Equal(7, events.OfType().Count()); Assert.Contains(events, e => e is ResponseFunctionCallArgumentsDoneEvent); Assert.Equal(2, events.OfType().Count()); Assert.IsType(events[^1]); } // ===== Helpers ===== private static (AgentFrameworkResponseHandler handler, CreateResponse request, ResponseContext context) CreateHandlerWithAgent(AIAgent agent, string userMessage) { var services = new ServiceCollection(); services.AddSingleton(new InMemoryAgentSessionStore()); services.AddSingleton(agent); services.AddSingleton>(NullLogger.Instance); var sp = services.BuildServiceProvider(); var handler = new AgentFrameworkResponseHandler(sp, NullLogger.Instance); var request = new CreateResponse { Model = "test" }; request.Input = CreateUserInput(userMessage); var mockContext = CreateMockContext(); return (handler, request, mockContext.Object); } private static BinaryData CreateUserInput(string text) { return BinaryData.FromObjectAsJson(new[] { new { type = "message", id = "msg_in_1", status = "completed", role = "user", content = new[] { new { type = "input_text", text } } } }); } private static Mock CreateMockContext() { var mock = new Mock("resp_" + new string('0', 46)) { CallBase = true }; mock.Setup(x => x.GetHistoryAsync(It.IsAny())) .ReturnsAsync(Array.Empty()); mock.Setup(x => x.GetInputItemsAsync(It.IsAny(), It.IsAny())) .ReturnsAsync(Array.Empty()); return mock; } private static (ResponseEventStream stream, Mock mockContext) CreateTestStream() { var mockContext = new Mock("resp_" + new string('0', 46)) { CallBase = true }; var request = new CreateResponse { Model = "test-model" }; var stream = new ResponseEventStream(mockContext.Object, request); return (stream, mockContext); } private static async Task> CollectEventsAsync( AgentFrameworkResponseHandler handler, CreateResponse request, ResponseContext context) { var events = new List(); await foreach (var evt in handler.CreateAsync(request, context, CancellationToken.None)) { events.Add(evt); } return events; } private static async IAsyncEnumerable ToAsync(IEnumerable source) { foreach (var item in source) { yield return item; } await Task.CompletedTask; } // ===== Test Agent Types ===== /// /// A test agent that streams a single text update. /// private sealed class StreamingTextAgent(string id, string responseText) : AIAgent { public new string Id => id; protected override async IAsyncEnumerable RunCoreStreamingAsync( IEnumerable messages, AgentSession? session, AgentRunOptions? options, [EnumeratorCancellation] CancellationToken cancellationToken = default) { yield return new AgentResponseUpdate { MessageId = $"msg_{id}", Contents = [new MeaiTextContent(responseText)] }; await Task.CompletedTask; } protected override Task RunCoreAsync( IEnumerable messages, AgentSession? session, AgentRunOptions? options, CancellationToken cancellationToken = default) => throw new NotImplementedException(); protected override ValueTask CreateSessionCoreAsync( CancellationToken cancellationToken = default) => throw new NotImplementedException(); protected override ValueTask SerializeSessionCoreAsync( AgentSession session, JsonSerializerOptions? jsonSerializerOptions, CancellationToken cancellationToken = default) => throw new NotImplementedException(); protected override ValueTask DeserializeSessionCoreAsync( JsonElement serializedState, JsonSerializerOptions? jsonSerializerOptions, CancellationToken cancellationToken = default) => throw new NotImplementedException(); } /// /// A test agent that always throws an exception during streaming. /// private sealed class ThrowingStreamingAgent(string id, Exception exception) : AIAgent { public new string Id => id; protected override IAsyncEnumerable RunCoreStreamingAsync( IEnumerable messages, AgentSession? session, AgentRunOptions? options, CancellationToken cancellationToken = default) => throw exception; protected override Task RunCoreAsync( IEnumerable messages, AgentSession? session, AgentRunOptions? options, CancellationToken cancellationToken = default) => throw new NotImplementedException(); protected override ValueTask CreateSessionCoreAsync( CancellationToken cancellationToken = default) => throw new NotImplementedException(); protected override ValueTask SerializeSessionCoreAsync( AgentSession session, JsonSerializerOptions? jsonSerializerOptions, CancellationToken cancellationToken = default) => throw new NotImplementedException(); protected override ValueTask DeserializeSessionCoreAsync( JsonElement serializedState, JsonSerializerOptions? jsonSerializerOptions, CancellationToken cancellationToken = default) => throw new NotImplementedException(); } }