From 0e2f9b8ae551c43c0cea2bd29c38be83061c742f Mon Sep 17 00:00:00 2001 From: David Kaya Date: Wed, 8 Apr 2026 11:35:50 +0200 Subject: [PATCH] refactor: generalize sidecar workflow runner Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- ARCHITECTURE.md | 8 +- .../Services/AgentHostOptionsFactory.cs | 19 + .../Services/AgentWorkflowTurnRunner.cs | 763 ++++++++++++++++++ .../Services/CopilotTurnExecutionState.cs | 653 +-------------- .../Services/IProviderTranscriptProjector.cs | 21 + .../Services/IProviderTurnSupport.cs | 34 + .../Services/ProviderAgentBundle.cs | 14 + .../Providers/Copilot/CopilotAgentBundle.cs | 182 +---- .../Providers/Copilot/CopilotAgentProvider.cs | 2 +- .../Copilot/CopilotTranscriptProjector.cs | 35 + .../Copilot/CopilotTurnRunnerSupport.cs | 107 +++ .../Copilot/CopilotWorkflowRunner.cs | 750 +---------------- .../Services/TurnExecutionState.cs | 674 ++++++++++++++++ .../Services/WorkflowOrchestrationFactory.cs | 196 +++++ .../Aryx.AgentHost/Services/WorkflowRunner.cs | 2 +- 15 files changed, 1912 insertions(+), 1548 deletions(-) create mode 100644 sidecar/src/Aryx.AgentHost/Services/AgentHostOptionsFactory.cs create mode 100644 sidecar/src/Aryx.AgentHost/Services/AgentWorkflowTurnRunner.cs create mode 100644 sidecar/src/Aryx.AgentHost/Services/IProviderTranscriptProjector.cs create mode 100644 sidecar/src/Aryx.AgentHost/Services/IProviderTurnSupport.cs create mode 100644 sidecar/src/Aryx.AgentHost/Services/ProviderAgentBundle.cs create mode 100644 sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotTranscriptProjector.cs create mode 100644 sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotTurnRunnerSupport.cs create mode 100644 sidecar/src/Aryx.AgentHost/Services/TurnExecutionState.cs create mode 100644 sidecar/src/Aryx.AgentHost/Services/WorkflowOrchestrationFactory.cs diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index a6acc5f..35bc4a3 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -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: diff --git a/sidecar/src/Aryx.AgentHost/Services/AgentHostOptionsFactory.cs b/sidecar/src/Aryx.AgentHost/Services/AgentHostOptionsFactory.cs new file mode 100644 index 0000000..a751f28 --- /dev/null +++ b/sidecar/src/Aryx.AgentHost/Services/AgentHostOptionsFactory.cs @@ -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, + }; + } +} diff --git a/sidecar/src/Aryx.AgentHost/Services/AgentWorkflowTurnRunner.cs b/sidecar/src/Aryx.AgentHost/Services/AgentWorkflowTurnRunner.cs new file mode 100644 index 0000000..ae18219 --- /dev/null +++ b/sidecar/src/Aryx.AgentHost/Services/AgentWorkflowTurnRunner.cs @@ -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> RunTurnAsync( + RunTurnCommandDto command, + Func onDelta, + Func onEvent, + Func onApproval, + Func onUserInput, + Func onMcpOAuthRequired, + Func 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 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 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 OpenWorkflowRunAsync( + RunTurnCommandDto command, + Workflow workflow, + IReadOnlyList 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 onEvent) + { + foreach (SidecarEventDto pendingEvent in state.DrainPendingEvents()) + { + await onEvent(pendingEvent).ConfigureAwait(false); + } + } + + private static async Task EmitPendingMcpOauthRequestsAsync( + TurnExecutionState state, + Func 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 TryHandleRequestPortRequestAsync( + RunTurnCommandDto command, + RequestInfoEvent requestInfo, + StreamingRun run, + Func 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 HandleWorkflowEventAsync( + RunTurnCommandDto command, + WorkflowEvent evt, + IReadOnlyList inputMessages, + TurnExecutionState state, + IProviderTranscriptProjector transcriptProjector, + Func onDelta, + Func 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 allMessages = outputEvent.As>() ?? []; + 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 onDelta, + Func onEvent) + { + AgentIdentity? updateAgent = null; + string authorName = update.ExecutorId; + string[] handoffFunctionCalls = update.Update.Contents + .OfType() + .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 ?? ""}'."); + } + + 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 ?? ""}'."); + } + + 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 onEvent) + { + state.ApplyEvent(activity); + TraceHandoff( + command, + $"Activity emitted: {activity.ActivityType} -> {activity.AgentName ?? activity.AgentId ?? ""}."); + 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(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}"); + } +} diff --git a/sidecar/src/Aryx.AgentHost/Services/CopilotTurnExecutionState.cs b/sidecar/src/Aryx.AgentHost/Services/CopilotTurnExecutionState.cs index fd10b34..0e31d2f 100644 --- a/sidecar/src/Aryx.AgentHost/Services/CopilotTurnExecutionState.cs +++ b/sidecar/src/Aryx.AgentHost/Services/CopilotTurnExecutionState.cs @@ -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 _startedAgents = new(StringComparer.OrdinalIgnoreCase); - private readonly HashSet _reclassifiedMessageIds = new(StringComparer.Ordinal); - private readonly ConcurrentQueue _pendingEvents = new(); - private readonly ConcurrentQueue _pendingMcpOauthRequests = new(); - private readonly ConcurrentDictionary _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 ToolNamesByCallId { get; } = new(StringComparer.Ordinal); - - public AgentIdentity? ActiveAgent { get; private set; } - - public List CompletedMessages { get; private set; } = []; - - public bool HasPendingExitPlanModeRequest { get; private set; } - - public bool SuppressHookLifecycleEvents { get; set; } - - public async Task EmitThinkingIfNeeded( - AgentIdentity agent, - Func 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 DrainPendingEvents() - { - List 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 DrainPendingMcpOauthRequests() - { - List 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? 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 allMessages, IReadOnlyList inputMessages) { - List newMessages = WorkflowTranscriptProjector.SelectNewOutputMessages(allMessages, inputMessages); - CompletedMessages = WorkflowTranscriptProjector.ProjectCompletedMessagesFromSegments( - _command, - newMessages, - _transcriptBuffer.Snapshot(), - ActiveAgent); + base.UpdateCompletedMessages(allMessages, inputMessages, CopilotTranscriptProjector.Instance); } public IReadOnlyList 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); } } diff --git a/sidecar/src/Aryx.AgentHost/Services/IProviderTranscriptProjector.cs b/sidecar/src/Aryx.AgentHost/Services/IProviderTranscriptProjector.cs new file mode 100644 index 0000000..793d8a7 --- /dev/null +++ b/sidecar/src/Aryx.AgentHost/Services/IProviderTranscriptProjector.cs @@ -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 messages, string? messageMode); + + List SelectNewOutputMessages( + IReadOnlyList outputMessages, + IReadOnlyList inputMessages); + + List ProjectCompletedMessagesFromSegments( + RunTurnCommandDto command, + IReadOnlyList newMessages, + IReadOnlyList segments, + AgentIdentity? fallbackAgent = null); +} diff --git a/sidecar/src/Aryx.AgentHost/Services/IProviderTurnSupport.cs b/sidecar/src/Aryx.AgentHost/Services/IProviderTurnSupport.cs new file mode 100644 index 0000000..e89c4f1 --- /dev/null +++ b/sidecar/src/Aryx.AgentHost/Services/IProviderTurnSupport.cs @@ -0,0 +1,34 @@ +using Aryx.AgentHost.Contracts; +using GitHub.Copilot.SDK; + +namespace Aryx.AgentHost.Services; + +internal interface IProviderTurnSupport +{ + Task CreateAgentBundleAsync( + RunTurnCommandDto command, + TurnExecutionState state, + Func onEvent, + Func onApproval, + Func onUserInput, + CancellationTokenSource runCancellation, + CancellationToken cancellationToken); + + Task ResolveApprovalAsync( + ResolveApprovalCommandDto command, + CancellationToken cancellationToken); + + Task ResolveUserInputAsync( + ResolveUserInputCommandDto command, + CancellationToken cancellationToken); + + Task RequestRequestPortUserInputAsync( + RunTurnCommandDto command, + UserInputRequest request, + Func onUserInput, + CancellationToken cancellationToken); + + ExitPlanModeRequestedEventDto? ConsumePendingExitPlanModeRequest(string requestId); + + void ClearRequestState(string requestId); +} diff --git a/sidecar/src/Aryx.AgentHost/Services/ProviderAgentBundle.cs b/sidecar/src/Aryx.AgentHost/Services/ProviderAgentBundle.cs new file mode 100644 index 0000000..bb33254 --- /dev/null +++ b/sidecar/src/Aryx.AgentHost/Services/ProviderAgentBundle.cs @@ -0,0 +1,14 @@ +using Microsoft.Agents.AI; + +namespace Aryx.AgentHost.Services; + +internal abstract class ProviderAgentBundle : IAsyncDisposable +{ + public abstract IReadOnlyList Agents { get; } + + public abstract bool HasConfiguredHooks { get; } + + public abstract IProviderTranscriptProjector TranscriptProjector { get; } + + public abstract ValueTask DisposeAsync(); +} diff --git a/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotAgentBundle.cs b/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotAgentBundle.cs index e817984..9f1c207 100644 --- a/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotAgentBundle.cs +++ b/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotAgentBundle.cs @@ -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 Agents { get; } + public override IReadOnlyList Agents { get; } - public bool HasConfiguredHooks { get; } + public override bool HasConfiguredHooks { get; } + + public override IProviderTranscriptProjector TranscriptProjector { get; } = CopilotTranscriptProjector.Instance; public static async Task 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 agents) { - ArgumentNullException.ThrowIfNull(workflowDefinition); - ArgumentNullException.ThrowIfNull(agents); - - IReadOnlyList agentNodes = workflowDefinition.GetAgentNodes(); - Dictionary 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 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 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 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 CreateAgentMap(IReadOnlyList agents) - { - Dictionary 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 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 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) diff --git a/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotAgentProvider.cs b/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotAgentProvider.cs index a091c22..b083f27 100644 --- a/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotAgentProvider.cs +++ b/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotAgentProvider.cs @@ -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 GetCapabilitiesAsync(CancellationToken cancellationToken) diff --git a/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotTranscriptProjector.cs b/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotTranscriptProjector.cs new file mode 100644 index 0000000..9d236c6 --- /dev/null +++ b/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotTranscriptProjector.cs @@ -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 messages, string? messageMode) + => WorkflowTranscriptProjector.AttachMessageMode(messages, messageMode); + + public List SelectNewOutputMessages( + IReadOnlyList outputMessages, + IReadOnlyList inputMessages) + => WorkflowTranscriptProjector.SelectNewOutputMessages(outputMessages, inputMessages); + + public List ProjectCompletedMessagesFromSegments( + RunTurnCommandDto command, + IReadOnlyList newMessages, + IReadOnlyList segments, + AgentIdentity? fallbackAgent = null) + => WorkflowTranscriptProjector.ProjectCompletedMessagesFromSegments( + command, + newMessages, + segments, + fallbackAgent); +} diff --git a/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotTurnRunnerSupport.cs b/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotTurnRunnerSupport.cs new file mode 100644 index 0000000..ce84cec --- /dev/null +++ b/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotTurnRunnerSupport.cs @@ -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 CreateAgentBundleAsync( + RunTurnCommandDto command, + TurnExecutionState state, + Func onEvent, + Func onApproval, + Func 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 RequestRequestPortUserInputAsync( + RunTurnCommandDto command, + UserInputRequest request, + Func 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(); + } + } +} diff --git a/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotWorkflowRunner.cs b/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotWorkflowRunner.cs index 70cddb3..a99d20d 100644 --- a/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotWorkflowRunner.cs +++ b/sidecar/src/Aryx.AgentHost/Services/Providers/Copilot/CopilotWorkflowRunner.cs @@ -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> RunTurnAsync( - RunTurnCommandDto command, - Func onDelta, - Func onEvent, - Func onApproval, - Func onUserInput, - Func onMcpOAuthRequired, - Func 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 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 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 OpenWorkflowRunAsync( - RunTurnCommandDto command, - Workflow workflow, - IReadOnlyList 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 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 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 TryHandleRequestPortRequestAsync( - RunTurnCommandDto command, - RequestInfoEvent requestInfo, - StreamingRun run, - Func 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 HandleWorkflowEventAsync( + private static Task HandleWorkflowEventAsync( RunTurnCommandDto command, WorkflowEvent evt, IReadOnlyList inputMessages, @@ -314,469 +74,13 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner Func onDelta, Func 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 allMessages = outputEvent.As>() ?? []; - 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 onDelta, - Func onEvent) - { - AgentIdentity? updateAgent = null; - string authorName = update.ExecutorId; - string[] handoffFunctionCalls = update.Update.Contents - .OfType() - .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 ?? ""}'."); - } - - 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 ?? ""}'."); - } - - 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 onEvent) - { - state.ApplyEvent(activity); - TraceHandoff( + return AgentWorkflowTurnRunner.HandleWorkflowEventAsync( command, - $"Activity emitted: {activity.ActivityType} -> {activity.AgentName ?? activity.AgentId ?? ""}."); - 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(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); } } diff --git a/sidecar/src/Aryx.AgentHost/Services/TurnExecutionState.cs b/sidecar/src/Aryx.AgentHost/Services/TurnExecutionState.cs new file mode 100644 index 0000000..887ba86 --- /dev/null +++ b/sidecar/src/Aryx.AgentHost/Services/TurnExecutionState.cs @@ -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 _startedAgents = new(StringComparer.OrdinalIgnoreCase); + private readonly HashSet _reclassifiedMessageIds = new(StringComparer.Ordinal); + private readonly ConcurrentQueue _pendingEvents = new(); + private readonly ConcurrentQueue _pendingMcpOauthRequests = new(); + private readonly ConcurrentDictionary _observedAgentsByMessageId = new(StringComparer.Ordinal); + private readonly StreamingTranscriptBuffer _transcriptBuffer = new(); + private int _fallbackMessageIndex; + private string? _lastObservedMessageId; + + public TurnExecutionState(RunTurnCommandDto command) + { + _command = command; + } + + public ConcurrentDictionary ToolNamesByCallId { get; } = new(StringComparer.Ordinal); + + public AgentIdentity? ActiveAgent { get; private set; } + + public List CompletedMessages { get; private set; } = []; + + public bool HasPendingExitPlanModeRequest { get; private set; } + + public bool SuppressHookLifecycleEvents { get; set; } + + public async Task EmitThinkingIfNeeded( + AgentIdentity agent, + Func 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 DrainPendingEvents() + { + List 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 DrainPendingMcpOauthRequests() + { + List 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? 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 allMessages, + IReadOnlyList inputMessages, + IProviderTranscriptProjector transcriptProjector) + { + ArgumentNullException.ThrowIfNull(transcriptProjector); + + List newMessages = transcriptProjector.SelectNewOutputMessages(allMessages, inputMessages); + CompletedMessages = transcriptProjector.ProjectCompletedMessagesFromSegments( + _command, + newMessages, + _transcriptBuffer.Snapshot(), + ActiveAgent); + } + + public IReadOnlyList 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, + }; + } +} diff --git a/sidecar/src/Aryx.AgentHost/Services/WorkflowOrchestrationFactory.cs b/sidecar/src/Aryx.AgentHost/Services/WorkflowOrchestrationFactory.cs new file mode 100644 index 0000000..946297c --- /dev/null +++ b/sidecar/src/Aryx.AgentHost/Services/WorkflowOrchestrationFactory.cs @@ -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 agents) + { + ArgumentNullException.ThrowIfNull(workflowDefinition); + ArgumentNullException.ThrowIfNull(agents); + + IReadOnlyList agentNodes = workflowDefinition.GetAgentNodes(); + Dictionary 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 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 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 agents) + { + return CreateGroupChatWorkflowBuilder(workflowDefinition, agents).Build(); + } + + private static Dictionary CreateAgentMap(IReadOnlyList agents) + { + Dictionary 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 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 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(); + } +} diff --git a/sidecar/src/Aryx.AgentHost/Services/WorkflowRunner.cs b/sidecar/src/Aryx.AgentHost/Services/WorkflowRunner.cs index 45d0adf..7b52f1b 100644 --- a/sidecar/src/Aryx.AgentHost/Services/WorkflowRunner.cs +++ b/sidecar/src/Aryx.AgentHost/Services/WorkflowRunner.cs @@ -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))