mirror of
https://github.com/davidkaya/aryx.git
synced 2026-07-30 08:28:48 +02:00
refactor: generalize sidecar workflow runner
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
David Kaya
+5
-3
@@ -141,13 +141,15 @@ Workflows describe how agents collaborate. The architecture supports:
|
||||
|
||||
Their runtime semantics follow the Agent Framework orchestration model: sequential and group chat preserve a visible shared conversation, concurrent aggregates multiple independent responses into one turn, and handoff turns can end once the active agent has responded and is waiting for the next user input.
|
||||
|
||||
For Copilot-backed agents, Aryx uses a repo-local adapter around the Copilot SDK session layer so workflow agent routes still behave like Agent Framework handoffs. This is necessary because the upstream `GitHubCopilotAgent` does not currently project run-time handoff tool declarations into Copilot sessions or surface Copilot tool requests back as `FunctionCallContent` for the workflow runtime.
|
||||
For Copilot-backed agents, Aryx uses a repo-local provider module around the Copilot SDK session layer so workflow agent routes still behave like Agent Framework handoffs. This is necessary because the upstream `GitHubCopilotAgent` does not currently project run-time handoff tool declarations into Copilot sessions or surface Copilot tool requests back as `FunctionCallContent` for the workflow runtime.
|
||||
|
||||
The sidecar now keeps that Copilot-specific behavior behind provider seams. Core execution uses shared `IAgentProvider`, `IProviderTurnSupport`, `ProviderSessionEvent`, `ProviderAgentBundle`, `TurnExecutionState`, and `AgentWorkflowTurnRunner` abstractions, while `Services/Providers/Copilot/` owns SDK-specific bundle creation, transcript projection, approvals, user input, MCP OAuth, exit-plan-mode handling, CLI/session management, and event adaptation.
|
||||
|
||||
Workflows are shared application data, not renderer-only configuration. The same workflow definition now drives validation, persistence, session execution, and sidecar orchestration.
|
||||
|
||||
Each workflow persists an explicit graph-backed topology. Agent nodes carry stable ids, ordering, and layout metadata, while start/end, fan-out/fan-in, sub-workflow, function, and request-port nodes make execution structure visible in the saved contract.
|
||||
|
||||
That graph remains the execution contract for the sidecar, but orchestration mode is now a first-class backend concept. Graph-based modes (`single`, `sequential`, `concurrent`) still execute directly from saved edges. Builder-based modes (`handoff`, `group-chat`) additionally persist mode-specific `settings.modeSettings` data for handoff filtering, triage selection, return behavior, and group-chat round limits, and the sidecar translates those settings into specialized Agent Framework workflow builders at run time.
|
||||
That graph remains the execution contract for the sidecar, but orchestration mode is now a first-class backend concept. Graph-based modes (`single`, `sequential`, `concurrent`) still execute directly from saved edges. Builder-based modes (`handoff`, `group-chat`) additionally persist mode-specific `settings.modeSettings` data for handoff filtering, triage selection, return behavior, and group-chat round limits, and the sidecar translates those settings into specialized Agent Framework workflow builders at run time through shared orchestration helpers rather than Copilot-specific workflow code.
|
||||
|
||||
Workflow templates remain a first-class shared-domain contract. The shared layer owns workflow definitions, workflow template definitions, and workflow import/export helpers (YAML import/export plus Mermaid and DOT export). Built-in workflows seed workspace state directly, while built-in and custom templates let the main process create additional saved workflows without expanding the sidecar protocol.
|
||||
|
||||
@@ -214,7 +216,7 @@ This is a structured stdio protocol used for:
|
||||
- streaming partial output
|
||||
- streaming agent activity
|
||||
|
||||
This protocol boundary keeps the AI execution runtime replaceable and prevents the Electron main process from becoming overloaded with workflow-specific behavior.
|
||||
This protocol boundary keeps the AI execution runtime replaceable and prevents the Electron main process from becoming overloaded with workflow-specific behavior. On the sidecar side, raw provider events are first normalized into sidecar-owned provider event records before they become streamed run activity, so future providers can plug into the same transport without reshaping the main-process contract.
|
||||
|
||||
The protocol also carries **turn-scoped lifecycle events** alongside output deltas. These events let the UI visualize execution internals without the main process having to interpret AI workflow semantics:
|
||||
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
|
||||
namespace Aryx.AgentHost.Services;
|
||||
|
||||
internal static class AgentHostOptionsFactory
|
||||
{
|
||||
public static AIAgentHostOptions CreateDefault()
|
||||
{
|
||||
return new AIAgentHostOptions
|
||||
{
|
||||
EmitAgentUpdateEvents = null,
|
||||
EmitAgentResponseEvents = false,
|
||||
InterceptUserInputRequests = false,
|
||||
InterceptUnterminatedFunctionCalls = false,
|
||||
ReassignOtherAgentsAsUsers = true,
|
||||
ForwardIncomingMessages = true,
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,763 @@
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Globalization;
|
||||
using System.Text.Json;
|
||||
using Aryx.AgentHost.Contracts;
|
||||
using GitHub.Copilot.SDK;
|
||||
using Microsoft.Agents.AI;
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
using Microsoft.Agents.AI.Workflows.Checkpointing;
|
||||
using Microsoft.Agents.AI.Workflows.InProc;
|
||||
using Microsoft.Extensions.AI;
|
||||
|
||||
namespace Aryx.AgentHost.Services;
|
||||
|
||||
public class AgentWorkflowTurnRunner : ITurnWorkflowRunner
|
||||
{
|
||||
private const string HandoffFunctionPrefix = "handoff_to_";
|
||||
private readonly WorkflowValidator _workflowValidator;
|
||||
private readonly WorkflowRunner _workflowRunner = new();
|
||||
private readonly IProviderTurnSupport _providerTurnSupport;
|
||||
|
||||
internal AgentWorkflowTurnRunner(
|
||||
IProviderTurnSupport providerTurnSupport,
|
||||
WorkflowValidator? workflowValidator = null)
|
||||
{
|
||||
_providerTurnSupport = providerTurnSupport ?? throw new ArgumentNullException(nameof(providerTurnSupport));
|
||||
_workflowValidator = workflowValidator ?? new WorkflowValidator();
|
||||
}
|
||||
|
||||
public async Task<IReadOnlyList<ChatMessageDto>> RunTurnAsync(
|
||||
RunTurnCommandDto command,
|
||||
Func<TurnDeltaEventDto, Task> onDelta,
|
||||
Func<SidecarEventDto, Task> onEvent,
|
||||
Func<ApprovalRequestedEventDto, Task> onApproval,
|
||||
Func<UserInputRequestedEventDto, Task> onUserInput,
|
||||
Func<McpOauthRequiredEventDto, Task> onMcpOAuthRequired,
|
||||
Func<ExitPlanModeRequestedEventDto, Task> onExitPlanMode,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
string? validationError = _workflowValidator.Validate(command.Workflow, command.WorkflowLibrary)
|
||||
.FirstOrDefault()?.Message;
|
||||
if (validationError is not null)
|
||||
{
|
||||
throw new InvalidOperationException(validationError);
|
||||
}
|
||||
|
||||
TurnExecutionState state = new(command);
|
||||
using CancellationTokenSource runCancellation =
|
||||
CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
IProviderTranscriptProjector? transcriptProjector = null;
|
||||
|
||||
try
|
||||
{
|
||||
await using ProviderAgentBundle bundle = await _providerTurnSupport.CreateAgentBundleAsync(
|
||||
command,
|
||||
state,
|
||||
onEvent,
|
||||
onApproval,
|
||||
onUserInput,
|
||||
runCancellation,
|
||||
runCancellation.Token)
|
||||
.ConfigureAwait(false);
|
||||
transcriptProjector = bundle.TranscriptProjector;
|
||||
ConfigureHookLifecycleEventSuppression(state, bundle);
|
||||
Workflow workflow = BuildWorkflowForCommand(command, bundle.Agents, _workflowRunner);
|
||||
List<ChatMessage> inputMessages = command.Messages.Select(transcriptProjector.ToChatMessage).ToList();
|
||||
transcriptProjector.AttachMessageMode(inputMessages, command.MessageMode);
|
||||
|
||||
using FileSystemJsonCheckpointStore? checkpointStore = CreateCheckpointStore(command);
|
||||
CheckpointManager? checkpointManager = checkpointStore is not null
|
||||
? CheckpointManager.CreateJson(checkpointStore)
|
||||
: null;
|
||||
|
||||
await using StreamingRun run = await OpenWorkflowRunAsync(
|
||||
command,
|
||||
workflow,
|
||||
inputMessages,
|
||||
checkpointManager).ConfigureAwait(false);
|
||||
await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false);
|
||||
|
||||
await foreach (WorkflowEvent evt in run.WatchStreamAsync(runCancellation.Token).ConfigureAwait(false))
|
||||
{
|
||||
if (evt is RequestInfoEvent requestInfo
|
||||
&& await TryHandleRequestPortRequestAsync(
|
||||
command,
|
||||
requestInfo,
|
||||
run,
|
||||
onUserInput,
|
||||
runCancellation.Token).ConfigureAwait(false))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
bool shouldEndTurn = await HandleWorkflowEventAsync(
|
||||
command,
|
||||
evt,
|
||||
inputMessages,
|
||||
state,
|
||||
transcriptProjector,
|
||||
onDelta,
|
||||
onEvent)
|
||||
.ConfigureAwait(false);
|
||||
await EmitPendingEventsAsync(state, onEvent).ConfigureAwait(false);
|
||||
await EmitPendingMcpOauthRequestsAsync(state, onMcpOAuthRequired).ConfigureAwait(false);
|
||||
if (shouldEndTurn)
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
await EmitPendingEventsAsync(state, onEvent).ConfigureAwait(false);
|
||||
await EmitPendingMcpOauthRequestsAsync(state, onMcpOAuthRequired).ConfigureAwait(false);
|
||||
return state.FinalizeCompletedMessages(transcriptProjector);
|
||||
}
|
||||
catch (OperationCanceledException) when (runCancellation.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
await EmitPendingEventsAsync(state, onEvent).ConfigureAwait(false);
|
||||
await EmitPendingMcpOauthRequestsAsync(state, onMcpOAuthRequired).ConfigureAwait(false);
|
||||
ExitPlanModeRequestedEventDto? exitPlanModeEvent =
|
||||
_providerTurnSupport.ConsumePendingExitPlanModeRequest(command.RequestId);
|
||||
if (exitPlanModeEvent is null || !state.HasPendingExitPlanModeRequest)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
|
||||
await onExitPlanMode(exitPlanModeEvent).ConfigureAwait(false);
|
||||
return state.FinalizeCompletedMessages(
|
||||
transcriptProjector ?? throw new InvalidOperationException("Provider transcript projector was not initialized."));
|
||||
}
|
||||
finally
|
||||
{
|
||||
_providerTurnSupport.ClearRequestState(command.RequestId);
|
||||
}
|
||||
}
|
||||
|
||||
internal static Workflow BuildWorkflowForCommand(
|
||||
RunTurnCommandDto command,
|
||||
IReadOnlyList<AIAgent> agents,
|
||||
WorkflowRunner? workflowRunner = null)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(command);
|
||||
ArgumentNullException.ThrowIfNull(agents);
|
||||
|
||||
return NormalizeOrchestrationMode(command.Workflow.Settings.OrchestrationMode) switch
|
||||
{
|
||||
"handoff" => WorkflowOrchestrationFactory.CreateHandoffWorkflow(command.Workflow, agents),
|
||||
"group-chat" => WorkflowOrchestrationFactory.CreateGroupChatWorkflow(command.Workflow, agents),
|
||||
_ => (workflowRunner ?? new WorkflowRunner()).BuildWorkflow(command.Workflow, agents, command.WorkflowLibrary),
|
||||
};
|
||||
}
|
||||
|
||||
internal static FileSystemJsonCheckpointStore? CreateCheckpointStore(RunTurnCommandDto command)
|
||||
{
|
||||
if (!ShouldEnableWorkflowCheckpointing(command))
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
DirectoryInfo checkpointDirectory = new(GetCheckpointStorePath(command));
|
||||
return new FileSystemJsonCheckpointStore(checkpointDirectory);
|
||||
}
|
||||
|
||||
internal static bool ShouldEnableWorkflowCheckpointing(RunTurnCommandDto command)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(command);
|
||||
return command.Workflow.Settings.Checkpointing.Enabled;
|
||||
}
|
||||
|
||||
internal static string GetCheckpointStorePath(RunTurnCommandDto command)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(command);
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(command.ResumeFromCheckpoint?.StorePath))
|
||||
{
|
||||
return command.ResumeFromCheckpoint.StorePath;
|
||||
}
|
||||
|
||||
string localAppData = Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData);
|
||||
return Path.Combine(localAppData, "Aryx", "workflow-checkpoints", command.SessionId, command.RequestId);
|
||||
}
|
||||
|
||||
private static string? NormalizeOrchestrationMode(string? value)
|
||||
{
|
||||
return string.IsNullOrWhiteSpace(value) ? null : value.Trim().ToLowerInvariant();
|
||||
}
|
||||
|
||||
private static ValueTask<StreamingRun> OpenWorkflowRunAsync(
|
||||
RunTurnCommandDto command,
|
||||
Workflow workflow,
|
||||
IReadOnlyList<ChatMessage> inputMessages,
|
||||
CheckpointManager? checkpointManager)
|
||||
{
|
||||
InProcessExecutionEnvironment environment = CreateExecutionEnvironment(command, checkpointManager);
|
||||
if (checkpointManager is not null && command.ResumeFromCheckpoint is { } resumeFromCheckpoint)
|
||||
{
|
||||
return environment.ResumeStreamingAsync(
|
||||
workflow,
|
||||
new CheckpointInfo(resumeFromCheckpoint.WorkflowSessionId, resumeFromCheckpoint.CheckpointId));
|
||||
}
|
||||
|
||||
return environment.RunStreamingAsync(workflow, inputMessages.ToList(), command.RequestId);
|
||||
}
|
||||
|
||||
internal static InProcessExecutionEnvironment CreateExecutionEnvironment(
|
||||
RunTurnCommandDto command,
|
||||
CheckpointManager? checkpointManager)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(command);
|
||||
|
||||
string executionMode = command.Workflow.Settings.ExecutionMode?.Trim() ?? "off-thread";
|
||||
InProcessExecutionEnvironment environment = string.Equals(
|
||||
executionMode,
|
||||
"lockstep",
|
||||
StringComparison.OrdinalIgnoreCase)
|
||||
? InProcessExecution.Lockstep
|
||||
: InProcessExecution.OffThread;
|
||||
|
||||
return checkpointManager is null ? environment : environment.WithCheckpointing(checkpointManager);
|
||||
}
|
||||
|
||||
internal static void ConfigureHookLifecycleEventSuppression(
|
||||
TurnExecutionState state,
|
||||
ProviderAgentBundle bundle)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(state);
|
||||
ArgumentNullException.ThrowIfNull(bundle);
|
||||
|
||||
state.SuppressHookLifecycleEvents = !bundle.HasConfiguredHooks;
|
||||
}
|
||||
|
||||
private static async Task EmitPendingEventsAsync(
|
||||
TurnExecutionState state,
|
||||
Func<SidecarEventDto, Task> onEvent)
|
||||
{
|
||||
foreach (SidecarEventDto pendingEvent in state.DrainPendingEvents())
|
||||
{
|
||||
await onEvent(pendingEvent).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task EmitPendingMcpOauthRequestsAsync(
|
||||
TurnExecutionState state,
|
||||
Func<McpOauthRequiredEventDto, Task> onMcpOAuthRequired)
|
||||
{
|
||||
foreach (McpOauthRequiredEventDto request in state.DrainPendingMcpOauthRequests())
|
||||
{
|
||||
await onMcpOAuthRequired(request).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
public Task ResolveApprovalAsync(
|
||||
ResolveApprovalCommandDto command,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return _providerTurnSupport.ResolveApprovalAsync(command, cancellationToken);
|
||||
}
|
||||
|
||||
public Task ResolveUserInputAsync(
|
||||
ResolveUserInputCommandDto command,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return _providerTurnSupport.ResolveUserInputAsync(command, cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<bool> TryHandleRequestPortRequestAsync(
|
||||
RunTurnCommandDto command,
|
||||
RequestInfoEvent requestInfo,
|
||||
StreamingRun run,
|
||||
Func<UserInputRequestedEventDto, Task> onUserInput,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (!TryResolveRequestPortMetadata(command.Workflow, requestInfo, out WorkflowRequestPortMetadata? metadata))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
UserInputRequest userInputRequest = CreateRequestPortUserInputRequest(metadata!, requestInfo);
|
||||
UserInputResponse response = await _providerTurnSupport.RequestRequestPortUserInputAsync(
|
||||
command,
|
||||
userInputRequest,
|
||||
onUserInput,
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
|
||||
object coercedResponse = CoerceRequestPortResponse(metadata!.ResponseType, response.Answer);
|
||||
await run.SendResponseAsync(requestInfo.Request.CreateResponse(coercedResponse)).ConfigureAwait(false);
|
||||
return true;
|
||||
}
|
||||
|
||||
internal static async Task<bool> HandleWorkflowEventAsync(
|
||||
RunTurnCommandDto command,
|
||||
WorkflowEvent evt,
|
||||
IReadOnlyList<ChatMessage> inputMessages,
|
||||
TurnExecutionState state,
|
||||
IProviderTranscriptProjector transcriptProjector,
|
||||
Func<TurnDeltaEventDto, Task> onDelta,
|
||||
Func<SidecarEventDto, Task> onEvent)
|
||||
{
|
||||
if (evt is ExecutorInvokedEvent invoked)
|
||||
{
|
||||
if (AgentIdentityResolver.TryResolveKnownAgentIdentity(
|
||||
command.Workflow,
|
||||
invoked.ExecutorId,
|
||||
out AgentIdentity invokedAgent))
|
||||
{
|
||||
TraceHandoff(command, $"Executor invoked: {invoked.ExecutorId} -> {invokedAgent.AgentName} ({invokedAgent.AgentId}).");
|
||||
await state.EmitThinkingIfNeeded(invokedAgent, onEvent).ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
TraceHandoff(command, $"Executor invoked without a known agent match: {invoked.ExecutorId}.");
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
if (evt is RequestInfoEvent requestInfo)
|
||||
{
|
||||
AgentActivityEventDto? activity = WorkflowRequestInfoInterpreter.TryCreateActivityFromRequest(
|
||||
command,
|
||||
requestInfo,
|
||||
state.ActiveAgent,
|
||||
state.ToolNamesByCallId);
|
||||
|
||||
if (activity is null)
|
||||
{
|
||||
bool requiresBoundary = WorkflowRequestInfoInterpreter.RequiresUserInputTurnBoundary(command, requestInfo);
|
||||
TraceHandoff(
|
||||
command,
|
||||
$"Request info produced no activity for data type '{requestInfo.Request.Data.TypeId}'. Requires boundary: {requiresBoundary}.");
|
||||
return requiresBoundary;
|
||||
}
|
||||
|
||||
await EmitActivityAsync(command, state, activity, onEvent).ConfigureAwait(false);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (TryCreateWorkflowCheckpointSavedEvent(command, evt, out WorkflowCheckpointSavedEventDto? checkpointSaved))
|
||||
{
|
||||
await onEvent(checkpointSaved).ConfigureAwait(false);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (TryCreateWorkflowDiagnosticEvent(command, evt, state, out WorkflowDiagnosticEventDto? diagnostic))
|
||||
{
|
||||
await onEvent(diagnostic).ConfigureAwait(false);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (evt is AgentResponseUpdateEvent update)
|
||||
{
|
||||
await HandleAgentResponseUpdateAsync(command, update, state, onDelta, onEvent).ConfigureAwait(false);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (evt is ExecutorCompletedEvent completed)
|
||||
{
|
||||
if (AgentIdentityResolver.TryResolveObservedAgentIdentity(
|
||||
command.Workflow,
|
||||
completed.ExecutorId,
|
||||
state.ActiveAgent,
|
||||
out AgentIdentity completedAgent))
|
||||
{
|
||||
TraceHandoff(command, $"Executor completed: {completed.ExecutorId} -> {completedAgent.AgentName} ({completedAgent.AgentId}).");
|
||||
state.QueueCompletedActivity(completedAgent);
|
||||
state.ClearActiveAgentIfMatching(completedAgent);
|
||||
}
|
||||
else
|
||||
{
|
||||
TraceHandoff(command, $"Executor completed without a known agent match: {completed.ExecutorId}.");
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
if (evt is WorkflowOutputEvent outputEvent)
|
||||
{
|
||||
List<ChatMessage> allMessages = outputEvent.As<List<ChatMessage>>() ?? [];
|
||||
state.UpdateCompletedMessages(allMessages, inputMessages, transcriptProjector);
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
internal static UserInputRequest CreateRequestPortUserInputRequest(
|
||||
WorkflowRequestPortMetadata metadata,
|
||||
RequestInfoEvent requestInfo)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(metadata);
|
||||
ArgumentNullException.ThrowIfNull(requestInfo);
|
||||
|
||||
string question = metadata.Prompt
|
||||
?? BuildRequestPortFallbackQuestion(metadata, requestInfo);
|
||||
|
||||
bool expectsBoolean = IsBooleanResponseType(metadata.ResponseType);
|
||||
return new UserInputRequest
|
||||
{
|
||||
Question = question,
|
||||
Choices = expectsBoolean ? ["true", "false"] : null,
|
||||
AllowFreeform = true,
|
||||
};
|
||||
}
|
||||
|
||||
internal static object CoerceRequestPortResponse(string responseType, string? answer)
|
||||
{
|
||||
string normalizedResponseType = responseType.Trim();
|
||||
string trimmedAnswer = answer?.Trim() ?? string.Empty;
|
||||
|
||||
if (IsStringResponseType(normalizedResponseType))
|
||||
{
|
||||
return trimmedAnswer;
|
||||
}
|
||||
|
||||
if (IsBooleanResponseType(normalizedResponseType))
|
||||
{
|
||||
return trimmedAnswer.ToLowerInvariant() switch
|
||||
{
|
||||
"true" or "t" or "yes" or "y" or "1" => true,
|
||||
"false" or "f" or "no" or "n" or "0" => false,
|
||||
_ => throw new InvalidOperationException(
|
||||
$"Request port response type \"{responseType}\" requires a boolean answer."),
|
||||
};
|
||||
}
|
||||
|
||||
if (IsNumericResponseType(normalizedResponseType))
|
||||
{
|
||||
if (double.TryParse(trimmedAnswer, NumberStyles.Float, CultureInfo.InvariantCulture, out double numeric))
|
||||
{
|
||||
return numeric;
|
||||
}
|
||||
|
||||
throw new InvalidOperationException(
|
||||
$"Request port response type \"{responseType}\" requires a numeric answer.");
|
||||
}
|
||||
|
||||
if (IsJsonResponseType(normalizedResponseType))
|
||||
{
|
||||
try
|
||||
{
|
||||
return JsonDocument.Parse(trimmedAnswer).RootElement.Clone();
|
||||
}
|
||||
catch (JsonException ex)
|
||||
{
|
||||
throw new InvalidOperationException(
|
||||
$"Request port response type \"{responseType}\" requires a valid JSON answer.",
|
||||
ex);
|
||||
}
|
||||
}
|
||||
|
||||
return trimmedAnswer;
|
||||
}
|
||||
|
||||
private static async Task HandleAgentResponseUpdateAsync(
|
||||
RunTurnCommandDto command,
|
||||
AgentResponseUpdateEvent update,
|
||||
TurnExecutionState state,
|
||||
Func<TurnDeltaEventDto, Task> onDelta,
|
||||
Func<SidecarEventDto, Task> onEvent)
|
||||
{
|
||||
AgentIdentity? updateAgent = null;
|
||||
string authorName = update.ExecutorId;
|
||||
string[] handoffFunctionCalls = update.Update.Contents
|
||||
.OfType<FunctionCallContent>()
|
||||
.Select(content => content.Name)
|
||||
.Where(IsHandoffFunctionName)
|
||||
.Distinct(StringComparer.Ordinal)
|
||||
.ToArray();
|
||||
if (state.TryResolveObservedAgentForMessage(update.Update.MessageId, out AgentIdentity observedMessageAgent))
|
||||
{
|
||||
updateAgent = observedMessageAgent;
|
||||
authorName = observedMessageAgent.AgentName;
|
||||
}
|
||||
else if (AgentIdentityResolver.TryResolveObservedAgentIdentity(
|
||||
command.Workflow,
|
||||
update.ExecutorId,
|
||||
state.ActiveAgent,
|
||||
out AgentIdentity resolvedUpdateAgent))
|
||||
{
|
||||
updateAgent = resolvedUpdateAgent;
|
||||
authorName = resolvedUpdateAgent.AgentName;
|
||||
}
|
||||
else if (state.ActiveAgent is AgentIdentity activeAgent)
|
||||
{
|
||||
updateAgent = activeAgent;
|
||||
authorName = activeAgent.AgentName;
|
||||
TraceHandoff(
|
||||
command,
|
||||
$"Agent response update fell back to active agent {activeAgent.AgentName} ({activeAgent.AgentId}) for executor '{update.ExecutorId}' and message '{update.Update.MessageId ?? "<none>"}'.");
|
||||
}
|
||||
|
||||
if (updateAgent.HasValue)
|
||||
{
|
||||
if (handoffFunctionCalls.Length > 0)
|
||||
{
|
||||
TraceHandoff(
|
||||
command,
|
||||
$"Agent response update from {updateAgent.Value.AgentName} ({updateAgent.Value.AgentId}) requested handoff via {string.Join(", ", handoffFunctionCalls)}.");
|
||||
}
|
||||
|
||||
await state.EmitThinkingIfNeeded(updateAgent.Value, onEvent).ConfigureAwait(false);
|
||||
}
|
||||
else if (!string.IsNullOrEmpty(update.Update.Text) || handoffFunctionCalls.Length > 0)
|
||||
{
|
||||
TraceHandoff(
|
||||
command,
|
||||
$"Agent response update could not resolve agent for executor '{update.ExecutorId}' and message '{update.Update.MessageId ?? "<none>"}'.");
|
||||
}
|
||||
|
||||
if (string.IsNullOrEmpty(update.Update.Text))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
string messageId = state.CreateMessageId(update.Update.MessageId);
|
||||
(string _, string currentAuthorName, string currentContent) = state.AppendDelta(
|
||||
messageId,
|
||||
authorName,
|
||||
update.Update.Text);
|
||||
|
||||
await onDelta(new TurnDeltaEventDto
|
||||
{
|
||||
Type = "turn-delta",
|
||||
RequestId = command.RequestId,
|
||||
SessionId = command.SessionId,
|
||||
MessageId = messageId,
|
||||
AuthorName = currentAuthorName,
|
||||
ContentDelta = update.Update.Text,
|
||||
Content = currentContent,
|
||||
}).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
internal static async Task EmitActivityAsync(
|
||||
RunTurnCommandDto command,
|
||||
TurnExecutionState state,
|
||||
AgentActivityEventDto activity,
|
||||
Func<SidecarEventDto, Task> onEvent)
|
||||
{
|
||||
state.ApplyEvent(activity);
|
||||
TraceHandoff(
|
||||
command,
|
||||
$"Activity emitted: {activity.ActivityType} -> {activity.AgentName ?? activity.AgentId ?? "<unknown>"}.");
|
||||
await onEvent(activity).ConfigureAwait(false);
|
||||
|
||||
if (string.Equals(activity.ActivityType, "handoff", StringComparison.Ordinal)
|
||||
&& !string.IsNullOrWhiteSpace(activity.AgentId)
|
||||
&& !string.IsNullOrWhiteSpace(activity.AgentName))
|
||||
{
|
||||
TraceHandoff(
|
||||
command,
|
||||
$"Promoting handoff target to thinking: {activity.AgentName} ({activity.AgentId}).");
|
||||
await state.EmitThinkingIfNeeded(
|
||||
new AgentIdentity(activity.AgentId, activity.AgentName),
|
||||
onEvent).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
internal static bool TryCreateWorkflowCheckpointSavedEvent(
|
||||
RunTurnCommandDto command,
|
||||
WorkflowEvent evt,
|
||||
out WorkflowCheckpointSavedEventDto checkpointSaved)
|
||||
{
|
||||
checkpointSaved = default!;
|
||||
|
||||
if (!ShouldEnableWorkflowCheckpointing(command)
|
||||
|| evt is not SuperStepCompletedEvent superStepCompleted
|
||||
|| superStepCompleted.CompletionInfo?.Checkpoint is not CheckpointInfo checkpoint)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
checkpointSaved = new WorkflowCheckpointSavedEventDto
|
||||
{
|
||||
Type = "workflow-checkpoint-saved",
|
||||
RequestId = command.RequestId,
|
||||
SessionId = command.SessionId,
|
||||
WorkflowSessionId = checkpoint.SessionId,
|
||||
CheckpointId = checkpoint.CheckpointId,
|
||||
StorePath = GetCheckpointStorePath(command),
|
||||
StepNumber = superStepCompleted.StepNumber,
|
||||
};
|
||||
return true;
|
||||
}
|
||||
|
||||
private static bool TryCreateWorkflowDiagnosticEvent(
|
||||
RunTurnCommandDto command,
|
||||
WorkflowEvent evt,
|
||||
TurnExecutionState state,
|
||||
out WorkflowDiagnosticEventDto diagnostic)
|
||||
{
|
||||
diagnostic = default!;
|
||||
|
||||
switch (evt)
|
||||
{
|
||||
case ExecutorFailedEvent executorFailed:
|
||||
{
|
||||
AgentIdentity? agent = AgentIdentityResolver.TryResolveObservedAgentIdentity(
|
||||
command.Workflow,
|
||||
executorFailed.ExecutorId,
|
||||
state.ActiveAgent,
|
||||
out AgentIdentity resolvedAgent)
|
||||
? resolvedAgent
|
||||
: null;
|
||||
Exception? exception = executorFailed.Data;
|
||||
diagnostic = new WorkflowDiagnosticEventDto
|
||||
{
|
||||
Type = "workflow-diagnostic",
|
||||
RequestId = command.RequestId,
|
||||
SessionId = command.SessionId,
|
||||
Severity = "error",
|
||||
DiagnosticKind = "executor-failed",
|
||||
Message = ResolveDiagnosticMessage(exception, "Executor failed."),
|
||||
AgentId = agent?.AgentId,
|
||||
AgentName = agent?.AgentName,
|
||||
ExecutorId = executorFailed.ExecutorId,
|
||||
ExceptionType = exception?.GetBaseException().GetType().Name,
|
||||
};
|
||||
return true;
|
||||
}
|
||||
case WorkflowWarningEvent workflowWarning:
|
||||
diagnostic = new WorkflowDiagnosticEventDto
|
||||
{
|
||||
Type = "workflow-diagnostic",
|
||||
RequestId = command.RequestId,
|
||||
SessionId = command.SessionId,
|
||||
Severity = "warning",
|
||||
DiagnosticKind = workflowWarning is SubworkflowWarningEvent
|
||||
? "subworkflow-warning"
|
||||
: "workflow-warning",
|
||||
Message = ResolveDiagnosticMessage(workflowWarning.Data as string, "Workflow warning."),
|
||||
SubworkflowId = workflowWarning is SubworkflowWarningEvent subworkflowWarning
|
||||
? subworkflowWarning.SubWorkflowId
|
||||
: null,
|
||||
};
|
||||
return true;
|
||||
case WorkflowErrorEvent workflowError:
|
||||
{
|
||||
Exception? exception = workflowError.Exception;
|
||||
diagnostic = new WorkflowDiagnosticEventDto
|
||||
{
|
||||
Type = "workflow-diagnostic",
|
||||
RequestId = command.RequestId,
|
||||
SessionId = command.SessionId,
|
||||
Severity = "error",
|
||||
DiagnosticKind = workflowError is SubworkflowErrorEvent
|
||||
? "subworkflow-error"
|
||||
: "workflow-error",
|
||||
Message = ResolveDiagnosticMessage(exception, "Workflow failed."),
|
||||
SubworkflowId = workflowError is SubworkflowErrorEvent subworkflowError
|
||||
? subworkflowError.SubworkflowId
|
||||
: null,
|
||||
ExceptionType = exception?.GetBaseException().GetType().Name,
|
||||
};
|
||||
return true;
|
||||
}
|
||||
default:
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private static bool TryResolveRequestPortMetadata(
|
||||
WorkflowDefinitionDto? workflow,
|
||||
RequestInfoEvent requestInfo,
|
||||
out WorkflowRequestPortMetadata? metadata)
|
||||
{
|
||||
metadata = null;
|
||||
if (workflow is null)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
string portId = requestInfo.Request.PortInfo.PortId;
|
||||
WorkflowNodeDto? node = workflow.Graph.Nodes.FirstOrDefault(candidate =>
|
||||
string.Equals(candidate.Kind, "request-port", StringComparison.OrdinalIgnoreCase)
|
||||
&& string.Equals(candidate.Config.PortId, portId, StringComparison.OrdinalIgnoreCase));
|
||||
|
||||
if (node is null)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
metadata = new WorkflowRequestPortMetadata(
|
||||
node.Id,
|
||||
string.IsNullOrWhiteSpace(node.Label) ? node.Id : node.Label,
|
||||
node.Config.PortId ?? portId,
|
||||
node.Config.RequestType ?? string.Empty,
|
||||
node.Config.ResponseType ?? string.Empty,
|
||||
string.IsNullOrWhiteSpace(node.Config.Prompt) ? null : node.Config.Prompt.Trim());
|
||||
return true;
|
||||
}
|
||||
|
||||
private static string BuildRequestPortFallbackQuestion(
|
||||
WorkflowRequestPortMetadata metadata,
|
||||
RequestInfoEvent requestInfo)
|
||||
{
|
||||
if (requestInfo.Request.Data.Is<WorkflowRequestPortPromptRequest>(out WorkflowRequestPortPromptRequest? promptRequest))
|
||||
{
|
||||
string baseQuestion = $"Provide a {metadata.ResponseType} response for \"{promptRequest.NodeLabel}\".";
|
||||
if (!string.IsNullOrWhiteSpace(promptRequest.InputSummary))
|
||||
{
|
||||
return $"{baseQuestion} Current input: {promptRequest.InputSummary}";
|
||||
}
|
||||
|
||||
return baseQuestion;
|
||||
}
|
||||
|
||||
return $"Provide a {metadata.ResponseType} response for request port \"{metadata.NodeLabel}\" ({metadata.PortId}).";
|
||||
}
|
||||
|
||||
private static bool IsStringResponseType(string responseType)
|
||||
=> string.Equals(responseType, "string", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "text", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
private static bool IsBooleanResponseType(string responseType)
|
||||
=> string.Equals(responseType, "bool", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "boolean", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
private static bool IsNumericResponseType(string responseType)
|
||||
=> string.Equals(responseType, "number", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "int", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "float", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "double", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "decimal", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
private static bool IsJsonResponseType(string responseType)
|
||||
=> string.Equals(responseType, "json", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "object", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "array", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
internal sealed record WorkflowRequestPortMetadata(
|
||||
string NodeId,
|
||||
string NodeLabel,
|
||||
string PortId,
|
||||
string RequestType,
|
||||
string ResponseType,
|
||||
string? Prompt);
|
||||
|
||||
private static string ResolveDiagnosticMessage(Exception? exception, string fallback)
|
||||
{
|
||||
return ResolveDiagnosticMessage(
|
||||
exception?.GetBaseException().Message,
|
||||
fallback);
|
||||
}
|
||||
|
||||
private static string ResolveDiagnosticMessage(string? message, string fallback)
|
||||
{
|
||||
return string.IsNullOrWhiteSpace(message) ? fallback : message;
|
||||
}
|
||||
|
||||
private static bool IsHandoffFunctionName(string? candidate)
|
||||
{
|
||||
return !string.IsNullOrWhiteSpace(candidate)
|
||||
&& candidate.StartsWith(HandoffFunctionPrefix, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
private static void TraceHandoff(RunTurnCommandDto command, string message)
|
||||
{
|
||||
if (!command.Workflow.IsOrchestrationMode("handoff"))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
Console.Error.WriteLine($"[aryx handoff] {message}");
|
||||
}
|
||||
}
|
||||
@@ -1,669 +1,24 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Aryx.AgentHost.Contracts;
|
||||
using Microsoft.Extensions.AI;
|
||||
|
||||
namespace Aryx.AgentHost.Services;
|
||||
|
||||
internal sealed class CopilotTurnExecutionState
|
||||
internal sealed class CopilotTurnExecutionState : TurnExecutionState
|
||||
{
|
||||
private readonly RunTurnCommandDto _command;
|
||||
private readonly HashSet<string> _startedAgents = new(StringComparer.OrdinalIgnoreCase);
|
||||
private readonly HashSet<string> _reclassifiedMessageIds = new(StringComparer.Ordinal);
|
||||
private readonly ConcurrentQueue<SidecarEventDto> _pendingEvents = new();
|
||||
private readonly ConcurrentQueue<McpOauthRequiredEventDto> _pendingMcpOauthRequests = new();
|
||||
private readonly ConcurrentDictionary<string, AgentIdentity> _observedAgentsByMessageId = new(StringComparer.Ordinal);
|
||||
private readonly StreamingTranscriptBuffer _transcriptBuffer = new();
|
||||
private int _fallbackMessageIndex;
|
||||
private string? _lastObservedMessageId;
|
||||
|
||||
public CopilotTurnExecutionState(RunTurnCommandDto command)
|
||||
: base(command)
|
||||
{
|
||||
_command = command;
|
||||
}
|
||||
|
||||
public ConcurrentDictionary<string, string> ToolNamesByCallId { get; } = new(StringComparer.Ordinal);
|
||||
|
||||
public AgentIdentity? ActiveAgent { get; private set; }
|
||||
|
||||
public List<ChatMessageDto> CompletedMessages { get; private set; } = [];
|
||||
|
||||
public bool HasPendingExitPlanModeRequest { get; private set; }
|
||||
|
||||
public bool SuppressHookLifecycleEvents { get; set; }
|
||||
|
||||
public async Task EmitThinkingIfNeeded(
|
||||
AgentIdentity agent,
|
||||
Func<SidecarEventDto, Task> onEvent)
|
||||
{
|
||||
AgentActivityEventDto? thinkingActivity = CreateThinkingActivityIfNeeded(agent);
|
||||
if (thinkingActivity is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await onEvent(thinkingActivity).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
public void QueueThinkingIfNeeded(AgentIdentity agent)
|
||||
{
|
||||
AgentActivityEventDto? thinkingActivity = CreateThinkingActivityIfNeeded(agent);
|
||||
if (thinkingActivity is not null)
|
||||
{
|
||||
_pendingEvents.Enqueue(thinkingActivity);
|
||||
}
|
||||
}
|
||||
|
||||
public void QueueCompletedActivity(AgentIdentity agent)
|
||||
{
|
||||
_pendingEvents.Enqueue(CreateCompletedActivity(agent));
|
||||
}
|
||||
|
||||
public void ApplyEvent(SidecarEventDto evt)
|
||||
{
|
||||
if (evt is AgentActivityEventDto activity
|
||||
&& string.Equals(activity.ActivityType, "handoff", StringComparison.Ordinal)
|
||||
&& !string.IsNullOrWhiteSpace(activity.AgentId)
|
||||
&& !string.IsNullOrWhiteSpace(activity.AgentName))
|
||||
{
|
||||
ActiveAgent = new AgentIdentity(activity.AgentId, activity.AgentName);
|
||||
}
|
||||
}
|
||||
|
||||
public void ObserveSessionEvent(WorkflowNodeDto agentDefinition, ProviderSessionEvent sessionEvent)
|
||||
{
|
||||
AgentIdentity agent = AgentIdentityResolver.ResolveAgentIdentity(
|
||||
_command.Workflow,
|
||||
agentDefinition.GetAgentId(),
|
||||
agentDefinition.GetAgentName());
|
||||
|
||||
switch (sessionEvent)
|
||||
{
|
||||
case ProviderAssistantMessageDeltaEvent messageDelta:
|
||||
RecordObservedAgentForMessage(agent, messageDelta.MessageId);
|
||||
QueueThinkingIfNeeded(agent);
|
||||
break;
|
||||
case ProviderAssistantMessageEvent assistantMessage:
|
||||
RecordObservedAgentForMessage(agent, assistantMessage.MessageId);
|
||||
QueueThinkingIfNeeded(agent);
|
||||
if (assistantMessage.HasToolRequests)
|
||||
{
|
||||
QueueMessageReclassifiedIfNeeded(assistantMessage.MessageId);
|
||||
}
|
||||
break;
|
||||
case ProviderToolExecutionStartEvent toolExecutionStart:
|
||||
string toolCallId = toolExecutionStart.ToolCallId;
|
||||
string toolName = toolExecutionStart.ToolName;
|
||||
ToolNamesByCallId[toolCallId] = toolName;
|
||||
ActiveAgent = agent;
|
||||
AgentActivityEventDto? toolActivity = CreateToolCallingActivity(
|
||||
agent, toolName, toolCallId, toolExecutionStart.ToolArguments);
|
||||
if (toolActivity is not null)
|
||||
{
|
||||
_pendingEvents.Enqueue(toolActivity);
|
||||
}
|
||||
|
||||
QueueMessageReclassifiedIfNeeded(_lastObservedMessageId);
|
||||
break;
|
||||
case ProviderAssistantIntentEvent intentEvent:
|
||||
ActiveAgent = agent;
|
||||
QueueThinkingIfNeeded(agent);
|
||||
AssistantIntentEventDto? assistantIntent = CreateAssistantIntentEvent(agent, intentEvent.Intent);
|
||||
if (assistantIntent is not null)
|
||||
{
|
||||
_pendingEvents.Enqueue(assistantIntent);
|
||||
}
|
||||
break;
|
||||
case ProviderAssistantReasoningDeltaEvent reasoningDelta:
|
||||
ActiveAgent = agent;
|
||||
QueueThinkingIfNeeded(agent);
|
||||
ReasoningDeltaEventDto? reasoningDeltaEvent = CreateReasoningDeltaEvent(
|
||||
agent,
|
||||
reasoningDelta.ReasoningId,
|
||||
reasoningDelta.DeltaContent);
|
||||
if (reasoningDeltaEvent is not null)
|
||||
{
|
||||
_pendingEvents.Enqueue(reasoningDeltaEvent);
|
||||
}
|
||||
break;
|
||||
case ProviderSubagentStartedEvent started:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateSubagentStartedEvent(agent, started));
|
||||
break;
|
||||
case ProviderSubagentCompletedEvent completed:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateSubagentCompletedEvent(agent, completed));
|
||||
break;
|
||||
case ProviderSubagentFailedEvent failed:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateSubagentFailedEvent(agent, failed));
|
||||
break;
|
||||
case ProviderSubagentSelectedEvent selected:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateSubagentSelectedEvent(agent, selected));
|
||||
break;
|
||||
case ProviderSubagentDeselectedEvent:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateSubagentDeselectedEvent(agent));
|
||||
break;
|
||||
case ProviderSkillInvokedEvent skillInvoked:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateSkillInvokedEvent(agent, skillInvoked));
|
||||
break;
|
||||
case ProviderHookStartEvent hookStart:
|
||||
ActiveAgent = agent;
|
||||
if (!SuppressHookLifecycleEvents)
|
||||
{
|
||||
_pendingEvents.Enqueue(CreateHookLifecycleEvent(
|
||||
agent,
|
||||
"start",
|
||||
hookStart.HookInvocationId,
|
||||
hookStart.HookType,
|
||||
input: hookStart.Input));
|
||||
}
|
||||
break;
|
||||
case ProviderHookEndEvent hookEnd:
|
||||
ActiveAgent = agent;
|
||||
if (!SuppressHookLifecycleEvents)
|
||||
{
|
||||
_pendingEvents.Enqueue(CreateHookLifecycleEvent(
|
||||
agent,
|
||||
"end",
|
||||
hookEnd.HookInvocationId,
|
||||
hookEnd.HookType,
|
||||
success: hookEnd.Success,
|
||||
output: hookEnd.Output,
|
||||
error: hookEnd.Error));
|
||||
}
|
||||
break;
|
||||
case ProviderAssistantUsageEvent assistantUsage:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateAssistantUsageEvent(agent, assistantUsage));
|
||||
break;
|
||||
case ProviderSessionUsageEvent usageInfo:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateUsageEvent(agent, usageInfo));
|
||||
break;
|
||||
case ProviderSessionCompactionStartEvent compactionStart:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateCompactionStartEvent(agent, compactionStart));
|
||||
break;
|
||||
case ProviderSessionCompactionCompleteEvent compactionComplete:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateCompactionCompleteEvent(agent, compactionComplete));
|
||||
break;
|
||||
case ProviderPendingMessagesModifiedEvent:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreatePendingMessagesModifiedEvent(agent));
|
||||
break;
|
||||
case ProviderMcpOauthRequiredEvent:
|
||||
ActiveAgent = agent;
|
||||
break;
|
||||
case ProviderExitPlanModeRequestedEvent:
|
||||
HasPendingExitPlanModeRequest = true;
|
||||
ActiveAgent = agent;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
public IReadOnlyList<SidecarEventDto> DrainPendingEvents()
|
||||
{
|
||||
List<SidecarEventDto> pending = [];
|
||||
while (_pendingEvents.TryDequeue(out SidecarEventDto? pendingEvent))
|
||||
{
|
||||
pending.Add(pendingEvent);
|
||||
}
|
||||
|
||||
return pending;
|
||||
}
|
||||
|
||||
public void EnqueuePendingMcpOauthRequest(McpOauthRequiredEventDto request)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(request);
|
||||
_pendingMcpOauthRequests.Enqueue(request);
|
||||
}
|
||||
|
||||
public IReadOnlyList<McpOauthRequiredEventDto> DrainPendingMcpOauthRequests()
|
||||
{
|
||||
List<McpOauthRequiredEventDto> pending = [];
|
||||
while (_pendingMcpOauthRequests.TryDequeue(out McpOauthRequiredEventDto? request))
|
||||
{
|
||||
pending.Add(request);
|
||||
}
|
||||
|
||||
return pending;
|
||||
}
|
||||
|
||||
public bool TryResolveObservedAgentForMessage(string? messageId, out AgentIdentity agent)
|
||||
{
|
||||
agent = default;
|
||||
return !string.IsNullOrWhiteSpace(messageId)
|
||||
&& _observedAgentsByMessageId.TryGetValue(messageId, out agent);
|
||||
}
|
||||
|
||||
public string CreateMessageId(string? messageId)
|
||||
{
|
||||
return messageId ?? $"{_command.RequestId}-delta-{_fallbackMessageIndex++}";
|
||||
}
|
||||
|
||||
public TranscriptSegment AppendDelta(
|
||||
string messageId,
|
||||
string authorName,
|
||||
string delta)
|
||||
{
|
||||
return _transcriptBuffer.AppendDelta(messageId, authorName, delta);
|
||||
}
|
||||
|
||||
public void ClearActiveAgentIfMatching(AgentIdentity completedAgent)
|
||||
{
|
||||
if (ActiveAgent.HasValue
|
||||
&& string.Equals(ActiveAgent.Value.AgentId, completedAgent.AgentId, StringComparison.Ordinal))
|
||||
{
|
||||
ActiveAgent = null;
|
||||
}
|
||||
}
|
||||
|
||||
private void RecordObservedAgentForMessage(AgentIdentity agent, string messageId)
|
||||
{
|
||||
ActiveAgent = agent;
|
||||
_observedAgentsByMessageId[messageId] = agent;
|
||||
_lastObservedMessageId = messageId;
|
||||
}
|
||||
|
||||
private void QueueMessageReclassifiedIfNeeded(string? messageId)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(messageId))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
string normalizedMessageId = messageId.Trim();
|
||||
if (!_reclassifiedMessageIds.Add(normalizedMessageId))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_pendingEvents.Enqueue(CreateMessageReclassifiedEvent(normalizedMessageId));
|
||||
}
|
||||
|
||||
private AgentActivityEventDto? CreateThinkingActivityIfNeeded(AgentIdentity agent)
|
||||
{
|
||||
ActiveAgent = agent;
|
||||
|
||||
if (!_startedAgents.Add(agent.AgentId))
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
return new AgentActivityEventDto
|
||||
{
|
||||
Type = "agent-activity",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
ActivityType = "thinking",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
};
|
||||
}
|
||||
|
||||
private AgentActivityEventDto? CreateToolCallingActivity(
|
||||
AgentIdentity agent,
|
||||
string toolName,
|
||||
string toolCallId,
|
||||
IReadOnlyDictionary<string, object?>? toolArguments = null)
|
||||
{
|
||||
if (toolName.StartsWith("handoff_to_", StringComparison.Ordinal))
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
return new AgentActivityEventDto
|
||||
{
|
||||
Type = "agent-activity",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
ActivityType = "tool-calling",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
ToolName = toolName,
|
||||
ToolCallId = toolCallId,
|
||||
ToolArguments = toolArguments,
|
||||
};
|
||||
}
|
||||
|
||||
private AgentActivityEventDto CreateCompletedActivity(AgentIdentity agent)
|
||||
{
|
||||
return new AgentActivityEventDto
|
||||
{
|
||||
Type = "agent-activity",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
ActivityType = "completed",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
};
|
||||
}
|
||||
|
||||
private MessageReclassifiedEventDto CreateMessageReclassifiedEvent(string messageId)
|
||||
{
|
||||
return new MessageReclassifiedEventDto
|
||||
{
|
||||
Type = "message-reclassified",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
MessageId = messageId,
|
||||
NewKind = "thinking",
|
||||
};
|
||||
}
|
||||
|
||||
public void UpdateCompletedMessages(
|
||||
IReadOnlyList<ChatMessage> allMessages,
|
||||
IReadOnlyList<ChatMessage> inputMessages)
|
||||
{
|
||||
List<ChatMessage> newMessages = WorkflowTranscriptProjector.SelectNewOutputMessages(allMessages, inputMessages);
|
||||
CompletedMessages = WorkflowTranscriptProjector.ProjectCompletedMessagesFromSegments(
|
||||
_command,
|
||||
newMessages,
|
||||
_transcriptBuffer.Snapshot(),
|
||||
ActiveAgent);
|
||||
base.UpdateCompletedMessages(allMessages, inputMessages, CopilotTranscriptProjector.Instance);
|
||||
}
|
||||
|
||||
public IReadOnlyList<ChatMessageDto> FinalizeCompletedMessages()
|
||||
{
|
||||
if (CompletedMessages.Count == 0 && _transcriptBuffer.Count > 0)
|
||||
{
|
||||
CompletedMessages = WorkflowTranscriptProjector.ProjectCompletedMessagesFromSegments(
|
||||
_command,
|
||||
[],
|
||||
_transcriptBuffer.Snapshot(),
|
||||
ActiveAgent);
|
||||
}
|
||||
|
||||
foreach (ChatMessageDto message in CompletedMessages)
|
||||
{
|
||||
if (_reclassifiedMessageIds.Contains(message.Id))
|
||||
{
|
||||
message.MessageKind = "thinking";
|
||||
}
|
||||
}
|
||||
|
||||
return CompletedMessages;
|
||||
}
|
||||
|
||||
private SubagentEventDto CreateSubagentStartedEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderSubagentStartedEvent data)
|
||||
{
|
||||
return new SubagentEventDto
|
||||
{
|
||||
Type = "subagent-event",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
EventKind = "started",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
ToolCallId = data.ToolCallId,
|
||||
CustomAgentName = data.AgentName,
|
||||
CustomAgentDisplayName = data.AgentDisplayName,
|
||||
CustomAgentDescription = data.AgentDescription,
|
||||
};
|
||||
}
|
||||
|
||||
private SubagentEventDto CreateSubagentCompletedEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderSubagentCompletedEvent data)
|
||||
{
|
||||
return new SubagentEventDto
|
||||
{
|
||||
Type = "subagent-event",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
EventKind = "completed",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
ToolCallId = data.ToolCallId,
|
||||
CustomAgentName = data.AgentName,
|
||||
CustomAgentDisplayName = data.AgentDisplayName,
|
||||
};
|
||||
}
|
||||
|
||||
private SubagentEventDto CreateSubagentFailedEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderSubagentFailedEvent data)
|
||||
{
|
||||
return new SubagentEventDto
|
||||
{
|
||||
Type = "subagent-event",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
EventKind = "failed",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
ToolCallId = data.ToolCallId,
|
||||
CustomAgentName = data.AgentName,
|
||||
CustomAgentDisplayName = data.AgentDisplayName,
|
||||
Error = data.Error,
|
||||
};
|
||||
}
|
||||
|
||||
private SubagentEventDto CreateSubagentSelectedEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderSubagentSelectedEvent data)
|
||||
{
|
||||
return new SubagentEventDto
|
||||
{
|
||||
Type = "subagent-event",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
EventKind = "selected",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
CustomAgentName = data.AgentName,
|
||||
CustomAgentDisplayName = data.AgentDisplayName,
|
||||
Tools = data.Tools,
|
||||
};
|
||||
}
|
||||
|
||||
private SubagentEventDto CreateSubagentDeselectedEvent(AgentIdentity agent)
|
||||
{
|
||||
return new SubagentEventDto
|
||||
{
|
||||
Type = "subagent-event",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
EventKind = "deselected",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
};
|
||||
}
|
||||
|
||||
private AssistantIntentEventDto? CreateAssistantIntentEvent(
|
||||
AgentIdentity agent,
|
||||
string? intent)
|
||||
{
|
||||
string? normalizedIntent = intent?.Trim();
|
||||
if (string.IsNullOrWhiteSpace(normalizedIntent))
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
return new AssistantIntentEventDto
|
||||
{
|
||||
Type = "assistant-intent",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
Intent = normalizedIntent,
|
||||
};
|
||||
}
|
||||
|
||||
private ReasoningDeltaEventDto? CreateReasoningDeltaEvent(
|
||||
AgentIdentity agent,
|
||||
string? reasoningId,
|
||||
string? deltaContent)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(reasoningId)
|
||||
|| string.IsNullOrEmpty(deltaContent))
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
return new ReasoningDeltaEventDto
|
||||
{
|
||||
Type = "reasoning-delta",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
ReasoningId = reasoningId,
|
||||
ContentDelta = deltaContent,
|
||||
};
|
||||
}
|
||||
|
||||
private SkillInvokedEventDto CreateSkillInvokedEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderSkillInvokedEvent data)
|
||||
{
|
||||
return new SkillInvokedEventDto
|
||||
{
|
||||
Type = "skill-invoked",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
SkillName = data.SkillName,
|
||||
Path = data.Path,
|
||||
Content = data.Content,
|
||||
AllowedTools = data.AllowedTools,
|
||||
PluginName = data.PluginName,
|
||||
PluginVersion = data.PluginVersion,
|
||||
};
|
||||
}
|
||||
|
||||
private HookLifecycleEventDto CreateHookLifecycleEvent(
|
||||
AgentIdentity agent,
|
||||
string phase,
|
||||
string hookInvocationId,
|
||||
string hookType,
|
||||
object? input = null,
|
||||
bool? success = null,
|
||||
object? output = null,
|
||||
string? error = null)
|
||||
{
|
||||
return new HookLifecycleEventDto
|
||||
{
|
||||
Type = "hook-lifecycle",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
HookInvocationId = hookInvocationId,
|
||||
HookType = hookType,
|
||||
Phase = phase,
|
||||
Input = input,
|
||||
Success = success,
|
||||
Output = output,
|
||||
Error = error,
|
||||
};
|
||||
}
|
||||
|
||||
private AssistantUsageEventDto CreateAssistantUsageEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderAssistantUsageEvent data)
|
||||
{
|
||||
return new AssistantUsageEventDto
|
||||
{
|
||||
Type = "assistant-usage",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
Model = data.Model,
|
||||
InputTokens = data.InputTokens,
|
||||
OutputTokens = data.OutputTokens,
|
||||
CacheReadTokens = data.CacheReadTokens,
|
||||
CacheWriteTokens = data.CacheWriteTokens,
|
||||
Cost = data.Cost,
|
||||
Duration = data.Duration,
|
||||
TotalNanoAiu = data.TotalNanoAiu,
|
||||
QuotaSnapshots = data.QuotaSnapshots,
|
||||
};
|
||||
}
|
||||
|
||||
private SessionUsageEventDto CreateUsageEvent(AgentIdentity agent, ProviderSessionUsageEvent data)
|
||||
{
|
||||
return new SessionUsageEventDto
|
||||
{
|
||||
Type = "session-usage",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
TokenLimit = data.TokenLimit,
|
||||
CurrentTokens = data.CurrentTokens,
|
||||
MessagesLength = data.MessagesLength,
|
||||
SystemTokens = data.SystemTokens,
|
||||
ConversationTokens = data.ConversationTokens,
|
||||
ToolDefinitionsTokens = data.ToolDefinitionsTokens,
|
||||
IsInitial = data.IsInitial,
|
||||
};
|
||||
}
|
||||
|
||||
private SessionCompactionEventDto CreateCompactionStartEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderSessionCompactionStartEvent data)
|
||||
{
|
||||
return new SessionCompactionEventDto
|
||||
{
|
||||
Type = "session-compaction",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
Phase = "start",
|
||||
SystemTokens = data.SystemTokens,
|
||||
ConversationTokens = data.ConversationTokens,
|
||||
ToolDefinitionsTokens = data.ToolDefinitionsTokens,
|
||||
};
|
||||
}
|
||||
|
||||
private SessionCompactionEventDto CreateCompactionCompleteEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderSessionCompactionCompleteEvent data)
|
||||
{
|
||||
return new SessionCompactionEventDto
|
||||
{
|
||||
Type = "session-compaction",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
Phase = "complete",
|
||||
Success = data.Success,
|
||||
Error = data.Error,
|
||||
SystemTokens = data.SystemTokens,
|
||||
ConversationTokens = data.ConversationTokens,
|
||||
ToolDefinitionsTokens = data.ToolDefinitionsTokens,
|
||||
PreCompactionTokens = data.PreCompactionTokens,
|
||||
PostCompactionTokens = data.PostCompactionTokens,
|
||||
PreCompactionMessagesLength = data.PreCompactionMessagesLength,
|
||||
MessagesRemoved = data.MessagesRemoved,
|
||||
TokensRemoved = data.TokensRemoved,
|
||||
SummaryContent = data.SummaryContent,
|
||||
CheckpointNumber = data.CheckpointNumber,
|
||||
CheckpointPath = data.CheckpointPath,
|
||||
};
|
||||
}
|
||||
|
||||
private PendingMessagesModifiedEventDto CreatePendingMessagesModifiedEvent(AgentIdentity agent)
|
||||
{
|
||||
return new PendingMessagesModifiedEventDto
|
||||
{
|
||||
Type = "pending-messages-modified",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
};
|
||||
return base.FinalizeCompletedMessages(CopilotTranscriptProjector.Instance);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
using Aryx.AgentHost.Contracts;
|
||||
using Microsoft.Extensions.AI;
|
||||
|
||||
namespace Aryx.AgentHost.Services;
|
||||
|
||||
internal interface IProviderTranscriptProjector
|
||||
{
|
||||
ChatMessage ToChatMessage(ChatMessageDto message);
|
||||
|
||||
void AttachMessageMode(IList<ChatMessage> messages, string? messageMode);
|
||||
|
||||
List<ChatMessage> SelectNewOutputMessages(
|
||||
IReadOnlyList<ChatMessage> outputMessages,
|
||||
IReadOnlyList<ChatMessage> inputMessages);
|
||||
|
||||
List<ChatMessageDto> ProjectCompletedMessagesFromSegments(
|
||||
RunTurnCommandDto command,
|
||||
IReadOnlyList<ChatMessage> newMessages,
|
||||
IReadOnlyList<TranscriptSegment> segments,
|
||||
AgentIdentity? fallbackAgent = null);
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
using Aryx.AgentHost.Contracts;
|
||||
using GitHub.Copilot.SDK;
|
||||
|
||||
namespace Aryx.AgentHost.Services;
|
||||
|
||||
internal interface IProviderTurnSupport
|
||||
{
|
||||
Task<ProviderAgentBundle> CreateAgentBundleAsync(
|
||||
RunTurnCommandDto command,
|
||||
TurnExecutionState state,
|
||||
Func<SidecarEventDto, Task> onEvent,
|
||||
Func<ApprovalRequestedEventDto, Task> onApproval,
|
||||
Func<UserInputRequestedEventDto, Task> onUserInput,
|
||||
CancellationTokenSource runCancellation,
|
||||
CancellationToken cancellationToken);
|
||||
|
||||
Task ResolveApprovalAsync(
|
||||
ResolveApprovalCommandDto command,
|
||||
CancellationToken cancellationToken);
|
||||
|
||||
Task ResolveUserInputAsync(
|
||||
ResolveUserInputCommandDto command,
|
||||
CancellationToken cancellationToken);
|
||||
|
||||
Task<UserInputResponse> RequestRequestPortUserInputAsync(
|
||||
RunTurnCommandDto command,
|
||||
UserInputRequest request,
|
||||
Func<UserInputRequestedEventDto, Task> onUserInput,
|
||||
CancellationToken cancellationToken);
|
||||
|
||||
ExitPlanModeRequestedEventDto? ConsumePendingExitPlanModeRequest(string requestId);
|
||||
|
||||
void ClearRequestState(string requestId);
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
using Microsoft.Agents.AI;
|
||||
|
||||
namespace Aryx.AgentHost.Services;
|
||||
|
||||
internal abstract class ProviderAgentBundle : IAsyncDisposable
|
||||
{
|
||||
public abstract IReadOnlyList<AIAgent> Agents { get; }
|
||||
|
||||
public abstract bool HasConfiguredHooks { get; }
|
||||
|
||||
public abstract IProviderTranscriptProjector TranscriptProjector { get; }
|
||||
|
||||
public abstract ValueTask DisposeAsync();
|
||||
}
|
||||
@@ -8,7 +8,7 @@ using Microsoft.Extensions.AI;
|
||||
|
||||
namespace Aryx.AgentHost.Services;
|
||||
|
||||
internal sealed class CopilotAgentBundle : IAsyncDisposable
|
||||
internal sealed class CopilotAgentBundle : ProviderAgentBundle
|
||||
{
|
||||
private static readonly string[] RequiredPromptTools =
|
||||
[
|
||||
@@ -25,9 +25,11 @@ internal sealed class CopilotAgentBundle : IAsyncDisposable
|
||||
HasConfiguredHooks = hasConfiguredHooks;
|
||||
}
|
||||
|
||||
public IReadOnlyList<AIAgent> Agents { get; }
|
||||
public override IReadOnlyList<AIAgent> Agents { get; }
|
||||
|
||||
public bool HasConfiguredHooks { get; }
|
||||
public override bool HasConfiguredHooks { get; }
|
||||
|
||||
public override IProviderTranscriptProjector TranscriptProjector { get; } = CopilotTranscriptProjector.Instance;
|
||||
|
||||
public static async Task<CopilotAgentBundle> CreateAsync(
|
||||
RunTurnCommandDto command,
|
||||
@@ -215,116 +217,38 @@ internal sealed class CopilotAgentBundle : IAsyncDisposable
|
||||
|
||||
internal static AIAgentHostOptions CreateAgentHostOptions()
|
||||
{
|
||||
return new AIAgentHostOptions
|
||||
{
|
||||
EmitAgentUpdateEvents = null,
|
||||
EmitAgentResponseEvents = false,
|
||||
InterceptUserInputRequests = false,
|
||||
InterceptUnterminatedFunctionCalls = false,
|
||||
ReassignOtherAgentsAsUsers = true,
|
||||
ForwardIncomingMessages = true,
|
||||
};
|
||||
return AgentHostOptionsFactory.CreateDefault();
|
||||
}
|
||||
|
||||
internal static HandoffWorkflowBuilder CreateHandoffWorkflowBuilder(
|
||||
AIAgent entryAgent,
|
||||
HandoffModeSettingsDto? settings = null)
|
||||
{
|
||||
HandoffModeSettingsDto effectiveSettings = settings ?? new HandoffModeSettingsDto();
|
||||
HandoffWorkflowBuilder builder = AgentWorkflowBuilder.CreateHandoffBuilderWith(entryAgent)
|
||||
.WithToolCallFilteringBehavior(MapHandoffToolCallFiltering(effectiveSettings.ToolCallFiltering))
|
||||
.WithHandoffInstructions(NormalizeOptionalString(effectiveSettings.HandoffInstructions)
|
||||
?? HandoffWorkflowGuidance.CreateWorkflowInstructions());
|
||||
|
||||
if (effectiveSettings.ReturnToPrevious)
|
||||
{
|
||||
builder = builder.EnableReturnToPrevious();
|
||||
}
|
||||
|
||||
return builder;
|
||||
return WorkflowOrchestrationFactory.CreateHandoffWorkflowBuilder(entryAgent, settings);
|
||||
}
|
||||
|
||||
internal static Workflow CreateHandoffWorkflow(
|
||||
WorkflowDefinitionDto workflowDefinition,
|
||||
IReadOnlyList<AIAgent> agents)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(workflowDefinition);
|
||||
ArgumentNullException.ThrowIfNull(agents);
|
||||
|
||||
IReadOnlyList<WorkflowNodeDto> agentNodes = workflowDefinition.GetAgentNodes();
|
||||
Dictionary<string, AIAgent> agentsById = CreateAgentMap(agents);
|
||||
WorkflowNodeDto triageNode = ResolveTriageAgentNode(workflowDefinition, agentNodes);
|
||||
AIAgent triageAgent = ResolveAgentForNode(triageNode, agentsById);
|
||||
HandoffModeSettingsDto? settings = workflowDefinition.Settings.ModeSettings?.Handoff;
|
||||
HandoffWorkflowBuilder builder = CreateHandoffWorkflowBuilder(triageAgent, settings);
|
||||
|
||||
List<WorkflowNodeDto> specialistNodes = agentNodes
|
||||
.Where(node => !string.Equals(node.Id, triageNode.Id, StringComparison.Ordinal))
|
||||
.ToList();
|
||||
|
||||
if (specialistNodes.Count == 0)
|
||||
{
|
||||
throw new InvalidOperationException("Handoff workflows require at least one specialist agent in addition to the triage agent.");
|
||||
}
|
||||
|
||||
foreach (WorkflowNodeDto specialistNode in specialistNodes)
|
||||
{
|
||||
AIAgent specialistAgent = ResolveAgentForNode(specialistNode, agentsById);
|
||||
builder.WithHandoff(
|
||||
triageAgent,
|
||||
specialistAgent,
|
||||
HandoffWorkflowGuidance.CreateForwardReason(specialistNode));
|
||||
|
||||
if (settings?.ReturnToPrevious != true)
|
||||
{
|
||||
builder.WithHandoff(
|
||||
specialistAgent,
|
||||
triageAgent,
|
||||
HandoffWorkflowGuidance.CreateReturnReason(triageNode));
|
||||
}
|
||||
}
|
||||
|
||||
return builder.Build();
|
||||
return WorkflowOrchestrationFactory.CreateHandoffWorkflow(workflowDefinition, agents);
|
||||
}
|
||||
|
||||
internal static GroupChatWorkflowBuilder CreateGroupChatWorkflowBuilder(
|
||||
WorkflowDefinitionDto workflowDefinition,
|
||||
IReadOnlyList<AIAgent> agents)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(workflowDefinition);
|
||||
ArgumentNullException.ThrowIfNull(agents);
|
||||
|
||||
int maxRounds = ResolveGroupChatMaxRounds(workflowDefinition);
|
||||
GroupChatWorkflowBuilder builder = AgentWorkflowBuilder.CreateGroupChatBuilderWith(
|
||||
participants => new RoundRobinGroupChatManager(participants)
|
||||
{
|
||||
MaximumIterationCount = maxRounds,
|
||||
})
|
||||
.AddParticipants(agents);
|
||||
|
||||
string? name = NormalizeOptionalString(workflowDefinition.Name);
|
||||
if (name is not null)
|
||||
{
|
||||
builder.WithName(name);
|
||||
}
|
||||
|
||||
string? description = NormalizeOptionalString(workflowDefinition.Description);
|
||||
if (description is not null)
|
||||
{
|
||||
builder.WithDescription(description);
|
||||
}
|
||||
|
||||
return builder;
|
||||
return WorkflowOrchestrationFactory.CreateGroupChatWorkflowBuilder(workflowDefinition, agents);
|
||||
}
|
||||
|
||||
internal static Workflow CreateGroupChatWorkflow(
|
||||
WorkflowDefinitionDto workflowDefinition,
|
||||
IReadOnlyList<AIAgent> agents)
|
||||
{
|
||||
return CreateGroupChatWorkflowBuilder(workflowDefinition, agents).Build();
|
||||
return WorkflowOrchestrationFactory.CreateGroupChatWorkflow(workflowDefinition, agents);
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
public override async ValueTask DisposeAsync()
|
||||
{
|
||||
foreach (IAsyncDisposable disposable in _disposables)
|
||||
{
|
||||
@@ -380,90 +304,6 @@ internal sealed class CopilotAgentBundle : IAsyncDisposable
|
||||
.ToList();
|
||||
}
|
||||
|
||||
private static Dictionary<string, AIAgent> CreateAgentMap(IReadOnlyList<AIAgent> agents)
|
||||
{
|
||||
Dictionary<string, AIAgent> agentMap = new(StringComparer.OrdinalIgnoreCase);
|
||||
foreach (AIAgent agent in agents)
|
||||
{
|
||||
if (!string.IsNullOrWhiteSpace(agent.Id))
|
||||
{
|
||||
agentMap[agent.Id] = agent;
|
||||
}
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(agent.Name))
|
||||
{
|
||||
agentMap[agent.Name] = agent;
|
||||
}
|
||||
}
|
||||
|
||||
return agentMap;
|
||||
}
|
||||
|
||||
private static AIAgent ResolveAgentForNode(
|
||||
WorkflowNodeDto node,
|
||||
IReadOnlyDictionary<string, AIAgent> agentsById)
|
||||
{
|
||||
string agentId = node.GetAgentId();
|
||||
if (agentsById.TryGetValue(agentId, out AIAgent? agent))
|
||||
{
|
||||
return agent;
|
||||
}
|
||||
|
||||
string agentName = node.GetAgentName();
|
||||
if (agentsById.TryGetValue(agentName, out agent))
|
||||
{
|
||||
return agent;
|
||||
}
|
||||
|
||||
throw new InvalidOperationException($"Workflow agent \"{agentId}\" could not be resolved from the constructed Copilot agents.");
|
||||
}
|
||||
|
||||
private static WorkflowNodeDto ResolveTriageAgentNode(
|
||||
WorkflowDefinitionDto workflowDefinition,
|
||||
IReadOnlyList<WorkflowNodeDto> agentNodes)
|
||||
{
|
||||
if (agentNodes.Count == 0)
|
||||
{
|
||||
throw new InvalidOperationException("Handoff workflows require at least one agent node.");
|
||||
}
|
||||
|
||||
string? triageAgentNodeId = NormalizeOptionalString(workflowDefinition.Settings.ModeSettings?.Handoff?.TriageAgentNodeId);
|
||||
if (triageAgentNodeId is null)
|
||||
{
|
||||
return agentNodes[0];
|
||||
}
|
||||
|
||||
WorkflowNodeDto? triageNode = agentNodes.FirstOrDefault(node => string.Equals(node.Id, triageAgentNodeId, StringComparison.Ordinal));
|
||||
return triageNode ?? throw new InvalidOperationException(
|
||||
$"Handoff workflow triage agent node \"{triageAgentNodeId}\" was not found in the workflow graph.");
|
||||
}
|
||||
|
||||
private static HandoffToolCallFilteringBehavior MapHandoffToolCallFiltering(string? value)
|
||||
{
|
||||
return value?.Trim().ToLowerInvariant() switch
|
||||
{
|
||||
"none" => HandoffToolCallFilteringBehavior.None,
|
||||
"all" => HandoffToolCallFilteringBehavior.All,
|
||||
_ => HandoffToolCallFilteringBehavior.HandoffOnly,
|
||||
};
|
||||
}
|
||||
|
||||
private static int ResolveGroupChatMaxRounds(WorkflowDefinitionDto workflowDefinition)
|
||||
{
|
||||
int? configuredMaxRounds = workflowDefinition.Settings.ModeSettings?.GroupChat?.MaxRounds;
|
||||
if (configuredMaxRounds is > 0)
|
||||
{
|
||||
return configuredMaxRounds.Value;
|
||||
}
|
||||
|
||||
if (workflowDefinition.Settings.MaxIterations is > 0)
|
||||
{
|
||||
return workflowDefinition.Settings.MaxIterations.Value;
|
||||
}
|
||||
|
||||
return 5;
|
||||
}
|
||||
|
||||
private static string? ResolveEffectiveAgent(
|
||||
string? defaultAgent,
|
||||
RunTurnPromptInvocationDto? promptInvocation)
|
||||
|
||||
@@ -30,7 +30,7 @@ internal sealed class CopilotAgentProvider : IAgentProvider
|
||||
public ITurnWorkflowRunner CreateWorkflowRunner(WorkflowValidator workflowValidator)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(workflowValidator);
|
||||
return new CopilotWorkflowRunner(workflowValidator);
|
||||
return new AgentWorkflowTurnRunner(new CopilotTurnRunnerSupport(), workflowValidator);
|
||||
}
|
||||
|
||||
public Task<SidecarCapabilitiesDto> GetCapabilitiesAsync(CancellationToken cancellationToken)
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
using Aryx.AgentHost.Contracts;
|
||||
using Microsoft.Extensions.AI;
|
||||
|
||||
namespace Aryx.AgentHost.Services;
|
||||
|
||||
internal sealed class CopilotTranscriptProjector : IProviderTranscriptProjector
|
||||
{
|
||||
public static CopilotTranscriptProjector Instance { get; } = new();
|
||||
|
||||
private CopilotTranscriptProjector()
|
||||
{
|
||||
}
|
||||
|
||||
public ChatMessage ToChatMessage(ChatMessageDto message)
|
||||
=> WorkflowTranscriptProjector.ToChatMessage(message);
|
||||
|
||||
public void AttachMessageMode(IList<ChatMessage> messages, string? messageMode)
|
||||
=> WorkflowTranscriptProjector.AttachMessageMode(messages, messageMode);
|
||||
|
||||
public List<ChatMessage> SelectNewOutputMessages(
|
||||
IReadOnlyList<ChatMessage> outputMessages,
|
||||
IReadOnlyList<ChatMessage> inputMessages)
|
||||
=> WorkflowTranscriptProjector.SelectNewOutputMessages(outputMessages, inputMessages);
|
||||
|
||||
public List<ChatMessageDto> ProjectCompletedMessagesFromSegments(
|
||||
RunTurnCommandDto command,
|
||||
IReadOnlyList<ChatMessage> newMessages,
|
||||
IReadOnlyList<TranscriptSegment> segments,
|
||||
AgentIdentity? fallbackAgent = null)
|
||||
=> WorkflowTranscriptProjector.ProjectCompletedMessagesFromSegments(
|
||||
command,
|
||||
newMessages,
|
||||
segments,
|
||||
fallbackAgent);
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
using Aryx.AgentHost.Contracts;
|
||||
using GitHub.Copilot.SDK;
|
||||
|
||||
namespace Aryx.AgentHost.Services;
|
||||
|
||||
internal sealed class CopilotTurnRunnerSupport : IProviderTurnSupport
|
||||
{
|
||||
private readonly CopilotApprovalCoordinator _approvalCoordinator = new();
|
||||
private readonly CopilotUserInputCoordinator _userInputCoordinator = new();
|
||||
private readonly CopilotMcpOAuthCoordinator _mcpOAuthCoordinator = new();
|
||||
private readonly CopilotExitPlanModeCoordinator _exitPlanModeCoordinator = new();
|
||||
private readonly IProviderEventAdapter _providerEventAdapter = new CopilotEventAdapter();
|
||||
|
||||
public async Task<ProviderAgentBundle> CreateAgentBundleAsync(
|
||||
RunTurnCommandDto command,
|
||||
TurnExecutionState state,
|
||||
Func<SidecarEventDto, Task> onEvent,
|
||||
Func<ApprovalRequestedEventDto, Task> onApproval,
|
||||
Func<UserInputRequestedEventDto, Task> onUserInput,
|
||||
CancellationTokenSource runCancellation,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return await CopilotAgentBundle.CreateAsync(
|
||||
command,
|
||||
(agent, request, invocation) => _approvalCoordinator.RequestApprovalAsync(
|
||||
command,
|
||||
agent,
|
||||
request,
|
||||
invocation,
|
||||
state.ToolNamesByCallId,
|
||||
activity => AgentWorkflowTurnRunner.EmitActivityAsync(command, state, activity, onEvent),
|
||||
onApproval,
|
||||
runCancellation.Token),
|
||||
(agent, request, invocation) => _userInputCoordinator.RequestUserInputAsync(
|
||||
command,
|
||||
agent,
|
||||
request,
|
||||
invocation,
|
||||
onUserInput,
|
||||
runCancellation.Token),
|
||||
(agent, sessionEvent) => ObserveSessionEvent(command, state, runCancellation, agent, sessionEvent),
|
||||
cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
public Task ResolveApprovalAsync(
|
||||
ResolveApprovalCommandDto command,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return _approvalCoordinator.ResolveApprovalAsync(command, cancellationToken);
|
||||
}
|
||||
|
||||
public Task ResolveUserInputAsync(
|
||||
ResolveUserInputCommandDto command,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return _userInputCoordinator.ResolveUserInputAsync(command, cancellationToken);
|
||||
}
|
||||
|
||||
public Task<UserInputResponse> RequestRequestPortUserInputAsync(
|
||||
RunTurnCommandDto command,
|
||||
UserInputRequest request,
|
||||
Func<UserInputRequestedEventDto, Task> onUserInput,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return _userInputCoordinator.RequestUserInputAsync(
|
||||
command,
|
||||
request,
|
||||
onUserInput,
|
||||
cancellationToken);
|
||||
}
|
||||
|
||||
public ExitPlanModeRequestedEventDto? ConsumePendingExitPlanModeRequest(string requestId)
|
||||
{
|
||||
return _exitPlanModeCoordinator.ConsumePendingRequest(requestId);
|
||||
}
|
||||
|
||||
public void ClearRequestState(string requestId)
|
||||
{
|
||||
_approvalCoordinator.ClearRequestApprovals(requestId);
|
||||
}
|
||||
|
||||
private void ObserveSessionEvent(
|
||||
RunTurnCommandDto command,
|
||||
TurnExecutionState state,
|
||||
CancellationTokenSource runCancellation,
|
||||
WorkflowNodeDto agent,
|
||||
SessionEvent sessionEvent)
|
||||
{
|
||||
if (_providerEventAdapter.TryAdapt(sessionEvent) is { } providerEvent)
|
||||
{
|
||||
state.ObserveSessionEvent(agent, providerEvent);
|
||||
}
|
||||
|
||||
if (sessionEvent is McpOauthRequiredEvent mcpOauthRequired)
|
||||
{
|
||||
state.EnqueuePendingMcpOauthRequest(
|
||||
_mcpOAuthCoordinator.BuildMcpOauthRequiredEvent(command, agent, mcpOauthRequired));
|
||||
}
|
||||
|
||||
if (sessionEvent is ExitPlanModeRequestedEvent exitPlanModeRequested)
|
||||
{
|
||||
_exitPlanModeCoordinator.RecordExitPlanModeRequest(command, agent, exitPlanModeRequested);
|
||||
runCancellation.Cancel();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,4 @@
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Globalization;
|
||||
using System.Text.Json;
|
||||
using Aryx.AgentHost.Contracts;
|
||||
using GitHub.Copilot.SDK;
|
||||
@@ -12,301 +10,63 @@ using Microsoft.Extensions.AI;
|
||||
|
||||
namespace Aryx.AgentHost.Services;
|
||||
|
||||
public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner
|
||||
public sealed class CopilotWorkflowRunner : AgentWorkflowTurnRunner
|
||||
{
|
||||
private const string HandoffFunctionPrefix = "handoff_to_";
|
||||
private readonly WorkflowValidator _workflowValidator;
|
||||
private readonly WorkflowRunner _workflowRunner = new();
|
||||
private readonly CopilotApprovalCoordinator _approvalCoordinator = new();
|
||||
private readonly CopilotUserInputCoordinator _userInputCoordinator = new();
|
||||
private readonly CopilotMcpOAuthCoordinator _mcpOAuthCoordinator = new();
|
||||
private readonly CopilotExitPlanModeCoordinator _exitPlanModeCoordinator = new();
|
||||
|
||||
public CopilotWorkflowRunner(WorkflowValidator? workflowValidator = null)
|
||||
: base(new CopilotTurnRunnerSupport(), workflowValidator)
|
||||
{
|
||||
_workflowValidator = workflowValidator ?? new WorkflowValidator();
|
||||
}
|
||||
|
||||
public async Task<IReadOnlyList<ChatMessageDto>> RunTurnAsync(
|
||||
RunTurnCommandDto command,
|
||||
Func<TurnDeltaEventDto, Task> onDelta,
|
||||
Func<SidecarEventDto, Task> onEvent,
|
||||
Func<ApprovalRequestedEventDto, Task> onApproval,
|
||||
Func<UserInputRequestedEventDto, Task> onUserInput,
|
||||
Func<McpOauthRequiredEventDto, Task> onMcpOAuthRequired,
|
||||
Func<ExitPlanModeRequestedEventDto, Task> onExitPlanMode,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
string? validationError = _workflowValidator.Validate(command.Workflow, command.WorkflowLibrary)
|
||||
.FirstOrDefault()?.Message;
|
||||
if (validationError is not null)
|
||||
{
|
||||
throw new InvalidOperationException(validationError);
|
||||
}
|
||||
|
||||
IProviderEventAdapter providerEventAdapter = new CopilotEventAdapter();
|
||||
CopilotTurnExecutionState state = new(command);
|
||||
using CancellationTokenSource runCancellation =
|
||||
CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
|
||||
try
|
||||
{
|
||||
await using CopilotAgentBundle bundle = await CopilotAgentBundle.CreateAsync(
|
||||
command,
|
||||
(agent, request, invocation) => _approvalCoordinator.RequestApprovalAsync(
|
||||
command,
|
||||
agent,
|
||||
request,
|
||||
invocation,
|
||||
state.ToolNamesByCallId,
|
||||
activity => EmitActivityAsync(command, state, activity, onEvent),
|
||||
onApproval,
|
||||
runCancellation.Token),
|
||||
(agent, request, invocation) => _userInputCoordinator.RequestUserInputAsync(
|
||||
command,
|
||||
agent,
|
||||
request,
|
||||
invocation,
|
||||
onUserInput,
|
||||
runCancellation.Token),
|
||||
(agent, sessionEvent) =>
|
||||
{
|
||||
if (providerEventAdapter.TryAdapt(sessionEvent) is { } providerEvent)
|
||||
{
|
||||
state.ObserveSessionEvent(agent, providerEvent);
|
||||
}
|
||||
|
||||
if (sessionEvent is McpOauthRequiredEvent mcpOauthRequired)
|
||||
{
|
||||
state.EnqueuePendingMcpOauthRequest(
|
||||
_mcpOAuthCoordinator.BuildMcpOauthRequiredEvent(command, agent, mcpOauthRequired));
|
||||
}
|
||||
|
||||
if (sessionEvent is ExitPlanModeRequestedEvent exitPlanModeRequested)
|
||||
{
|
||||
_exitPlanModeCoordinator.RecordExitPlanModeRequest(command, agent, exitPlanModeRequested);
|
||||
runCancellation.Cancel();
|
||||
}
|
||||
},
|
||||
runCancellation.Token);
|
||||
ConfigureHookLifecycleEventSuppression(state, bundle);
|
||||
Workflow workflow = BuildWorkflowForCommand(command, bundle.Agents, _workflowRunner);
|
||||
List<ChatMessage> inputMessages = command.Messages.Select(WorkflowTranscriptProjector.ToChatMessage).ToList();
|
||||
WorkflowTranscriptProjector.AttachMessageMode(inputMessages, command.MessageMode);
|
||||
|
||||
using FileSystemJsonCheckpointStore? checkpointStore = CreateCheckpointStore(command);
|
||||
CheckpointManager? checkpointManager = checkpointStore is not null
|
||||
? CheckpointManager.CreateJson(checkpointStore)
|
||||
: null;
|
||||
|
||||
await using StreamingRun run = await OpenWorkflowRunAsync(
|
||||
command,
|
||||
workflow,
|
||||
inputMessages,
|
||||
checkpointManager).ConfigureAwait(false);
|
||||
await run.TrySendMessageAsync(new TurnToken(emitEvents: true)).ConfigureAwait(false);
|
||||
|
||||
await foreach (WorkflowEvent evt in run.WatchStreamAsync(runCancellation.Token).ConfigureAwait(false))
|
||||
{
|
||||
if (evt is RequestInfoEvent requestInfo
|
||||
&& await TryHandleRequestPortRequestAsync(
|
||||
command,
|
||||
requestInfo,
|
||||
run,
|
||||
onUserInput,
|
||||
runCancellation.Token).ConfigureAwait(false))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
bool shouldEndTurn = await HandleWorkflowEventAsync(command, evt, inputMessages, state, onDelta, onEvent)
|
||||
.ConfigureAwait(false);
|
||||
await EmitPendingEventsAsync(state, onEvent).ConfigureAwait(false);
|
||||
await EmitPendingMcpOauthRequestsAsync(state, onMcpOAuthRequired).ConfigureAwait(false);
|
||||
if (shouldEndTurn)
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
await EmitPendingEventsAsync(state, onEvent).ConfigureAwait(false);
|
||||
await EmitPendingMcpOauthRequestsAsync(state, onMcpOAuthRequired).ConfigureAwait(false);
|
||||
return state.FinalizeCompletedMessages();
|
||||
}
|
||||
catch (OperationCanceledException) when (runCancellation.IsCancellationRequested && !cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
await EmitPendingEventsAsync(state, onEvent).ConfigureAwait(false);
|
||||
await EmitPendingMcpOauthRequestsAsync(state, onMcpOAuthRequired).ConfigureAwait(false);
|
||||
ExitPlanModeRequestedEventDto? exitPlanModeEvent =
|
||||
_exitPlanModeCoordinator.ConsumePendingRequest(command.RequestId);
|
||||
if (exitPlanModeEvent is null || !state.HasPendingExitPlanModeRequest)
|
||||
{
|
||||
throw;
|
||||
}
|
||||
|
||||
await onExitPlanMode(exitPlanModeEvent).ConfigureAwait(false);
|
||||
return state.FinalizeCompletedMessages();
|
||||
}
|
||||
finally
|
||||
{
|
||||
_approvalCoordinator.ClearRequestApprovals(command.RequestId);
|
||||
}
|
||||
}
|
||||
|
||||
internal static Workflow BuildWorkflowForCommand(
|
||||
internal new static Workflow BuildWorkflowForCommand(
|
||||
RunTurnCommandDto command,
|
||||
IReadOnlyList<AIAgent> agents,
|
||||
WorkflowRunner? workflowRunner = null)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(command);
|
||||
ArgumentNullException.ThrowIfNull(agents);
|
||||
|
||||
return NormalizeOrchestrationMode(command.Workflow.Settings.OrchestrationMode) switch
|
||||
{
|
||||
"handoff" => CopilotAgentBundle.CreateHandoffWorkflow(command.Workflow, agents),
|
||||
"group-chat" => CopilotAgentBundle.CreateGroupChatWorkflow(command.Workflow, agents),
|
||||
_ => (workflowRunner ?? new WorkflowRunner()).BuildWorkflow(command.Workflow, agents, command.WorkflowLibrary),
|
||||
};
|
||||
return AgentWorkflowTurnRunner.BuildWorkflowForCommand(command, agents, workflowRunner);
|
||||
}
|
||||
|
||||
internal static FileSystemJsonCheckpointStore? CreateCheckpointStore(RunTurnCommandDto command)
|
||||
internal new static FileSystemJsonCheckpointStore? CreateCheckpointStore(RunTurnCommandDto command)
|
||||
{
|
||||
if (!ShouldEnableWorkflowCheckpointing(command))
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
DirectoryInfo checkpointDirectory = new(GetCheckpointStorePath(command));
|
||||
return new FileSystemJsonCheckpointStore(checkpointDirectory);
|
||||
return AgentWorkflowTurnRunner.CreateCheckpointStore(command);
|
||||
}
|
||||
|
||||
internal static bool ShouldEnableWorkflowCheckpointing(RunTurnCommandDto command)
|
||||
internal new static bool ShouldEnableWorkflowCheckpointing(RunTurnCommandDto command)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(command);
|
||||
return command.Workflow.Settings.Checkpointing.Enabled;
|
||||
return AgentWorkflowTurnRunner.ShouldEnableWorkflowCheckpointing(command);
|
||||
}
|
||||
|
||||
internal static string GetCheckpointStorePath(RunTurnCommandDto command)
|
||||
internal new static string GetCheckpointStorePath(RunTurnCommandDto command)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(command);
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(command.ResumeFromCheckpoint?.StorePath))
|
||||
{
|
||||
return command.ResumeFromCheckpoint.StorePath;
|
||||
}
|
||||
|
||||
string localAppData = Environment.GetFolderPath(Environment.SpecialFolder.LocalApplicationData);
|
||||
return Path.Combine(localAppData, "Aryx", "workflow-checkpoints", command.SessionId, command.RequestId);
|
||||
return AgentWorkflowTurnRunner.GetCheckpointStorePath(command);
|
||||
}
|
||||
|
||||
private static string? NormalizeOrchestrationMode(string? value)
|
||||
{
|
||||
return string.IsNullOrWhiteSpace(value) ? null : value.Trim().ToLowerInvariant();
|
||||
}
|
||||
|
||||
private static ValueTask<StreamingRun> OpenWorkflowRunAsync(
|
||||
RunTurnCommandDto command,
|
||||
Workflow workflow,
|
||||
IReadOnlyList<ChatMessage> inputMessages,
|
||||
CheckpointManager? checkpointManager)
|
||||
{
|
||||
InProcessExecutionEnvironment environment = CreateExecutionEnvironment(command, checkpointManager);
|
||||
if (checkpointManager is not null && command.ResumeFromCheckpoint is { } resumeFromCheckpoint)
|
||||
{
|
||||
return environment.ResumeStreamingAsync(
|
||||
workflow,
|
||||
new CheckpointInfo(resumeFromCheckpoint.WorkflowSessionId, resumeFromCheckpoint.CheckpointId));
|
||||
}
|
||||
|
||||
return environment.RunStreamingAsync(workflow, inputMessages.ToList(), command.RequestId);
|
||||
}
|
||||
|
||||
internal static InProcessExecutionEnvironment CreateExecutionEnvironment(
|
||||
internal new static InProcessExecutionEnvironment CreateExecutionEnvironment(
|
||||
RunTurnCommandDto command,
|
||||
CheckpointManager? checkpointManager)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(command);
|
||||
|
||||
string executionMode = command.Workflow.Settings.ExecutionMode?.Trim() ?? "off-thread";
|
||||
InProcessExecutionEnvironment environment = string.Equals(
|
||||
executionMode,
|
||||
"lockstep",
|
||||
StringComparison.OrdinalIgnoreCase)
|
||||
? InProcessExecution.Lockstep
|
||||
: InProcessExecution.OffThread;
|
||||
|
||||
return checkpointManager is null ? environment : environment.WithCheckpointing(checkpointManager);
|
||||
return AgentWorkflowTurnRunner.CreateExecutionEnvironment(command, checkpointManager);
|
||||
}
|
||||
|
||||
internal static void ConfigureHookLifecycleEventSuppression(
|
||||
CopilotTurnExecutionState state,
|
||||
CopilotAgentBundle bundle)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(state);
|
||||
ArgumentNullException.ThrowIfNull(bundle);
|
||||
|
||||
state.SuppressHookLifecycleEvents = !bundle.HasConfiguredHooks;
|
||||
AgentWorkflowTurnRunner.ConfigureHookLifecycleEventSuppression(state, bundle);
|
||||
}
|
||||
|
||||
private static async Task EmitPendingEventsAsync(
|
||||
CopilotTurnExecutionState state,
|
||||
Func<SidecarEventDto, Task> onEvent)
|
||||
internal new static UserInputRequest CreateRequestPortUserInputRequest(
|
||||
AgentWorkflowTurnRunner.WorkflowRequestPortMetadata metadata,
|
||||
RequestInfoEvent requestInfo)
|
||||
{
|
||||
foreach (SidecarEventDto pendingEvent in state.DrainPendingEvents())
|
||||
{
|
||||
await onEvent(pendingEvent).ConfigureAwait(false);
|
||||
}
|
||||
return AgentWorkflowTurnRunner.CreateRequestPortUserInputRequest(metadata, requestInfo);
|
||||
}
|
||||
|
||||
private static async Task EmitPendingMcpOauthRequestsAsync(
|
||||
CopilotTurnExecutionState state,
|
||||
Func<McpOauthRequiredEventDto, Task> onMcpOAuthRequired)
|
||||
internal new static object CoerceRequestPortResponse(string responseType, string? answer)
|
||||
{
|
||||
foreach (McpOauthRequiredEventDto request in state.DrainPendingMcpOauthRequests())
|
||||
{
|
||||
await onMcpOAuthRequired(request).ConfigureAwait(false);
|
||||
}
|
||||
return AgentWorkflowTurnRunner.CoerceRequestPortResponse(responseType, answer);
|
||||
}
|
||||
|
||||
public Task ResolveApprovalAsync(
|
||||
ResolveApprovalCommandDto command,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return _approvalCoordinator.ResolveApprovalAsync(command, cancellationToken);
|
||||
}
|
||||
|
||||
public Task ResolveUserInputAsync(
|
||||
ResolveUserInputCommandDto command,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
return _userInputCoordinator.ResolveUserInputAsync(command, cancellationToken);
|
||||
}
|
||||
|
||||
private async Task<bool> TryHandleRequestPortRequestAsync(
|
||||
RunTurnCommandDto command,
|
||||
RequestInfoEvent requestInfo,
|
||||
StreamingRun run,
|
||||
Func<UserInputRequestedEventDto, Task> onUserInput,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (!TryResolveRequestPortMetadata(command.Workflow, requestInfo, out WorkflowRequestPortMetadata? metadata))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
UserInputRequest userInputRequest = CreateRequestPortUserInputRequest(metadata!, requestInfo);
|
||||
UserInputResponse response = await _userInputCoordinator.RequestUserInputAsync(
|
||||
command,
|
||||
userInputRequest,
|
||||
onUserInput,
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
|
||||
object coercedResponse = CoerceRequestPortResponse(metadata!.ResponseType, response.Answer);
|
||||
await run.SendResponseAsync(requestInfo.Request.CreateResponse(coercedResponse)).ConfigureAwait(false);
|
||||
return true;
|
||||
}
|
||||
|
||||
private static async Task<bool> HandleWorkflowEventAsync(
|
||||
private static Task<bool> HandleWorkflowEventAsync(
|
||||
RunTurnCommandDto command,
|
||||
WorkflowEvent evt,
|
||||
IReadOnlyList<ChatMessage> inputMessages,
|
||||
@@ -314,469 +74,13 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner
|
||||
Func<TurnDeltaEventDto, Task> onDelta,
|
||||
Func<SidecarEventDto, Task> onEvent)
|
||||
{
|
||||
if (evt is ExecutorInvokedEvent invoked)
|
||||
{
|
||||
if (AgentIdentityResolver.TryResolveKnownAgentIdentity(
|
||||
command.Workflow,
|
||||
invoked.ExecutorId,
|
||||
out AgentIdentity invokedAgent))
|
||||
{
|
||||
TraceHandoff(command, $"Executor invoked: {invoked.ExecutorId} -> {invokedAgent.AgentName} ({invokedAgent.AgentId}).");
|
||||
await state.EmitThinkingIfNeeded(invokedAgent, onEvent).ConfigureAwait(false);
|
||||
}
|
||||
else
|
||||
{
|
||||
TraceHandoff(command, $"Executor invoked without a known agent match: {invoked.ExecutorId}.");
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
if (evt is RequestInfoEvent requestInfo)
|
||||
{
|
||||
AgentActivityEventDto? activity = WorkflowRequestInfoInterpreter.TryCreateActivityFromRequest(
|
||||
command,
|
||||
requestInfo,
|
||||
state.ActiveAgent,
|
||||
state.ToolNamesByCallId);
|
||||
|
||||
if (activity is null)
|
||||
{
|
||||
bool requiresBoundary = WorkflowRequestInfoInterpreter.RequiresUserInputTurnBoundary(command, requestInfo);
|
||||
TraceHandoff(
|
||||
command,
|
||||
$"Request info produced no activity for data type '{requestInfo.Request.Data.TypeId}'. Requires boundary: {requiresBoundary}.");
|
||||
return requiresBoundary;
|
||||
}
|
||||
|
||||
await EmitActivityAsync(command, state, activity, onEvent).ConfigureAwait(false);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (TryCreateWorkflowCheckpointSavedEvent(command, evt, out WorkflowCheckpointSavedEventDto? checkpointSaved))
|
||||
{
|
||||
await onEvent(checkpointSaved).ConfigureAwait(false);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (TryCreateWorkflowDiagnosticEvent(command, evt, state, out WorkflowDiagnosticEventDto? diagnostic))
|
||||
{
|
||||
await onEvent(diagnostic).ConfigureAwait(false);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (evt is AgentResponseUpdateEvent update)
|
||||
{
|
||||
await HandleAgentResponseUpdateAsync(command, update, state, onDelta, onEvent).ConfigureAwait(false);
|
||||
return false;
|
||||
}
|
||||
|
||||
if (evt is ExecutorCompletedEvent completed)
|
||||
{
|
||||
if (AgentIdentityResolver.TryResolveObservedAgentIdentity(
|
||||
command.Workflow,
|
||||
completed.ExecutorId,
|
||||
state.ActiveAgent,
|
||||
out AgentIdentity completedAgent))
|
||||
{
|
||||
TraceHandoff(command, $"Executor completed: {completed.ExecutorId} -> {completedAgent.AgentName} ({completedAgent.AgentId}).");
|
||||
state.QueueCompletedActivity(completedAgent);
|
||||
state.ClearActiveAgentIfMatching(completedAgent);
|
||||
}
|
||||
else
|
||||
{
|
||||
TraceHandoff(command, $"Executor completed without a known agent match: {completed.ExecutorId}.");
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
if (evt is WorkflowOutputEvent outputEvent)
|
||||
{
|
||||
List<ChatMessage> allMessages = outputEvent.As<List<ChatMessage>>() ?? [];
|
||||
state.UpdateCompletedMessages(allMessages, inputMessages);
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
internal static UserInputRequest CreateRequestPortUserInputRequest(
|
||||
WorkflowRequestPortMetadata metadata,
|
||||
RequestInfoEvent requestInfo)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(metadata);
|
||||
ArgumentNullException.ThrowIfNull(requestInfo);
|
||||
|
||||
string question = metadata.Prompt
|
||||
?? BuildRequestPortFallbackQuestion(metadata, requestInfo);
|
||||
|
||||
bool expectsBoolean = IsBooleanResponseType(metadata.ResponseType);
|
||||
return new UserInputRequest
|
||||
{
|
||||
Question = question,
|
||||
Choices = expectsBoolean ? ["true", "false"] : null,
|
||||
AllowFreeform = true,
|
||||
};
|
||||
}
|
||||
|
||||
internal static object CoerceRequestPortResponse(string responseType, string? answer)
|
||||
{
|
||||
string normalizedResponseType = responseType.Trim();
|
||||
string trimmedAnswer = answer?.Trim() ?? string.Empty;
|
||||
|
||||
if (IsStringResponseType(normalizedResponseType))
|
||||
{
|
||||
return trimmedAnswer;
|
||||
}
|
||||
|
||||
if (IsBooleanResponseType(normalizedResponseType))
|
||||
{
|
||||
return trimmedAnswer.ToLowerInvariant() switch
|
||||
{
|
||||
"true" or "t" or "yes" or "y" or "1" => true,
|
||||
"false" or "f" or "no" or "n" or "0" => false,
|
||||
_ => throw new InvalidOperationException(
|
||||
$"Request port response type \"{responseType}\" requires a boolean answer."),
|
||||
};
|
||||
}
|
||||
|
||||
if (IsNumericResponseType(normalizedResponseType))
|
||||
{
|
||||
if (double.TryParse(trimmedAnswer, NumberStyles.Float, CultureInfo.InvariantCulture, out double numeric))
|
||||
{
|
||||
return numeric;
|
||||
}
|
||||
|
||||
throw new InvalidOperationException(
|
||||
$"Request port response type \"{responseType}\" requires a numeric answer.");
|
||||
}
|
||||
|
||||
if (IsJsonResponseType(normalizedResponseType))
|
||||
{
|
||||
try
|
||||
{
|
||||
return JsonDocument.Parse(trimmedAnswer).RootElement.Clone();
|
||||
}
|
||||
catch (JsonException ex)
|
||||
{
|
||||
throw new InvalidOperationException(
|
||||
$"Request port response type \"{responseType}\" requires a valid JSON answer.",
|
||||
ex);
|
||||
}
|
||||
}
|
||||
|
||||
return trimmedAnswer;
|
||||
}
|
||||
|
||||
private static async Task HandleAgentResponseUpdateAsync(
|
||||
RunTurnCommandDto command,
|
||||
AgentResponseUpdateEvent update,
|
||||
CopilotTurnExecutionState state,
|
||||
Func<TurnDeltaEventDto, Task> onDelta,
|
||||
Func<SidecarEventDto, Task> onEvent)
|
||||
{
|
||||
AgentIdentity? updateAgent = null;
|
||||
string authorName = update.ExecutorId;
|
||||
string[] handoffFunctionCalls = update.Update.Contents
|
||||
.OfType<FunctionCallContent>()
|
||||
.Select(content => content.Name)
|
||||
.Where(IsHandoffFunctionName)
|
||||
.Distinct(StringComparer.Ordinal)
|
||||
.ToArray();
|
||||
if (state.TryResolveObservedAgentForMessage(update.Update.MessageId, out AgentIdentity observedMessageAgent))
|
||||
{
|
||||
updateAgent = observedMessageAgent;
|
||||
authorName = observedMessageAgent.AgentName;
|
||||
}
|
||||
else if (AgentIdentityResolver.TryResolveObservedAgentIdentity(
|
||||
command.Workflow,
|
||||
update.ExecutorId,
|
||||
state.ActiveAgent,
|
||||
out AgentIdentity resolvedUpdateAgent))
|
||||
{
|
||||
updateAgent = resolvedUpdateAgent;
|
||||
authorName = resolvedUpdateAgent.AgentName;
|
||||
}
|
||||
else if (state.ActiveAgent is AgentIdentity activeAgent)
|
||||
{
|
||||
updateAgent = activeAgent;
|
||||
authorName = activeAgent.AgentName;
|
||||
TraceHandoff(
|
||||
command,
|
||||
$"Agent response update fell back to active agent {activeAgent.AgentName} ({activeAgent.AgentId}) for executor '{update.ExecutorId}' and message '{update.Update.MessageId ?? "<none>"}'.");
|
||||
}
|
||||
|
||||
if (updateAgent.HasValue)
|
||||
{
|
||||
if (handoffFunctionCalls.Length > 0)
|
||||
{
|
||||
TraceHandoff(
|
||||
command,
|
||||
$"Agent response update from {updateAgent.Value.AgentName} ({updateAgent.Value.AgentId}) requested handoff via {string.Join(", ", handoffFunctionCalls)}.");
|
||||
}
|
||||
|
||||
await state.EmitThinkingIfNeeded(updateAgent.Value, onEvent).ConfigureAwait(false);
|
||||
}
|
||||
else if (!string.IsNullOrEmpty(update.Update.Text) || handoffFunctionCalls.Length > 0)
|
||||
{
|
||||
TraceHandoff(
|
||||
command,
|
||||
$"Agent response update could not resolve agent for executor '{update.ExecutorId}' and message '{update.Update.MessageId ?? "<none>"}'.");
|
||||
}
|
||||
|
||||
if (string.IsNullOrEmpty(update.Update.Text))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
string messageId = state.CreateMessageId(update.Update.MessageId);
|
||||
(string _, string currentAuthorName, string currentContent) = state.AppendDelta(
|
||||
messageId,
|
||||
authorName,
|
||||
update.Update.Text);
|
||||
|
||||
await onDelta(new TurnDeltaEventDto
|
||||
{
|
||||
Type = "turn-delta",
|
||||
RequestId = command.RequestId,
|
||||
SessionId = command.SessionId,
|
||||
MessageId = messageId,
|
||||
AuthorName = currentAuthorName,
|
||||
ContentDelta = update.Update.Text,
|
||||
Content = currentContent,
|
||||
}).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private static async Task EmitActivityAsync(
|
||||
RunTurnCommandDto command,
|
||||
CopilotTurnExecutionState state,
|
||||
AgentActivityEventDto activity,
|
||||
Func<SidecarEventDto, Task> onEvent)
|
||||
{
|
||||
state.ApplyEvent(activity);
|
||||
TraceHandoff(
|
||||
return AgentWorkflowTurnRunner.HandleWorkflowEventAsync(
|
||||
command,
|
||||
$"Activity emitted: {activity.ActivityType} -> {activity.AgentName ?? activity.AgentId ?? "<unknown>"}.");
|
||||
await onEvent(activity).ConfigureAwait(false);
|
||||
|
||||
if (string.Equals(activity.ActivityType, "handoff", StringComparison.Ordinal)
|
||||
&& !string.IsNullOrWhiteSpace(activity.AgentId)
|
||||
&& !string.IsNullOrWhiteSpace(activity.AgentName))
|
||||
{
|
||||
TraceHandoff(
|
||||
command,
|
||||
$"Promoting handoff target to thinking: {activity.AgentName} ({activity.AgentId}).");
|
||||
await state.EmitThinkingIfNeeded(
|
||||
new AgentIdentity(activity.AgentId, activity.AgentName),
|
||||
onEvent).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
internal static bool TryCreateWorkflowCheckpointSavedEvent(
|
||||
RunTurnCommandDto command,
|
||||
WorkflowEvent evt,
|
||||
out WorkflowCheckpointSavedEventDto checkpointSaved)
|
||||
{
|
||||
checkpointSaved = default!;
|
||||
|
||||
if (!ShouldEnableWorkflowCheckpointing(command)
|
||||
|| evt is not SuperStepCompletedEvent superStepCompleted
|
||||
|| superStepCompleted.CompletionInfo?.Checkpoint is not CheckpointInfo checkpoint)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
checkpointSaved = new WorkflowCheckpointSavedEventDto
|
||||
{
|
||||
Type = "workflow-checkpoint-saved",
|
||||
RequestId = command.RequestId,
|
||||
SessionId = command.SessionId,
|
||||
WorkflowSessionId = checkpoint.SessionId,
|
||||
CheckpointId = checkpoint.CheckpointId,
|
||||
StorePath = GetCheckpointStorePath(command),
|
||||
StepNumber = superStepCompleted.StepNumber,
|
||||
};
|
||||
return true;
|
||||
}
|
||||
|
||||
private static bool TryCreateWorkflowDiagnosticEvent(
|
||||
RunTurnCommandDto command,
|
||||
WorkflowEvent evt,
|
||||
CopilotTurnExecutionState state,
|
||||
out WorkflowDiagnosticEventDto diagnostic)
|
||||
{
|
||||
diagnostic = default!;
|
||||
|
||||
switch (evt)
|
||||
{
|
||||
case ExecutorFailedEvent executorFailed:
|
||||
{
|
||||
AgentIdentity? agent = AgentIdentityResolver.TryResolveObservedAgentIdentity(
|
||||
command.Workflow,
|
||||
executorFailed.ExecutorId,
|
||||
state.ActiveAgent,
|
||||
out AgentIdentity resolvedAgent)
|
||||
? resolvedAgent
|
||||
: null;
|
||||
Exception? exception = executorFailed.Data;
|
||||
diagnostic = new WorkflowDiagnosticEventDto
|
||||
{
|
||||
Type = "workflow-diagnostic",
|
||||
RequestId = command.RequestId,
|
||||
SessionId = command.SessionId,
|
||||
Severity = "error",
|
||||
DiagnosticKind = "executor-failed",
|
||||
Message = ResolveDiagnosticMessage(exception, "Executor failed."),
|
||||
AgentId = agent?.AgentId,
|
||||
AgentName = agent?.AgentName,
|
||||
ExecutorId = executorFailed.ExecutorId,
|
||||
ExceptionType = exception?.GetBaseException().GetType().Name,
|
||||
};
|
||||
return true;
|
||||
}
|
||||
case WorkflowWarningEvent workflowWarning:
|
||||
diagnostic = new WorkflowDiagnosticEventDto
|
||||
{
|
||||
Type = "workflow-diagnostic",
|
||||
RequestId = command.RequestId,
|
||||
SessionId = command.SessionId,
|
||||
Severity = "warning",
|
||||
DiagnosticKind = workflowWarning is SubworkflowWarningEvent
|
||||
? "subworkflow-warning"
|
||||
: "workflow-warning",
|
||||
Message = ResolveDiagnosticMessage(workflowWarning.Data as string, "Workflow warning."),
|
||||
SubworkflowId = workflowWarning is SubworkflowWarningEvent subworkflowWarning
|
||||
? subworkflowWarning.SubWorkflowId
|
||||
: null,
|
||||
};
|
||||
return true;
|
||||
case WorkflowErrorEvent workflowError:
|
||||
{
|
||||
Exception? exception = workflowError.Exception;
|
||||
diagnostic = new WorkflowDiagnosticEventDto
|
||||
{
|
||||
Type = "workflow-diagnostic",
|
||||
RequestId = command.RequestId,
|
||||
SessionId = command.SessionId,
|
||||
Severity = "error",
|
||||
DiagnosticKind = workflowError is SubworkflowErrorEvent
|
||||
? "subworkflow-error"
|
||||
: "workflow-error",
|
||||
Message = ResolveDiagnosticMessage(exception, "Workflow failed."),
|
||||
SubworkflowId = workflowError is SubworkflowErrorEvent subworkflowError
|
||||
? subworkflowError.SubworkflowId
|
||||
: null,
|
||||
ExceptionType = exception?.GetBaseException().GetType().Name,
|
||||
};
|
||||
return true;
|
||||
}
|
||||
default:
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private static bool TryResolveRequestPortMetadata(
|
||||
WorkflowDefinitionDto? workflow,
|
||||
RequestInfoEvent requestInfo,
|
||||
out WorkflowRequestPortMetadata? metadata)
|
||||
{
|
||||
metadata = null;
|
||||
if (workflow is null)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
string portId = requestInfo.Request.PortInfo.PortId;
|
||||
WorkflowNodeDto? node = workflow.Graph.Nodes.FirstOrDefault(candidate =>
|
||||
string.Equals(candidate.Kind, "request-port", StringComparison.OrdinalIgnoreCase)
|
||||
&& string.Equals(candidate.Config.PortId, portId, StringComparison.OrdinalIgnoreCase));
|
||||
|
||||
if (node is null)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
metadata = new WorkflowRequestPortMetadata(
|
||||
node.Id,
|
||||
string.IsNullOrWhiteSpace(node.Label) ? node.Id : node.Label,
|
||||
node.Config.PortId ?? portId,
|
||||
node.Config.RequestType ?? string.Empty,
|
||||
node.Config.ResponseType ?? string.Empty,
|
||||
string.IsNullOrWhiteSpace(node.Config.Prompt) ? null : node.Config.Prompt.Trim());
|
||||
return true;
|
||||
}
|
||||
|
||||
private static string BuildRequestPortFallbackQuestion(
|
||||
WorkflowRequestPortMetadata metadata,
|
||||
RequestInfoEvent requestInfo)
|
||||
{
|
||||
if (requestInfo.Request.Data.Is<WorkflowRequestPortPromptRequest>(out WorkflowRequestPortPromptRequest? promptRequest))
|
||||
{
|
||||
string baseQuestion = $"Provide a {metadata.ResponseType} response for \"{promptRequest.NodeLabel}\".";
|
||||
if (!string.IsNullOrWhiteSpace(promptRequest.InputSummary))
|
||||
{
|
||||
return $"{baseQuestion} Current input: {promptRequest.InputSummary}";
|
||||
}
|
||||
|
||||
return baseQuestion;
|
||||
}
|
||||
|
||||
return $"Provide a {metadata.ResponseType} response for request port \"{metadata.NodeLabel}\" ({metadata.PortId}).";
|
||||
}
|
||||
|
||||
private static bool IsStringResponseType(string responseType)
|
||||
=> string.Equals(responseType, "string", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "text", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
private static bool IsBooleanResponseType(string responseType)
|
||||
=> string.Equals(responseType, "bool", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "boolean", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
private static bool IsNumericResponseType(string responseType)
|
||||
=> string.Equals(responseType, "number", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "int", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "float", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "double", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "decimal", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
private static bool IsJsonResponseType(string responseType)
|
||||
=> string.Equals(responseType, "json", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "object", StringComparison.OrdinalIgnoreCase)
|
||||
|| string.Equals(responseType, "array", StringComparison.OrdinalIgnoreCase);
|
||||
|
||||
internal sealed record WorkflowRequestPortMetadata(
|
||||
string NodeId,
|
||||
string NodeLabel,
|
||||
string PortId,
|
||||
string RequestType,
|
||||
string ResponseType,
|
||||
string? Prompt);
|
||||
|
||||
private static string ResolveDiagnosticMessage(Exception? exception, string fallback)
|
||||
{
|
||||
return ResolveDiagnosticMessage(
|
||||
exception?.GetBaseException().Message,
|
||||
fallback);
|
||||
}
|
||||
|
||||
private static string ResolveDiagnosticMessage(string? message, string fallback)
|
||||
{
|
||||
return string.IsNullOrWhiteSpace(message) ? fallback : message;
|
||||
}
|
||||
|
||||
private static bool IsHandoffFunctionName(string? candidate)
|
||||
{
|
||||
return !string.IsNullOrWhiteSpace(candidate)
|
||||
&& candidate.StartsWith(HandoffFunctionPrefix, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
private static void TraceHandoff(RunTurnCommandDto command, string message)
|
||||
{
|
||||
if (!command.Workflow.IsOrchestrationMode("handoff"))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
Console.Error.WriteLine($"[aryx handoff] {message}");
|
||||
evt,
|
||||
inputMessages,
|
||||
state,
|
||||
CopilotTranscriptProjector.Instance,
|
||||
onDelta,
|
||||
onEvent);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,674 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Aryx.AgentHost.Contracts;
|
||||
using Microsoft.Extensions.AI;
|
||||
|
||||
namespace Aryx.AgentHost.Services;
|
||||
|
||||
internal class TurnExecutionState
|
||||
{
|
||||
private readonly RunTurnCommandDto _command;
|
||||
private readonly HashSet<string> _startedAgents = new(StringComparer.OrdinalIgnoreCase);
|
||||
private readonly HashSet<string> _reclassifiedMessageIds = new(StringComparer.Ordinal);
|
||||
private readonly ConcurrentQueue<SidecarEventDto> _pendingEvents = new();
|
||||
private readonly ConcurrentQueue<McpOauthRequiredEventDto> _pendingMcpOauthRequests = new();
|
||||
private readonly ConcurrentDictionary<string, AgentIdentity> _observedAgentsByMessageId = new(StringComparer.Ordinal);
|
||||
private readonly StreamingTranscriptBuffer _transcriptBuffer = new();
|
||||
private int _fallbackMessageIndex;
|
||||
private string? _lastObservedMessageId;
|
||||
|
||||
public TurnExecutionState(RunTurnCommandDto command)
|
||||
{
|
||||
_command = command;
|
||||
}
|
||||
|
||||
public ConcurrentDictionary<string, string> ToolNamesByCallId { get; } = new(StringComparer.Ordinal);
|
||||
|
||||
public AgentIdentity? ActiveAgent { get; private set; }
|
||||
|
||||
public List<ChatMessageDto> CompletedMessages { get; private set; } = [];
|
||||
|
||||
public bool HasPendingExitPlanModeRequest { get; private set; }
|
||||
|
||||
public bool SuppressHookLifecycleEvents { get; set; }
|
||||
|
||||
public async Task EmitThinkingIfNeeded(
|
||||
AgentIdentity agent,
|
||||
Func<SidecarEventDto, Task> onEvent)
|
||||
{
|
||||
AgentActivityEventDto? thinkingActivity = CreateThinkingActivityIfNeeded(agent);
|
||||
if (thinkingActivity is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await onEvent(thinkingActivity).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
public void QueueThinkingIfNeeded(AgentIdentity agent)
|
||||
{
|
||||
AgentActivityEventDto? thinkingActivity = CreateThinkingActivityIfNeeded(agent);
|
||||
if (thinkingActivity is not null)
|
||||
{
|
||||
_pendingEvents.Enqueue(thinkingActivity);
|
||||
}
|
||||
}
|
||||
|
||||
public void QueueCompletedActivity(AgentIdentity agent)
|
||||
{
|
||||
_pendingEvents.Enqueue(CreateCompletedActivity(agent));
|
||||
}
|
||||
|
||||
public void ApplyEvent(SidecarEventDto evt)
|
||||
{
|
||||
if (evt is AgentActivityEventDto activity
|
||||
&& string.Equals(activity.ActivityType, "handoff", StringComparison.Ordinal)
|
||||
&& !string.IsNullOrWhiteSpace(activity.AgentId)
|
||||
&& !string.IsNullOrWhiteSpace(activity.AgentName))
|
||||
{
|
||||
ActiveAgent = new AgentIdentity(activity.AgentId, activity.AgentName);
|
||||
}
|
||||
}
|
||||
|
||||
public void ObserveSessionEvent(WorkflowNodeDto agentDefinition, ProviderSessionEvent sessionEvent)
|
||||
{
|
||||
AgentIdentity agent = AgentIdentityResolver.ResolveAgentIdentity(
|
||||
_command.Workflow,
|
||||
agentDefinition.GetAgentId(),
|
||||
agentDefinition.GetAgentName());
|
||||
|
||||
switch (sessionEvent)
|
||||
{
|
||||
case ProviderAssistantMessageDeltaEvent messageDelta:
|
||||
RecordObservedAgentForMessage(agent, messageDelta.MessageId);
|
||||
QueueThinkingIfNeeded(agent);
|
||||
break;
|
||||
case ProviderAssistantMessageEvent assistantMessage:
|
||||
RecordObservedAgentForMessage(agent, assistantMessage.MessageId);
|
||||
QueueThinkingIfNeeded(agent);
|
||||
if (assistantMessage.HasToolRequests)
|
||||
{
|
||||
QueueMessageReclassifiedIfNeeded(assistantMessage.MessageId);
|
||||
}
|
||||
break;
|
||||
case ProviderToolExecutionStartEvent toolExecutionStart:
|
||||
string toolCallId = toolExecutionStart.ToolCallId;
|
||||
string toolName = toolExecutionStart.ToolName;
|
||||
ToolNamesByCallId[toolCallId] = toolName;
|
||||
ActiveAgent = agent;
|
||||
AgentActivityEventDto? toolActivity = CreateToolCallingActivity(
|
||||
agent, toolName, toolCallId, toolExecutionStart.ToolArguments);
|
||||
if (toolActivity is not null)
|
||||
{
|
||||
_pendingEvents.Enqueue(toolActivity);
|
||||
}
|
||||
|
||||
QueueMessageReclassifiedIfNeeded(_lastObservedMessageId);
|
||||
break;
|
||||
case ProviderAssistantIntentEvent intentEvent:
|
||||
ActiveAgent = agent;
|
||||
QueueThinkingIfNeeded(agent);
|
||||
AssistantIntentEventDto? assistantIntent = CreateAssistantIntentEvent(agent, intentEvent.Intent);
|
||||
if (assistantIntent is not null)
|
||||
{
|
||||
_pendingEvents.Enqueue(assistantIntent);
|
||||
}
|
||||
break;
|
||||
case ProviderAssistantReasoningDeltaEvent reasoningDelta:
|
||||
ActiveAgent = agent;
|
||||
QueueThinkingIfNeeded(agent);
|
||||
ReasoningDeltaEventDto? reasoningDeltaEvent = CreateReasoningDeltaEvent(
|
||||
agent,
|
||||
reasoningDelta.ReasoningId,
|
||||
reasoningDelta.DeltaContent);
|
||||
if (reasoningDeltaEvent is not null)
|
||||
{
|
||||
_pendingEvents.Enqueue(reasoningDeltaEvent);
|
||||
}
|
||||
break;
|
||||
case ProviderSubagentStartedEvent started:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateSubagentStartedEvent(agent, started));
|
||||
break;
|
||||
case ProviderSubagentCompletedEvent completed:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateSubagentCompletedEvent(agent, completed));
|
||||
break;
|
||||
case ProviderSubagentFailedEvent failed:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateSubagentFailedEvent(agent, failed));
|
||||
break;
|
||||
case ProviderSubagentSelectedEvent selected:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateSubagentSelectedEvent(agent, selected));
|
||||
break;
|
||||
case ProviderSubagentDeselectedEvent:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateSubagentDeselectedEvent(agent));
|
||||
break;
|
||||
case ProviderSkillInvokedEvent skillInvoked:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateSkillInvokedEvent(agent, skillInvoked));
|
||||
break;
|
||||
case ProviderHookStartEvent hookStart:
|
||||
ActiveAgent = agent;
|
||||
if (!SuppressHookLifecycleEvents)
|
||||
{
|
||||
_pendingEvents.Enqueue(CreateHookLifecycleEvent(
|
||||
agent,
|
||||
"start",
|
||||
hookStart.HookInvocationId,
|
||||
hookStart.HookType,
|
||||
input: hookStart.Input));
|
||||
}
|
||||
break;
|
||||
case ProviderHookEndEvent hookEnd:
|
||||
ActiveAgent = agent;
|
||||
if (!SuppressHookLifecycleEvents)
|
||||
{
|
||||
_pendingEvents.Enqueue(CreateHookLifecycleEvent(
|
||||
agent,
|
||||
"end",
|
||||
hookEnd.HookInvocationId,
|
||||
hookEnd.HookType,
|
||||
success: hookEnd.Success,
|
||||
output: hookEnd.Output,
|
||||
error: hookEnd.Error));
|
||||
}
|
||||
break;
|
||||
case ProviderAssistantUsageEvent assistantUsage:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateAssistantUsageEvent(agent, assistantUsage));
|
||||
break;
|
||||
case ProviderSessionUsageEvent usageInfo:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateUsageEvent(agent, usageInfo));
|
||||
break;
|
||||
case ProviderSessionCompactionStartEvent compactionStart:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateCompactionStartEvent(agent, compactionStart));
|
||||
break;
|
||||
case ProviderSessionCompactionCompleteEvent compactionComplete:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreateCompactionCompleteEvent(agent, compactionComplete));
|
||||
break;
|
||||
case ProviderPendingMessagesModifiedEvent:
|
||||
ActiveAgent = agent;
|
||||
_pendingEvents.Enqueue(CreatePendingMessagesModifiedEvent(agent));
|
||||
break;
|
||||
case ProviderMcpOauthRequiredEvent:
|
||||
ActiveAgent = agent;
|
||||
break;
|
||||
case ProviderExitPlanModeRequestedEvent:
|
||||
HasPendingExitPlanModeRequest = true;
|
||||
ActiveAgent = agent;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
public IReadOnlyList<SidecarEventDto> DrainPendingEvents()
|
||||
{
|
||||
List<SidecarEventDto> pending = [];
|
||||
while (_pendingEvents.TryDequeue(out SidecarEventDto? pendingEvent))
|
||||
{
|
||||
pending.Add(pendingEvent);
|
||||
}
|
||||
|
||||
return pending;
|
||||
}
|
||||
|
||||
public void EnqueuePendingMcpOauthRequest(McpOauthRequiredEventDto request)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(request);
|
||||
_pendingMcpOauthRequests.Enqueue(request);
|
||||
}
|
||||
|
||||
public IReadOnlyList<McpOauthRequiredEventDto> DrainPendingMcpOauthRequests()
|
||||
{
|
||||
List<McpOauthRequiredEventDto> pending = [];
|
||||
while (_pendingMcpOauthRequests.TryDequeue(out McpOauthRequiredEventDto? request))
|
||||
{
|
||||
pending.Add(request);
|
||||
}
|
||||
|
||||
return pending;
|
||||
}
|
||||
|
||||
public bool TryResolveObservedAgentForMessage(string? messageId, out AgentIdentity agent)
|
||||
{
|
||||
agent = default;
|
||||
return !string.IsNullOrWhiteSpace(messageId)
|
||||
&& _observedAgentsByMessageId.TryGetValue(messageId, out agent);
|
||||
}
|
||||
|
||||
public string CreateMessageId(string? messageId)
|
||||
{
|
||||
return messageId ?? $"{_command.RequestId}-delta-{_fallbackMessageIndex++}";
|
||||
}
|
||||
|
||||
public TranscriptSegment AppendDelta(
|
||||
string messageId,
|
||||
string authorName,
|
||||
string delta)
|
||||
{
|
||||
return _transcriptBuffer.AppendDelta(messageId, authorName, delta);
|
||||
}
|
||||
|
||||
public void ClearActiveAgentIfMatching(AgentIdentity completedAgent)
|
||||
{
|
||||
if (ActiveAgent.HasValue
|
||||
&& string.Equals(ActiveAgent.Value.AgentId, completedAgent.AgentId, StringComparison.Ordinal))
|
||||
{
|
||||
ActiveAgent = null;
|
||||
}
|
||||
}
|
||||
|
||||
private void RecordObservedAgentForMessage(AgentIdentity agent, string messageId)
|
||||
{
|
||||
ActiveAgent = agent;
|
||||
_observedAgentsByMessageId[messageId] = agent;
|
||||
_lastObservedMessageId = messageId;
|
||||
}
|
||||
|
||||
private void QueueMessageReclassifiedIfNeeded(string? messageId)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(messageId))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
string normalizedMessageId = messageId.Trim();
|
||||
if (!_reclassifiedMessageIds.Add(normalizedMessageId))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
_pendingEvents.Enqueue(CreateMessageReclassifiedEvent(normalizedMessageId));
|
||||
}
|
||||
|
||||
private AgentActivityEventDto? CreateThinkingActivityIfNeeded(AgentIdentity agent)
|
||||
{
|
||||
ActiveAgent = agent;
|
||||
|
||||
if (!_startedAgents.Add(agent.AgentId))
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
return new AgentActivityEventDto
|
||||
{
|
||||
Type = "agent-activity",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
ActivityType = "thinking",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
};
|
||||
}
|
||||
|
||||
private AgentActivityEventDto? CreateToolCallingActivity(
|
||||
AgentIdentity agent,
|
||||
string toolName,
|
||||
string toolCallId,
|
||||
IReadOnlyDictionary<string, object?>? toolArguments = null)
|
||||
{
|
||||
if (toolName.StartsWith("handoff_to_", StringComparison.Ordinal))
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
return new AgentActivityEventDto
|
||||
{
|
||||
Type = "agent-activity",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
ActivityType = "tool-calling",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
ToolName = toolName,
|
||||
ToolCallId = toolCallId,
|
||||
ToolArguments = toolArguments,
|
||||
};
|
||||
}
|
||||
|
||||
private AgentActivityEventDto CreateCompletedActivity(AgentIdentity agent)
|
||||
{
|
||||
return new AgentActivityEventDto
|
||||
{
|
||||
Type = "agent-activity",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
ActivityType = "completed",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
};
|
||||
}
|
||||
|
||||
private MessageReclassifiedEventDto CreateMessageReclassifiedEvent(string messageId)
|
||||
{
|
||||
return new MessageReclassifiedEventDto
|
||||
{
|
||||
Type = "message-reclassified",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
MessageId = messageId,
|
||||
NewKind = "thinking",
|
||||
};
|
||||
}
|
||||
|
||||
public void UpdateCompletedMessages(
|
||||
IReadOnlyList<ChatMessage> allMessages,
|
||||
IReadOnlyList<ChatMessage> inputMessages,
|
||||
IProviderTranscriptProjector transcriptProjector)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(transcriptProjector);
|
||||
|
||||
List<ChatMessage> newMessages = transcriptProjector.SelectNewOutputMessages(allMessages, inputMessages);
|
||||
CompletedMessages = transcriptProjector.ProjectCompletedMessagesFromSegments(
|
||||
_command,
|
||||
newMessages,
|
||||
_transcriptBuffer.Snapshot(),
|
||||
ActiveAgent);
|
||||
}
|
||||
|
||||
public IReadOnlyList<ChatMessageDto> FinalizeCompletedMessages(IProviderTranscriptProjector transcriptProjector)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(transcriptProjector);
|
||||
|
||||
if (CompletedMessages.Count == 0 && _transcriptBuffer.Count > 0)
|
||||
{
|
||||
CompletedMessages = transcriptProjector.ProjectCompletedMessagesFromSegments(
|
||||
_command,
|
||||
[],
|
||||
_transcriptBuffer.Snapshot(),
|
||||
ActiveAgent);
|
||||
}
|
||||
|
||||
foreach (ChatMessageDto message in CompletedMessages)
|
||||
{
|
||||
if (_reclassifiedMessageIds.Contains(message.Id))
|
||||
{
|
||||
message.MessageKind = "thinking";
|
||||
}
|
||||
}
|
||||
|
||||
return CompletedMessages;
|
||||
}
|
||||
|
||||
private SubagentEventDto CreateSubagentStartedEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderSubagentStartedEvent data)
|
||||
{
|
||||
return new SubagentEventDto
|
||||
{
|
||||
Type = "subagent-event",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
EventKind = "started",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
ToolCallId = data.ToolCallId,
|
||||
CustomAgentName = data.AgentName,
|
||||
CustomAgentDisplayName = data.AgentDisplayName,
|
||||
CustomAgentDescription = data.AgentDescription,
|
||||
};
|
||||
}
|
||||
|
||||
private SubagentEventDto CreateSubagentCompletedEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderSubagentCompletedEvent data)
|
||||
{
|
||||
return new SubagentEventDto
|
||||
{
|
||||
Type = "subagent-event",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
EventKind = "completed",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
ToolCallId = data.ToolCallId,
|
||||
CustomAgentName = data.AgentName,
|
||||
CustomAgentDisplayName = data.AgentDisplayName,
|
||||
};
|
||||
}
|
||||
|
||||
private SubagentEventDto CreateSubagentFailedEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderSubagentFailedEvent data)
|
||||
{
|
||||
return new SubagentEventDto
|
||||
{
|
||||
Type = "subagent-event",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
EventKind = "failed",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
ToolCallId = data.ToolCallId,
|
||||
CustomAgentName = data.AgentName,
|
||||
CustomAgentDisplayName = data.AgentDisplayName,
|
||||
Error = data.Error,
|
||||
};
|
||||
}
|
||||
|
||||
private SubagentEventDto CreateSubagentSelectedEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderSubagentSelectedEvent data)
|
||||
{
|
||||
return new SubagentEventDto
|
||||
{
|
||||
Type = "subagent-event",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
EventKind = "selected",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
CustomAgentName = data.AgentName,
|
||||
CustomAgentDisplayName = data.AgentDisplayName,
|
||||
Tools = data.Tools,
|
||||
};
|
||||
}
|
||||
|
||||
private SubagentEventDto CreateSubagentDeselectedEvent(AgentIdentity agent)
|
||||
{
|
||||
return new SubagentEventDto
|
||||
{
|
||||
Type = "subagent-event",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
EventKind = "deselected",
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
};
|
||||
}
|
||||
|
||||
private AssistantIntentEventDto? CreateAssistantIntentEvent(
|
||||
AgentIdentity agent,
|
||||
string? intent)
|
||||
{
|
||||
string? normalizedIntent = intent?.Trim();
|
||||
if (string.IsNullOrWhiteSpace(normalizedIntent))
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
return new AssistantIntentEventDto
|
||||
{
|
||||
Type = "assistant-intent",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
Intent = normalizedIntent,
|
||||
};
|
||||
}
|
||||
|
||||
private ReasoningDeltaEventDto? CreateReasoningDeltaEvent(
|
||||
AgentIdentity agent,
|
||||
string? reasoningId,
|
||||
string? deltaContent)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(reasoningId)
|
||||
|| string.IsNullOrEmpty(deltaContent))
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
return new ReasoningDeltaEventDto
|
||||
{
|
||||
Type = "reasoning-delta",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
ReasoningId = reasoningId,
|
||||
ContentDelta = deltaContent,
|
||||
};
|
||||
}
|
||||
|
||||
private SkillInvokedEventDto CreateSkillInvokedEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderSkillInvokedEvent data)
|
||||
{
|
||||
return new SkillInvokedEventDto
|
||||
{
|
||||
Type = "skill-invoked",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
SkillName = data.SkillName,
|
||||
Path = data.Path,
|
||||
Content = data.Content,
|
||||
AllowedTools = data.AllowedTools,
|
||||
PluginName = data.PluginName,
|
||||
PluginVersion = data.PluginVersion,
|
||||
};
|
||||
}
|
||||
|
||||
private HookLifecycleEventDto CreateHookLifecycleEvent(
|
||||
AgentIdentity agent,
|
||||
string phase,
|
||||
string hookInvocationId,
|
||||
string hookType,
|
||||
object? input = null,
|
||||
bool? success = null,
|
||||
object? output = null,
|
||||
string? error = null)
|
||||
{
|
||||
return new HookLifecycleEventDto
|
||||
{
|
||||
Type = "hook-lifecycle",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
HookInvocationId = hookInvocationId,
|
||||
HookType = hookType,
|
||||
Phase = phase,
|
||||
Input = input,
|
||||
Success = success,
|
||||
Output = output,
|
||||
Error = error,
|
||||
};
|
||||
}
|
||||
|
||||
private AssistantUsageEventDto CreateAssistantUsageEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderAssistantUsageEvent data)
|
||||
{
|
||||
return new AssistantUsageEventDto
|
||||
{
|
||||
Type = "assistant-usage",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
Model = data.Model,
|
||||
InputTokens = data.InputTokens,
|
||||
OutputTokens = data.OutputTokens,
|
||||
CacheReadTokens = data.CacheReadTokens,
|
||||
CacheWriteTokens = data.CacheWriteTokens,
|
||||
Cost = data.Cost,
|
||||
Duration = data.Duration,
|
||||
TotalNanoAiu = data.TotalNanoAiu,
|
||||
QuotaSnapshots = data.QuotaSnapshots,
|
||||
};
|
||||
}
|
||||
|
||||
private SessionUsageEventDto CreateUsageEvent(AgentIdentity agent, ProviderSessionUsageEvent data)
|
||||
{
|
||||
return new SessionUsageEventDto
|
||||
{
|
||||
Type = "session-usage",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
TokenLimit = data.TokenLimit,
|
||||
CurrentTokens = data.CurrentTokens,
|
||||
MessagesLength = data.MessagesLength,
|
||||
SystemTokens = data.SystemTokens,
|
||||
ConversationTokens = data.ConversationTokens,
|
||||
ToolDefinitionsTokens = data.ToolDefinitionsTokens,
|
||||
IsInitial = data.IsInitial,
|
||||
};
|
||||
}
|
||||
|
||||
private SessionCompactionEventDto CreateCompactionStartEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderSessionCompactionStartEvent data)
|
||||
{
|
||||
return new SessionCompactionEventDto
|
||||
{
|
||||
Type = "session-compaction",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
Phase = "start",
|
||||
SystemTokens = data.SystemTokens,
|
||||
ConversationTokens = data.ConversationTokens,
|
||||
ToolDefinitionsTokens = data.ToolDefinitionsTokens,
|
||||
};
|
||||
}
|
||||
|
||||
private SessionCompactionEventDto CreateCompactionCompleteEvent(
|
||||
AgentIdentity agent,
|
||||
ProviderSessionCompactionCompleteEvent data)
|
||||
{
|
||||
return new SessionCompactionEventDto
|
||||
{
|
||||
Type = "session-compaction",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
Phase = "complete",
|
||||
Success = data.Success,
|
||||
Error = data.Error,
|
||||
SystemTokens = data.SystemTokens,
|
||||
ConversationTokens = data.ConversationTokens,
|
||||
ToolDefinitionsTokens = data.ToolDefinitionsTokens,
|
||||
PreCompactionTokens = data.PreCompactionTokens,
|
||||
PostCompactionTokens = data.PostCompactionTokens,
|
||||
PreCompactionMessagesLength = data.PreCompactionMessagesLength,
|
||||
MessagesRemoved = data.MessagesRemoved,
|
||||
TokensRemoved = data.TokensRemoved,
|
||||
SummaryContent = data.SummaryContent,
|
||||
CheckpointNumber = data.CheckpointNumber,
|
||||
CheckpointPath = data.CheckpointPath,
|
||||
};
|
||||
}
|
||||
|
||||
private PendingMessagesModifiedEventDto CreatePendingMessagesModifiedEvent(AgentIdentity agent)
|
||||
{
|
||||
return new PendingMessagesModifiedEventDto
|
||||
{
|
||||
Type = "pending-messages-modified",
|
||||
RequestId = _command.RequestId,
|
||||
SessionId = _command.SessionId,
|
||||
AgentId = agent.AgentId,
|
||||
AgentName = agent.AgentName,
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,196 @@
|
||||
using System.Linq;
|
||||
using Aryx.AgentHost.Contracts;
|
||||
using Microsoft.Agents.AI;
|
||||
using Microsoft.Agents.AI.Workflows;
|
||||
|
||||
namespace Aryx.AgentHost.Services;
|
||||
|
||||
internal static class WorkflowOrchestrationFactory
|
||||
{
|
||||
public static HandoffWorkflowBuilder CreateHandoffWorkflowBuilder(
|
||||
AIAgent entryAgent,
|
||||
HandoffModeSettingsDto? settings = null)
|
||||
{
|
||||
HandoffModeSettingsDto effectiveSettings = settings ?? new HandoffModeSettingsDto();
|
||||
HandoffWorkflowBuilder builder = AgentWorkflowBuilder.CreateHandoffBuilderWith(entryAgent)
|
||||
.WithToolCallFilteringBehavior(MapHandoffToolCallFiltering(effectiveSettings.ToolCallFiltering))
|
||||
.WithHandoffInstructions(NormalizeOptionalString(effectiveSettings.HandoffInstructions)
|
||||
?? HandoffWorkflowGuidance.CreateWorkflowInstructions());
|
||||
|
||||
if (effectiveSettings.ReturnToPrevious)
|
||||
{
|
||||
builder = builder.EnableReturnToPrevious();
|
||||
}
|
||||
|
||||
return builder;
|
||||
}
|
||||
|
||||
public static Workflow CreateHandoffWorkflow(
|
||||
WorkflowDefinitionDto workflowDefinition,
|
||||
IReadOnlyList<AIAgent> agents)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(workflowDefinition);
|
||||
ArgumentNullException.ThrowIfNull(agents);
|
||||
|
||||
IReadOnlyList<WorkflowNodeDto> agentNodes = workflowDefinition.GetAgentNodes();
|
||||
Dictionary<string, AIAgent> agentsById = CreateAgentMap(agents);
|
||||
WorkflowNodeDto triageNode = ResolveTriageAgentNode(workflowDefinition, agentNodes);
|
||||
AIAgent triageAgent = ResolveAgentForNode(triageNode, agentsById);
|
||||
HandoffModeSettingsDto? settings = workflowDefinition.Settings.ModeSettings?.Handoff;
|
||||
HandoffWorkflowBuilder builder = CreateHandoffWorkflowBuilder(triageAgent, settings);
|
||||
|
||||
List<WorkflowNodeDto> specialistNodes = agentNodes
|
||||
.Where(node => !string.Equals(node.Id, triageNode.Id, StringComparison.Ordinal))
|
||||
.ToList();
|
||||
|
||||
if (specialistNodes.Count == 0)
|
||||
{
|
||||
throw new InvalidOperationException("Handoff workflows require at least one specialist agent in addition to the triage agent.");
|
||||
}
|
||||
|
||||
foreach (WorkflowNodeDto specialistNode in specialistNodes)
|
||||
{
|
||||
AIAgent specialistAgent = ResolveAgentForNode(specialistNode, agentsById);
|
||||
builder.WithHandoff(
|
||||
triageAgent,
|
||||
specialistAgent,
|
||||
HandoffWorkflowGuidance.CreateForwardReason(specialistNode));
|
||||
|
||||
if (settings?.ReturnToPrevious != true)
|
||||
{
|
||||
builder.WithHandoff(
|
||||
specialistAgent,
|
||||
triageAgent,
|
||||
HandoffWorkflowGuidance.CreateReturnReason(triageNode));
|
||||
}
|
||||
}
|
||||
|
||||
return builder.Build();
|
||||
}
|
||||
|
||||
public static GroupChatWorkflowBuilder CreateGroupChatWorkflowBuilder(
|
||||
WorkflowDefinitionDto workflowDefinition,
|
||||
IReadOnlyList<AIAgent> agents)
|
||||
{
|
||||
ArgumentNullException.ThrowIfNull(workflowDefinition);
|
||||
ArgumentNullException.ThrowIfNull(agents);
|
||||
|
||||
int maxRounds = ResolveGroupChatMaxRounds(workflowDefinition);
|
||||
GroupChatWorkflowBuilder builder = AgentWorkflowBuilder.CreateGroupChatBuilderWith(
|
||||
participants => new RoundRobinGroupChatManager(participants)
|
||||
{
|
||||
MaximumIterationCount = maxRounds,
|
||||
})
|
||||
.AddParticipants(agents);
|
||||
|
||||
string? name = NormalizeOptionalString(workflowDefinition.Name);
|
||||
if (name is not null)
|
||||
{
|
||||
builder.WithName(name);
|
||||
}
|
||||
|
||||
string? description = NormalizeOptionalString(workflowDefinition.Description);
|
||||
if (description is not null)
|
||||
{
|
||||
builder.WithDescription(description);
|
||||
}
|
||||
|
||||
return builder;
|
||||
}
|
||||
|
||||
public static Workflow CreateGroupChatWorkflow(
|
||||
WorkflowDefinitionDto workflowDefinition,
|
||||
IReadOnlyList<AIAgent> agents)
|
||||
{
|
||||
return CreateGroupChatWorkflowBuilder(workflowDefinition, agents).Build();
|
||||
}
|
||||
|
||||
private static Dictionary<string, AIAgent> CreateAgentMap(IReadOnlyList<AIAgent> agents)
|
||||
{
|
||||
Dictionary<string, AIAgent> agentMap = new(StringComparer.OrdinalIgnoreCase);
|
||||
foreach (AIAgent agent in agents)
|
||||
{
|
||||
if (!string.IsNullOrWhiteSpace(agent.Id))
|
||||
{
|
||||
agentMap[agent.Id] = agent;
|
||||
}
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(agent.Name))
|
||||
{
|
||||
agentMap[agent.Name] = agent;
|
||||
}
|
||||
}
|
||||
|
||||
return agentMap;
|
||||
}
|
||||
|
||||
private static AIAgent ResolveAgentForNode(
|
||||
WorkflowNodeDto node,
|
||||
IReadOnlyDictionary<string, AIAgent> agentsById)
|
||||
{
|
||||
string agentId = node.GetAgentId();
|
||||
if (agentsById.TryGetValue(agentId, out AIAgent? agent))
|
||||
{
|
||||
return agent;
|
||||
}
|
||||
|
||||
string agentName = node.GetAgentName();
|
||||
if (agentsById.TryGetValue(agentName, out agent))
|
||||
{
|
||||
return agent;
|
||||
}
|
||||
|
||||
throw new InvalidOperationException($"Workflow agent \"{agentId}\" could not be resolved from the constructed agents.");
|
||||
}
|
||||
|
||||
private static WorkflowNodeDto ResolveTriageAgentNode(
|
||||
WorkflowDefinitionDto workflowDefinition,
|
||||
IReadOnlyList<WorkflowNodeDto> agentNodes)
|
||||
{
|
||||
if (agentNodes.Count == 0)
|
||||
{
|
||||
throw new InvalidOperationException("Handoff workflows require at least one agent node.");
|
||||
}
|
||||
|
||||
string? triageAgentNodeId = NormalizeOptionalString(workflowDefinition.Settings.ModeSettings?.Handoff?.TriageAgentNodeId);
|
||||
if (triageAgentNodeId is null)
|
||||
{
|
||||
return agentNodes[0];
|
||||
}
|
||||
|
||||
WorkflowNodeDto? triageNode = agentNodes.FirstOrDefault(node => string.Equals(node.Id, triageAgentNodeId, StringComparison.Ordinal));
|
||||
return triageNode ?? throw new InvalidOperationException(
|
||||
$"Handoff workflow triage agent node \"{triageAgentNodeId}\" was not found in the workflow graph.");
|
||||
}
|
||||
|
||||
private static HandoffToolCallFilteringBehavior MapHandoffToolCallFiltering(string? value)
|
||||
{
|
||||
return value?.Trim().ToLowerInvariant() switch
|
||||
{
|
||||
"none" => HandoffToolCallFilteringBehavior.None,
|
||||
"all" => HandoffToolCallFilteringBehavior.All,
|
||||
_ => HandoffToolCallFilteringBehavior.HandoffOnly,
|
||||
};
|
||||
}
|
||||
|
||||
private static int ResolveGroupChatMaxRounds(WorkflowDefinitionDto workflowDefinition)
|
||||
{
|
||||
int? configuredMaxRounds = workflowDefinition.Settings.ModeSettings?.GroupChat?.MaxRounds;
|
||||
if (configuredMaxRounds is > 0)
|
||||
{
|
||||
return configuredMaxRounds.Value;
|
||||
}
|
||||
|
||||
if (workflowDefinition.Settings.MaxIterations is > 0)
|
||||
{
|
||||
return workflowDefinition.Settings.MaxIterations.Value;
|
||||
}
|
||||
|
||||
return 5;
|
||||
}
|
||||
|
||||
private static string? NormalizeOptionalString(string? value)
|
||||
{
|
||||
return string.IsNullOrWhiteSpace(value) ? null : value.Trim();
|
||||
}
|
||||
}
|
||||
@@ -141,7 +141,7 @@ internal sealed class WorkflowRunner
|
||||
throw new InvalidOperationException($"Workflow node \"{node.Id}\" references unknown agent \"{agentId}\".");
|
||||
}
|
||||
|
||||
return new WorkflowNodeRoute(agent.BindAsExecutor(CopilotAgentBundle.CreateAgentHostOptions()));
|
||||
return new WorkflowNodeRoute(agent.BindAsExecutor(AgentHostOptionsFactory.CreateDefault()));
|
||||
}
|
||||
|
||||
if (string.Equals(node.Kind, "code-executor", StringComparison.OrdinalIgnoreCase))
|
||||
|
||||
Reference in New Issue
Block a user