Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,9 @@ internal ExternalInputRequest(string text)
/// <c>requireApproval: true</c>) over <see cref="FunctionCallContent"/> so that
/// hosts which speak the approval protocol see the approval-bearing content.
/// </remarks>
AIContent? IExternalRequestEnvelope.GetInnerRequestContent()
AIContent? IExternalRequestEnvelope.GetInnerRequestContent() => this.GetInnerRequestContent();

private AIContent? GetInnerRequestContent()
{
IList<ChatMessage>? messages = this.AgentResponse?.Messages;
if (messages is null)
Expand Down Expand Up @@ -73,5 +75,16 @@ internal ExternalInputRequest(string text)

/// <inheritdoc />
object IExternalRequestEnvelope.CreateResponse(IList<ChatMessage> messages)
=> new ExternalInputResponse(messages);
=> new ExternalInputResponse(messages)
{
RequestId = GetContentId(this.GetInnerRequestContent()),
};

private static string? GetContentId(AIContent? content)
=> content switch
{
ToolCallContent functionCall => functionCall.CallId,
InputRequestContent toolApprovalRequest => toolApprovalRequest.RequestId,
_ => null,
};
}
Original file line number Diff line number Diff line change
Expand Up @@ -16,15 +16,21 @@ public sealed class ExternalInputResponse
/// </summary>
public IList<ChatMessage> Messages { get; }

/// <summary>
/// Gets the request id that produced this response, if available.
/// </summary>
[JsonInclude]
public string? RequestId { get; internal init; }

internal bool HasMessages => this.Messages?.Count > 0;

/// <summary>
/// Initializes a new instance of <see cref="ExternalInputResponse"/>.
/// </summary>
/// <param name="message">The external input message being provided to the workflow.</param>
public ExternalInputResponse(ChatMessage message)
: this([message], requestId: null)
{
this.Messages = [message];
}

/// <summary>
Expand All @@ -33,7 +39,14 @@ public ExternalInputResponse(ChatMessage message)
/// <param name="messages">The external input messages being provided to the workflow.</param>
[JsonConstructor]
public ExternalInputResponse(IList<ChatMessage> messages)
: this(messages, requestId: null)
{
}

