mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Address PR comments on sending turn token for normal messages and handle contentId collision by source agent
This commit is contained in:
Peter Ibekwe
@@ -53,6 +53,9 @@ internal sealed class AsyncRunHandle : ICheckpointingHandle, IAsyncDisposable
|
||||
public ValueTask<RunStatus> GetStatusAsync(CancellationToken cancellationToken = default)
|
||||
=> this._eventStream.GetStatusAsync(cancellationToken);
|
||||
|
||||
internal bool TryGetResponsePortExecutorId(string portId, out string? executorId)
|
||||
=> this._stepRunner.TryGetResponsePortExecutorId(portId, out executorId);
|
||||
|
||||
public async IAsyncEnumerable<WorkflowEvent> TakeEventStreamAsync(bool blockOnPendingRequest, [EnumeratorCancellation] CancellationToken cancellationToken = default)
|
||||
{
|
||||
//Debug.Assert(breakOnHalt);
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
using System;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Collections.Generic;
|
||||
using System.Diagnostics.CodeAnalysis;
|
||||
using System.Linq;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
@@ -95,6 +96,18 @@ internal sealed class EdgeMap
|
||||
return portRunner.ChaseEdgeAsync(new MessageEnvelope(response, ExecutorIdentity.None), this._stepTracer, cancellationToken);
|
||||
}
|
||||
|
||||
internal bool TryGetResponsePortExecutorId(string portId, [NotNullWhen(true)] out string? executorId)
|
||||
{
|
||||
if (this._portEdgeRunners.TryGetValue(portId, out ResponseEdgeRunner? portRunner))
|
||||
{
|
||||
executorId = portRunner.ExecutorId;
|
||||
return true;
|
||||
}
|
||||
|
||||
executorId = null;
|
||||
return false;
|
||||
}
|
||||
|
||||
internal async ValueTask<Dictionary<EdgeId, PortableValue>> ExportStateAsync()
|
||||
{
|
||||
Dictionary<EdgeId, PortableValue> exportedStates = [];
|
||||
|
||||
@@ -19,6 +19,7 @@ internal interface ISuperStepRunner
|
||||
bool HasUnprocessedMessages { get; }
|
||||
|
||||
ValueTask EnqueueResponseAsync(ExternalResponse response, CancellationToken cancellationToken = default);
|
||||
bool TryGetResponsePortExecutorId(string portId, out string? executorId);
|
||||
|
||||
ValueTask<bool> IsValidInputTypeAsync<T>(CancellationToken cancellationToken = default);
|
||||
ValueTask<bool> EnqueueMessageAsync<T>(T message, CancellationToken cancellationToken = default);
|
||||
|
||||
@@ -160,6 +160,8 @@ internal sealed class InProcessRunner : ISuperStepRunner, ICheckpointingHandle
|
||||
|
||||
bool ISuperStepRunner.HasUnservicedRequests => this.RunContext.HasUnservicedRequests;
|
||||
bool ISuperStepRunner.HasUnprocessedMessages => this.RunContext.NextStepHasActions;
|
||||
bool ISuperStepRunner.TryGetResponsePortExecutorId(string portId, out string? executorId)
|
||||
=> this.RunContext.TryGetResponsePortExecutorId(portId, out executorId);
|
||||
|
||||
public bool IsCheckpointingEnabled => this.RunContext.IsCheckpointingEnabled;
|
||||
|
||||
|
||||
@@ -296,6 +296,9 @@ internal sealed class InProcessRunnerContext : IRunnerContext
|
||||
return this._externalRequests.TryRemove(requestId, out _);
|
||||
}
|
||||
|
||||
internal bool TryGetResponsePortExecutorId(string portId, [NotNullWhen(true)] out string? executorId)
|
||||
=> this._edgeMap.TryGetResponsePortExecutorId(portId, out executorId);
|
||||
|
||||
private IEventSink OutgoingEvents { get; }
|
||||
|
||||
internal StateManager StateManager { get; } = new();
|
||||
|
||||
@@ -16,10 +16,12 @@ internal sealed class AIContentExternalHandler<TRequestContent, TResponseContent
|
||||
where TResponseContent : AIContent
|
||||
{
|
||||
private readonly PortBinding? _portBinding;
|
||||
private readonly string _portId;
|
||||
private ConcurrentDictionary<string, TRequestContent> _pendingRequests = new();
|
||||
|
||||
public AIContentExternalHandler(ref ProtocolBuilder protocolBuilder, string portId, bool intercepted, Func<TResponseContent, IWorkflowContext, CancellationToken, ValueTask> handler)
|
||||
{
|
||||
this._portId = portId;
|
||||
PortBinding? portBinding = null;
|
||||
protocolBuilder = protocolBuilder.ConfigureRoutes(routeBuilder => ConfigureRoutes(routeBuilder, out portBinding));
|
||||
this._portBinding = portBinding;
|
||||
@@ -65,7 +67,7 @@ internal sealed class AIContentExternalHandler<TRequestContent, TResponseContent
|
||||
|
||||
return this.IsIntercepted
|
||||
? context.SendMessageAsync(requestContent, cancellationToken: cancellationToken)
|
||||
: this._portBinding.PostRequestAsync(requestContent, id, cancellationToken);
|
||||
: this._portBinding.PostRequestAsync(requestContent, this.CreateExternalRequestId(id), cancellationToken);
|
||||
}
|
||||
|
||||
public bool MarkRequestAsHandled(string id)
|
||||
@@ -76,6 +78,8 @@ internal sealed class AIContentExternalHandler<TRequestContent, TResponseContent
|
||||
[MemberNotNullWhen(false, nameof(_portBinding))]
|
||||
private bool IsIntercepted => this._portBinding == null;
|
||||
|
||||
private string CreateExternalRequestId(string requestId) => $"{this._portId.Length}:{this._portId}:{requestId}";
|
||||
|
||||
private static string MakeKey(string id) => $"{id}_PendingRequests";
|
||||
|
||||
public async ValueTask OnCheckpointingAsync(string id, IWorkflowContext context, CancellationToken cancellationToken = default)
|
||||
|
||||
@@ -60,6 +60,9 @@ public sealed class StreamingRun : CheckpointableRunBase, IAsyncDisposable
|
||||
internal ValueTask<bool> TrySendMessageUntypedAsync(object message, Type? declaredType = null)
|
||||
=> this._runHandle.EnqueueMessageUntypedAsync(message, declaredType);
|
||||
|
||||
internal bool TryGetResponsePortExecutorId(string portId, out string? executorId)
|
||||
=> this._runHandle.TryGetResponsePortExecutorId(portId, out executorId);
|
||||
|
||||
/// <summary>
|
||||
/// Asynchronously streams workflow events as they occur during workflow execution.
|
||||
/// </summary>
|
||||
|
||||
@@ -26,9 +26,9 @@ internal sealed class WorkflowSession : AgentSession
|
||||
private InMemoryCheckpointManager? _inMemoryCheckpointManager;
|
||||
|
||||
/// <summary>
|
||||
/// Tracks pending external requests by their content ID (e.g., <see cref="FunctionCallContent.CallId"/>
|
||||
/// or <see cref="UserInputRequestContent.Id"/>). This mapping enables converting incoming response
|
||||
/// content back to <see cref="ExternalResponse"/> when resuming a workflow from a checkpoint.
|
||||
/// Tracks pending external requests by their workflow-facing request ID.
|
||||
/// This mapping enables converting incoming response content back to <see cref="ExternalResponse"/>
|
||||
/// when resuming a workflow from a checkpoint.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
@@ -176,8 +176,8 @@ internal sealed class WorkflowSession : AgentSession
|
||||
.ConfigureAwait(false);
|
||||
|
||||
// Process messages: convert response content to ExternalResponse, send regular messages as-is
|
||||
bool hasMatchedExternalResponses = await this.SendMessagesWithResponseConversionAsync(run, messages).ConfigureAwait(false);
|
||||
return new ResumeRunResult(run, hasMatchedExternalResponses: hasMatchedExternalResponses);
|
||||
ResumeDispatchInfo dispatchInfo = await this.SendMessagesWithResponseConversionAsync(run, messages).ConfigureAwait(false);
|
||||
return new ResumeRunResult(run, dispatchInfo);
|
||||
}
|
||||
|
||||
StreamingRun newRun = await this._executionEnvironment
|
||||
@@ -194,15 +194,15 @@ internal sealed class WorkflowSession : AgentSession
|
||||
/// to ExternalResponse when there's a matching pending request.
|
||||
/// </summary>
|
||||
/// <returns>
|
||||
/// <see langword="true"/> if any external responses were sent; otherwise, <see langword="false"/>.
|
||||
/// Structured information about how resume content was dispatched.
|
||||
/// </returns>
|
||||
private async ValueTask<bool> SendMessagesWithResponseConversionAsync(StreamingRun run, List<ChatMessage> messages)
|
||||
private async ValueTask<ResumeDispatchInfo> SendMessagesWithResponseConversionAsync(StreamingRun run, List<ChatMessage> messages)
|
||||
{
|
||||
List<ChatMessage> regularMessages = [];
|
||||
// Responses are deferred until after regular messages are queued so response handlers
|
||||
// can merge buffered regular content in the same continuation turn.
|
||||
List<(ExternalResponse Response, string? ContentId)> externalResponses = [];
|
||||
bool hasMatchedExternalResponses = false;
|
||||
List<(ExternalResponse Response, string RequestId)> externalResponses = [];
|
||||
bool hasMatchedResponseForStartExecutor = false;
|
||||
|
||||
// Tracks content IDs already matched to pending requests within this invocation,
|
||||
// preventing duplicate responses for the same ID from being sent to the workflow engine.
|
||||
@@ -225,8 +225,16 @@ internal sealed class WorkflowSession : AgentSession
|
||||
if (contentId != null
|
||||
&& this.TryGetPendingRequest(contentId) is ExternalRequest pendingRequest)
|
||||
{
|
||||
externalResponses.Add((pendingRequest.CreateResponse(content), contentId));
|
||||
(matchedContentIds ??= new(StringComparer.OrdinalIgnoreCase)).Add(contentId);
|
||||
if (!run.TryGetResponsePortExecutorId(pendingRequest.PortInfo.PortId, out string? responseExecutorId))
|
||||
{
|
||||
throw new InvalidOperationException(
|
||||
$"Matched pending request '{pendingRequest.RequestId}' refers to unknown response port '{pendingRequest.PortInfo.PortId}'.");
|
||||
}
|
||||
|
||||
AIContent normalizedResponseContent = NormalizeResponseContentForDelivery(content, pendingRequest);
|
||||
externalResponses.Add((pendingRequest.CreateResponse(normalizedResponseContent), pendingRequest.RequestId));
|
||||
(matchedContentIds ??= new(StringComparer.Ordinal)).Add(contentId);
|
||||
hasMatchedResponseForStartExecutor |= string.Equals(responseExecutorId, this._workflow.StartExecutorId, StringComparison.Ordinal);
|
||||
}
|
||||
else
|
||||
{
|
||||
@@ -243,28 +251,54 @@ internal sealed class WorkflowSession : AgentSession
|
||||
}
|
||||
|
||||
// Send regular messages first so response handlers can merge them with responses.
|
||||
if (regularMessages.Count > 0)
|
||||
bool hasRegularMessages = regularMessages.Count > 0;
|
||||
if (hasRegularMessages)
|
||||
{
|
||||
await run.TrySendMessageAsync(regularMessages).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
// Send external responses after regular messages.
|
||||
foreach ((ExternalResponse response, string? contentId) in externalResponses)
|
||||
bool hasMatchedExternalResponses = false;
|
||||
foreach ((ExternalResponse response, string requestId) in externalResponses)
|
||||
{
|
||||
await run.SendResponseAsync(response).ConfigureAwait(false);
|
||||
hasMatchedExternalResponses = true;
|
||||
|
||||
if (contentId is string id)
|
||||
{
|
||||
this.RemovePendingRequest(id);
|
||||
}
|
||||
this.RemovePendingRequest(requestId);
|
||||
}
|
||||
|
||||
return hasMatchedExternalResponses;
|
||||
return new ResumeDispatchInfo(
|
||||
hasRegularMessages,
|
||||
hasMatchedExternalResponses,
|
||||
hasMatchedResponseForStartExecutor);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the content ID from response content types.
|
||||
/// Creates the workflow-facing request content surfaced in response updates.
|
||||
/// </summary>
|
||||
private static AIContent CreateRequestContentForDelivery(ExternalRequest request) => request switch
|
||||
{
|
||||
ExternalRequest externalRequest when externalRequest.TryGetDataAs(out FunctionCallContent? functionCallContent)
|
||||
=> CloneFunctionCallContent(functionCallContent, externalRequest.RequestId),
|
||||
ExternalRequest externalRequest when externalRequest.TryGetDataAs(out UserInputRequestContent? userInputRequestContent)
|
||||
=> CloneUserInputRequestContent(userInputRequestContent, externalRequest.RequestId),
|
||||
ExternalRequest externalRequest
|
||||
=> externalRequest.ToFunctionCall(),
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// Rewrites workflow-facing response content back to the original agent-owned content ID.
|
||||
/// </summary>
|
||||
private static AIContent NormalizeResponseContentForDelivery(AIContent content, ExternalRequest request) => content switch
|
||||
{
|
||||
FunctionResultContent functionResultContent when request.TryGetDataAs(out FunctionCallContent? functionCallContent)
|
||||
=> CloneFunctionResultContent(functionResultContent, functionCallContent.CallId),
|
||||
UserInputResponseContent userInputResponseContent when request.TryGetDataAs(out UserInputRequestContent? userInputRequestContent)
|
||||
=> CloneUserInputResponseContent(userInputResponseContent, userInputRequestContent.Id),
|
||||
_ => content,
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// Gets the workflow-facing request ID from response content types.
|
||||
/// </summary>
|
||||
private static string? GetResponseContentId(AIContent content) => content switch
|
||||
{
|
||||
@@ -274,22 +308,21 @@ internal sealed class WorkflowSession : AgentSession
|
||||
};
|
||||
|
||||
/// <summary>
|
||||
/// Tries to get a pending request by content ID.
|
||||
/// Tries to get a pending request by workflow-facing request ID.
|
||||
/// </summary>
|
||||
private ExternalRequest? TryGetPendingRequest(string contentId) =>
|
||||
this._pendingRequests.TryGetValue(contentId, out ExternalRequest? request) ? request : null;
|
||||
private ExternalRequest? TryGetPendingRequest(string requestId) =>
|
||||
this._pendingRequests.TryGetValue(requestId, out ExternalRequest? request) ? request : null;
|
||||
|
||||
/// <summary>
|
||||
/// Adds a pending request indexed by content ID.
|
||||
/// Adds a pending request indexed by workflow-facing request ID.
|
||||
/// </summary>
|
||||
private void AddPendingRequest(string contentId, ExternalRequest request) =>
|
||||
this._pendingRequests[contentId] = request;
|
||||
private void AddPendingRequest(string requestId, ExternalRequest request) => this._pendingRequests[requestId] = request;
|
||||
|
||||
/// <summary>
|
||||
/// Removes a pending request by content ID.
|
||||
/// Removes a pending request by workflow-facing request ID.
|
||||
/// </summary>
|
||||
private void RemovePendingRequest(string contentId) =>
|
||||
this._pendingRequests.Remove(contentId);
|
||||
private void RemovePendingRequest(string requestId) =>
|
||||
this._pendingRequests.Remove(requestId);
|
||||
|
||||
internal async
|
||||
IAsyncEnumerable<AgentResponseUpdate> InvokeStageAsync(
|
||||
@@ -306,10 +339,11 @@ internal sealed class WorkflowSession : AgentSession
|
||||
await using StreamingRun run = resumeResult.Run;
|
||||
#pragma warning restore CA2007
|
||||
|
||||
// Send a TurnToken only when no external responses were delivered.
|
||||
// External response handlers already drive continuation turns and can merge
|
||||
// buffered regular messages, so an extra TurnToken would cause a redundant turn.
|
||||
if (!resumeResult.HasMatchedExternalResponses)
|
||||
ResumeDispatchInfo dispatchInfo = resumeResult.DispatchInfo;
|
||||
bool shouldSendTurnToken =
|
||||
!dispatchInfo.HasMatchedExternalResponses
|
||||
|| (dispatchInfo.HasRegularMessages && !dispatchInfo.HasMatchedResponseForStartExecutor);
|
||||
if (shouldSendTurnToken)
|
||||
{
|
||||
await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false);
|
||||
}
|
||||
@@ -324,18 +358,11 @@ internal sealed class WorkflowSession : AgentSession
|
||||
break;
|
||||
|
||||
case RequestInfoEvent requestInfo:
|
||||
(AIContent requestContent, string? contentId) = requestInfo.Request switch
|
||||
{
|
||||
ExternalRequest externalRequest when externalRequest.TryGetDataAs(out FunctionCallContent? fcc) => (fcc, fcc.CallId),
|
||||
ExternalRequest externalRequest when externalRequest.TryGetDataAs(out UserInputRequestContent? uic) => (uic, uic.Id),
|
||||
ExternalRequest externalRequest => ((AIContent)externalRequest.ToFunctionCall(), externalRequest.RequestId)
|
||||
};
|
||||
AIContent requestContent = CreateRequestContentForDelivery(requestInfo.Request);
|
||||
|
||||
// Track the pending request so we can convert incoming responses back to ExternalResponse
|
||||
if (contentId != null)
|
||||
{
|
||||
this.AddPendingRequest(contentId, requestInfo.Request);
|
||||
}
|
||||
// Track the pending request so we can convert incoming responses back to ExternalResponse.
|
||||
// External callers respond using the workflow-facing request ID, which is always RequestId.
|
||||
this.AddPendingRequest(requestInfo.Request.RequestId, requestInfo.Request);
|
||||
|
||||
AgentResponseUpdate update = this.CreateUpdate(this.LastResponseId, evt, requestContent);
|
||||
yield return update;
|
||||
@@ -420,16 +447,113 @@ internal sealed class WorkflowSession : AgentSession
|
||||
/// <summary>The streaming run that was created or resumed.</summary>
|
||||
public StreamingRun Run { get; }
|
||||
|
||||
/// <summary>Whether any external responses (e.g., <see cref="FunctionResultContent"/>) were delivered.</summary>
|
||||
public bool HasMatchedExternalResponses { get; }
|
||||
/// <summary>How resume-time content was dispatched into the workflow runtime.</summary>
|
||||
public ResumeDispatchInfo DispatchInfo { get; }
|
||||
|
||||
public ResumeRunResult(StreamingRun run, bool hasMatchedExternalResponses = false)
|
||||
public ResumeRunResult(StreamingRun run, ResumeDispatchInfo dispatchInfo = default)
|
||||
{
|
||||
this.Run = Throw.IfNull(run);
|
||||
this.HasMatchedExternalResponses = hasMatchedExternalResponses;
|
||||
this.DispatchInfo = dispatchInfo;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Captures how resumed input was split across regular-message and external-response delivery paths.
|
||||
/// </summary>
|
||||
private readonly struct ResumeDispatchInfo
|
||||
{
|
||||
public ResumeDispatchInfo(bool hasRegularMessages, bool hasMatchedExternalResponses, bool hasMatchedResponseForStartExecutor)
|
||||
{
|
||||
this.HasRegularMessages = hasRegularMessages;
|
||||
this.HasMatchedExternalResponses = hasMatchedExternalResponses;
|
||||
this.HasMatchedResponseForStartExecutor = hasMatchedResponseForStartExecutor;
|
||||
}
|
||||
|
||||
public bool HasRegularMessages { get; }
|
||||
|
||||
public bool HasMatchedExternalResponses { get; }
|
||||
|
||||
public bool HasMatchedResponseForStartExecutor { get; }
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Clones a <see cref="FunctionCallContent"/> with a workflow-facing call ID.
|
||||
/// </summary>
|
||||
private static FunctionCallContent CloneFunctionCallContent(FunctionCallContent content, string callId)
|
||||
{
|
||||
FunctionCallContent clone = new(callId, content.Name, content.Arguments)
|
||||
{
|
||||
Exception = content.Exception,
|
||||
InformationalOnly = content.InformationalOnly,
|
||||
};
|
||||
|
||||
return CopyContentMetadata(content, clone);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Clones a <see cref="FunctionResultContent"/> with an agent-owned call ID.
|
||||
/// </summary>
|
||||
private static FunctionResultContent CloneFunctionResultContent(FunctionResultContent content, string callId)
|
||||
{
|
||||
FunctionResultContent clone = new(callId, content.Result)
|
||||
{
|
||||
Exception = content.Exception,
|
||||
};
|
||||
|
||||
return CopyContentMetadata(content, clone);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Clones a <see cref="UserInputRequestContent"/> with a workflow-facing request ID.
|
||||
/// </summary>
|
||||
private static UserInputRequestContent CloneUserInputRequestContent(UserInputRequestContent content, string id)
|
||||
{
|
||||
UserInputRequestContent clone = content switch
|
||||
{
|
||||
FunctionApprovalRequestContent functionApprovalRequestContent =>
|
||||
new FunctionApprovalRequestContent(id, functionApprovalRequestContent.FunctionCall),
|
||||
McpServerToolApprovalRequestContent mcpApprovalRequestContent =>
|
||||
new McpServerToolApprovalRequestContent(id, mcpApprovalRequestContent.ToolCall),
|
||||
_ => throw new NotSupportedException(
|
||||
$"Unsupported user input request content type '{content.GetType().Name}' for workflow request ID rewriting."),
|
||||
};
|
||||
|
||||
return CopyContentMetadata(content, clone);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Clones a <see cref="UserInputResponseContent"/> with an agent-owned request ID.
|
||||
/// </summary>
|
||||
private static UserInputResponseContent CloneUserInputResponseContent(UserInputResponseContent content, string id)
|
||||
{
|
||||
UserInputResponseContent clone = content switch
|
||||
{
|
||||
FunctionApprovalResponseContent functionApprovalResponseContent =>
|
||||
new FunctionApprovalResponseContent(id, functionApprovalResponseContent.Approved, functionApprovalResponseContent.FunctionCall)
|
||||
{
|
||||
Reason = functionApprovalResponseContent.Reason,
|
||||
},
|
||||
McpServerToolApprovalResponseContent mcpApprovalResponseContent =>
|
||||
new McpServerToolApprovalResponseContent(id, mcpApprovalResponseContent.Approved),
|
||||
_ => throw new NotSupportedException(
|
||||
$"Unsupported user input response content type '{content.GetType().Name}' for workflow response ID rewriting."),
|
||||
};
|
||||
|
||||
return CopyContentMetadata(content, clone);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Copies shared <see cref="AIContent"/> metadata to a cloned content instance.
|
||||
/// </summary>
|
||||
private static TContent CopyContentMetadata<TContent>(AIContent source, TContent target)
|
||||
where TContent : AIContent
|
||||
{
|
||||
target.AdditionalProperties = source.AdditionalProperties;
|
||||
target.Annotations = source.Annotations;
|
||||
target.RawRepresentation = source.RawRepresentation;
|
||||
return target;
|
||||
}
|
||||
|
||||
internal sealed class SessionState(
|
||||
string sessionId,
|
||||
CheckpointInfo? lastCheckpoint,
|
||||
|
||||
@@ -85,6 +85,71 @@ internal sealed class RequestEmittingAgent : AIAgent
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed class KickoffOnStartExecutor : ChatProtocolExecutor
|
||||
{
|
||||
private static readonly ChatProtocolExecutorOptions s_options = new()
|
||||
{
|
||||
AutoSendTurnToken = false,
|
||||
};
|
||||
|
||||
private readonly string _downstreamExecutorId;
|
||||
private readonly string _kickoffInputText;
|
||||
private readonly string _kickoffMessageText;
|
||||
private readonly string _regularResumeText;
|
||||
private readonly string _regularProcessedText;
|
||||
|
||||
public KickoffOnStartExecutor(
|
||||
string id,
|
||||
string downstreamExecutorId,
|
||||
string kickoffInputText,
|
||||
string kickoffMessageText,
|
||||
string regularResumeText,
|
||||
string regularProcessedText)
|
||||
: base(id, s_options)
|
||||
{
|
||||
this._downstreamExecutorId = downstreamExecutorId;
|
||||
this._kickoffInputText = kickoffInputText;
|
||||
this._kickoffMessageText = kickoffMessageText;
|
||||
this._regularResumeText = regularResumeText;
|
||||
this._regularProcessedText = regularProcessedText;
|
||||
}
|
||||
|
||||
protected override async ValueTask TakeTurnAsync(List<ChatMessage> messages, IWorkflowContext context, bool? emitEvents, CancellationToken cancellationToken = default)
|
||||
{
|
||||
List<string> textContents =
|
||||
[
|
||||
.. messages
|
||||
.SelectMany(message => message.Contents.OfType<TextContent>())
|
||||
.Select(content => content.Text)
|
||||
];
|
||||
|
||||
if (textContents.Contains(this._kickoffInputText, StringComparer.Ordinal))
|
||||
{
|
||||
await context.SendMessageAsync(
|
||||
new List<ChatMessage> { new(ChatRole.User, this._kickoffMessageText) },
|
||||
this._downstreamExecutorId,
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
await context.SendMessageAsync(
|
||||
new TurnToken(emitEvents),
|
||||
this._downstreamExecutorId,
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
if (textContents.Contains(this._regularResumeText, StringComparer.Ordinal))
|
||||
{
|
||||
AgentResponseUpdate update = new(ChatRole.Assistant, [new TextContent(this._regularProcessedText)])
|
||||
{
|
||||
CreatedAt = DateTimeOffset.UtcNow,
|
||||
MessageId = Guid.NewGuid().ToString("N"),
|
||||
ResponseId = Guid.NewGuid().ToString("N"),
|
||||
Role = ChatRole.Assistant,
|
||||
};
|
||||
|
||||
await context.AddEventAsync(new AgentResponseUpdateEvent(this.Id, update), cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public class WorkflowHostSmokeTests
|
||||
{
|
||||
private sealed class AlwaysFailsAIAgent(bool failByThrowing) : AIAgent
|
||||
@@ -193,7 +258,7 @@ public class WorkflowHostSmokeTests
|
||||
|
||||
// Assert
|
||||
AgentResponseUpdate? updateWithFunctionCall = updates.FirstOrDefault(u =>
|
||||
u.Contents.Any(c => c is FunctionCallContent));
|
||||
u.RawRepresentation is RequestInfoEvent && u.Contents.Any(c => c is FunctionCallContent));
|
||||
|
||||
updateWithFunctionCall.Should().NotBeNull("a FunctionCallContent should be present in the response updates");
|
||||
FunctionCallContent retrievedContent = updateWithFunctionCall!.Contents
|
||||
@@ -201,7 +266,8 @@ public class WorkflowHostSmokeTests
|
||||
.Should().ContainSingle()
|
||||
.Which;
|
||||
|
||||
retrievedContent.CallId.Should().Be(CallId);
|
||||
retrievedContent.CallId.Should().NotBe(CallId);
|
||||
retrievedContent.CallId.Should().EndWith($":{CallId}");
|
||||
retrievedContent.Name.Should().Be(FunctionName);
|
||||
}
|
||||
|
||||
@@ -228,7 +294,7 @@ public class WorkflowHostSmokeTests
|
||||
|
||||
// Assert
|
||||
AgentResponseUpdate? updateWithUserInput = updates.FirstOrDefault(u =>
|
||||
u.Contents.Any(c => c is UserInputRequestContent));
|
||||
u.RawRepresentation is RequestInfoEvent && u.Contents.Any(c => c is UserInputRequestContent));
|
||||
|
||||
updateWithUserInput.Should().NotBeNull("a UserInputRequestContent should be present in the response updates");
|
||||
UserInputRequestContent retrievedContent = updateWithUserInput!.Contents
|
||||
@@ -237,7 +303,8 @@ public class WorkflowHostSmokeTests
|
||||
.Which;
|
||||
|
||||
retrievedContent.Should().BeOfType<McpServerToolApprovalRequestContent>();
|
||||
retrievedContent.Id.Should().Be(RequestId);
|
||||
retrievedContent.Id.Should().NotBe(RequestId);
|
||||
retrievedContent.Id.Should().EndWith($":{RequestId}");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -264,16 +331,16 @@ public class WorkflowHostSmokeTests
|
||||
|
||||
// Assert 1: We should have received a FunctionCallContent
|
||||
AgentResponseUpdate? updateWithRequest = firstCallUpdates.FirstOrDefault(u =>
|
||||
u.Contents.Any(c => c is FunctionCallContent));
|
||||
u.RawRepresentation is RequestInfoEvent && u.Contents.Any(c => c is FunctionCallContent));
|
||||
updateWithRequest.Should().NotBeNull("a FunctionCallContent should be present in the response updates");
|
||||
|
||||
FunctionCallContent receivedRequest = updateWithRequest!.Contents
|
||||
.OfType<FunctionCallContent>()
|
||||
.First();
|
||||
receivedRequest.CallId.Should().Be(CallId);
|
||||
receivedRequest.CallId.Should().EndWith($":{CallId}");
|
||||
|
||||
// Act 2: Send the response back
|
||||
FunctionResultContent responseContent = new(CallId, "test result");
|
||||
FunctionResultContent responseContent = new(receivedRequest.CallId, "test result");
|
||||
ChatMessage responseMessage = new(ChatRole.Tool, [responseContent]);
|
||||
|
||||
// Act 2: Run the workflow with the response and capture the resulting updates
|
||||
@@ -284,9 +351,10 @@ public class WorkflowHostSmokeTests
|
||||
secondCallUpdates.Should().NotBeNull("processing the response should produce updates");
|
||||
secondCallUpdates.Should().NotBeEmpty("processing the response should progress the workflow");
|
||||
secondCallUpdates
|
||||
.Where(u => u.RawRepresentation is RequestInfoEvent)
|
||||
.SelectMany(u => u.Contents.OfType<FunctionCallContent>())
|
||||
.Should()
|
||||
.NotContain(c => c.CallId == CallId, "the original FunctionCallContent request should be cleared after processing the response");
|
||||
.NotContain(c => c.CallId == receivedRequest.CallId, "the external FunctionCallContent request should be cleared after processing the response");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
@@ -314,16 +382,17 @@ public class WorkflowHostSmokeTests
|
||||
|
||||
// Assert 1: We should have received a UserInputRequestContent
|
||||
AgentResponseUpdate? updateWithRequest = firstCallUpdates.FirstOrDefault(u =>
|
||||
u.Contents.Any(c => c is UserInputRequestContent));
|
||||
u.RawRepresentation is RequestInfoEvent && u.Contents.Any(c => c is UserInputRequestContent));
|
||||
updateWithRequest.Should().NotBeNull("a UserInputRequestContent should be present in the response updates");
|
||||
|
||||
UserInputRequestContent receivedRequest = updateWithRequest!.Contents
|
||||
.OfType<UserInputRequestContent>()
|
||||
.First();
|
||||
receivedRequest.Id.Should().Be(RequestId);
|
||||
receivedRequest.Id.Should().EndWith($":{RequestId}");
|
||||
receivedRequest.Should().BeOfType<McpServerToolApprovalRequestContent>();
|
||||
|
||||
// Act 2: Send the response back - use CreateResponse to get the right response type
|
||||
UserInputResponseContent responseContent = requestContent.CreateResponse(approved: true);
|
||||
UserInputResponseContent responseContent = ((McpServerToolApprovalRequestContent)receivedRequest).CreateResponse(approved: true);
|
||||
ChatMessage responseMessage = new(ChatRole.User, [responseContent]);
|
||||
|
||||
// Act 2: Run the workflow again with the response and capture the updates
|
||||
@@ -331,7 +400,9 @@ public class WorkflowHostSmokeTests
|
||||
|
||||
// Assert 2: The response should be applied so that the original request is no longer pending
|
||||
secondCallUpdates.Should().NotBeEmpty("handling the user input response should produce follow-up updates");
|
||||
bool requestStillPresent = secondCallUpdates.Any(u => u.Contents.OfType<UserInputRequestContent>().Any(r => r.Id == RequestId));
|
||||
bool requestStillPresent = secondCallUpdates.Any(u =>
|
||||
u.RawRepresentation is RequestInfoEvent
|
||||
&& u.Contents.OfType<UserInputRequestContent>().Any(r => r.Id == receivedRequest.Id));
|
||||
requestStillPresent.Should().BeFalse("the original UserInputRequestContent should not be re-emitted after its response is processed");
|
||||
}
|
||||
|
||||
@@ -363,11 +434,15 @@ public class WorkflowHostSmokeTests
|
||||
session).ToListAsync();
|
||||
|
||||
// Assert 1: We should have received a FunctionCallContent
|
||||
AgentResponseUpdate requestUpdate = firstCallUpdates.First(u =>
|
||||
u.RawRepresentation is RequestInfoEvent && u.Contents.Any(c => c is FunctionCallContent));
|
||||
FunctionCallContent emittedRequest = requestUpdate.Contents.OfType<FunctionCallContent>().Single();
|
||||
|
||||
firstCallUpdates.Should().Contain(u => u.Contents.Any(c => c is FunctionCallContent),
|
||||
"the first call should emit a FunctionCallContent request");
|
||||
|
||||
// Act 2: Send a mixed message containing both the function result AND regular non-response content
|
||||
FunctionResultContent responseContent = new(CallId, "tool output");
|
||||
FunctionResultContent responseContent = new(emittedRequest.CallId, "tool output");
|
||||
ChatMessage mixedMessage = new(ChatRole.Tool, [responseContent, new TextContent("additional context")]);
|
||||
|
||||
List<AgentResponseUpdate> secondCallUpdates = await agent.RunStreamingAsync(mixedMessage, session).ToListAsync();
|
||||
@@ -375,9 +450,10 @@ public class WorkflowHostSmokeTests
|
||||
// Assert 2: The workflow should have processed both parts without errors
|
||||
secondCallUpdates.Should().NotBeEmpty("the mixed message should produce follow-up updates");
|
||||
secondCallUpdates
|
||||
.Where(u => u.RawRepresentation is RequestInfoEvent)
|
||||
.SelectMany(u => u.Contents.OfType<FunctionCallContent>())
|
||||
.Should()
|
||||
.NotContain(c => c.CallId == CallId, "the original FunctionCallContent should be cleared after the response is processed");
|
||||
.NotContain(c => c.CallId == emittedRequest.CallId, "the external FunctionCallContent should be cleared after the response is processed");
|
||||
secondCallUpdates
|
||||
.SelectMany(u => u.Contents.OfType<ErrorContent>())
|
||||
.Should()
|
||||
@@ -398,11 +474,14 @@ public class WorkflowHostSmokeTests
|
||||
|
||||
AgentSession session = await agent.CreateSessionAsync();
|
||||
List<AgentResponseUpdate> firstCallUpdates = await agent.RunStreamingAsync(new ChatMessage(ChatRole.User, "Start"), session).ToListAsync();
|
||||
firstCallUpdates.Should().Contain(u => u.Contents.Any(c => c is FunctionCallContent));
|
||||
FunctionCallContent emittedRequest = firstCallUpdates
|
||||
.Where(u => u.RawRepresentation is RequestInfoEvent)
|
||||
.SelectMany(u => u.Contents.OfType<FunctionCallContent>())
|
||||
.Single();
|
||||
|
||||
ChatMessage[] resumeMessages =
|
||||
[
|
||||
new(ChatRole.Tool, [new FunctionResultContent(CallId, "tool output")]),
|
||||
new(ChatRole.Tool, [new FunctionResultContent(emittedRequest.CallId, "tool output")]),
|
||||
new(ChatRole.Tool, [new TextContent("extra context in separate message")])
|
||||
];
|
||||
|
||||
@@ -410,9 +489,10 @@ public class WorkflowHostSmokeTests
|
||||
|
||||
secondCallUpdates.Should().NotBeEmpty();
|
||||
secondCallUpdates
|
||||
.Where(u => u.RawRepresentation is RequestInfoEvent)
|
||||
.SelectMany(u => u.Contents.OfType<FunctionCallContent>())
|
||||
.Should()
|
||||
.NotContain(c => c.CallId == CallId, "response+regular content split across messages should not re-emit the handled request");
|
||||
.NotContain(c => c.CallId == emittedRequest.CallId, "response+regular content split across messages should not re-emit the handled external request");
|
||||
secondCallUpdates
|
||||
.SelectMany(u => u.Contents.OfType<ErrorContent>())
|
||||
.Should()
|
||||
@@ -433,20 +513,82 @@ public class WorkflowHostSmokeTests
|
||||
|
||||
AgentSession session = await agent.CreateSessionAsync();
|
||||
List<AgentResponseUpdate> firstCallUpdates = await agent.RunStreamingAsync(new ChatMessage(ChatRole.User, "Start"), session).ToListAsync();
|
||||
firstCallUpdates.Should().Contain(u => u.Contents.Any(c => c is FunctionCallContent));
|
||||
FunctionCallContent emittedRequest = firstCallUpdates
|
||||
.Where(u => u.RawRepresentation is RequestInfoEvent)
|
||||
.SelectMany(u => u.Contents.OfType<FunctionCallContent>())
|
||||
.Single();
|
||||
|
||||
List<AgentResponseUpdate> secondCallUpdates = await agent.RunStreamingAsync(
|
||||
new ChatMessage(ChatRole.Tool, [new FunctionResultContent(CallId, "tool output")]),
|
||||
new ChatMessage(ChatRole.Tool, [new FunctionResultContent(emittedRequest.CallId, "tool output")]),
|
||||
session).ToListAsync();
|
||||
|
||||
int functionCallCount = secondCallUpdates
|
||||
.Where(u => u.RawRepresentation is RequestInfoEvent)
|
||||
.SelectMany(u => u.Contents.OfType<FunctionCallContent>())
|
||||
.Count(c => c.CallId == CallId);
|
||||
.Count(c => c.CallId == emittedRequest.CallId);
|
||||
|
||||
functionCallCount.Should().Be(1, "a matching external response should not trigger an extra TurnToken-driven turn");
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Test_AsAgent_MixedResponseAndRegularMessage_CrossExecutorStartExecutorIsReawakenedAsync()
|
||||
{
|
||||
const string StartExecutorId = "start-executor";
|
||||
const string KickoffInputText = "Start";
|
||||
const string KickoffMessageText = "kickoff downstream";
|
||||
const string ResumeRegularText = "resume regular";
|
||||
const string ResumeProcessedText = "regular message processed";
|
||||
const string CallId = "cross-executor-call-id";
|
||||
const string FunctionName = "crossExecutorFunction";
|
||||
|
||||
RequestEmittingAgent requestAgent = new(new FunctionCallContent(CallId, FunctionName), completeOnResponse: true);
|
||||
ExecutorBinding requestBinding = requestAgent.BindAsExecutor(
|
||||
new AIAgentHostOptions { InterceptUnterminatedFunctionCalls = false, EmitAgentUpdateEvents = true });
|
||||
|
||||
KickoffOnStartExecutor startExecutor = new(
|
||||
StartExecutorId,
|
||||
requestBinding.Id,
|
||||
KickoffInputText,
|
||||
KickoffMessageText,
|
||||
ResumeRegularText,
|
||||
ResumeProcessedText);
|
||||
ExecutorBinding startBinding = startExecutor.BindExecutor();
|
||||
|
||||
Workflow workflow = new WorkflowBuilder(startBinding)
|
||||
.AddEdge<List<ChatMessage>>(startBinding, requestBinding, messages =>
|
||||
messages?.Any(message => message.Contents.OfType<TextContent>().Any(content => content.Text == KickoffMessageText)) == true)
|
||||
.AddEdge<TurnToken>(startBinding, requestBinding, _ => true)
|
||||
.Build();
|
||||
AIAgent agent = workflow.AsAIAgent("WorkflowAgent");
|
||||
|
||||
AgentSession session = await agent.CreateSessionAsync();
|
||||
List<AgentResponseUpdate> firstCallUpdates = await agent.RunStreamingAsync(
|
||||
new ChatMessage(ChatRole.User, KickoffInputText),
|
||||
session).ToListAsync();
|
||||
FunctionCallContent emittedRequest = firstCallUpdates
|
||||
.Where(u => u.RawRepresentation is RequestInfoEvent)
|
||||
.SelectMany(u => u.Contents.OfType<FunctionCallContent>())
|
||||
.Single();
|
||||
|
||||
ChatMessage[] resumeMessages =
|
||||
[
|
||||
new(ChatRole.Tool, [new FunctionResultContent(emittedRequest.CallId, "tool output")]),
|
||||
new(ChatRole.User, ResumeRegularText)
|
||||
];
|
||||
|
||||
List<AgentResponseUpdate> secondCallUpdates = await agent.RunStreamingAsync(resumeMessages, session).ToListAsync();
|
||||
List<string> textContents = [.. secondCallUpdates.SelectMany(update => update.Contents.OfType<TextContent>()).Select(content => content.Text)];
|
||||
|
||||
textContents.Should().Contain(ResumeProcessedText, "the start executor should receive an explicit TurnToken when the matched response wakes a different executor");
|
||||
textContents.Should().Contain("Request processed", "the matched external response should still be delivered to the downstream request owner");
|
||||
secondCallUpdates
|
||||
.Where(u => u.RawRepresentation is RequestInfoEvent)
|
||||
.SelectMany(u => u.Contents.OfType<FunctionCallContent>())
|
||||
.Should()
|
||||
.NotContain(c => c.CallId == emittedRequest.CallId, "the handled external request should not be re-emitted while waking the start executor");
|
||||
secondCallUpdates.SelectMany(u => u.Contents.OfType<ErrorContent>()).Should().BeEmpty();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task Test_AsAgent_UnmatchedResponse_TriggersTurnAndKeepsProgressingAsync()
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user