// Copyright (c) Microsoft. All rights reserved. using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Text.Json; using System.Threading.Tasks; using FluentAssertions; using Microsoft.Agents.AI.Workflows.InProc; using Microsoft.Agents.AI.Workflows.Specialized.Magentic; using Microsoft.Extensions.AI; namespace Microsoft.Agents.AI.Workflows.UnitTests; /// /// End-to-end tests for the Magentic orchestrator workflow. /// public class MagenticOrchestrationTests { [Fact] public async Task Task_Completes_When_RequestSatisfiedAsync() { // Arrange: Manager reports task satisfied on first coordination round // Each response must have unique message IDs, so create separate instances List factsResponse = CreatePlanResponse("Facts about the task"); List planResponse = CreatePlanResponse("Step 1: Do the task"); List progressLedgerResponse = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Complete the task"); List finalAnswerResponse = CreateFinalAnswerResponse("Task completed successfully!"); TestReplayAgent manager = new( [factsResponse, planResponse, progressLedgerResponse, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync(workflow, [new ChatMessage(ChatRole.User, "Do the task")]); // Assert: Check the result contains the final answer runResult.Result.Should().NotBeNull(); runResult.Result.Should().ContainSingle(); runResult.Result![0].Text.Should().Contain("Task completed successfully!"); runResult.PendingRequests.Should().BeEmpty(); } [Fact] public async Task PlanReview_Approved_ProceedsAsync() { // Arrange: Human approves initial plan List factsResponse = CreatePlanResponse("Facts about executing the plan"); List planResponse = CreatePlanResponse("Step 1: Execute the plan"); List progressLedgerResponse = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Execute"); List finalAnswerResponse = CreateFinalAnswerResponse("Plan executed successfully"); TestReplayAgent manager = new( [factsResponse, planResponse, progressLedgerResponse, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(true) .Build(); CheckpointManager checkpointManager = CheckpointManager.CreateInMemory(); // Act: First run - should pause for plan review WorkflowRunResult firstResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Execute plan")], checkpointManager: checkpointManager); firstResult.PendingRequests.Should().ContainSingle(); ExternalRequest request = firstResult.PendingRequests[0].Request; MagenticPlanReviewRequest? reviewRequest = request.Data.As(); reviewRequest.Should().NotBeNull(); reviewRequest!.Plan.Text.Should().Contain("Execute the plan"); // Act: Resume with approval MagenticPlanReviewResponse approval = reviewRequest.Approve(); ExternalResponse response = request.CreateResponse(approval); WorkflowRunResult secondResult = await ResumeMagenticWorkflowAsync( workflow, response, checkpointManager, firstResult.LastCheckpoint); // Assert secondResult.Result.Should().NotBeNull(); secondResult.Result![0].Text.Should().Contain("Plan executed successfully"); } [Fact] public async Task Initial_Plan_Emits_PlanCreatedEventAsync() { // Arrange List factsResponse = CreatePlanResponse("Facts about the task"); List planResponse = CreatePlanResponse("Step 1: Initial plan"); List progressLedgerResponse = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Execute"); List finalAnswerResponse = CreateFinalAnswerResponse("Done"); TestReplayAgent manager = new( [factsResponse, planResponse, progressLedgerResponse, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); List collectedEvents = []; Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .Build(); // Act await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Do task")], eventCollector: collectedEvents); // Assert collectedEvents.OfType().Should().NotBeEmpty(); MagenticPlanCreatedEvent planEvent = collectedEvents.OfType().First(); planEvent.FullTaskLedger.Should().NotBeNull(); } [Fact] public async Task NextSpeaker_Invalid_Triggers_FinalAnswerAsync() { // Arrange: ProgressLedger returns invalid next_speaker List factsResponse = CreatePlanResponse("Facts about the task"); List planResponse = CreatePlanResponse("Step 1: Execute"); List invalidNextSpeakerLedger = CreateProgressLedgerResponse( isRequestSatisfied: false, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "NonExistentAgent", // Invalid - doesn't match any team member instructionOrQuestion: "Continue"); List finalAnswer = CreateFinalAnswerResponse("Forced to conclude due to invalid speaker"); TestReplayAgent manager = new( [factsResponse, planResponse, invalidNextSpeakerLedger, finalAnswer], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); List collectedEvents = []; Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Do task")], eventCollector: collectedEvents); // Assert: Warning should be emitted and final answer prepared collectedEvents.OfType() .Should().Contain(e => e.Data != null && e.Data.ToString()!.Contains("Invalid next speaker")); runResult.Result.Should().NotBeNull(); runResult.Result![0].Text.Should().Contain("Forced to conclude"); } [Fact] public async Task ProgressLedger_Updated_Event_EmittedAsync() { // Arrange List factsResponse = CreatePlanResponse("Facts about the task"); List planResponse = CreatePlanResponse("Step 1: Execute"); List progressLedgerResponse = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Execute"); List finalAnswerResponse = CreateFinalAnswerResponse("Done"); TestReplayAgent manager = new( [factsResponse, planResponse, progressLedgerResponse, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); List collectedEvents = []; Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .Build(); // Act await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Do task")], eventCollector: collectedEvents); // Assert collectedEvents.OfType().Should().NotBeEmpty(); MagenticProgressLedgerUpdatedEvent ledgerEvent = collectedEvents.OfType().First(); ledgerEvent.ProgressLedger.Should().NotBeNull(); ledgerEvent.ProgressLedger.IsRequestSatisfied.Should().BeTrue(); } [Fact] public async Task PlanSignoff_Disabled_Proceeds_ImmediatelyAsync() { // Arrange: requirePlanSignoff=false should mean no plan review request List factsResponse = CreatePlanResponse("Task facts"); List planResponse = CreatePlanResponse("Step 1: Execute immediately"); List progressLedgerResponse = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Go"); List finalAnswerResponse = CreateFinalAnswerResponse("Immediate completion"); TestReplayAgent manager = new( [factsResponse, planResponse, progressLedgerResponse, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); List collectedEvents = []; Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Do it now")], eventCollector: collectedEvents); // Assert: No plan review request, workflow completes immediately runResult.PendingRequests.Should().BeEmpty("plan signoff is disabled, so no review should be requested"); collectedEvents.OfType().Should().BeEmpty(); runResult.Result.Should().NotBeNull(); runResult.Result![0].Text.Should().Contain("Immediate completion"); } [Fact] public async Task NextSpeaker_Empty_Falls_Back_To_FirstAsync() { // Arrange: First progress ledger returns empty next_speaker, which should fall back to first participant. // Round 1: empty speaker → fallback to Worker (first participant) → Worker echoes // Round 2 (after Worker responds): RunCoordinationRoundAsync → satisfied ledger → final answer // Note: No replan on agent return — only progress ledger is created on subsequent turns. List factsResponse1 = CreatePlanResponse("Facts about the task"); List planResponse1 = CreatePlanResponse("Step 1: Execute"); List emptyNextSpeakerLedger = CreateProgressLedgerResponse( isRequestSatisfied: false, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "", // Empty - should fall back to first participant instructionOrQuestion: "Please help with this task"); // Round 2: satisfied ledger + final answer (no replan on normal agent return) List satisfiedLedger = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Done"); List finalAnswerResponse = CreateFinalAnswerResponse("Task completed after fallback"); TestReplayAgent manager = new( [factsResponse1, planResponse1, emptyNextSpeakerLedger, satisfiedLedger, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); List collectedEvents = []; Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Do the task")], eventCollector: collectedEvents); // Assert: Warning about empty next speaker should be emitted collectedEvents.OfType() .Should().Contain(e => e.Data != null && e.Data.ToString()!.Contains("empty")); runResult.Result.Should().NotBeNull(); runResult.Result![0].Text.Should().Contain("Task completed after fallback"); } [Fact] public async Task Task_Completes_After_Multiple_RoundsAsync() { // Arrange: Round 1 delegates to Worker (not satisfied), round 2 completes // Manager turn sequence: facts1, plan1, ledger1(not satisfied), ledger2(satisfied), finalAnswer // Note: No replan on agent return — only progress ledger is created on subsequent turns. List factsResponse1 = CreatePlanResponse("Initial facts"); List planResponse1 = CreatePlanResponse("Step 1: Delegate to worker"); List round1Ledger = CreateProgressLedgerResponse( isRequestSatisfied: false, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Please work on the task"); List round2Ledger = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Task is done"); List finalAnswerResponse = CreateFinalAnswerResponse("Multi-round task completed!"); TestReplayAgent manager = new( [factsResponse1, planResponse1, round1Ledger, round2Ledger, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); List collectedEvents = []; Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Complex multi-round task")], eventCollector: collectedEvents); // Assert: One plan created, one progress ledger per round, final answer collectedEvents.OfType().Should().HaveCount(2); collectedEvents.OfType().Should().ContainSingle("only one initial plan, no replan on agent return"); collectedEvents.OfType().Should().BeEmpty("no replan occurs on normal agent return"); runResult.Result.Should().NotBeNull(); runResult.Result![0].Text.Should().Contain("Multi-round task completed!"); } [Fact] public async Task PlanReview_Revised_Triggers_ReplanAsync() { // Arrange: Human rejects initial plan with revision, triggering a replan. // Flow: facts1, plan1 → PlanCreatedEvent → plan review (pending) // resume with revision → facts2, plan2 → MagenticReplannedEvent → plan review again (pending) // resume with approval → progressLedger(satisfied) → finalAnswer List factsResponse1 = CreatePlanResponse("Initial facts"); List planResponse1 = CreatePlanResponse("Initial plan - needs revision"); List factsResponse2 = CreatePlanResponse("Revised facts"); List planResponse2 = CreatePlanResponse("Revised plan - much better"); List progressLedgerResponse = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Execute revised plan"); List finalAnswerResponse = CreateFinalAnswerResponse("Revised plan executed successfully"); TestReplayAgent manager = new( [factsResponse1, planResponse1, factsResponse2, planResponse2, progressLedgerResponse, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(true) .Build(); CheckpointManager checkpointManager = CheckpointManager.CreateInMemory(); List allEvents = []; // Act 1: First run - should pause for plan review with initial plan WorkflowRunResult firstResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Execute task")], checkpointManager: checkpointManager, eventCollector: allEvents); firstResult.PendingRequests.Should().ContainSingle(); ExternalRequest request1 = firstResult.PendingRequests[0].Request; MagenticPlanReviewRequest? reviewRequest1 = request1.Data.As(); reviewRequest1.Should().NotBeNull(); reviewRequest1!.Plan.Text.Should().Contain("Initial plan"); // Act 2: Resume with revision (reject the plan) MagenticPlanReviewResponse revision = reviewRequest1.Revise("Please include more detail"); ExternalResponse revisionResponse = request1.CreateResponse(revision); WorkflowRunResult secondResult = await ResumeMagenticWorkflowAsync( workflow, revisionResponse, checkpointManager, firstResult.LastCheckpoint, eventCollector: allEvents); // Should pause again for review of the revised plan (stream may include prior request too) secondResult.PendingRequests.Should().NotBeEmpty(); ExternalRequest request2 = secondResult.PendingRequests[^1].Request; MagenticPlanReviewRequest? reviewRequest2 = request2.Data.As(); reviewRequest2.Should().NotBeNull(); reviewRequest2!.Plan.Text.Should().Contain("Revised plan"); // Act 3: Resume with approval MagenticPlanReviewResponse approval = reviewRequest2.Approve(); ExternalResponse approvalResponse = request2.CreateResponse(approval); WorkflowRunResult thirdResult = await ResumeMagenticWorkflowAsync( workflow, approvalResponse, checkpointManager, secondResult.LastCheckpoint, eventCollector: allEvents); // Assert: MagenticReplannedEvent should have been emitted, and final answer produced allEvents.OfType().Should().NotBeEmpty("initial plan emits PlanCreatedEvent"); allEvents.OfType().Should().NotBeEmpty("revision triggers ReplannedEvent"); thirdResult.Result.Should().NotBeNull(); thirdResult.Result![0].Text.Should().Contain("Revised plan executed successfully"); } [Fact] public async Task MaxRoundLimit_Terminates_WorkflowAsync() { // Arrange: MaxRounds=1, so round 1 delegates to Worker, round 2 hits limit and terminates. // Manager turns: facts1, plan1, ledger1(not satisfied→delegates), then limit hit before ledger. List factsResponse1 = CreatePlanResponse("Facts"); List planResponse1 = CreatePlanResponse("Plan"); List round1Ledger = CreateProgressLedgerResponse( isRequestSatisfied: false, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Work on it"); // No more turns needed: RunCoordinationRoundAsync hits round limit before calling UpdateProgressLedgerAsync TestReplayAgent manager = new( [factsResponse1, planResponse1, round1Ledger], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .WithMaxRounds(1) .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Do task")]); // Assert: Workflow terminates with round limit message runResult.Result.Should().NotBeNull(); runResult.Result![0].Text.Should().Contain("maximum round count limit"); } [Fact] public async Task MaxStallCount_Triggers_ResetAsync() { // Arrange: MaxStallCount=0, so one stall (isInLoop=true, StallCount=1 > 0) triggers ResetAndReplanAsync. // Flow: facts1, plan1 → round1 ledger(stall: isInLoop=true) → StallCount=1 → IsStalled → Reset // → facts2, plan2 (replan) → round2 ledger(satisfied) → finalAnswer List factsResponse1 = CreatePlanResponse("Initial facts"); List planResponse1 = CreatePlanResponse("Initial plan"); List stalledLedger = CreateProgressLedgerResponse( isRequestSatisfied: false, isInLoop: true, // This triggers stall isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Keep trying"); // After reset: ResetAndReplanAsync → UpdatePlanAndDelegateAsync → new plan List factsResponse2 = CreatePlanResponse("Fresh facts after reset"); List planResponse2 = CreatePlanResponse("Fresh plan after reset"); List satisfiedLedger = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Done"); List finalAnswerResponse = CreateFinalAnswerResponse("Recovered after stall reset"); TestReplayAgent manager = new( [factsResponse1, planResponse1, stalledLedger, factsResponse2, planResponse2, satisfiedLedger, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); List collectedEvents = []; Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .WithMaxStalls(0) // One stall triggers reset (StallCount 1 > 0) .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Do task")], eventCollector: collectedEvents); // Assert: MagenticReplannedEvent should be emitted (reset triggers replan), final answer produced collectedEvents.OfType().Should().NotBeEmpty("initial plan created"); collectedEvents.OfType().Should().NotBeEmpty("stall triggers reset and replan"); runResult.Result.Should().NotBeNull(); runResult.Result![0].Text.Should().Contain("Recovered after stall reset"); } [Fact] public async Task Instruction_Message_Sent_When_PresentAsync() { // Arrange: Progress ledger has a non-empty instruction_or_question. // The orchestrator should send the instruction as a ChatMessage before delegating to the next agent. // After Worker echoes, the second round completes. List factsResponse1 = CreatePlanResponse("Facts about the task"); List planResponse1 = CreatePlanResponse("Step 1: Instruct the worker"); List ledgerWithInstruction = CreateProgressLedgerResponse( isRequestSatisfied: false, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Please analyze the data carefully"); // Round 2 after Worker responds (no replan, just progress ledger) List satisfiedLedger = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Done"); List finalAnswerResponse = CreateFinalAnswerResponse("Task completed with instruction"); TestReplayAgent manager = new( [factsResponse1, planResponse1, ledgerWithInstruction, satisfiedLedger, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); List collectedEvents = []; Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Analyze data")], eventCollector: collectedEvents); // Assert: The workflow completed successfully, proving the instruction path executed without error. // The update text should contain the instruction text since it is sent to participants as a ChatMessage. runResult.Result.Should().NotBeNull(); runResult.Result![0].Text.Should().Contain("Task completed with instruction"); // Verify the delegation happened (two progress ledger events for two rounds) collectedEvents.OfType().Should().HaveCount(2); } [Fact] public async Task PlanReview_On_Stall_ReplanAsync() { // Arrange: Plan signoff enabled, stall triggers reset, replan requires new plan review. // Flow: facts1, plan1 → PlanCreatedEvent → plan review (pending) // resume with approval → ledger1(stall: isInLoop=true) → StallCount=1 → IsStalled → Reset // → facts2, plan2 → MagenticReplannedEvent → plan review again (pending) // resume with approval → ledger2(satisfied) → finalAnswer List factsResponse1 = CreatePlanResponse("Initial facts"); List planResponse1 = CreatePlanResponse("Initial plan"); List stalledLedger = CreateProgressLedgerResponse( isRequestSatisfied: false, isInLoop: true, // This triggers stall isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Keep trying"); // After reset: new plan List factsResponse2 = CreatePlanResponse("Fresh facts after stall reset"); List planResponse2 = CreatePlanResponse("Fresh plan after stall reset"); // After second approval: satisfied ledger + final answer List satisfiedLedger = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Done"); List finalAnswerResponse = CreateFinalAnswerResponse("Recovered after stall with plan review"); TestReplayAgent manager = new( [factsResponse1, planResponse1, stalledLedger, factsResponse2, planResponse2, satisfiedLedger, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(true) .WithMaxStalls(0) .Build(); CheckpointManager checkpointManager = CheckpointManager.CreateInMemory(); List allEvents = []; // Act 1: First run - should pause for initial plan review WorkflowRunResult firstResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Do task")], checkpointManager: checkpointManager, eventCollector: allEvents); firstResult.PendingRequests.Should().ContainSingle(); ExternalRequest request1 = firstResult.PendingRequests[0].Request; MagenticPlanReviewRequest? reviewRequest1 = request1.Data.As(); reviewRequest1.Should().NotBeNull(); reviewRequest1!.Plan.Text.Should().Contain("Initial plan"); reviewRequest1.IsStalled.Should().BeFalse("the initial plan review is not stall-triggered"); // Act 2: Approve initial plan → stall occurs → reset → replan → new plan review MagenticPlanReviewResponse approval1 = reviewRequest1.Approve(); ExternalResponse approvalResponse1 = request1.CreateResponse(approval1); WorkflowRunResult secondResult = await ResumeMagenticWorkflowAsync( workflow, approvalResponse1, checkpointManager, firstResult.LastCheckpoint, eventCollector: allEvents); // Should pause for review of the replanned plan secondResult.PendingRequests.Should().NotBeEmpty(); ExternalRequest request2 = secondResult.PendingRequests[^1].Request; MagenticPlanReviewRequest? reviewRequest2 = request2.Data.As(); reviewRequest2.Should().NotBeNull(); reviewRequest2!.Plan.Text.Should().Contain("Fresh plan after stall reset"); reviewRequest2.IsStalled.Should().BeTrue("the replan was triggered by a stall"); // Act 3: Approve the revised plan → satisfied → final answer MagenticPlanReviewResponse approval2 = reviewRequest2.Approve(); ExternalResponse approvalResponse2 = request2.CreateResponse(approval2); WorkflowRunResult thirdResult = await ResumeMagenticWorkflowAsync( workflow, approvalResponse2, checkpointManager, secondResult.LastCheckpoint, eventCollector: allEvents); // Assert allEvents.OfType().Should().NotBeEmpty("initial plan emits PlanCreatedEvent"); allEvents.OfType().Should().NotBeEmpty("stall reset triggers ReplannedEvent"); thirdResult.Result.Should().NotBeNull(); thirdResult.Result![0].Text.Should().Contain("Recovered after stall with plan review"); } [Fact] public async Task MaxResetLimit_Terminates_WorkflowAsync() { // Arrange: MaxStallCount=0, MaxResets=1. // Flow: facts1, plan1 → ledger1(stall: isInLoop=true) → StallCount=1 > 0 → IsStalled → ResetAndReplanAsync // → ResetCount becomes 1 → facts2, plan2 → DelegateToTeamAsync // → RunCoordinationRoundAsync: CheckLimits() detects ResetCount(1) >= MaxResetCount(1) → terminates List factsResponse1 = CreatePlanResponse("Initial facts"); List planResponse1 = CreatePlanResponse("Initial plan"); List stalledLedger = CreateProgressLedgerResponse( isRequestSatisfied: false, isInLoop: true, // This triggers stall isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Keep trying"); // After reset: ResetAndReplanAsync → UpdatePlanAndDelegateAsync → new plan List factsResponse2 = CreatePlanResponse("Fresh facts after reset"); List planResponse2 = CreatePlanResponse("Fresh plan after reset"); // No more turns needed: RunCoordinationRoundAsync hits reset limit before calling UpdateProgressLedgerAsync TestReplayAgent manager = new( [factsResponse1, planResponse1, stalledLedger, factsResponse2, planResponse2], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .WithMaxStalls(0) // One stall triggers reset (StallCount 1 > 0) .WithMaxResets(1) // One reset triggers termination .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Do task")]); // Assert: Workflow terminates with reset limit message runResult.Result.Should().NotBeNull(); runResult.Result![0].Text.Should().Contain("maximum reset count limit"); } [Fact] public async Task ProgressLedger_Retry_On_Parse_FailureAsync() { // Arrange: First progress ledger attempt returns invalid JSON (triggers parse failure + warning), // second attempt returns valid JSON (satisfied=true). // Manager turn sequence: facts, plan, INVALID_JSON, VALID_LEDGER(satisfied), finalAnswer // MagenticManager.UpdateProgressLedgerAsync retries internally: attempt 0 fails, attempt 1 succeeds. List factsResponse = CreatePlanResponse("Facts about the task"); List planResponse = CreatePlanResponse("Step 1: Execute"); List invalidLedgerResponse = CreatePlanResponse("This is not valid JSON for a progress ledger"); List validLedgerResponse = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Done after retry"); List finalAnswerResponse = CreateFinalAnswerResponse("Completed after ledger retry"); TestReplayAgent manager = new( [factsResponse, planResponse, invalidLedgerResponse, validLedgerResponse, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); List collectedEvents = []; Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Do task")], eventCollector: collectedEvents); // Assert: Warning emitted for parse failure, but workflow completes successfully collectedEvents.OfType() .Should().Contain(e => e.Data != null && e.Data.ToString()!.Contains("Progress ledger JSON parse failed")); runResult.Result.Should().NotBeNull(); runResult.Result![0].Text.Should().Contain("Completed after ledger retry"); } [Fact] public async Task ProgressLedger_Max_Retries_Triggers_ResetAsync() { // Arrange: All 3 progress ledger retry attempts return invalid JSON → exception → ResetAndReplanAsync. // After reset: new plan, valid ledger (satisfied), final answer. // Turn sequence: facts1, plan1, invalidJSON×3, facts2, plan2, validLedger(satisfied), finalAnswer List factsResponse1 = CreatePlanResponse("Initial facts"); List planResponse1 = CreatePlanResponse("Initial plan"); List invalidLedger1 = CreatePlanResponse("not json at all"); List invalidLedger2 = CreatePlanResponse("still not json"); List invalidLedger3 = CreatePlanResponse("definitely not json"); // After reset: ResetAndReplanAsync → UpdatePlanAndDelegateAsync → new plan List factsResponse2 = CreatePlanResponse("Fresh facts after reset"); List planResponse2 = CreatePlanResponse("Fresh plan after reset"); List validLedger = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Done after reset"); List finalAnswerResponse = CreateFinalAnswerResponse("Recovered after max retries reset"); TestReplayAgent manager = new( [factsResponse1, planResponse1, invalidLedger1, invalidLedger2, invalidLedger3, factsResponse2, planResponse2, validLedger, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); List collectedEvents = []; Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Do task")], eventCollector: collectedEvents); // Assert: Parse failure warnings emitted, reset triggered (ReplannedEvent), workflow completes collectedEvents.OfType() .Where(e => e.Data?.ToString()?.Contains("Progress ledger JSON parse failed") == true) .Should().HaveCountGreaterThanOrEqualTo(3, "all 3 retry attempts should emit warnings"); collectedEvents.OfType() .Should().Contain(e => e.Data != null && e.Data.ToString()!.Contains("triggering reset")); collectedEvents.OfType().Should().NotBeEmpty("reset triggers replan"); runResult.Result.Should().NotBeNull(); runResult.Result![0].Text.Should().Contain("Recovered after max retries reset"); } [Fact] public async Task Stall_NoProgress_Increments_StallCountAsync() { // Arrange: MaxStallCount=0, progress ledger reports IsProgressBeingMade=false (not IsInLoop). // This exercises the alternative stall trigger: !IsProgressBeingMade. // Flow: facts1, plan1 → ledger1(IsProgressBeingMade=false) → StallCount=1 > 0 → IsStalled → Reset // → facts2, plan2 (replan) → ledger2(satisfied) → finalAnswer List factsResponse1 = CreatePlanResponse("Initial facts"); List planResponse1 = CreatePlanResponse("Initial plan"); List noProgressLedger = CreateProgressLedgerResponse( isRequestSatisfied: false, isInLoop: false, // Not in loop isProgressBeingMade: false, // But no progress → stall nextSpeaker: "Worker", instructionOrQuestion: "Keep trying"); // After reset: ResetAndReplanAsync → UpdatePlanAndDelegateAsync → new plan List factsResponse2 = CreatePlanResponse("Fresh facts after no-progress reset"); List planResponse2 = CreatePlanResponse("Fresh plan after no-progress reset"); List satisfiedLedger = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Done"); List finalAnswerResponse = CreateFinalAnswerResponse("Recovered after no-progress stall"); TestReplayAgent manager = new( [factsResponse1, planResponse1, noProgressLedger, factsResponse2, planResponse2, satisfiedLedger, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); List collectedEvents = []; Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .WithMaxStalls(0) .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Do task")], eventCollector: collectedEvents); // Assert: Stall detected via no-progress, reset triggered, replan emitted, workflow completes collectedEvents.OfType().Should().NotBeEmpty("no-progress stall triggers reset and replan"); runResult.Result.Should().NotBeNull(); runResult.Result![0].Text.Should().Contain("Recovered after no-progress stall"); } [Fact] public async Task Task_Delegates_To_Correct_AgentAsync() { // Arrange: Two participants (WorkerA, WorkerB). Manager selects "WorkerA" as next speaker. // We verify that WorkerA produces a response update event and WorkerB does not. // Flow: facts1, plan1 → ledger1(nextSpeaker=WorkerA, not satisfied) → WorkerA runs // → RunCoordinationRoundAsync (no replan) → ledger2(satisfied) → finalAnswer List factsResponse1 = CreatePlanResponse("Task delegation facts"); List planResponse1 = CreatePlanResponse("Delegate to WorkerA"); List ledger1 = CreateProgressLedgerResponse( isRequestSatisfied: false, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "WorkerA", instructionOrQuestion: "WorkerA please handle this"); // After WorkerA responds, orchestrator goes directly to RunCoordinationRoundAsync (no replan) List ledger2 = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "WorkerA", instructionOrQuestion: "Done"); List finalAnswerResponse = CreateFinalAnswerResponse("Delegated correctly!"); TestReplayAgent manager = new( [factsResponse1, planResponse1, ledger1, ledger2, finalAnswerResponse], name: "Manager"); TestEchoAgent workerA = new(name: "WorkerA", prefix: "[A] "); TestEchoAgent workerB = new(name: "WorkerB", prefix: "[B] "); List collectedEvents = []; Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(workerA, workerB) .RequirePlanSignoff(false) .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Test delegation routing")], eventCollector: collectedEvents); // Assert: WorkerA should have produced response updates, WorkerB should not List agentUpdates = collectedEvents.OfType().ToList(); // WorkerA's executor should appear in the events agentUpdates.Should().Contain(e => e.Update.AuthorName == "WorkerA", "WorkerA was selected as next speaker and should have responded"); // WorkerB should NOT have responded agentUpdates.Should().NotContain(e => e.Update.AuthorName == "WorkerB", "WorkerB was not selected and should not have responded"); runResult.Result.Should().NotBeNull(); runResult.Result![0].Text.Should().Contain("Delegated correctly!"); } [Fact] public async Task Progress_Made_Decrements_StallCountAsync() { // Arrange: MaxStallCount=3, so a single stall won't trigger reset. // Round 1: isInLoop=true (stall count → 1), delegates to Worker // Round 2: progress being made (stall count → 0), delegates to Worker // Round 3: satisfied → final answer // No reset should occur because the stall count was decremented before reaching threshold. List facts1 = CreatePlanResponse("Initial facts"); List plan1 = CreatePlanResponse("Initial plan"); List ledger1 = CreateProgressLedgerResponse( isRequestSatisfied: false, isInLoop: true, // triggers stall increment → StallCount=1 isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Keep trying"); // After Worker responds → RunCoordinationRoundAsync (no replan) List ledger2 = CreateProgressLedgerResponse( isRequestSatisfied: false, isInLoop: false, isProgressBeingMade: true, // progress → stall count decrements → StallCount=0 nextSpeaker: "Worker", instructionOrQuestion: "Good progress"); // After Worker responds → RunCoordinationRoundAsync (no replan) List ledger3 = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "All done"); List finalAnswerResponse = CreateFinalAnswerResponse("Completed without reset!"); TestReplayAgent manager = new( [facts1, plan1, ledger1, ledger2, ledger3, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); List collectedEvents = []; Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .WithMaxStalls(3) // high threshold so single stall doesn't trigger reset .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Test stall decrement")], eventCollector: collectedEvents); // Assert: Three progress ledger updates, no stall-triggered reset collectedEvents.OfType().Should().HaveCount(3, "three coordination rounds should produce three progress ledger events"); // One initial plan, no replans (agent returns go directly to coordination, no replan) collectedEvents.OfType().Should().ContainSingle( "only one initial plan should be created"); collectedEvents.OfType().Should().BeEmpty( "no replan occurs on normal agent return; stall count never exceeded threshold"); runResult.Result.Should().NotBeNull(); runResult.Result![0].Text.Should().Contain("Completed without reset!"); } [Fact] public async Task Consecutive_Stalls_Trigger_ResetAsync() { // Arrange: MaxStallCount=1 — two consecutive stalls trigger reset (StallCount 2 > 1). // Round 1: isInLoop=true (stall count → 1), delegates to Worker // Round 2: isProgressBeingMade=false (stall count → 2 > 1 → IsStalled) → reset & replan // After reset: new plan → ledger(satisfied) → final answer List facts1 = CreatePlanResponse("Initial facts"); List plan1 = CreatePlanResponse("Initial plan"); List ledger1 = CreateProgressLedgerResponse( isRequestSatisfied: false, isInLoop: true, // stall #1 → StallCount=1 isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Keep trying"); // After Worker responds → RunCoordinationRoundAsync (no replan) List ledger2 = CreateProgressLedgerResponse( isRequestSatisfied: false, isInLoop: false, isProgressBeingMade: false, // stall #2 → StallCount=2 > 1 → IsStalled → reset nextSpeaker: "Worker", instructionOrQuestion: "No progress"); // Reset & replan: new facts + plan, then coordination round List resetFacts = CreatePlanResponse("Fresh facts after stall reset"); List resetPlan = CreatePlanResponse("Fresh plan after stall reset"); List postResetLedger = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Done"); List finalAnswerResponse = CreateFinalAnswerResponse("Recovered after consecutive stalls!"); TestReplayAgent manager = new( [facts1, plan1, ledger1, ledger2, resetFacts, resetPlan, postResetLedger, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); List collectedEvents = []; Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .WithMaxStalls(1) // requires 2 consecutive stalls (StallCount > 1) .Build(); // Act WorkflowRunResult runResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Test consecutive stalls")], eventCollector: collectedEvents); // Assert: Two pre-reset coordination rounds + one post-reset round = 3 ledger events collectedEvents.OfType().Should().HaveCount(3, "two pre-reset rounds and one post-reset round"); // One initial plan + one stall-triggered reset replan (no normal re-entry replans anymore) collectedEvents.OfType().Should().ContainSingle(); collectedEvents.OfType().Should().ContainSingle( "only one replan from stall-triggered reset; no replan on normal agent return"); runResult.Result.Should().NotBeNull(); runResult.Result![0].Text.Should().Contain("Recovered after consecutive stalls!"); } [Fact] public async Task PlanReview_Multiple_RevisionsAsync() { // Arrange: Human rejects the plan twice before approving on the third review. // Flow: facts1, plan1 → PlanCreatedEvent → plan review (pending) // resume with revision1 → facts2, plan2 → ReplannedEvent → plan review (pending) // resume with revision2 → facts3, plan3 → ReplannedEvent → plan review (pending) // resume with approval → ledger(satisfied) → finalAnswer List factsResponse1 = CreatePlanResponse("Initial facts"); List planResponse1 = CreatePlanResponse("Initial plan - too vague"); List factsResponse2 = CreatePlanResponse("Revised facts v2"); List planResponse2 = CreatePlanResponse("Revised plan v2 - still needs work"); List factsResponse3 = CreatePlanResponse("Revised facts v3"); List planResponse3 = CreatePlanResponse("Revised plan v3 - final version"); List progressLedgerResponse = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Execute final plan"); List finalAnswerResponse = CreateFinalAnswerResponse("Completed after multiple revisions"); TestReplayAgent manager = new( [factsResponse1, planResponse1, factsResponse2, planResponse2, factsResponse3, planResponse3, progressLedgerResponse, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(true) .Build(); CheckpointManager checkpointManager = CheckpointManager.CreateInMemory(); List allEvents = []; // Act 1: First run - should pause for plan review with initial plan WorkflowRunResult firstResult = await RunMagenticWorkflowAsync( workflow, [new ChatMessage(ChatRole.User, "Execute task")], checkpointManager: checkpointManager, eventCollector: allEvents); firstResult.PendingRequests.Should().ContainSingle(); ExternalRequest request1 = firstResult.PendingRequests[0].Request; MagenticPlanReviewRequest? reviewRequest1 = request1.Data.As(); reviewRequest1.Should().NotBeNull(); reviewRequest1!.Plan.Text.Should().Contain("Initial plan"); // Act 2: Resume with first revision MagenticPlanReviewResponse revision1 = reviewRequest1.Revise("Too vague, add more detail"); ExternalResponse revisionResponse1 = request1.CreateResponse(revision1); WorkflowRunResult secondResult = await ResumeMagenticWorkflowAsync( workflow, revisionResponse1, checkpointManager, firstResult.LastCheckpoint, eventCollector: allEvents); secondResult.PendingRequests.Should().NotBeEmpty(); ExternalRequest request2 = secondResult.PendingRequests[^1].Request; MagenticPlanReviewRequest? reviewRequest2 = request2.Data.As(); reviewRequest2.Should().NotBeNull(); reviewRequest2!.Plan.Text.Should().Contain("Revised plan v2"); // Act 3: Resume with second revision MagenticPlanReviewResponse revision2 = reviewRequest2.Revise("Still needs more work on step 3"); ExternalResponse revisionResponse2 = request2.CreateResponse(revision2); WorkflowRunResult thirdResult = await ResumeMagenticWorkflowAsync( workflow, revisionResponse2, checkpointManager, secondResult.LastCheckpoint, eventCollector: allEvents); thirdResult.PendingRequests.Should().NotBeEmpty(); ExternalRequest request3 = thirdResult.PendingRequests[^1].Request; MagenticPlanReviewRequest? reviewRequest3 = request3.Data.As(); reviewRequest3.Should().NotBeNull(); reviewRequest3!.Plan.Text.Should().Contain("Revised plan v3"); // Act 4: Resume with approval MagenticPlanReviewResponse approval = reviewRequest3.Approve(); ExternalResponse approvalResponse = request3.CreateResponse(approval); WorkflowRunResult fourthResult = await ResumeMagenticWorkflowAsync( workflow, approvalResponse, checkpointManager, thirdResult.LastCheckpoint, eventCollector: allEvents); // Assert: Multiple replan events emitted, final answer produced allEvents.OfType().Should().NotBeEmpty("initial plan emits PlanCreatedEvent"); allEvents.OfType().Should().HaveCountGreaterThanOrEqualTo(2, "two revisions should emit at least two ReplannedEvents"); fourthResult.Result.Should().NotBeNull(); fourthResult.Result![0].Text.Should().Contain("Completed after multiple revisions"); } [Fact] public void Empty_Team_Build_Throws() { // Arrange: No participants added to the builder. TestReplayAgent manager = new( [CreatePlanResponse("Facts"), CreatePlanResponse("Plan")], name: "Manager"); MagenticWorkflowBuilder builder = new MagenticWorkflowBuilder(manager) // No .AddParticipants() — empty team .RequirePlanSignoff(false); // Act & Assert: Build() should throw because the team is empty. Action buildAction = () => builder.Build(); buildAction.Should().Throw() .WithMessage("*participant*"); } [Fact] public async Task Terminated_Context_Rejects_New_MessagesAsync() { // Arrange: Run a workflow to completion so IsTerminated=true, then send another message. // The framework accepts the message (TrySendMessageAsync returns true), but Magentic // should error out internally, surfacing as a WorkflowErrorEvent. List factsResponse = CreatePlanResponse("Facts"); List planResponse = CreatePlanResponse("The plan"); List satisfiedLedger = CreateProgressLedgerResponse( isRequestSatisfied: true, isInLoop: false, isProgressBeingMade: true, nextSpeaker: "Worker", instructionOrQuestion: "Done"); List finalAnswerResponse = CreateFinalAnswerResponse("Task done"); TestReplayAgent manager = new( [factsResponse, planResponse, satisfiedLedger, finalAnswerResponse], name: "Manager"); TestEchoAgent worker = new(name: "Worker"); Workflow workflow = new MagenticWorkflowBuilder(manager) .AddParticipants(worker) .RequirePlanSignoff(false) .Build(); InProcessExecutionEnvironment environment = ExecutionEnvironment.InProcess_Lockstep .ToWorkflowExecutionEnvironment() .WithCheckpointing(CheckpointManager.CreateInMemory()); await using StreamingRun run = await environment.OpenStreamingAsync(workflow); // Send the initial messages and run to completion await run.TrySendMessageAsync(new List { new(ChatRole.User, "Do the task") }); await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); // Drain the stream to completion — workflow yields output and sets IsTerminated=true WorkflowOutputEvent? output = null; await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false).ConfigureAwait(false)) { if (evt is WorkflowOutputEvent o) { output = o; } } output.Should().NotBeNull("workflow should have completed with output"); // Act: Send a new message after termination — framework accepts it, but Magentic errors out bool accepted = await run.TrySendMessageAsync(new List { new(ChatRole.User, "Another message") }); accepted.Should().BeTrue("framework does not have a terminal state — it always queues messages"); await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); // Watch for the error WorkflowErrorEvent? errorEvent = null; await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false).ConfigureAwait(false)) { if (evt is WorkflowErrorEvent e) { errorEvent = e; } } // Assert: Magentic should have rejected the message with an InvalidOperationException // (may be wrapped in TargetInvocationException by the framework's reflection-based dispatch) errorEvent.Should().NotBeNull("sending a message after termination should produce a WorkflowErrorEvent"); Exception actual = errorEvent!.Exception is System.Reflection.TargetInvocationException tie && tie.InnerException != null ? tie.InnerException : errorEvent.Exception!; actual.Should().BeOfType(); actual.Message.Should().Contain("terminated"); } #region Helper Methods private sealed record WorkflowRunResult( string UpdateText, List? Result, CheckpointInfo? LastCheckpoint, List PendingRequests); private static List CreatePlanResponse(string plan) { return [ new ChatMessage(ChatRole.Assistant, plan) { MessageId = Guid.NewGuid().ToString("N"), CreatedAt = DateTimeOffset.UtcNow } ]; } private static List CreateProgressLedgerResponse( bool isRequestSatisfied, bool isInLoop, bool isProgressBeingMade, string nextSpeaker, string instructionOrQuestion) { string isRequestSatisfiedStr = isRequestSatisfied ? "true" : "false"; string isInLoopStr = isInLoop ? "true" : "false"; string isProgressBeingMadeStr = isProgressBeingMade ? "true" : "false"; string nextSpeakerJson = JsonSerializer.Serialize(nextSpeaker); string instructionJson = JsonSerializer.Serialize(instructionOrQuestion); string ledgerJson = $$""" { "is_request_satisfied": { "answer": {{isRequestSatisfiedStr}}, "reason": "test reason" }, "is_in_loop": { "answer": {{isInLoopStr}}, "reason": "test reason" }, "is_progress_being_made": { "answer": {{isProgressBeingMadeStr}}, "reason": "test reason" }, "next_speaker": { "answer": {{nextSpeakerJson}}, "reason": "test reason" }, "instruction_or_question": { "answer": {{instructionJson}}, "reason": "test reason" } } """; return [ new ChatMessage(ChatRole.Assistant, ledgerJson) { MessageId = Guid.NewGuid().ToString("N"), CreatedAt = DateTimeOffset.UtcNow } ]; } private static List CreateFinalAnswerResponse(string answer) { return [ new ChatMessage(ChatRole.Assistant, answer) { MessageId = Guid.NewGuid().ToString("N"), CreatedAt = DateTimeOffset.UtcNow } ]; } private static async Task ResumeMagenticWorkflowAsync( Workflow workflow, ExternalResponse response, CheckpointManager checkpointManager, CheckpointInfo? fromCheckpoint, List? eventCollector = null) { InProcessExecutionEnvironment environment = ExecutionEnvironment.InProcess_Lockstep .ToWorkflowExecutionEnvironment() .WithCheckpointing(checkpointManager); await using StreamingRun run = fromCheckpoint != null ? await environment.ResumeStreamingAsync(workflow, fromCheckpoint) : await environment.OpenStreamingAsync(workflow); await run.SendResponseAsync(response); return await ProcessWorkflowRunAsync(run, eventCollector); } private static async Task RunMagenticWorkflowAsync( Workflow workflow, List input, CheckpointManager? checkpointManager = null, List? eventCollector = null) { checkpointManager ??= CheckpointManager.CreateInMemory(); InProcessExecutionEnvironment environment = ExecutionEnvironment.InProcess_Lockstep .ToWorkflowExecutionEnvironment() .WithCheckpointing(checkpointManager); await using StreamingRun run = await environment.OpenStreamingAsync(workflow); await run.TrySendMessageAsync(input); await run.TrySendMessageAsync(new TurnToken(emitEvents: true)); return await ProcessWorkflowRunAsync(run, eventCollector); } private static async Task ProcessWorkflowRunAsync( StreamingRun run, List? eventCollector = null) { StringBuilder sb = new(); WorkflowOutputEvent? output = null; CheckpointInfo? lastCheckpoint = null; List pendingRequests = []; await foreach (WorkflowEvent evt in run.WatchStreamAsync(blockOnPendingRequest: false).ConfigureAwait(false)) { eventCollector?.Add(evt); switch (evt) { case AgentResponseUpdateEvent responseUpdate: sb.Append(responseUpdate.Data); break; case RequestInfoEvent requestInfo: pendingRequests.Add(requestInfo); break; case WorkflowOutputEvent e: output = e; break; case WorkflowErrorEvent errorEvent: Assert.Fail($"Workflow execution failed with error: {errorEvent.Exception}"); break; case SuperStepCompletedEvent stepCompleted: lastCheckpoint = stepCompleted.CompletionInfo?.Checkpoint; break; } } return new(sb.ToString(), output?.As>(), lastCheckpoint, pendingRequests); } #endregion }