internal ExternalInputResponse(IList<ChatMessage> messages, string? requestId)
{
this.Messages = messages;
this.RequestId = requestId;
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ internal sealed class InvokeFunctionToolExecutor(
DeclarativeActionExecutor<InvokeFunctionTool>(model, state)
{
private const string ApprovalSnapshotStateKey = nameof(_approvalSnapshots);
private const string HasApprovalRequiredInvocationStateKey = nameof(_hasApprovalRequiredInvocation);
private const string PendingCallIdsStateKey = nameof(_pendingNonApprovalCallIds);
private const string LegacyApprovalSnapshotStateKey = "_approvalSnapshot";

Expand All @@ -39,6 +40,8 @@ internal sealed class InvokeFunctionToolExecutor(
/// </summary>
private readonly ConcurrentDictionary<string, ApprovalSnapshot> _approvalSnapshots = new(StringComparer.Ordinal);

private bool _hasApprovalRequiredInvocation;

/// <summary>
/// Per-invocation call ids for in-flight non-approval requests; used to match the
/// returning <see cref="FunctionResultContent"/> on the response path. Used as a set;
Expand Down Expand Up @@ -94,6 +97,7 @@ public static class Steps
// Capture the evaluated parameters keyed by request id; the matching response
// resumes from this snapshot.
this._approvalSnapshots[requestId] = new ApprovalSnapshot(functionName, arguments);
this._hasApprovalRequiredInvocation = true;

requestMessage.Contents.Add(new ToolApprovalRequestContent(requestId, functionCall));
}
Expand Down Expand Up @@ -125,24 +129,29 @@ public async ValueTask CaptureResponseAsync(
{
bool autoSend = this.GetAutoSendValue();
string? conversationId = this.GetConversationId();
string? responseRequestId = response.RequestId;

// Match the inbound result by its per-invocation call id.
FunctionResultContent? matchingResult = response.Messages
.SelectMany(m => m.Contents)
.OfType<FunctionResultContent>()
.FirstOrDefault(r => this.IsKnownPendingId(r.CallId));
.FirstOrDefault(r => this.IsCorrelatedPendingNonApprovalResult(r, responseRequestId));
FunctionResultContent? approvedFunctionResult = null;

// Legacy non-approval backstop: when no pendings are tracked, accept a result
// whose CallId equals this.Id. The runtime has already routed the response to
// this executor's port and the framework does not invoke a function here.
// Legacy non-approval backstop: when no pendings are tracked and approval is
// not required, accept a result whose CallId equals this.Id. The runtime has
// already routed the response to this executor's port and the framework does
// not invoke a function here.
if (matchingResult is null
&& !this._hasApprovalRequiredInvocation
&& this._pendingNonApprovalCallIds.IsEmpty
&& this._approvalSnapshots.IsEmpty)
{
matchingResult = response.Messages
.SelectMany(m => m.Contents)
.OfType<FunctionResultContent>()
.FirstOrDefault(r => string.Equals(r.CallId, this.Id, StringComparison.Ordinal));
.FirstOrDefault(r => string.Equals(r.CallId, this.Id, StringComparison.Ordinal)
&& IsCorrelatedResponse(r.CallId, responseRequestId));
}

// When the caller approved an approval-required function call but didn't execute it
Expand All @@ -160,8 +169,9 @@ public async ValueTask CaptureResponseAsync(
// Prefer an approval matching a pending snapshot; otherwise take the first
// present approval.
ToolApprovalResponseContent? approval =
approvals.FirstOrDefault(r => this._approvalSnapshots.ContainsKey(r.RequestId))
?? approvals.FirstOrDefault();
approvals.FirstOrDefault(r => this._approvalSnapshots.ContainsKey(r.RequestId)
&& IsCorrelatedResponse(r.RequestId, responseRequestId))
?? approvals.FirstOrDefault(r => IsCorrelatedResponse(r.RequestId, responseRequestId));

if (approval is not null)
{
Expand All @@ -177,12 +187,17 @@ public async ValueTask CaptureResponseAsync(
else if (this._approvalSnapshots.TryRemove(approval.RequestId, out ApprovalSnapshot? snapshot))
{
matchingResult = await this.InvokeRegisteredFunctionAsync(approval.RequestId, snapshot, cancellationToken).ConfigureAwait(false);
approvedFunctionResult = matchingResult;
}
else
{
await this.AssignErrorAsync(context, "No pending approval matched the response.").ConfigureAwait(false);
}
}
else if (!this._approvalSnapshots.IsEmpty)
{
await this.AssignErrorAsync(context, "No pending approval matched the response.").ConfigureAwait(false);
}
}

if (matchingResult is not null)
Expand All @@ -196,16 +211,15 @@ public async ValueTask CaptureResponseAsync(
AgentResponse resultResponse = new([new ChatMessage(ChatRole.Tool, [matchingResult])]);
await context.AddEventAsync(new AgentResponseEvent(this.Id, resultResponse), cancellationToken).ConfigureAwait(false);
}

// Drop the per-invocation entry now that the response has been processed.
this._pendingNonApprovalCallIds.TryRemove(matchingResult.CallId, out _);
this._approvalSnapshots.TryRemove(matchingResult.CallId, out _);
}

// Store messages if output path is configured
if (this.Model.Output?.Messages is not null)
{
await this.AssignAsync(this.Model.Output.Messages?.Path, response.Messages.ToFormula(), context).ConfigureAwait(false);
await this.AssignAsync(
this.Model.Output.Messages?.Path,
this.GetSideEffectMessages(response).ToFormula(),
context).ConfigureAwait(false);
}

// Add messages to conversation if conversationId is provided
Expand All @@ -214,37 +228,60 @@ public async ValueTask CaptureResponseAsync(
// actual AI-generated tool calls and would be rejected by the API.
if (conversationId is not null)
{
foreach (ChatMessage message in TransformConversationMessages(response.Messages))
IEnumerable<ChatMessage> sideEffectMessages = this.GetSideEffectMessages(response);
if (approvedFunctionResult is not null)
{
sideEffectMessages = sideEffectMessages.Append(new ChatMessage(ChatRole.Tool, [approvedFunctionResult]));
}

foreach (ChatMessage message in TransformConversationMessages(sideEffectMessages))
{
await agentProvider.CreateMessageAsync(conversationId, message, cancellationToken).ConfigureAwait(false);
}
}

if (matchingResult is not null)
{
// Drop the per-invocation entry now that the response side effects have been processed.
this._pendingNonApprovalCallIds.TryRemove(matchingResult.CallId, out _);
this._approvalSnapshots.TryRemove(matchingResult.CallId, out _);
}

// Completes the action after processing the function result.
await context.RaiseCompletionEventAsync(this.Model, cancellationToken).ConfigureAwait(false);
}

private bool IsKnownPendingId(string callId) =>
this._pendingNonApprovalCallIds.ContainsKey(callId) || this._approvalSnapshots.ContainsKey(callId);
private IEnumerable<ChatMessage> GetSideEffectMessages(ExternalInputResponse response)
{
if (!this._hasApprovalRequiredInvocation)
{
return response.Messages;
}

return this.FilterRejectedApprovalResults(response);
}

/// <inheritdoc/>
public override ValueTask ResetAsync()
{
this._approvalSnapshots.Clear();
this._hasApprovalRequiredInvocation = false;
this._pendingNonApprovalCallIds.Clear();
return default;
}

/// <inheritdoc/>
/// <remarks>
/// Persists pending approval snapshots and non-approval call ids so they survive
/// checkpoint/restore cycles.
/// Persists approval classification, pending approval snapshots, and non-approval
/// call ids so they survive checkpoint/restore cycles.
/// </remarks>
protected override async ValueTask OnCheckpointingAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
{
Dictionary<string, ApprovalSnapshot> snapshotCopy = this._approvalSnapshots.ToDictionary(kvp => kvp.Key, kvp => kvp.Value, StringComparer.Ordinal);
await context.QueueStateUpdateAsync(ApprovalSnapshotStateKey, snapshotCopy, null, cancellationToken).ConfigureAwait(false);

await context.QueueStateUpdateAsync(HasApprovalRequiredInvocationStateKey, this._hasApprovalRequiredInvocation, null, cancellationToken).ConfigureAwait(false);

List<string> pendingCopy = [.. this._pendingNonApprovalCallIds.Keys];
await context.QueueStateUpdateAsync(PendingCallIdsStateKey, pendingCopy, null, cancellationToken).ConfigureAwait(false);

Expand All @@ -259,7 +296,6 @@ protected override async ValueTask OnCheckpointingAsync(IWorkflowContext context
protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext context, CancellationToken cancellationToken = default)
{
await base.OnCheckpointRestoredAsync(context, cancellationToken).ConfigureAwait(false);

this._approvalSnapshots.Clear();
Dictionary<string, ApprovalSnapshot>? snapshots = await context.ReadStateAsync<Dictionary<string, ApprovalSnapshot>>(
ApprovalSnapshotStateKey, null, cancellationToken).ConfigureAwait(false);
Expand All @@ -271,6 +307,10 @@ protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext co
}
}

this._hasApprovalRequiredInvocation =
await context.ReadStateAsync<bool>(HasApprovalRequiredInvocationStateKey, null, cancellationToken).ConfigureAwait(false)
|| !this._approvalSnapshots.IsEmpty;

this._pendingNonApprovalCallIds.Clear();
List<string>? pending = await context.ReadStateAsync<List<string>>(
PendingCallIdsStateKey, null, cancellationToken).ConfigureAwait(false);
Expand All @@ -289,6 +329,7 @@ protected override async ValueTask OnCheckpointRestoredAsync(IWorkflowContext co
if (legacy is not null)
{
this._approvalSnapshots.TryAdd(this.Id, legacy);
this._hasApprovalRequiredInvocation = true;
await context.QueueStateUpdateAsync<ApprovalSnapshot?>(
LegacyApprovalSnapshotStateKey, null, null, cancellationToken).ConfigureAwait(false);
}
Expand Down Expand Up @@ -341,6 +382,41 @@ private static IEnumerable<ChatMessage> TransformConversationMessages(IEnumerabl
}
}

private IEnumerable<ChatMessage> FilterRejectedApprovalResults(ExternalInputResponse response)
{
foreach (ChatMessage message in response.Messages)
{
List<AIContent> contents =
[.. message.Contents.Where(c => c is not ToolApprovalResponseContent
&& (c is not FunctionResultContent functionResult
|| this.IsCorrelatedPendingNonApprovalResult(functionResult, response.RequestId)))];
if (contents.Count == message.Contents.Count)
{
yield return message;
}
else if (contents.Count > 0)
{
yield return new ChatMessage
{
Role = message.Role,
AuthorName = message.AuthorName,
Contents = contents,
MessageId = message.MessageId,
CreatedAt = message.CreatedAt,
RawRepresentation = message.RawRepresentation,
AdditionalProperties = message.AdditionalProperties,
};
}
}
}

private bool IsCorrelatedPendingNonApprovalResult(FunctionResultContent functionResult, string? responseRequestId) =>
this._pendingNonApprovalCallIds.ContainsKey(functionResult.CallId)
&& IsCorrelatedResponse(functionResult.CallId, responseRequestId);

private static bool IsCorrelatedResponse(string contentId, string? responseRequestId) =>
responseRequestId is null || string.Equals(contentId, responseRequestId, StringComparison.Ordinal);
Comment thread
baywet marked this conversation as resolved.
Resolved

private async ValueTask AssignResultAsync(IWorkflowContext context, FunctionResultContent result)
{
if (this.Model.Output?.Result is null)
Expand Down
Original file line number Diff line number Diff line change
@@ -1 +1,2 @@
#nullable enable
Microsoft.Agents.AI.Workflows.Declarative.Events.ExternalInputResponse.RequestId.get -> string?
Original file line number Diff line number Diff line change
@@ -1 +1,2 @@
#nullable enable
Microsoft.Agents.AI.Workflows.Declarative.Events.ExternalInputResponse.RequestId.get -> string?
Original file line number Diff line number Diff line change
@@ -1 +1,2 @@
#nullable enable
Microsoft.Agents.AI.Workflows.Declarative.Events.ExternalInputResponse.RequestId.get -> string?
Original file line number Diff line number Diff line change
@@ -1 +1,2 @@
#nullable enable
Microsoft.Agents.AI.Workflows.Declarative.Events.ExternalInputResponse.RequestId.get -> string?
Original file line number Diff line number Diff line change
@@ -1 +1,2 @@
#nullable enable
Microsoft.Agents.AI.Workflows.Declarative.Events.ExternalInputResponse.RequestId.get -> string?
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,8 @@ public static ExternalRequest Create(RequestPort port, [NotNull] object data, st
/// <exception cref="InvalidOperationException">Thrown when the input data object does not match the expected response type.</exception>
public ExternalResponse CreateResponse(object data)
{
if (!Throw.IfNull(this.PortInfo).ResponseType.IsMatchPolymorphic(Throw.IfNull(data).GetType()))
data = Throw.IfNull(data);
if (!Throw.IfNull(this.PortInfo).ResponseType.IsMatchPolymorphic(data.GetType()))
{
throw new InvalidOperationException(
$"Message type {data.GetType().Name} does not match expected response type {this.PortInfo.ResponseType.TypeName} of input port {this.PortInfo.PortId}.");
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,8 @@ async ValueTask PrepareExternalDeliveryAsync()
$"Response port id '{response.PortInfo.PortId}' does not match the originating port id for request {response.RequestId}.");
}

response = pendingRequest.RewrapResponse(response);

// Consume only after validation so a rejected response leaves the legitimate one able to complete.
if (!this.CompleteRequest(response.RequestId))
{
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,7 @@ private async Task RunInvokeFunctionToolTestAsync(
string workflowPath = GetWorkflowPath(workflowFileName);
IEnumerable<AIFunction> functionTools = new MenuPlugin().GetTools();
Dictionary<string, AIFunction> functionMap = functionTools.ToDictionary(tool => tool.Name, tool => tool);
DeclarativeWorkflowOptions workflowOptions = await this.CreateOptionsAsync(externalConversation: false);
DeclarativeWorkflowOptions workflowOptions = await this.CreateOptionsAsync(externalConversation: false, functionTools);
Workflow workflow = DeclarativeWorkflowBuilder.Build<string>(workflowPath, workflowOptions);

WorkflowHarness harness = new(workflow, runId: Path.GetFileNameWithoutExtension(workflowPath));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -67,11 +67,12 @@ public void ExternalInputRequest_RoundTrip_WithApprovalAndFunctionCallContent()
[Fact]
public void ExternalInputResponse_RoundTrip()
{
ExternalInputResponse copy = RoundTrip(new ExternalInputResponse(new ChatMessage(ChatRole.User, "ok")));
ExternalInputResponse copy = RoundTrip(new ExternalInputResponse([new ChatMessage(ChatRole.User, "ok")], "request-1"));

ChatMessage messageCopy = Assert.Single(copy.Messages);
Assert.Equal(ChatRole.User, messageCopy.Role);
Assert.Equal("ok", messageCopy.Text);
Assert.Equal("request-1", copy.RequestId);
}

[Fact]
Expand Down
Loading
Loading