From 959650558321c16833ad30671e593e7d78ddb978 Mon Sep 17 00:00:00 2001 From: Saibernard Yogendran Date: Thu, 27 Aug 2026 20:02:02 -0400 Subject: [PATCH 1/2] .NET: Propagate AgentRunOptions to workflow agents --- .../ChatProtocolExecutor.cs | 13 +- .../IWorkflowContext.cs | 5 + .../IWorkflowContextExtensions.cs | 8 + .../InProc/InProcessExecutionEnvironment.cs | 43 +++++ .../InProc/InProcessRunner.cs | 23 ++- .../InProc/InProcessRunnerContext.cs | 21 ++- .../Specialized/AIAgentHostExecutor.cs | 26 ++- .../Specialized/GroupChatHost.cs | 11 +- .../Specialized/HandoffAgentExecutor.cs | 51 +++++- .../Specialized/HandoffStartExecutor.cs | 5 +- .../Specialized/Magentic/MagenticManager.cs | 24 ++- .../Magentic/MagenticOrchestrator.cs | 16 +- .../Specialized/WorkflowHostExecutor.cs | 7 +- .../TurnToken.cs | 36 +++- .../WorkflowHostAgent.cs | 4 +- .../WorkflowSession.cs | 13 +- .../ChatProtocolExecutorTests.cs | 99 ++++++++++ .../GroupChatOrchestrationTests.cs | 58 ++++++ .../HandoffOrchestrationTests.cs | 44 +++++ .../MagenticOrchestrationTests.cs | 97 ++++++++++ .../OrchestrationTestHelpers.cs | 25 +++ .../RecordingEchoAgent.cs | 2 + .../RecordingReplayAgent.cs | 2 + .../Sample/12_HandOff_HostAsAgent.cs | 4 + .../WorkflowHostSmokeTests.cs | 172 ++++++++++++++++++ 25 files changed, 768 insertions(+), 41 deletions(-) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/ChatProtocolExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/ChatProtocolExecutor.cs index 18541464c19..2d40f8ab41a 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/ChatProtocolExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/ChatProtocolExecutor.cs @@ -140,7 +140,7 @@ public ValueTask TakeTurnAsync(TurnToken token, IWorkflowContext context, Cancel async ValueTask?> InvokeTakeTurnAsync(List? maybePendingMessages, IWorkflowContext context, CancellationToken cancellationToken) { - await this.TakeTurnAsync(maybePendingMessages ?? s_initFunction(), context, token.EmitEvents, cancellationToken) + await this.TakeTurnAsync(maybePendingMessages ?? s_initFunction(), context, token, cancellationToken) .ConfigureAwait(false); if (this.AutoSendTurnToken) @@ -154,6 +154,17 @@ await this.TakeTurnAsync(maybePendingMessages ?? s_initFunction(), context, toke } } + /// + /// When overridden in a derived class, processes the accumulated chat messages and the full turn token. + /// + /// The list of chat messages accumulated since the last turn. + /// The workflow context in which the executor executes. + /// The token containing turn configuration, including any agent run options. + /// The to monitor for cancellation requests. + /// A representing the asynchronous operation. + protected virtual ValueTask TakeTurnAsync(List messages, IWorkflowContext context, TurnToken turnToken, CancellationToken cancellationToken = default) + => this.TakeTurnAsync(messages, context, turnToken.EmitEvents, cancellationToken); + /// /// Processes the current set of turn messages using the specified asynchronous processing function. /// diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowContext.cs index b8b35fffd64..e149e48c1d9 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowContext.cs @@ -195,3 +195,8 @@ ValueTask ReadOrInitStateAsync(string key, Func initialStateFactory, Ca /// bool ConcurrentRunsEnabled { get; } } + +internal interface IWorkflowAgentRunOptionsContext +{ + bool TryGetAgentRunOptions(out AgentRunOptions? runOptions); +} diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowContextExtensions.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowContextExtensions.cs index 950078cd3d0..ab4d1f69a95 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowContextExtensions.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/IWorkflowContextExtensions.cs @@ -11,6 +11,14 @@ namespace Microsoft.Agents.AI.Workflows; /// public static class IWorkflowContextExtensions { + internal static AgentRunOptions? GetAgentRunOptions(this IWorkflowContext context, AgentRunOptions? fallback) + { + return context is IWorkflowAgentRunOptionsContext invocationContext + && invocationContext.TryGetAgentRunOptions(out AgentRunOptions? runOptions) + ? runOptions + : fallback; + } + /// /// Invokes an asynchronous operation that reads, updates, and persists workflow state associated with the specified /// key. diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs index d08c23c0899..6653662339c 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessExecutionEnvironment.cs @@ -84,6 +84,26 @@ public async ValueTask RunStreamingAsync( return await runHandle.EnqueueAndStreamAsync(input, cancellationToken).ConfigureAwait(false); } + internal async ValueTask RunStreamingWithAgentRunOptionsAsync( + Workflow workflow, + TInput input, + AgentRunOptions? runOptions, + string? sessionId = null, + CancellationToken cancellationToken = default) where TInput : notnull + { + InProcessRunner runner = InProcessRunner.CreateTopLevelRunner( + workflow, + this.CheckpointManager, + sessionId, + this.EnableConcurrentRuns, + knownValidInputTypes: [], + hasAgentRunOptions: true, + runOptions); + AsyncRunHandle runHandle = await runner.BeginStreamAsync(this.ExecutionMode, cancellationToken).ConfigureAwait(false); + + return await runHandle.EnqueueAndStreamAsync(input, cancellationToken).ConfigureAwait(false); + } + [MemberNotNull(nameof(CheckpointManager))] private void VerifyCheckpointingConfigured() { @@ -133,6 +153,29 @@ internal async ValueTask ResumeStreamingInternalAsync( return new(runHandle); } + internal async ValueTask ResumeStreamingWithAgentRunOptionsInternalAsync( + Workflow workflow, + CheckpointInfo fromCheckpoint, + bool republishPendingEvents, + AgentRunOptions? runOptions, + CancellationToken cancellationToken = default) + { + this.VerifyCheckpointingConfigured(); + + InProcessRunner runner = InProcessRunner.CreateTopLevelRunner( + workflow, + this.CheckpointManager, + fromCheckpoint.SessionId, + this.EnableConcurrentRuns, + knownValidInputTypes: [], + hasAgentRunOptions: true, + runOptions); + AsyncRunHandle runHandle = await runner.ResumeStreamAsync(this.ExecutionMode, fromCheckpoint, republishPendingEvents, cancellationToken) + .ConfigureAwait(false); + + return new(runHandle); + } + private async ValueTask BeginRunHandlingChatProtocolAsync(Workflow workflow, TInput input, string? sessionId = null, diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs index 47167cd9e53..fe29b2c8a3e 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunner.cs @@ -22,16 +22,18 @@ namespace Microsoft.Agents.AI.Workflows.InProc; /// scenarios where workflow execution does not require executor distribution. internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle { - public static InProcessRunner CreateTopLevelRunner(Workflow workflow, ICheckpointManager? checkpointManager, string? sessionId = null, bool enableConcurrentRuns = false, IEnumerable? knownValidInputTypes = null) + public static InProcessRunner CreateTopLevelRunner(Workflow workflow, ICheckpointManager? checkpointManager, string? sessionId = null, bool enableConcurrentRuns = false, IEnumerable? knownValidInputTypes = null, bool hasAgentRunOptions = false, AgentRunOptions? agentRunOptions = null) { return new InProcessRunner(workflow, checkpointManager, sessionId, enableConcurrentRuns: enableConcurrentRuns, - knownValidInputTypes: knownValidInputTypes); + knownValidInputTypes: knownValidInputTypes, + hasAgentRunOptions: hasAgentRunOptions, + agentRunOptions: agentRunOptions); } - public static InProcessRunner CreateSubworkflowRunner(Workflow workflow, ICheckpointManager? checkpointManager, string? sessionId = null, object? existingOwnerSignoff = null, bool enableConcurrentRuns = false, IEnumerable? knownValidInputTypes = null) + public static InProcessRunner CreateSubworkflowRunner(Workflow workflow, ICheckpointManager? checkpointManager, string? sessionId = null, object? existingOwnerSignoff = null, bool enableConcurrentRuns = false, IEnumerable? knownValidInputTypes = null, bool hasAgentRunOptions = false, AgentRunOptions? agentRunOptions = null) { return new InProcessRunner(workflow, checkpointManager, @@ -39,10 +41,12 @@ public static InProcessRunner CreateSubworkflowRunner(Workflow workflow, ICheckp existingOwnerSignoff: existingOwnerSignoff, enableConcurrentRuns: enableConcurrentRuns, knownValidInputTypes: knownValidInputTypes, + hasAgentRunOptions: hasAgentRunOptions, + agentRunOptions: agentRunOptions, subworkflow: true); } - private InProcessRunner(Workflow workflow, ICheckpointManager? checkpointManager, string? sessionId = null, object? existingOwnerSignoff = null, bool subworkflow = false, bool enableConcurrentRuns = false, IEnumerable? knownValidInputTypes = null) + private InProcessRunner(Workflow workflow, ICheckpointManager? checkpointManager, string? sessionId = null, object? existingOwnerSignoff = null, bool subworkflow = false, bool enableConcurrentRuns = false, IEnumerable? knownValidInputTypes = null, bool hasAgentRunOptions = false, AgentRunOptions? agentRunOptions = null) { if (enableConcurrentRuns && !workflow.AllowConcurrent) { @@ -54,7 +58,7 @@ private InProcessRunner(Workflow workflow, ICheckpointManager? checkpointManager this.StartExecutorId = workflow.StartExecutorId; this.Workflow = Throw.IfNull(workflow); - this.RunContext = new InProcessRunnerContext(workflow, this.SessionId, checkpointingEnabled: checkpointManager != null, this.OutgoingEvents, this.StepTracer, existingOwnerSignoff, subworkflow, enableConcurrentRuns); + this.RunContext = new InProcessRunnerContext(workflow, this.SessionId, checkpointingEnabled: checkpointManager != null, this.OutgoingEvents, this.StepTracer, existingOwnerSignoff, subworkflow, enableConcurrentRuns, hasAgentRunOptions, agentRunOptions); this.CheckpointManager = checkpointManager; this._knownValidInputTypes = knownValidInputTypes != null @@ -259,6 +263,15 @@ await executor.OnMessageDeliveryStartingAsync(tracelessContext, cancellationToke { (object message, TypeId messageType) = await TranslateMessageAsync(envelope).ConfigureAwait(false); + // A restored token can contain options from an earlier in-memory run or no options after + // serialization. The options supplied for the current invocation take precedence. + if (message is TurnToken turnToken + && this.RunContext.TryGetAgentRunOptions(out AgentRunOptions? runOptions) + && !ReferenceEquals(turnToken.RunOptions, runOptions)) + { + message = new TurnToken(turnToken.EmitEvents, runOptions); + } + await executor.ExecuteCoreAsync( message, messageType, diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs index 8ccfb957294..c6f8f149614 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/InProc/InProcessRunnerContext.cs @@ -19,7 +19,7 @@ namespace Microsoft.Agents.AI.Workflows.InProc; -internal sealed class InProcessRunnerContext : IRunnerContext +internal sealed class InProcessRunnerContext : IRunnerContext, IWorkflowAgentRunOptionsContext { private int _runEnded; private readonly string _sessionId; @@ -37,6 +37,8 @@ internal sealed class InProcessRunnerContext : IRunnerContext private readonly ConcurrentDictionary _joinedSubworkflowRunners = new(); private readonly ConcurrentDictionary _externalRequests = new(); + private readonly bool _hasAgentRunOptions; + private readonly AgentRunOptions? _agentRunOptions; public InProcessRunnerContext( Workflow workflow, @@ -47,6 +49,8 @@ public InProcessRunnerContext( object? existingOwnershipSignoff = null, bool subworkflow = false, bool enableConcurrentRuns = false, + bool hasAgentRunOptions = false, + AgentRunOptions? agentRunOptions = null, ILogger? logger = null) { if (enableConcurrentRuns) @@ -62,6 +66,8 @@ public InProcessRunnerContext( this._workflow = workflow; this._sessionId = sessionId; + this._hasAgentRunOptions = hasAgentRunOptions; + this._agentRunOptions = agentRunOptions; this._edgeMap = new(this, this._workflow, stepTracer); this._outputFilter = new(workflow); @@ -310,6 +316,12 @@ public IWorkflowContext BindWorkflowContext(string executorId, Dictionary? traceContext) : IWorkflowContext + Dictionary? traceContext) : IWorkflowContext, IWorkflowAgentRunOptionsContext { + public bool TryGetAgentRunOptions(out AgentRunOptions? runOptions) + { + return RunnerContext.TryGetAgentRunOptions(out runOptions); + } + public ValueTask AddEventAsync(WorkflowEvent workflowEvent, CancellationToken cancellationToken = default) => RunnerContext.AddEventAsync(workflowEvent, cancellationToken); public ValueTask SendMessageAsync(object message, string? targetId = null, CancellationToken cancellationToken = default) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs index 85ff6e40528..421bd92dafb 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/AIAgentHostExecutor.cs @@ -30,6 +30,7 @@ internal class AIAgentHostExecutor : ChatProtocolExecutor private readonly AIAgentHostOptions _options; private AgentSession? _session; private bool? _currentTurnEmitEvents; + private AgentRunOptions? _currentTurnRunOptions; private AIContentExternalHandler? _userInputHandler; private AIContentExternalHandler? _functionCallHandler; @@ -77,6 +78,7 @@ internal void ResetChat(ResetChatSignal signal, IWorkflowContext context) { this._session = null; this._currentTurnEmitEvents = null; + this._currentTurnRunOptions = null; } private ValueTask HandleUserInputResponseAsync( @@ -99,7 +101,7 @@ private ValueTask HandleUserInputResponseAsync( MessageId = Guid.NewGuid().ToString("N"), }); - await this.ContinueTurnAsync(pendingMessages, ctx, this._currentTurnEmitEvents ?? false, ct).ConfigureAwait(false); + await this.ContinueTurnAsync(pendingMessages, ctx, this._currentTurnEmitEvents ?? false, ctx.GetAgentRunOptions(this._currentTurnRunOptions), ct).ConfigureAwait(false); // Clear the buffered turn messages because they were consumed by ContinueTurnAsync. return null; @@ -127,7 +129,7 @@ private ValueTask HandleFunctionResultAsync( MessageId = Guid.NewGuid().ToString("N"), }); - await this.ContinueTurnAsync(pendingMessages, ctx, this._currentTurnEmitEvents ?? false, ct).ConfigureAwait(false); + await this.ContinueTurnAsync(pendingMessages, ctx, this._currentTurnEmitEvents ?? false, ctx.GetAgentRunOptions(this._currentTurnRunOptions), ct).ConfigureAwait(false); // Clear the buffered turn messages because they were consumed by ContinueTurnAsync. return null; @@ -176,9 +178,10 @@ protected internal override async ValueTask OnCheckpointRestoredAsync(IWorkflowC || (this._functionCallHandler?.HasPendingRequests == true); // While we save this on the instance, we are not cross-run shareable, but as AgentBinding uses the factory pattern this is not an issue - private async ValueTask ContinueTurnAsync(List messages, IWorkflowContext context, bool emitEvents, CancellationToken cancellationToken) + private async ValueTask ContinueTurnAsync(List messages, IWorkflowContext context, bool emitEvents, AgentRunOptions? runOptions, CancellationToken cancellationToken) { this._currentTurnEmitEvents = emitEvents; + this._currentTurnRunOptions = runOptions; if (this._options.ForwardIncomingMessages) { await context.SendMessageAsync(messages, cancellationToken).ConfigureAwait(false); @@ -188,7 +191,7 @@ private async ValueTask ContinueTurnAsync(List messages, IWorkflowC ? messages.Select(m => m.ChatAssistantToUserIfNotFromNamed(this._agent.Name ?? this._agent.Id)) : messages; - AgentResponse response = await this.InvokeAgentAsync(filteredMessages, context, emitEvents, cancellationToken).ConfigureAwait(false); + AgentResponse response = await this.InvokeAgentAsync(filteredMessages, context, emitEvents, runOptions, cancellationToken).ConfigureAwait(false); // Filter out server-side artifacts (reasoning tokens, web search calls, etc.) // that are internal to this agent. Forwarding them to other agents in the workflow @@ -204,8 +207,9 @@ await context.SendMessageAsync(forwardableMessages, cancellationToken) // If we have no outstanding requests, we can yield a turn token back to the workflow. if (!this.HasOutstandingRequests) { - await context.SendMessageAsync(new TurnToken(this._currentTurnEmitEvents), cancellationToken).ConfigureAwait(false); + await context.SendMessageAsync(new TurnToken(this._currentTurnEmitEvents, this._currentTurnRunOptions), cancellationToken).ConfigureAwait(false); this._currentTurnEmitEvents = null; // Possibly not actually necessary, but cleaning this up makes it clearer when debugging + this._currentTurnRunOptions = null; } } @@ -213,9 +217,17 @@ protected override ValueTask TakeTurnAsync(List messages, IWorkflow => this.ContinueTurnAsync(messages, context, TurnExtensions.ShouldEmitStreamingEvents(turnTokenSetting: emitEvents, this._options.EmitAgentUpdateEvents), + runOptions: null, cancellationToken); - private async ValueTask InvokeAgentAsync(IEnumerable messages, IWorkflowContext context, bool emitUpdateEvents, CancellationToken cancellationToken = default) + protected override ValueTask TakeTurnAsync(List messages, IWorkflowContext context, TurnToken turnToken, CancellationToken cancellationToken = default) + => this.ContinueTurnAsync(messages, + context, + turnToken.ShouldEmitStreamingEvents(this._options.EmitAgentUpdateEvents), + context.GetAgentRunOptions(turnToken.RunOptions), + cancellationToken); + + private async ValueTask InvokeAgentAsync(IEnumerable messages, IWorkflowContext context, bool emitUpdateEvents, AgentRunOptions? runOptions, CancellationToken cancellationToken = default) { AgentResponse response; AIAgentUnservicedRequestsCollector collector = new(this._userInputHandler, this._functionCallHandler); @@ -226,6 +238,7 @@ private async ValueTask InvokeAgentAsync(IEnumerable IAsyncEnumerable agentStream = this._agent.RunStreamingAsync( messages, await this.EnsureSessionAsync(context, cancellationToken).ConfigureAwait(false), + runOptions, cancellationToken: cancellationToken); List updates = []; @@ -281,6 +294,7 @@ await this.EnsureSessionAsync(context, cancellationToken).ConfigureAwait(false), // Otherwise, run the agent in non-streaming mode. response = await this._agent.RunAsync(messages, await this.EnsureSessionAsync(context, cancellationToken).ConfigureAwait(false), + runOptions, cancellationToken: cancellationToken) .ConfigureAwait(false); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/GroupChatHost.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/GroupChatHost.cs index ec9ebdd86ff..bbf475aa8d3 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/GroupChatHost.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/GroupChatHost.cs @@ -43,7 +43,16 @@ protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBui => base.ConfigureProtocol(protocolBuilder).YieldsOutput>(); protected override async ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellationToken = default) + => await this.TakeTurnAsync(messages, context, new TurnToken(emitEvents), cancellationToken).ConfigureAwait(false); + + protected override async ValueTask TakeTurnAsync(List messages, IWorkflowContext context, TurnToken turnToken, CancellationToken cancellationToken = default) { + AgentRunOptions? runOptions = context.GetAgentRunOptions(turnToken.RunOptions); + if (!ReferenceEquals(runOptions, turnToken.RunOptions)) + { + turnToken = new TurnToken(turnToken.EmitEvents, runOptions); + } + this._manager ??= this._managerFactory(this._agents); // The delta arriving here is either the initial user input (turn 0) or the most recent speaker's @@ -86,7 +95,7 @@ protected override async ValueTask TakeTurnAsync(List messages, IWo this._manager.IterationCount++; this._currentSpeakerExecutorId = executor.Id; - await context.SendMessageAsync(new TurnToken(emitEvents), executor.Id, cancellationToken).ConfigureAwait(false); + await context.SendMessageAsync(turnToken, executor.Id, cancellationToken).ConfigureAwait(false); return; } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffAgentExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffAgentExecutor.cs index 50a6871aa2c..b4c87fdcbf7 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffAgentExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffAgentExecutor.cs @@ -146,6 +146,50 @@ public HandoffAgentExecutor(AIAgent agent, HashSet handoffs, Hand return result; } + private AgentRunOptions? MergeRunOptions(AgentRunOptions? runOptions) + { + if (this._agentOptions is null) + { + return runOptions; + } + + if (runOptions is null) + { + return this._agentOptions; + } + + ChatClientAgentRunOptions mergedOptions = runOptions is ChatClientAgentRunOptions chatClientOptions + ? (ChatClientAgentRunOptions)chatClientOptions.Clone() + : new ChatClientAgentRunOptions + { + AllowBackgroundResponses = runOptions.AllowBackgroundResponses, + ContinuationToken = runOptions.ContinuationToken, + AdditionalProperties = runOptions.AdditionalProperties?.Clone(), + ResponseFormat = runOptions.ResponseFormat, + }; + + ChatOptions handoffChatOptions = this._agentOptions.ChatOptions!; + ChatOptions mergedChatOptions = mergedOptions.ChatOptions ?? new(); + + mergedChatOptions.AllowMultipleToolCalls = handoffChatOptions.AllowMultipleToolCalls; + if (!string.IsNullOrWhiteSpace(handoffChatOptions.Instructions)) + { + mergedChatOptions.Instructions = !string.IsNullOrWhiteSpace(mergedChatOptions.Instructions) + ? $"{handoffChatOptions.Instructions}\n{mergedChatOptions.Instructions}" + : handoffChatOptions.Instructions; + } + + if (handoffChatOptions.Tools is { Count: > 0 }) + { + mergedChatOptions.Tools = mergedChatOptions.Tools is { Count: > 0 } runTools + ? [.. runTools, .. handoffChatOptions.Tools] + : [.. handoffChatOptions.Tools]; + } + + mergedOptions.ChatOptions = mergedChatOptions; + return mergedOptions; + } + private AIContentExternalHandler? _userInputHandler; private AIContentExternalHandler? _functionCallHandler; @@ -251,7 +295,8 @@ private ValueTask HandleFunctionResultAsync( .CopyWithAssistantToUserForOtherParticipants(this._agent.Name ?? this._agent.Id); bool emitUpdateEvents = state.IncomingState!.ShouldEmitStreamingEvents(this._options.EmitAgentResponseUpdateEvents); - AgentInvocationResult result = await this.InvokeAgentAsync(messagesForAgent, context, emitUpdateEvents, cancellationToken) + AgentRunOptions? runOptions = context.GetAgentRunOptions(state.IncomingState.TurnToken.RunOptions); + AgentInvocationResult result = await this.InvokeAgentAsync(messagesForAgent, context, emitUpdateEvents, runOptions, cancellationToken) .ConfigureAwait(false); if (this.HasOutstandingRequests && result.IsHandoffRequested) @@ -422,7 +467,7 @@ async Task RestoreAgentSessionAsync() private bool HasOutstandingRequests => (this._userInputHandler?.HasPendingRequests == true) || (this._functionCallHandler?.HasPendingRequests == true); - private async ValueTask InvokeAgentAsync(IEnumerable messages, IWorkflowContext context, bool emitUpdateEvents, CancellationToken cancellationToken = default) + private async ValueTask InvokeAgentAsync(IEnumerable messages, IWorkflowContext context, bool emitUpdateEvents, AgentRunOptions? runOptions, CancellationToken cancellationToken = default) { AgentResponse response; @@ -434,7 +479,7 @@ private async ValueTask InvokeAgentAsync(IEnumerable agentStream = - this._agent.RunStreamingAsync(messages, this._session, this._agentOptions, cancellationToken); + this._agent.RunStreamingAsync(messages, this._session, this.MergeRunOptions(runOptions), cancellationToken); await foreach (AgentResponseUpdate update in agentStream.ConfigureAwait(false)) { diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffStartExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffStartExecutor.cs index 47ef204d867..0508c6f03a0 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffStartExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/HandoffStartExecutor.cs @@ -68,6 +68,9 @@ protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBui base.ConfigureProtocol(protocolBuilder).SendsMessage(); protected override ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellationToken = default) + => this.TakeTurnAsync(messages, context, new TurnToken(emitEvents), cancellationToken); + + protected override ValueTask TakeTurnAsync(List messages, IWorkflowContext context, TurnToken turnToken, CancellationToken cancellationToken = default) { return context.InvokeWithStateAsync( async (HandoffSharedState? sharedState, IWorkflowContext context, CancellationToken cancellationToken) => @@ -83,7 +86,7 @@ protected override ValueTask TakeTurnAsync(List messages, IWorkflow // If we are configured to return to the previous agent, include the previous agent id in the handoff state. // If there was no previousAgent, it will still be null. - HandoffState turnState = new(new(emitEvents), null, returnToPrevious ? previousAgentId : null); + HandoffState turnState = new(turnToken, null, returnToPrevious ? previousAgentId : null); await context.SendMessageAsync(turnState, cancellationToken).ConfigureAwait(false); diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/Magentic/MagenticManager.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/Magentic/MagenticManager.cs index 936d9d951e7..8a30027611f 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/Magentic/MagenticManager.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/Magentic/MagenticManager.cs @@ -30,10 +30,13 @@ private static async ValueTask CheckResponseAsync(Task InvokeAgentAsync(IEnumerable messages, IWorkflowContext context, CancellationToken cancellationToken, AgentSession? session = null) - => CheckResponseAsync(managerAgent.RunAsync(messages, session, cancellationToken: cancellationToken), context, cancellationToken); + private ValueTask InvokeAgentAsync(IEnumerable messages, IWorkflowContext context, AgentRunOptions? runOptions, CancellationToken cancellationToken, AgentSession? session = null) + => CheckResponseAsync(managerAgent.RunAsync(messages, session, runOptions, cancellationToken), context, cancellationToken); - public async ValueTask UpdatePlanAsync(MagenticTaskContext taskContext, IWorkflowContext context, CancellationToken cancellationToken) + public ValueTask UpdatePlanAsync(MagenticTaskContext taskContext, IWorkflowContext context, CancellationToken cancellationToken) + => this.UpdatePlanAsync(taskContext, context, runOptions: null, cancellationToken); + + public async ValueTask UpdatePlanAsync(MagenticTaskContext taskContext, IWorkflowContext context, AgentRunOptions? runOptions, CancellationToken cancellationToken) { // If we already have a TaskLedger, we need to update the facts based on the existing factset; otherwise, we use the initial facts construction bool isReplan = taskContext.TaskLedger != null; @@ -44,6 +47,7 @@ public async ValueTask UpdatePlanAsync(MagenticTaskContext taskConte ChatMessage updatedFacts = await this.InvokeAgentAsync( messages: [.. taskContext.ChatHistory, factsRequest], context, + runOptions, cancellationToken, localSession) .ConfigureAwait(false); @@ -54,6 +58,7 @@ public async ValueTask UpdatePlanAsync(MagenticTaskContext taskConte // history, facts request, or updated facts in the messages list. messages: [planRequest], context, + runOptions, cancellationToken, localSession) .ConfigureAwait(false); @@ -63,7 +68,10 @@ public async ValueTask UpdatePlanAsync(MagenticTaskContext taskConte return new(updatedFacts, updatedPlan); } - public async ValueTask UpdateProgressLedgerAsync(MagenticTaskContext taskContext, IWorkflowContext context, CancellationToken cancellationToken) + public ValueTask UpdateProgressLedgerAsync(MagenticTaskContext taskContext, IWorkflowContext context, CancellationToken cancellationToken) + => this.UpdateProgressLedgerAsync(taskContext, context, runOptions: null, cancellationToken); + + public async ValueTask UpdateProgressLedgerAsync(MagenticTaskContext taskContext, IWorkflowContext context, AgentRunOptions? runOptions, CancellationToken cancellationToken) { ChatMessage progressRequest = new(ChatRole.User, taskContext.ToProgressLedgerPrompt()); @@ -74,6 +82,7 @@ public async ValueTask UpdateProgressLedgerAsync(MagenticTaskContext taskContext ChatMessage progressUpdateMessage = await this.InvokeAgentAsync( messages: [.. taskContext.ChatHistory, progressRequest], context, + runOptions, cancellationToken) .ConfigureAwait(false); @@ -105,10 +114,13 @@ public async ValueTask UpdateProgressLedgerAsync(MagenticTaskContext taskContext lastException?.Throw(); } - public async ValueTask PrepareFinalAnswerAsync(MagenticTaskContext taskContext, IWorkflowContext context, CancellationToken cancellationToken) + public ValueTask PrepareFinalAnswerAsync(MagenticTaskContext taskContext, IWorkflowContext context, CancellationToken cancellationToken) + => this.PrepareFinalAnswerAsync(taskContext, context, runOptions: null, cancellationToken); + + public async ValueTask PrepareFinalAnswerAsync(MagenticTaskContext taskContext, IWorkflowContext context, AgentRunOptions? runOptions, CancellationToken cancellationToken) { ChatMessage finalAnswerRequest = new(ChatRole.User, taskContext.ToFinalAnswerPrompt()); - ChatMessage finalAnswer = await this.InvokeAgentAsync([.. taskContext.ChatHistory, finalAnswerRequest], context, cancellationToken) + ChatMessage finalAnswer = await this.InvokeAgentAsync([.. taskContext.ChatHistory, finalAnswerRequest], context, runOptions, cancellationToken) .ConfigureAwait(false); return new(ChatRole.Assistant, finalAnswer.Text) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/Magentic/MagenticOrchestrator.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/Magentic/MagenticOrchestrator.cs index 7f1fa92d0f6..0feede800f8 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/Magentic/MagenticOrchestrator.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/Magentic/MagenticOrchestrator.cs @@ -93,6 +93,7 @@ internal class MagenticOrchestrator(AIAgent managerAgent, List team, Ta private MagenticTaskContext? _taskContext; private PortBinding? _planReviewPort; private string? _currentSpeakerExecutorId; + private AgentRunOptions? _currentTurnRunOptions; protected override ProtocolBuilder ConfigureProtocol(ProtocolBuilder protocolBuilder) { @@ -163,7 +164,7 @@ private async ValueTask UpdatePlanAndDelegateAsync(MagenticTaskContext taskConte { bool isReplan = taskContext.TaskLedger != null; - taskContext.TaskLedger = await this._manager.UpdatePlanAsync(taskContext, context, cancellationToken) + taskContext.TaskLedger = await this._manager.UpdatePlanAsync(taskContext, context, this._currentTurnRunOptions, cancellationToken) .ConfigureAwait(false); this._fullTaskLedgerMessage = new(ChatRole.User, taskContext.ToTaskLedgerFullPrompt()); @@ -184,7 +185,12 @@ await context.AddEventAsync(isReplan } protected override async ValueTask TakeTurnAsync(List messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellationToken = default) + => await this.TakeTurnAsync(messages, context, new TurnToken(emitEvents), cancellationToken).ConfigureAwait(false); + + protected override async ValueTask TakeTurnAsync(List messages, IWorkflowContext context, TurnToken turnToken, CancellationToken cancellationToken = default) { + this._currentTurnRunOptions = context.GetAgentRunOptions(turnToken.RunOptions); + if (this._taskContext?.IsTerminated == true) { throw new InvalidOperationException("This Magentic orchestration has already terminated. To process new messages, create a new workflow instance."); @@ -193,7 +199,7 @@ protected override async ValueTask TakeTurnAsync(List messages, IWo if (this._taskContext == null) { // First Turn: Initialize the task context and create the initial plan - this._taskContext = new(messages, team, limits, emitEvents, []) { ResponseLanguage = responseLanguage, PromptOverrides = promptOverrides }; + this._taskContext = new(messages, team, limits, turnToken.EmitEvents, []) { ResponseLanguage = responseLanguage, PromptOverrides = promptOverrides }; await this.UpdatePlanAndDelegateAsync(this._taskContext, context, cancellationToken).ConfigureAwait(false); } else @@ -265,7 +271,7 @@ private async ValueTask RunCoordinationRoundAsync(MagenticTaskContext taskContex // Update the Progress Ledger try { - await this._manager.UpdateProgressLedgerAsync(taskContext, context, cancellationToken).ConfigureAwait(false); + await this._manager.UpdateProgressLedgerAsync(taskContext, context, this._currentTurnRunOptions, cancellationToken).ConfigureAwait(false); await context.AddEventAsync(new MagenticProgressLedgerUpdatedEvent(taskContext.ProgressLedger), cancellationToken) .ConfigureAwait(false); @@ -333,7 +339,7 @@ await context.AddEventAsync(new WorkflowWarningEvent($"Invalid next speaker: {ne } this._currentSpeakerExecutorId = nextExecutorId; - await context.SendMessageAsync(new TurnToken(taskContext.EmitUpdateEvents), nextExecutorId, cancellationToken).ConfigureAwait(false); + await context.SendMessageAsync(new TurnToken(taskContext.EmitUpdateEvents, this._currentTurnRunOptions), nextExecutorId, cancellationToken).ConfigureAwait(false); } private async ValueTask ResetAndReplanAsync(MagenticTaskContext taskContext, IWorkflowContext context, CancellationToken cancellationToken) @@ -348,7 +354,7 @@ private async ValueTask ResetAndReplanAsync(MagenticTaskContext taskContext, IWo private async ValueTask PrepareFinalAnswerAsync(MagenticTaskContext taskContext, IWorkflowContext context, CancellationToken cancellationToken) { - List messages = [await this._manager.PrepareFinalAnswerAsync(taskContext, context, cancellationToken).ConfigureAwait(false)]; + List messages = [await this._manager.PrepareFinalAnswerAsync(taskContext, context, this._currentTurnRunOptions, cancellationToken).ConfigureAwait(false)]; await context.YieldOutputAsync(messages, cancellationToken).ConfigureAwait(false); taskContext.IsTerminated = true; this._currentSpeakerExecutorId = null; diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/WorkflowHostExecutor.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/WorkflowHostExecutor.cs index 58e3a9e5239..913f6f479c5 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/WorkflowHostExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/WorkflowHostExecutor.cs @@ -91,11 +91,16 @@ internal async ValueTask EnsureRunnerAsync() this._checkpointManager ??= new InMemoryCheckpointManager(); } + AgentRunOptions? agentRunOptions = null; + bool hasAgentRunOptions = this.JoinContext is IWorkflowAgentRunOptionsContext runOptionsContext + && runOptionsContext.TryGetAgentRunOptions(out agentRunOptions); this._activeRunner = InProcessRunner.CreateSubworkflowRunner(this._workflow, this._checkpointManager, this._sessionId, this._ownershipToken, - this.JoinContext.ConcurrentRunsEnabled); + this.JoinContext.ConcurrentRunsEnabled, + hasAgentRunOptions: hasAgentRunOptions, + agentRunOptions: agentRunOptions); } return this._activeRunner; diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/TurnToken.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/TurnToken.cs index 91a0833cc08..2be8321dc04 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/TurnToken.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/TurnToken.cs @@ -1,5 +1,6 @@ // Copyright (c) Microsoft. All rights reserved. +using System.Text.Json.Serialization; using Microsoft.Extensions.AI; namespace Microsoft.Agents.AI.Workflows; @@ -8,12 +9,41 @@ namespace Microsoft.Agents.AI.Workflows; /// Sent to an -based executor to request /// a response to accumulated . /// -/// Whether to raise AgentRunEvents for this executor. -public class TurnToken(bool? emitEvents = null) +public class TurnToken { + /// + /// Initializes a new instance of the class. + /// + /// Whether to raise agent run events for the receiving executor. + [JsonConstructor] + public TurnToken(bool? emitEvents = null) + { + this.EmitEvents = emitEvents; + } + + /// + /// Initializes a new instance of the class. + /// + /// Whether to raise agent run events for the receiving executor. + /// Options to pass to agents invoked during this turn. + public TurnToken(bool? emitEvents, AgentRunOptions? runOptions) + : this(emitEvents) + { + this.RunOptions = runOptions; + } + /// /// Gets a value indicating whether events are emitted by the receiving executor. If the /// value is not set, defaults to the configuration in the executor. /// - public bool? EmitEvents => emitEvents; + public bool? EmitEvents { get; } + + /// + /// Gets the options to pass to agents invoked during this turn. + /// + /// + /// Run options apply only to the current invocation and are not persisted in workflow checkpoints. + /// + [JsonIgnore] + public AgentRunOptions? RunOptions { get; } } diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs index d498d13b651..f62972f073c 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowHostAgent.cs @@ -160,7 +160,7 @@ Task RunCoreAsync( WorkflowSession workflowSession = await this.UpdateSessionAsync(messages, session, cancellationToken).ConfigureAwait(false); ResponseMergeState mergeState = new(); - await foreach (AgentResponseUpdate update in workflowSession.InvokeStageAsync(cancellationToken) + await foreach (AgentResponseUpdate update in workflowSession.InvokeStageAsync(options, cancellationToken) .ConfigureAwait(false) .WithCancellation(cancellationToken)) { @@ -186,7 +186,7 @@ IAsyncEnumerable RunCoreStreamingAsync( WorkflowSession workflowSession = await this.UpdateSessionAsync(messages, session, cancellationToken).ConfigureAwait(false); ResponseMergeState mergeState = new(); - await foreach (AgentResponseUpdate update in workflowSession.InvokeStageAsync(cancellationToken) + await foreach (AgentResponseUpdate update in workflowSession.InvokeStageAsync(options, cancellationToken) .ConfigureAwait(false) .WithCancellation(cancellationToken)) { diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs index 97134a94073..1e285efee9f 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/WorkflowSession.cs @@ -200,7 +200,7 @@ public AgentResponseUpdate CreateUpdate(string responseId, object raw, ChatMessa }; } - private async ValueTask CreateOrResumeRunAsync(List messages, CancellationToken cancellationToken = default) + private async ValueTask CreateOrResumeRunAsync(List messages, AgentRunOptions? runOptions, CancellationToken cancellationToken = default) { // The workflow is validated to be a ChatProtocol workflow by the WorkflowHostAgent before creating the session, // and does not need to be checked again here. @@ -212,9 +212,10 @@ private async ValueTask CreateOrResumeRunAsync(List internal async IAsyncEnumerable InvokeStageAsync( + AgentRunOptions? runOptions = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) { this.LastResponseId = Guid.NewGuid().ToString("N"); List messages = this.ChatHistoryProvider.GetFromBookmark(this).ToList(); ResumeRunResult resumeResult = - await this.CreateOrResumeRunAsync(messages, cancellationToken).ConfigureAwait(false); + await this.CreateOrResumeRunAsync(messages, runOptions, cancellationToken).ConfigureAwait(false); bool resumeWithoutNewTurn = this._resumeWithoutNewTurn; this._resumeWithoutNewTurn = false; @@ -496,7 +499,7 @@ IAsyncEnumerable InvokeStageAsync( || !dispatchInfo.HasMatchedResponseForStartExecutor); if (shouldSendTurnToken) { - await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false); + await run.TrySendMessageAsync(new TurnToken(emitEvents: true, runOptions)).ConfigureAwait(false); } AgentResponseUpdate CreateObservabilityUpdate(WorkflowEvent evt) diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ChatProtocolExecutorTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ChatProtocolExecutorTests.cs index 0f2ab2e7342..8c2ef59e15e 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ChatProtocolExecutorTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/ChatProtocolExecutorTests.cs @@ -3,10 +3,12 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Text.Json; using System.Threading; using System.Threading.Tasks; using FluentAssertions; using Microsoft.Agents.AI.Workflows.Checkpointing; +using Microsoft.Agents.AI.Workflows.InProc; using Microsoft.Extensions.AI; namespace Microsoft.Agents.AI.Workflows.UnitTests; @@ -20,6 +22,7 @@ private sealed class TestChatProtocolExecutor : ChatProtocolExecutor { public List ReceivedMessages { get; } = []; public int TurnCount { get; private set; } + public TurnToken? ReceivedTurnToken { get; private set; } public TestChatProtocolExecutor(string id = "test-executor", ChatProtocolExecutorOptions? options = null) : base(id, options) @@ -38,6 +41,16 @@ protected override async ValueTask TakeTurnAsync( // Send messages back to context so they can be collected await context.SendMessageAsync(messages, cancellationToken: cancellationToken); } + + protected override ValueTask TakeTurnAsync( + List messages, + IWorkflowContext context, + TurnToken turnToken, + CancellationToken cancellationToken = default) + { + this.ReceivedTurnToken = turnToken; + return base.TakeTurnAsync(messages, context, turnToken, cancellationToken); + } } [Fact] @@ -75,6 +88,92 @@ public async Task ChatProtocolExecutor_Handles_ListOfChatMessagesAsync() executor.TurnCount.Should().Be(1); } + [Fact] + public async Task ChatProtocolExecutor_ReceivesAndForwardsFullTurnTokenAsync() + { + // Arrange + TestChatProtocolExecutor executor = new(); + TestWorkflowContext context = new(executor.Id); + AgentRunOptions runOptions = new() { AdditionalProperties = new() { ["test-property"] = "test-value" } }; + TurnToken turnToken = new(emitEvents: false, runOptions); + + // Act + await executor.TakeTurnAsync(turnToken, context); + + // Assert + executor.ReceivedTurnToken.Should().BeSameAs(turnToken); + executor.ReceivedTurnToken!.RunOptions.Should().BeSameAs(runOptions); + context.SentMessages.OfType().Should().ContainSingle().Which.Should().BeSameAs(turnToken); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ChatProtocolExecutor_CheckpointRecoveryUsesCurrentRunOptionsAsync(bool serializeSession) + { + // Arrange + TestChatProtocolExecutor firstExecutor = new("first-executor"); + TestChatProtocolExecutor secondExecutor = new("second-executor"); + ExecutorBinding firstBinding = firstExecutor.BindExecutor(); + ExecutorBinding secondBinding = secondExecutor.BindExecutor(); + Workflow workflow = new WorkflowBuilder(firstBinding) + .AddEdge>(firstBinding, secondBinding, messages => messages is not null) + .AddEdge(firstBinding, secondBinding, token => token is not null) + .WithOutputFrom(secondBinding) + .Build(); + InProcessExecutionEnvironment environment = + InProcessExecution.Lockstep.WithCheckpointing(CheckpointManager.CreateInMemory()); + AIAgent workflowAgent = workflow.AsAIAgent(executionEnvironment: environment); + AgentSession session = await workflowAgent.CreateSessionAsync(); + AgentRunOptions firstRunOptions = new() { AdditionalProperties = new() { ["invocation"] = "first" } }; + AgentRunOptions recoveryRunOptions = new() { AdditionalProperties = new() { ["invocation"] = "recovery" } }; + + CheckpointInfo checkpoint = await OrchestrationTestHelpers.RunWorkflowAgentUntilCheckpointAsync( + workflowAgent, + session, + firstRunOptions, + checkpointNumber: 1); + if (serializeSession) + { + JsonElement serializedSession = await workflowAgent.SerializeSessionAsync(session); + session = await workflowAgent.DeserializeSessionAsync(serializedSession); + } + + WorkflowSessionCheckpointRecovery recovery = session.GetService() + ?? throw new InvalidOperationException("Workflow checkpoint recovery was not available."); + recovery.TryPrepare(checkpoint.CheckpointId).Should().BeTrue(); + + // Act + _ = await workflowAgent.RunStreamingAsync([], session, recoveryRunOptions).ToListAsync(); + + // Assert + firstExecutor.ReceivedTurnToken.Should().NotBeNull(); + firstExecutor.ReceivedTurnToken!.RunOptions.Should().BeSameAs(firstRunOptions); + secondExecutor.ReceivedTurnToken.Should().NotBeNull(); + secondExecutor.ReceivedTurnToken!.RunOptions.Should().BeSameAs(recoveryRunOptions); + } + + [Fact] + public void TurnToken_RunOptionsAreNotSerialized() + { + // Arrange + ChatClientAgentRunOptions runOptions = new() + { + ChatClientFactory = static chatClient => chatClient, + }; + TurnToken turnToken = new(emitEvents: false, runOptions); + + // Act + string json = JsonSerializer.Serialize(turnToken); + TurnToken? deserialized = JsonSerializer.Deserialize(json); + + // Assert + json.Should().NotContain(nameof(TurnToken.RunOptions)); + deserialized.Should().NotBeNull(); + deserialized!.EmitEvents.Should().BeFalse(); + deserialized.RunOptions.Should().BeNull(); + } + [Fact] public async Task ChatProtocolExecutor_Handles_ArrayOfChatMessagesAsync() { diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/GroupChatOrchestrationTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/GroupChatOrchestrationTests.cs index dbbcc1e44d3..7b31110186d 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/GroupChatOrchestrationTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/GroupChatOrchestrationTests.cs @@ -25,6 +25,64 @@ namespace Microsoft.Agents.AI.Workflows.UnitTests; /// public class GroupChatOrchestrationTests { + [Fact] + public async Task GroupChat_AsAgent_PropagatesRunOptionsToEveryParticipantAsync() + { + // Arrange + RecordingEchoAgent firstAgent = new("first-agent", "FirstAgent"); + RecordingEchoAgent secondAgent = new("second-agent", "SecondAgent"); + AIAgent workflowAgent = AgentWorkflowBuilder + .CreateGroupChatBuilderWith(agents => new RoundRobinGroupChatManager(agents) { MaximumIterationCount = 2 }) + .AddParticipants(firstAgent, secondAgent) + .Build() + .AsAIAgent(); + AgentRunOptions runOptions = new() { AdditionalProperties = new() { ["test-property"] = "test-value" } }; + + // Act + _ = await workflowAgent.RunAsync("Hello", options: runOptions); + + // Assert + firstAgent.RecordedRunOptions.Should().NotBeEmpty().And.OnlyContain(options => ReferenceEquals(options, runOptions)); + secondAgent.RecordedRunOptions.Should().NotBeEmpty().And.OnlyContain(options => ReferenceEquals(options, runOptions)); + } + + [Fact] + public async Task GroupChat_AsAgent_CheckpointRecoveryUsesCurrentRunOptionsAsync() + { + // Arrange + RecordingEchoAgent firstAgent = new("first-agent", "FirstAgent"); + RecordingEchoAgent secondAgent = new("second-agent", "SecondAgent"); + Workflow workflow = AgentWorkflowBuilder + .CreateGroupChatBuilderWith(agents => new RoundRobinGroupChatManager(agents) { MaximumIterationCount = 2 }) + .AddParticipants(firstAgent, secondAgent) + .Build(); + InProcessExecutionEnvironment environment = + InProcessExecution.Lockstep.WithCheckpointing(CheckpointManager.CreateInMemory()); + AIAgent workflowAgent = workflow.AsAIAgent(executionEnvironment: environment); + AgentSession session = await workflowAgent.CreateSessionAsync(); + AgentRunOptions firstRunOptions = new() { AdditionalProperties = new() { ["invocation"] = "first" } }; + AgentRunOptions recoveryRunOptions = new() { AdditionalProperties = new() { ["invocation"] = "recovery" } }; + + CheckpointInfo checkpoint = await OrchestrationTestHelpers.RunWorkflowAgentUntilCheckpointAsync( + workflowAgent, + session, + firstRunOptions, + checkpointNumber: 2); + WorkflowSessionCheckpointRecovery recovery = session.GetService() + ?? throw new InvalidOperationException("Workflow checkpoint recovery was not available."); + recovery.TryPrepare(checkpoint.CheckpointId).Should().BeTrue(); + + // Act + List recoveryUpdates = await workflowAgent + .RunStreamingAsync([], session, recoveryRunOptions) + .ToListAsync(); + + // Assert + recoveryUpdates.SelectMany(update => update.Contents.OfType()).Should().BeEmpty(); + firstAgent.RecordedRunOptions.Should().ContainSingle().Which.Should().BeSameAs(firstRunOptions); + secondAgent.RecordedRunOptions.Should().ContainSingle().Which.Should().BeSameAs(recoveryRunOptions); + } + /// /// End-to-end tool-approval checkpoint/resume scenario through a /// with a single participant. Mirrors the maximal repro added in PR #5952 (Track A2 in diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/HandoffOrchestrationTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/HandoffOrchestrationTests.cs index 21a5e46b98e..af1ad80ccea 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/HandoffOrchestrationTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/HandoffOrchestrationTests.cs @@ -11,6 +11,7 @@ using System.Threading.Tasks; using FluentAssertions; using Microsoft.Agents.AI.Workflows.InProc; +using Microsoft.Agents.AI.Workflows.Sample; using Microsoft.Agents.AI.Workflows.Specialized; using Microsoft.Agents.AI.Workflows.Specialized.Magentic; using Microsoft.Extensions.AI; @@ -357,6 +358,49 @@ public async Task Handoffs_MultipleTransfers_AsAgentPreservesCallResultOrderAsyn GetMessageSequence(streamingWorkflowSession.ChatHistoryProvider.GetAllMessages(streamingWorkflowSession).Skip(1)).Should().Equal(expected); } + [Fact] + public async Task Handoffs_AsAgent_PropagatesRunOptionsAndPreservesHandoffToolsAsync() + { + // Arrange + HandoffTestEchoAgent firstAgent = new("first-agent", "FirstAgent"); + HandoffTestEchoAgent secondAgent = new("second-agent", "SecondAgent"); + AITool callerTool = AIFunctionFactory.CreateDeclaration( + "CallerTool", + description: null, + AIFunctionFactory.Create(() => { }).JsonSchema); + Func chatClientFactory = static chatClient => chatClient; + AIAgent workflowAgent = AgentWorkflowBuilder.CreateHandoffBuilderWith(firstAgent) + .WithHandoff(firstAgent, secondAgent) + .Build() + .AsAIAgent(); + ChatClientAgentRunOptions runOptions = new(new ChatOptions + { + ModelId = "test-model", + Instructions = "Caller instructions", + Tools = [callerTool], + }) + { + AdditionalProperties = new() { ["test-property"] = "test-value" }, + ChatClientFactory = chatClientFactory, + }; + + // Act + _ = await workflowAgent.RunAsync("Hello", options: runOptions); + + // Assert + ChatClientAgentRunOptions firstAgentOptions = firstAgent.RecordedRunOptions.Should().ContainSingle() + .Which.Should().BeOfType().Subject; + firstAgentOptions.Should().NotBeSameAs(runOptions); + firstAgentOptions.AdditionalProperties.Should().ContainKey("test-property").WhoseValue.Should().Be("test-value"); + firstAgentOptions.ChatClientFactory.Should().BeSameAs(chatClientFactory); + firstAgentOptions.ChatOptions.Should().NotBeNull(); + firstAgentOptions.ChatOptions!.ModelId.Should().Be("test-model"); + firstAgentOptions.ChatOptions.Instructions.Should().EndWith("Caller instructions"); + firstAgentOptions.ChatOptions.Tools.Should().Contain(callerTool); + firstAgentOptions.ChatOptions.Tools.Should().Contain(tool => tool.Name.StartsWith(HandoffWorkflowBuilder.FunctionPrefix, StringComparison.Ordinal)); + secondAgent.RecordedRunOptions.Should().ContainSingle().Which.Should().BeSameAs(runOptions); + } + [Fact] public async Task Handoffs_ReturnToInitialAgent_AsAgentKeepsInvocationsSeparateAsync() { diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/MagenticOrchestrationTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/MagenticOrchestrationTests.cs index 7c5260f507e..262cafd3331 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/MagenticOrchestrationTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/MagenticOrchestrationTests.cs @@ -18,6 +18,103 @@ namespace Microsoft.Agents.AI.Workflows.UnitTests; /// public class MagenticOrchestrationTests { + [Fact] + public async Task Magentic_AsAgent_PropagatesRunOptionsToManagerAndParticipantsAsync() + { + // Arrange + RecordingReplayAgent manager = new( + [ + CreatePlanResponse("Facts about the task"), + CreatePlanResponse("Ask Worker to complete the task"), + CreateProgressLedgerResponse( + isRequestSatisfied: false, + isInLoop: false, + isProgressBeingMade: true, + nextSpeaker: "Worker", + instructionOrQuestion: "Complete the task"), + CreateProgressLedgerResponse( + isRequestSatisfied: true, + isInLoop: false, + isProgressBeingMade: true, + nextSpeaker: "Worker", + instructionOrQuestion: "The task is complete"), + CreateFinalAnswerResponse("Task completed successfully!"), + ], + name: "Manager"); + RecordingEchoAgent worker = new(name: "Worker"); + AIAgent workflowAgent = new MagenticWorkflowBuilder(manager) + .AddParticipants(worker) + .RequirePlanSignoff(false) + .Build() + .AsAIAgent(); + AgentRunOptions runOptions = new() { AdditionalProperties = new() { ["test-property"] = "test-value" } }; + + // Act + _ = await workflowAgent.RunAsync("Do the task", options: runOptions); + + // Assert + manager.RecordedRunOptions.Should().NotBeEmpty().And.OnlyContain(options => ReferenceEquals(options, runOptions)); + worker.RecordedRunOptions.Should().NotBeEmpty().And.OnlyContain(options => ReferenceEquals(options, runOptions)); + } + + [Fact] + public async Task Magentic_AsAgent_CheckpointRecoveryUsesCurrentRunOptionsAsync() + { + // Arrange + RecordingReplayAgent manager = new( + [ + CreatePlanResponse("Facts about the task"), + CreatePlanResponse("Ask Worker to complete the task"), + CreateProgressLedgerResponse( + isRequestSatisfied: false, + isInLoop: false, + isProgressBeingMade: true, + nextSpeaker: "Worker", + instructionOrQuestion: "Complete the task"), + CreateProgressLedgerResponse( + isRequestSatisfied: true, + isInLoop: false, + isProgressBeingMade: true, + nextSpeaker: "Worker", + instructionOrQuestion: "The task is complete"), + CreateFinalAnswerResponse("Task completed successfully!"), + ], + name: "Manager"); + RecordingEchoAgent worker = new(name: "Worker"); + Workflow workflow = new MagenticWorkflowBuilder(manager) + .AddParticipants(worker) + .RequirePlanSignoff(false) + .Build(); + InProcessExecutionEnvironment environment = + InProcessExecution.Lockstep.WithCheckpointing(CheckpointManager.CreateInMemory()); + AIAgent workflowAgent = workflow.AsAIAgent(executionEnvironment: environment); + AgentSession session = await workflowAgent.CreateSessionAsync(); + AgentRunOptions firstRunOptions = new() { AdditionalProperties = new() { ["invocation"] = "first" } }; + AgentRunOptions recoveryRunOptions = new() { AdditionalProperties = new() { ["invocation"] = "recovery" } }; + + CheckpointInfo checkpoint = await OrchestrationTestHelpers.RunWorkflowAgentUntilCheckpointAsync( + workflowAgent, + session, + firstRunOptions, + checkpointNumber: 2); + int managerCallsBeforeRecovery = manager.RecordedRunOptions.Count; + WorkflowSessionCheckpointRecovery recovery = session.GetService() + ?? throw new InvalidOperationException("Workflow checkpoint recovery was not available."); + recovery.TryPrepare(checkpoint.CheckpointId).Should().BeTrue(); + + // Act + List recoveryUpdates = await workflowAgent + .RunStreamingAsync([], session, recoveryRunOptions) + .ToListAsync(); + + // Assert + recoveryUpdates.SelectMany(update => update.Contents.OfType()).Should().BeEmpty(); + manager.RecordedRunOptions.Take(managerCallsBeforeRecovery) + .Should().OnlyContain(options => ReferenceEquals(options, firstRunOptions)); + manager.RecordedRunOptions.Skip(managerCallsBeforeRecovery) + .Should().NotBeEmpty().And.OnlyContain(options => ReferenceEquals(options, recoveryRunOptions)); + } + [Fact] public async Task Task_Completes_When_RequestSatisfiedAsync() { diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/OrchestrationTestHelpers.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/OrchestrationTestHelpers.cs index a8838380293..7789651cba6 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/OrchestrationTestHelpers.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/OrchestrationTestHelpers.cs @@ -75,6 +75,31 @@ protected override async IAsyncEnumerable RunCoreStreamingA internal sealed record WorkflowRunResult(string UpdateText, List? Result, CheckpointInfo? LastCheckpoint, List PendingRequests); + internal static async Task RunWorkflowAgentUntilCheckpointAsync( + AIAgent workflowAgent, + AgentSession session, + AgentRunOptions runOptions, + int checkpointNumber) + { + int observedCheckpoints = 0; + await foreach (AgentResponseUpdate update in workflowAgent.RunStreamingAsync( + new ChatMessage(ChatRole.User, "Start"), + session, + runOptions)) + { + if (update.RawRepresentation is SuperStepCompletedEvent + { + CompletionInfo.Checkpoint: { } checkpoint + } + && ++observedCheckpoints == checkpointNumber) + { + return checkpoint; + } + } + + throw new InvalidOperationException($"The workflow completed before checkpoint {checkpointNumber} was observed."); + } + internal static async Task RunWorkflowCheckpointedAsync( Workflow workflow, List input, InProcessExecutionEnvironment environment, CheckpointInfo? fromCheckpoint = null) { diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RecordingEchoAgent.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RecordingEchoAgent.cs index 83a0e994a14..9a8af1aedaa 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RecordingEchoAgent.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RecordingEchoAgent.cs @@ -18,6 +18,7 @@ internal sealed class RecordingEchoAgent(string? id = null, string? name = null, : TestEchoAgent(id, name, prefix) { public List> RecordedInputs { get; } = []; + public List RecordedRunOptions { get; } = []; protected override async IAsyncEnumerable RunCoreStreamingAsync( IEnumerable messages, @@ -28,6 +29,7 @@ protected override async IAsyncEnumerable RunCoreStreamingA // Materialize once so the deferred input is recorded and replayed identically. List recorded = messages.ToList(); this.RecordedInputs.Add(recorded); + this.RecordedRunOptions.Add(options); await foreach (AgentResponseUpdate update in base.RunCoreStreamingAsync(recorded, session, options, cancellationToken)) { diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RecordingReplayAgent.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RecordingReplayAgent.cs index ff4386a461c..27295ce26c4 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RecordingReplayAgent.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/RecordingReplayAgent.cs @@ -16,6 +16,7 @@ internal sealed class RecordingReplayAgent(List> messages, str : TestReplayAgent(messages, id, name) { public List> RecordedInputs { get; } = []; + public List RecordedRunOptions { get; } = []; protected override async IAsyncEnumerable RunCoreStreamingAsync( IEnumerable messages, @@ -24,6 +25,7 @@ protected override async IAsyncEnumerable RunCoreStreamingA [EnumeratorCancellation] CancellationToken cancellationToken = default) { this.RecordedInputs.Add(messages.ToList()); + this.RecordedRunOptions.Add(options); await foreach (AgentResponseUpdate update in base.RunCoreStreamingAsync(messages, session, options, cancellationToken)) { yield return update; diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/12_HandOff_HostAsAgent.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/12_HandOff_HostAsAgent.cs index dc1072aa727..16b3d736765 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/12_HandOff_HostAsAgent.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/Sample/12_HandOff_HostAsAgent.cs @@ -13,8 +13,12 @@ namespace Microsoft.Agents.AI.Workflows.Sample; internal sealed class HandoffTestEchoAgent(string id, string name, string prefix = "") : TestEchoAgent(id, name, prefix) { + public List RecordedRunOptions { get; } = []; + protected override IEnumerable GetEpilogueMessages(AgentRunOptions? options = null) { + this.RecordedRunOptions.Add(options); + if (options is ChatClientAgentRunOptions chatClientOptions && chatClientOptions.ChatOptions != null) { diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs index 92ebc5915d8..6a80b673b85 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/WorkflowHostSmokeTests.cs @@ -8,6 +8,8 @@ using System.Threading; using System.Threading.Tasks; using FluentAssertions; +using Microsoft.Agents.AI.Workflows.InProc; +using Microsoft.Agents.AI.Workflows.Sample; using Microsoft.Extensions.AI; namespace Microsoft.Agents.AI.Workflows.UnitTests; @@ -37,6 +39,8 @@ internal sealed class RequestEmittingAgent : AIAgent private readonly AIContent _requestContent; private readonly bool _completeOnResponse; + public List RecordedRunOptions { get; } = []; + /// /// Creates a new that emits the given request content. /// @@ -72,6 +76,8 @@ protected override Task RunCoreAsync(IEnumerable mes protected override async IAsyncEnumerable RunCoreStreamingAsync(IEnumerable messages, AgentSession? session = null, AgentRunOptions? options = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) { + this.RecordedRunOptions.Add(options); + if (this._completeOnResponse && messages.Any(m => m.Contents.Any(c => c is FunctionResultContent || c is ToolApprovalResponseContent))) { @@ -230,6 +236,31 @@ public override ValueTask HandleAsync( public class WorkflowHostSmokeTests : AIAgentHostingExecutorTestsBase { + private sealed class RunOptionsRecordingAgent(AIAgent innerAgent) : DelegatingAIAgent(innerAgent) + { + public List RecordedRunOptions { get; } = []; + + protected override Task RunCoreAsync( + IEnumerable messages, + AgentSession? session = null, + AgentRunOptions? options = null, + CancellationToken cancellationToken = default) + { + this.RecordedRunOptions.Add(options); + return base.RunCoreAsync(messages, session, options, cancellationToken); + } + + protected override IAsyncEnumerable RunCoreStreamingAsync( + IEnumerable messages, + AgentSession? session = null, + AgentRunOptions? options = null, + CancellationToken cancellationToken = default) + { + this.RecordedRunOptions.Add(options); + return base.RunCoreStreamingAsync(messages, session, options, cancellationToken); + } + } + private sealed class AlwaysFailsAIAgent(bool failByThrowing) : AIAgent { private sealed class Session : AgentSession @@ -277,6 +308,100 @@ private static Workflow CreateWorkflow(bool failByThrowing) return new WorkflowBuilder(agent).Build(); } + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Test_AsAgent_AgentRunOptionsFlowToEverySequentialAgentAsync(bool streaming) + { + // Arrange + RunOptionsRecordingAgent firstAgent = new(new TestEchoAgent("first-agent", "FirstAgent")); + RunOptionsRecordingAgent secondAgent = new(new TestEchoAgent("second-agent", "SecondAgent")); + AIAgent workflowAgent = AgentWorkflowBuilder.BuildSequential(firstAgent, secondAgent).AsAIAgent(); + ChatClientAgentRunOptions runOptions = new(new ChatOptions { ModelId = "test-model" }) + { + AdditionalProperties = new() { ["test-property"] = "test-value" }, + }; + + // Act + if (streaming) + { + _ = await workflowAgent.RunStreamingAsync("Hello", options: runOptions).ToListAsync(); + } + else + { + _ = await workflowAgent.RunAsync("Hello", options: runOptions); + } + + // Assert + firstAgent.RecordedRunOptions.Should().ContainSingle().Which.Should().BeSameAs(runOptions); + secondAgent.RecordedRunOptions.Should().ContainSingle().Which.Should().BeSameAs(runOptions); + } + + [Fact] + public async Task Test_AsAgent_AgentRunOptionsFlowToEveryConcurrentAgentAsync() + { + // Arrange + RunOptionsRecordingAgent firstAgent = new(new TestEchoAgent("first-agent", "FirstAgent")); + RunOptionsRecordingAgent secondAgent = new(new TestEchoAgent("second-agent", "SecondAgent")); + AIAgent workflowAgent = AgentWorkflowBuilder.BuildConcurrent([firstAgent, secondAgent]).AsAIAgent(); + AgentRunOptions runOptions = new() { AdditionalProperties = new() { ["test-property"] = "test-value" } }; + + // Act + _ = await workflowAgent.RunAsync("Hello", options: runOptions); + + // Assert + firstAgent.RecordedRunOptions.Should().ContainSingle().Which.Should().BeSameAs(runOptions); + secondAgent.RecordedRunOptions.Should().ContainSingle().Which.Should().BeSameAs(runOptions); + } + + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task Test_AsAgent_CheckpointRecoveryUsesCurrentAgentRunOptionsAsync( + bool useSubworkflow, + bool specifyRecoveryOptions) + { + // Arrange + RunOptionsRecordingAgent firstAgent = new(new TestEchoAgent("first-agent", "FirstAgent")); + RunOptionsRecordingAgent secondAgent = new(new TestEchoAgent("second-agent", "SecondAgent")); + Workflow workflow = AgentWorkflowBuilder.BuildSequential(firstAgent, secondAgent); + if (useSubworkflow) + { + ExecutorBinding subworkflow = workflow.BindAsExecutor("NestedWorkflow"); + workflow = new WorkflowBuilder(subworkflow).Build(); + } + + InProcessExecutionEnvironment environment = + InProcessExecution.Lockstep.WithCheckpointing(CheckpointManager.CreateInMemory()); + AIAgent workflowAgent = workflow.AsAIAgent(executionEnvironment: environment); + AgentSession session = await workflowAgent.CreateSessionAsync(); + AgentRunOptions firstRunOptions = new() { AdditionalProperties = new() { ["invocation"] = "first" } }; + AgentRunOptions? recoveryRunOptions = specifyRecoveryOptions + ? new() { AdditionalProperties = new() { ["invocation"] = "recovery" } } + : null; + + CheckpointInfo checkpoint = await OrchestrationTestHelpers.RunWorkflowAgentUntilCheckpointAsync( + workflowAgent, + session, + firstRunOptions, + checkpointNumber: 1); + WorkflowSessionCheckpointRecovery recovery = session.GetService() + ?? throw new InvalidOperationException("Workflow checkpoint recovery was not available."); + recovery.TryPrepare(checkpoint.CheckpointId).Should().BeTrue(); + + // Act + List recoveryUpdates = await workflowAgent + .RunStreamingAsync([], session, recoveryRunOptions) + .ToListAsync(); + + // Assert + recoveryUpdates.SelectMany(update => update.Contents.OfType()).Should().BeEmpty(); + firstAgent.RecordedRunOptions.Should().ContainSingle().Which.Should().BeSameAs(firstRunOptions); + secondAgent.RecordedRunOptions.Should().ContainSingle().Which.Should().BeSameAs(recoveryRunOptions); + } + [Theory] [InlineData(true, true)] [InlineData(true, false)] @@ -435,6 +560,53 @@ public async Task Test_AsAgent_FunctionCallRoundtrip_ResponseIsProcessedAsync() .NotContain(c => c.CallId == receivedRequest.CallId, "the external FunctionCallContent request should be cleared after processing the response"); } + [Theory] + [InlineData(false, false)] + [InlineData(false, true)] + [InlineData(true, false)] + [InlineData(true, true)] + public async Task Test_AsAgent_AgentRunOptionsFlowToExternalResponseContinuationAsync(bool useSubworkflow, bool specifyContinuationOptions) + { + // Arrange + const string CallId = "run-options-continuation-call-id"; + RequestEmittingAgent requestAgent = new(new FunctionCallContent(CallId, "testFunction"), completeOnResponse: true); + Workflow requestWorkflow = new WorkflowBuilder(requestAgent.BindAsExecutor( + new AIAgentHostOptions { InterceptUnterminatedFunctionCalls = false, EmitAgentUpdateEvents = true })).Build(); + Workflow workflow = requestWorkflow; + if (useSubworkflow) + { + ExecutorBinding subworkflow = requestWorkflow.BindAsExecutor("NestedWorkflow"); + workflow = new WorkflowBuilder(subworkflow) + .AddExternalRequest(subworkflow, "NestedFunctionCall") + .Build(); + } + AIAgent agent = workflow.AsAIAgent("WorkflowAgent"); + AgentSession session = await agent.CreateSessionAsync(); + AgentRunOptions firstRunOptions = new() { AdditionalProperties = new() { ["invocation"] = "first" } }; + AgentRunOptions? secondRunOptions = specifyContinuationOptions + ? new() { AdditionalProperties = new() { ["invocation"] = "second" } } + : null; + + List firstCallUpdates = await agent.RunStreamingAsync( + new ChatMessage(ChatRole.User, "Start"), session, firstRunOptions).ToListAsync(); + firstCallUpdates.SelectMany(update => update.Contents.OfType()).Should().BeEmpty(); + FunctionCallContent request = firstCallUpdates + .Where(update => update.RawRepresentation is RequestInfoEvent) + .SelectMany(update => update.Contents.OfType()) + .Should().ContainSingle().Subject; + + // Act + _ = await agent.RunStreamingAsync( + new ChatMessage(ChatRole.Tool, [new FunctionResultContent(request.CallId, "tool output")]), + session, + secondRunOptions).ToListAsync(); + + // Assert + requestAgent.RecordedRunOptions.Should().HaveCountGreaterThanOrEqualTo(2); + requestAgent.RecordedRunOptions[0].Should().BeSameAs(firstRunOptions); + requestAgent.RecordedRunOptions.Skip(1).Should().OnlyContain(options => ReferenceEquals(options, secondRunOptions)); + } + /// /// Tests the full roundtrip for ToolApprovalRequestContent: workflow emits request, external caller responds. /// Verifying inbound ToolApprovalResponseContent conversion. From 3b94fc099b833e4249f28426bcd9ec8226bac477 Mon Sep 17 00:00:00 2001 From: Saibernard Yogendran Date: Thu, 27 Aug 2026 20:21:53 -0400 Subject: [PATCH 2/2] .NET: Preserve run options across Magentic plan review --- .../Magentic/MagenticOrchestrator.cs | 2 + .../MagenticOrchestrationTests.cs | 102 ++++++++++++++++++ 2 files changed, 104 insertions(+) diff --git a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/Magentic/MagenticOrchestrator.cs b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/Magentic/MagenticOrchestrator.cs index 0feede800f8..00922e21cc0 100644 --- a/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/Magentic/MagenticOrchestrator.cs +++ b/dotnet/src/Microsoft.Agents.AI.Workflows/Specialized/Magentic/MagenticOrchestrator.cs @@ -148,6 +148,8 @@ to the conversation and enters the inner loop. throw new InvalidOperationException("This Magentic orchestration has already terminated. To process new messages, create a new workflow instance."); } + this._currentTurnRunOptions = context.GetAgentRunOptions(this._currentTurnRunOptions); + if (response.IsApproved) { await this.DelegateToTeamAsync(this._taskContext, context, cancellationToken).ConfigureAwait(false); diff --git a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/MagenticOrchestrationTests.cs b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/MagenticOrchestrationTests.cs index 262cafd3331..6cf7e773604 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/MagenticOrchestrationTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Workflows.UnitTests/MagenticOrchestrationTests.cs @@ -115,6 +115,108 @@ public async Task Magentic_AsAgent_CheckpointRecoveryUsesCurrentRunOptionsAsync( .Should().NotBeEmpty().And.OnlyContain(options => ReferenceEquals(options, recoveryRunOptions)); } + [Theory] + [InlineData(true, false)] + [InlineData(true, true)] + [InlineData(false, false)] + [InlineData(false, true)] + public async Task Magentic_AsAgent_PlanReviewContinuationUsesCurrentRunOptionsAsync(bool approvePlan, bool serializeSession) + { + // Arrange + List> managerResponses = + [ + CreatePlanResponse("Initial facts"), + CreatePlanResponse("Initial plan"), + ]; + if (approvePlan) + { + managerResponses.Add(CreateProgressLedgerResponse( + isRequestSatisfied: false, + isInLoop: false, + isProgressBeingMade: true, + nextSpeaker: "Worker", + instructionOrQuestion: "Complete the task")); + managerResponses.Add(CreateProgressLedgerResponse( + isRequestSatisfied: true, + isInLoop: false, + isProgressBeingMade: true, + nextSpeaker: "Worker", + instructionOrQuestion: "The task is complete")); + managerResponses.Add(CreateFinalAnswerResponse("Task completed successfully!")); + } + else + { + managerResponses.Add(CreatePlanResponse("Revised facts")); + managerResponses.Add(CreatePlanResponse("Revised plan")); + } + + RecordingReplayAgent manager = new(managerResponses, name: "Manager"); + RecordingEchoAgent worker = new(name: "Worker"); + InProcessExecutionEnvironment environment = + InProcessExecution.Lockstep.WithCheckpointing(CheckpointManager.CreateInMemory()); + AIAgent workflowAgent = new MagenticWorkflowBuilder(manager) + .AddParticipants(worker) + .RequirePlanSignoff(true) + .Build() + .AsAIAgent(executionEnvironment: environment); + AgentSession session = await workflowAgent.CreateSessionAsync(); + AgentRunOptions firstRunOptions = new() { AdditionalProperties = new() { ["invocation"] = "first" } }; + AgentRunOptions continuationRunOptions = new() { AdditionalProperties = new() { ["invocation"] = "continuation" } }; + + List firstUpdates = await workflowAgent + .RunStreamingAsync("Do the task", session, firstRunOptions) + .ToListAsync(); + RequestInfoEvent requestEvent = firstUpdates + .Select(update => update.RawRepresentation) + .OfType() + .Should().ContainSingle().Subject; + MagenticPlanReviewRequest reviewRequest = requestEvent.Request.Data.As() + ?? throw new InvalidOperationException("The plan review request was not available."); + FunctionCallContent functionCall = firstUpdates + .SelectMany(update => update.Contents) + .OfType() + .Should().ContainSingle().Subject; + int managerCallsBeforeContinuation = manager.RecordedRunOptions.Count; + + if (serializeSession) + { + JsonElement serializedSession = await workflowAgent.SerializeSessionAsync(session); + session = await workflowAgent.DeserializeSessionAsync(serializedSession); + } + + MagenticPlanReviewResponse reviewResponse = approvePlan + ? reviewRequest.Approve() + : reviewRequest.Revise("Please revise the plan"); + ChatMessage responseMessage = new( + ChatRole.Tool, + [new FunctionResultContent(functionCall.CallId, reviewResponse)]); + + // Act + List continuationUpdates = await workflowAgent + .RunStreamingAsync(responseMessage, session, continuationRunOptions) + .ToListAsync(); + + // Assert + continuationUpdates.SelectMany(update => update.Contents.OfType()).Should().BeEmpty(); + manager.RecordedRunOptions.Take(managerCallsBeforeContinuation) + .Should().OnlyContain(options => ReferenceEquals(options, firstRunOptions)); + manager.RecordedRunOptions.Skip(managerCallsBeforeContinuation) + .Should().NotBeEmpty().And.OnlyContain(options => ReferenceEquals(options, continuationRunOptions)); + + if (approvePlan) + { + worker.RecordedRunOptions + .Should().NotBeEmpty().And.OnlyContain(options => ReferenceEquals(options, continuationRunOptions)); + } + else + { + continuationUpdates + .Select(update => update.RawRepresentation) + .OfType() + .Should().ContainSingle(); + } + } + [Fact] public async Task Task_Completes_When_RequestSatisfiedAsync() {